Broadcast Traffic Scheduling Automation

Python Script for Conflict Detection in Avails

This guide delivers one deployable artifact: a Python script that scans a set of avails and emits a deterministic, schema-strict verdict on every collision — temporal overlap, buffer-band breach, break over-capacity, and competitive-category adjacency — before the schedule is ever committed to automation. It is the runnable drill-down beneath Detecting Time Slot Conflicts in Traffic Logs, which is itself one stage of the Spot Scheduling Validation & Rule Engines pipeline. Where the parent page establishes the sweep-line model, this page hands you the full module — models, event generation, the single-pass detector, and a CLI wrapper — with the buffer and capacity logic expanded in full. Getting it right is a revenue and compliance concern, not a cosmetic one: an undetected overlap drops a paid unit at playout, and the resulting credit surfaces only during as-run reconciliation, long after the money is gone. Every verdict this script emits is anchored to a spot_id so it can be joined against the billing code standardization layer during an audit.

An avail is the channel-specific inventory container defined in Avails Mapping Strategies for Linear TV; each spot placed into one carries the identity and duration fields from the canonical broadcast spot schema. This detector assumes those records are already typed and normalized upstream by Pydantic schema validation — it re-asserts the invariants it depends on but does not re-parse raw vendor exports.

Prerequisites

  • Python 3.11+ — required for the X | None union syntax, enum.auto, and timezone-aware datetime.now(timezone.utc) used below.
  • Standard library only — dataclasses, datetime, enum, logging, json, argparse, and pathlib cover the whole detector; no third-party runtime dependency is needed for the interval math itself.
  • Timezone-aware input — every spot’s start_utc must be a tz-aware UTC datetime. Naive local timestamps are rejected at ingress, not silently coerced.
  • Read-only access to the traffic log source — a mounted export or a DB role scoped to SELECT, governed by the same security boundaries for traffic database access as every other pipeline stage.
  • A writable audit path — an append-only location (default /var/log/broadcast/avail_audit.json) where each validation run is recorded as an immutable JSON line.
  • Threshold configuration — buffer minutes, competitive-separation minutes, and per-break capacity injected as config, owned by threshold tuning for scheduling accuracy rather than hard-coded here.

Step-by-Step Implementation

The detector runs as a pure function of well-formed input: build two events per spot, sort them chronologically, sweep once, and emit a typed conflict for every rule that fires. The diagram below is the exact control flow each OPEN event follows.

Sweep-line conflict detection pass Each spot becomes an OPEN and a CLOSE event; events are sorted so CLOSE precedes OPEN at a tied instant. A CLOSE event drops its spot from the active set. An OPEN event compares the newcomer against the active set and emits one typed conflict per fired rule: TEMPORAL_OVERLAP (CRITICAL) when the gap is negative, BUFFER_VIOLATION (MEDIUM) when the gap is under the buffer band, COMPETITIVE_SEPARATION (HIGH) for a same-category spot still inside the window, and CAPACITY_EXCEEDED (HIGH) when the break is already full. Every finding flows into a single conflict report. Build OPEN / CLOSE events per spot Sort events CLOSE before OPEN on tie Event type? CLOSE Drop spot from active set OPEN Compare newcomer against the active set gap < 0 gap < buffer same category break full TEMPORAL_OVERLAP intervals intersect CRITICAL BUFFER_VIOLATION gap under buffer band MEDIUM COMPETITIVE_SEPARATION same category in window HIGH CAPACITY_EXCEEDED break over ceiling HIGH Conflict report one typed finding per fired rule

Figure — Sweep-line pass: OPEN/CLOSE events are sorted, and each OPEN compares the gap against the active set to emit overlap, buffer, competitive, or capacity findings.

Step 1 — Models, enums, and structured audit logging

Goal: fix the typed entities that carry a placement through the pipeline and stand up the audit logger in the traffic-ops timestamp | level | module | spot_id shape. The audit line is the compliance artifact — every validation run must be reconstructable from it alone.

python 
#!/usr/bin/env python3
"""Broadcast avail conflict detector — deterministic pre-commit validation."""
from __future__ import annotations

import argparse
import json
import logging
import sys
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum, auto
from pathlib import Path

# Traffic-ops structured logging: timestamp | level | module | message(spot_id=...)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.validation.avail_conflict_detector")

DEFAULT_BUFFER_MINUTES = 3            # minimum gap between adjacent spots
DEFAULT_MAX_SPOTS_PER_BREAK = 4       # capacity ceiling for one break
DEFAULT_COMP_SEPARATION_MINUTES = 15  # same-category exclusion window
AUDIT_LOG_PATH = Path("/var/log/broadcast/avail_audit.json")


class ConflictType(Enum):
    TEMPORAL_OVERLAP = auto()        # intervals physically intersect
    BUFFER_VIOLATION = auto()        # gap smaller than the buffer band
    COMPETITIVE_SEPARATION = auto()  # same category inside exclusion window
    CAPACITY_EXCEEDED = auto()       # more units than the break can hold


class SpotPriority(Enum):
    HARD_STOP = 100      # guaranteed contract — never preempted
    PROGRAMMATIC = 50
    MAKE_GOOD = 20
    LOCAL_INSERT = 10    # first to yield when a break is over-subscribed


@dataclass(frozen=True)
class SpotPlacement:
    spot_id: str
    campaign_id: str
    category: str
    start_utc: datetime
    duration_minutes: int
    priority: SpotPriority
    break_id: str

    @property
    def end_utc(self) -> datetime:
        return self.start_utc + timedelta(minutes=self.duration_minutes)


@dataclass(frozen=True)
class ConflictReport:
    spot_id: str
    conflict_type: ConflictType
    conflicting_ids: list[str]
    severity: str
    detected_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

Expected: importing the module emits no output; instantiating a SpotPlacement with a naive start_utc is legal at this layer but is rejected in Step 3’s input guard.

Step 2 — Build padded OPEN/CLOSE events

Goal: turn each placement into two events whose boundaries already include the competitive-separation window, so the sweep never has to special-case padding later. Padding once, at generation, keeps the effective interval that produced a verdict fully reconstructable from the log.

python 
def build_events(
    placements: list[SpotPlacement], comp_separation_minutes: int
) -> list[tuple[datetime, bool, SpotPlacement]]:
    """Two events per spot; the CLOSE carries the competitive-separation pad."""
    events: list[tuple[datetime, bool, SpotPlacement]] = []
    for spot in placements:
        events.append((spot.start_utc, True, spot))  # True = OPEN
        pad_end = spot.end_utc + timedelta(minutes=comp_separation_minutes)
        events.append((pad_end, False, spot))         # False = CLOSE
    # Sort CLOSE before OPEN at an identical instant so a spot ending exactly
    # when the next begins reads as adjacent, not overlapping.
    events.sort(key=lambda e: (e[0], e[1]))
    return events

Expected: for N placements the function returns 2 * N events; a CLOSE and an OPEN sharing a timestamp always order the CLOSE first because False < True.

Step 3 — Sweep once and emit typed conflicts

Goal: in a single left-to-right pass, hold the set of active spots and, on every OPEN, test the newcomer against that set for overlap, buffer breach, and same-category adjacency, plus assert the break’s capacity ceiling. Complexity is O(n log n), dominated by the sort.

python 
class AvailValidationError(ValueError):
    """Raised when input placements violate a structural invariant."""


class AvailConflictDetector:
    def __init__(self, config: dict | None = None) -> None:
        cfg = config or {}
        self.buffer_minutes: int = cfg.get("buffer_minutes", DEFAULT_BUFFER_MINUTES)
        self.comp_minutes: int = cfg.get(
            "comp_separation_minutes", DEFAULT_COMP_SEPARATION_MINUTES
        )
        self.max_spots: int = cfg.get("max_spots_per_break", DEFAULT_MAX_SPOTS_PER_BREAK)

    def validate_schedule(self, placements: list[SpotPlacement]) -> list[ConflictReport]:
        if not placements:
            logger.warning("empty schedule — nothing to validate")
            return []
        self._assert_invariants(placements)

        active: list[SpotPlacement] = []
        conflicts: list[ConflictReport] = []
        for _, is_open, spot in build_events(placements, self.comp_minutes):
            if not is_open:
                active = [s for s in active if s.spot_id != spot.spot_id]
                continue
            conflicts.extend(self._check(spot, active))
            active.append(spot)

        self._write_audit(len(placements), len(conflicts))
        logger.info("scan-complete | spots=%d | conflicts=%d",
                    len(placements), len(conflicts))
        return conflicts

    def _check(
        self, spot: SpotPlacement, active: list[SpotPlacement]
    ) -> list[ConflictReport]:
        found: list[ConflictReport] = []

        break_peers = [s for s in active if s.break_id == spot.break_id]
        if len(break_peers) >= self.max_spots:
            found.append(ConflictReport(
                spot.spot_id, ConflictType.CAPACITY_EXCEEDED,
                [s.spot_id for s in break_peers], "HIGH"))
            logger.error("%s | capacity exceeded on break=%s (%d active)",
                         spot.spot_id, spot.break_id, len(break_peers))

        for other in active:
            gap = (spot.start_utc - other.end_utc).total_seconds() / 60.0
            if gap < 0:
                found.append(ConflictReport(
                    spot.spot_id, ConflictType.TEMPORAL_OVERLAP, [other.spot_id],
                    "CRITICAL"))
                logger.error("%s | overlaps %s by %.1f min",
                             spot.spot_id, other.spot_id, -gap)
            elif gap < self.buffer_minutes:
                found.append(ConflictReport(
                    spot.spot_id, ConflictType.BUFFER_VIOLATION, [other.spot_id],
                    "MEDIUM"))
            if other.category == spot.category and gap < self.comp_minutes:
                found.append(ConflictReport(
                    spot.spot_id, ConflictType.COMPETITIVE_SEPARATION,
                    [other.spot_id], "HIGH"))
                logger.warning("%s | competitive breach vs %s (category=%s)",
                               spot.spot_id, other.spot_id, spot.category)
        return found

    def _assert_invariants(self, placements: list[SpotPlacement]) -> None:
        for spot in placements:
            if spot.duration_minutes <= 0:
                raise AvailValidationError(f"{spot.spot_id}: duration must be > 0")
            if spot.start_utc.tzinfo is None:
                raise AvailValidationError(f"{spot.spot_id}: start_utc must be tz-aware")

    def _write_audit(self, total: int, conflicts: int) -> None:
        AUDIT_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        record = {
            "action": "validation_complete",
            "total_spots": total,
            "conflicts_found": conflicts,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        with AUDIT_LOG_PATH.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(record) + "\n")

Expected: a clean two-spot break logs ... | INFO | traffic.validation.avail_conflict_detector | scan-complete | spots=2 | conflicts=0; an overlapping pair logs a CRITICAL line naming both spot_id values and the overlap magnitude in minutes.

Step 4 — Wrap it as a pre-commit CLI gate

Goal: expose the detector as a standalone worker that reads a JSON schedule, writes a conflict report, and exits non-zero when any CRITICAL conflict is present — the exact contract a CI stage or cron gate asserts against before promoting a log to automation.

python 
def main() -> None:
    parser = argparse.ArgumentParser(description="Broadcast avail conflict detector")
    parser.add_argument("--input", type=Path, required=True, help="JSON schedule")
    parser.add_argument("--output", type=Path, default=Path("conflicts.json"))
    args = parser.parse_args()

    detector = AvailConflictDetector()
    try:
        raw = json.loads(args.input.read_text(encoding="utf-8"))
        placements = [
            SpotPlacement(
                spot_id=r["spot_id"], campaign_id=r["campaign_id"],
                category=r["category"],
                start_utc=datetime.fromisoformat(r["start_utc"]),
                duration_minutes=r["duration_minutes"],
                priority=SpotPriority[r["priority"]], break_id=r["break_id"],
            )
            for r in raw
        ]
        conflicts = detector.validate_schedule(placements)
        args.output.write_text(
            json.dumps([c.__dict__ for c in conflicts], indent=2, default=str),
            encoding="utf-8",
        )
        criticals = sum(c.severity == "CRITICAL" for c in conflicts)
        logger.info("wrote %d conflicts (%d critical) to %s",
                    len(conflicts), criticals, args.output)
        sys.exit(1 if criticals else 0)  # fail the gate on any hard collision
    except (AvailValidationError, KeyError, ValueError):
        logger.exception("validation aborted")
        sys.exit(2)


if __name__ == "__main__":
    main()

Expected: python avail_detector.py --input day.json --output out.json exits 0 on a clean log and 1 when a CRITICAL overlap is written — the signal a scheduler promotion step blocks on.

Verification & Testing

Correctness rests on two properties: adjacency is never flagged as overlap, and every rule fires deterministically. Both are assertable against fixtures drawn from a real break. Run this suite in CI so a change to the padding or sort rule that would silently re-key verdicts fails the build instead of shipping.

python 
from datetime import datetime, timezone

def _spot(sid: str, minute: int, dur: int, cat: str, brk: str = "B1") -> SpotPlacement:
    return SpotPlacement(
        spot_id=sid, campaign_id="C-" + sid, category=cat,
        start_utc=datetime(2026, 7, 3, 20, minute, tzinfo=timezone.utc),
        duration_minutes=dur, priority=SpotPriority.LOCAL_INSERT, break_id=brk,
    )

detector = AvailConflictDetector({"comp_separation_minutes": 0, "buffer_minutes": 0})

# 1. Adjacency is not a conflict: spot B starts exactly when A ends.
clean = detector.validate_schedule([_spot("A", 0, 1, "auto"), _spot("B", 1, 1, "food")])
assert clean == []

# 2. A true 1-minute overlap raises exactly one CRITICAL temporal conflict.
overlap = detector.validate_schedule([_spot("A", 0, 2, "auto"), _spot("B", 1, 1, "food")])
assert any(c.conflict_type is ConflictType.TEMPORAL_OVERLAP for c in overlap)
assert overlap[0].severity == "CRITICAL"

# 3. Same-category spots inside the window breach competitive separation.
comp = AvailConflictDetector({"comp_separation_minutes": 15}).validate_schedule(
    [_spot("A", 0, 1, "auto"), _spot("B", 5, 1, "auto")])
assert any(c.conflict_type is ConflictType.COMPETITIVE_SEPARATION for c in comp)

Because build_events and the sweep are pure functions of their input, replaying an archived schedule during an audit reproduces byte-identical ConflictReport sets — the reproducibility SOC 2 change-integrity controls expect.

Edge Cases & Failure Handling

  • Naive or DST-ambiguous timestamps. A spot logged as 02:30 on a spring-forward night, or a regional feed delivered without an offset, produces phantom overlaps. _assert_invariants rejects any naive start_utc before the sweep runs; quarantine and re-normalize the record upstream rather than defaulting a zone. This is the same tz-safety discipline the parent conflict-detection gate enforces at ingress.
  • Capacity breach on a valid-looking break. Five 30-second units fit inside a 120-second break’s clock yet exceed its unit ceiling. Pairwise overlap testing alone misses this; the len(break_peers) >= self.max_spots check on every OPEN catches it and emits CAPACITY_EXCEEDED. Tune max_spots_per_break per break format instead of loosening the guard.
  • Preemption tiers and make-good routing. Not every collision is an error — a LOCAL_INSERT yielding to a HARD_STOP contract is expected attrition. The SpotPriority on each placement encodes the tier so the downstream make-good routing for preemptions stage knows which spot to rebook, and the cleaned timeline flows into rule engines for spot rotation for frequency-cap enforcement.

FAQ

Why a sweep-line pass instead of comparing every spot to every other?

A pairwise comparison is O(n²) and collapses on a real station-day carrying thousands of placements across hundreds of breaks. The sweep sorts 2N events once and tracks only the currently active set, dropping the cost to O(n log n). It also gives a single natural place to check capacity (the active count) and competitive separation (categories in the active window) in the same pass, which the parent Detecting Time Slot Conflicts in Traffic Logs explains in more depth.

How is a temporal overlap different from a buffer violation?

A temporal overlap means the two intervals physically intersect — the gap between them is negative — so real airtime is double-booked; that is always CRITICAL. A buffer violation means the spots do not overlap but the gap is smaller than the required separation band (default three minutes), which is a MEDIUM soft conflict routed to an exception queue rather than halting the log. The band itself is owned by threshold tuning for scheduling accuracy.

Why must every conflict carry a spot_id, and where does it come from?

The spot_id is the primary key that joins detection to billing. A preempted spot’s make-good must be traceable from the conflict record through rebooking to an invoice credit without an identity break, so the spot_id here must be the same one produced by the canonical broadcast spot schema and normalized in the billing code standardization layer.

Can I run this on a live schedule that is edited spot by spot?

Full-log batch validation is the correct default for pre-commit gating because it is deterministic and produces a complete report. An event-driven variant that revalidates only the affected break on each edit is worth building for live editing, but it must fall back to a full sweep on any structural change (a moved break boundary, a duration edit) to stay correct — otherwise a stale active set produces a false clean verdict.

How do I keep the audit log admissible for a financial audit?

Write it append-only and never mutate a record in place; a re-scan appends a new line referencing the prior run rather than overwriting it, so the full history of how a collision was detected and resolved is reconstructable. Ship the file to centralized, write-once storage, and validate that records requiring Pydantic schema validation upstream were typed before they reached the detector.

Related content

Up one levelDetecting Time Slot Conflicts in Traffic Logs