Broadcast Traffic Scheduling Automation

Validating Spot Durations Against Broadcast Standards

Spot duration mismatches are silent pipeline killers. When an Avion export delivers a duration as 00:00:29.97, 30.1s, or a malformed :30, the downstream scheduler either rejects the traffic log, truncates the creative, or clears a break that runs long — and a break that runs long is a black-frame or a dropped spot the as-run log can never reconcile. This guide solves one exact task: deterministically validating and normalizing every spot duration at ingestion against NAB/ATSC standard lengths, then routing anything non-compliant to quarantine before it reaches the scheduling engine. It is the boundary check inside Schema Validation with Pydantic for Traffic Data, itself a phase of the broader Avion & Avstar Ingestion Pipelines. Getting it right is not cosmetic: a wrong duration that clears silently corrupts billed seconds, breaks make-good accounting, and leaves an audit trail that fails FCC record-keeping scrutiny.

Broadcast operations depend on rigid standard lengths — 5, 10, 15, 30, 60, and 120 seconds — yet raw Avion CSV/XML exports routinely mix HH:MM:SS, MM:SS, raw floats, and frame counts in the same file. Three failure modes recur: frame-rate drift, where 29.97 fps exports round 30.0 down to 29.97; non-standard buys, where a :45 or :90 placement violates network clearance rules; and malformed exports, where trailing whitespace, a UTF-8 BOM, or a localized decimal separator (29,97) defeats a naive parser. Deterministic validation at the boundary is the only remedy that keeps these out of the log without post-scheduling patching.

Prerequisites

  • Python 3.11+ — required for the model_validator semantics and timezone-aware datetime.now(timezone.utc) used below.
  • Pinned dependencies — pydantic==2.7.1, aiohttp==3.9.5. Pin exactly; Pydantic v2 validator signatures and aiohttp timeout defaults both shift across minor releases.
  • Avstar API access — a service bearer token scoped to ingest:write plus the published request budget for your contract tier (see Avstar API Authentication and Rate Limits).
  • A parsed export — records already lifted out of the raw file by the Avion export parsers and resolved to the canonical spot schema, with billing code normalization applied upstream.
  • A writable quarantine directory — a local path (or object-store prefix) where non-compliant records are isolated for traffic-desk review.

Step-by-Step Implementation

The validator runs as a pure boundary function: each raw record is parsed to seconds, snapped to the nearest standard length within a frame tolerance, and stamped COMPLIANT, NON_STANDARD, or REJECTED. Only COMPLIANT records advance to the Avstar sync; everything else is quarantined with an actionable note.

Spot duration compliance decision flow A left-to-right pipeline. A raw Avion duration string is parsed into a float of seconds, then converted and snapped to the nearest NAB standard length of 5, 10, 15, 30, 60, or 120 seconds. A decision node tests whether the snapped value is within the one-frame tolerance of plus or minus 0.034 seconds. If yes, the record is stamped COMPLIANT and advances to the Avstar sync. If no, it is stamped NON_STANDARD or REJECTED and written to quarantine for traffic-desk review. Parse HH:MM:SS.ff · raw s Convert to seconds Snap to NAB 5·10·15·30·60·120s Within tolerance? ±0.034s COMPLIANT → Avstar sync NON_STANDARD / REJECTED → Quarantine yes no

Figure — Duration strings are parsed, converted to seconds, snapped to the nearest NAB standard length, and either synced to Avstar or quarantined based on frame tolerance.

Step 1 — Structured audit logging and broadcast constants

Goal: emit machine-parseable audit lines in the traffic-ops timestamp | level | module | spot_id shape, and fix the compliance constants — the allowed lengths and a one-frame tolerance at 29.97 fps.

python 
import json
import logging
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional

# Audit trail: timestamp | level | module | spot_id-bearing message
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler("duration_validation_audit.log", mode="a", encoding="utf-8"),
    ],
)
logger = logging.getLogger("duration_validator")

# NAB/ATSC standard commercial lengths, in seconds.
ALLOWED_DURATIONS_SEC: set[int] = {5, 10, 15, 30, 60, 120}
# One frame at 29.97 fps is 0.03336s; round up to 0.034s so a clean
# 29.97-rounded export snaps to 30 without admitting real short-clears.
FRAME_TOLERANCE_SEC: float = 0.034

Expected log line: 2026-07-03 02:14:07,881 | INFO | duration_validator | batch start | spot_id=- records=1204

Step 2 — The duration-parsing helper

Goal: collapse every Avion representation — clock time, frame-suffixed time, or raw seconds — into a single float of seconds, and return None for anything unparseable so it fails closed rather than silently. Drop-frame timecode uses a ; delimiter, which this parser deliberately does not accept (see Edge Cases).

python 
def parse_duration_string(raw: str) -> Optional[float]:
    """Normalize a raw Avion duration to seconds, or None if unparseable."""
    # Strip BOM, trim, and normalize a localized decimal comma to a dot.
    cleaned = raw.replace("\ufeff", "").strip().replace(",", ".")

    # HH:MM:SS or MM:SS with optional fractional seconds (29.97-frame exports).
    clock = re.match(r"^(?:(\d{1,2}):)?(\d{1,2}):(\d{1,2})(?:\.(\d+))?$", cleaned)
    if clock:
        h = int(clock.group(1) or 0)
        m = int(clock.group(2))
        s = int(clock.group(3))
        frac = float(f"0.{clock.group(4)}") if clock.group(4) else 0.0
        return h * 3600 + m * 60 + s + frac

    # Raw seconds: "30", "30.1", "60s".
    seconds = re.match(r"^(\d+(?:\.\d+)?)s?$", cleaned)
    if seconds:
        return float(seconds.group(1))

    return None

Step 3 — The Pydantic SpotRecord contract with the compliance gate

Goal: attach the compliance decision to the record itself. A model_validator(mode="before") parses the raw duration, snaps it to the nearest standard length, and sets compliance_status before the record is ever trusted downstream — the same Pydantic validator pattern used across the ingestion pipeline.

python 
from pydantic import BaseModel, Field, model_validator


class SpotRecord(BaseModel):
    spot_id: str = Field(..., description="Unique traffic-log identifier")
    raw_duration: str = Field(..., description="Raw duration string from the Avion export")
    client_code: str
    creative_title: str
    parsed_seconds: Optional[float] = Field(default=None)
    compliance_status: str = Field(default="PENDING")
    validation_notes: List[str] = Field(default_factory=list)

    @model_validator(mode="before")
    @classmethod
    def enforce_broadcast_standard(cls, data: dict) -> dict:
        # Guard against re-validation of an already-built model instance.
        if not isinstance(data, dict):
            return data

        spot_id = data.get("spot_id", "?")
        notes = list(data.get("validation_notes") or [])
        parsed = parse_duration_string(str(data.get("raw_duration", "")))
        data["parsed_seconds"] = parsed

        if parsed is None:
            data["compliance_status"] = "REJECTED"
            notes.append("MALFORMED_DURATION: could not parse duration string")
            logger.warning("reject malformed | spot_id=%s", spot_id)
            data["validation_notes"] = notes
            return data

        # Snap to the nearest NAB standard and measure the frame delta.
        nearest = min(ALLOWED_DURATIONS_SEC, key=lambda std: abs(std - parsed))
        delta = abs(parsed - nearest)

        if parsed in ALLOWED_DURATIONS_SEC or delta <= FRAME_TOLERANCE_SEC:
            data["parsed_seconds"] = float(nearest)  # canonicalize 29.97 -> 30.0
            data["compliance_status"] = "COMPLIANT"
        else:
            data["compliance_status"] = "NON_STANDARD"
            notes.append(
                f"DEVIATION: {parsed:.3f}s is {delta:.3f}s from nearest standard {nearest}s"
            )
            logger.warning("non-standard length | spot_id=%s parsed=%.3f", spot_id, parsed)

        data["validation_notes"] = notes
        return data

Step 4 — Async quarantine routing and Avstar sync

Goal: validate a batch without blocking the ingestion thread, transmit only COMPLIANT records to Avstar, and isolate everything else in a timestamped quarantine file. A sync failure never loses data — the batch is dumped for deterministic retry.

python 
import asyncio

import aiohttp

QUARANTINE_DIR = Path("./quarantine")
QUARANTINE_DIR.mkdir(exist_ok=True)


async def process_batch(records: List[dict], avstar_endpoint: str) -> None:
    compliant: List[SpotRecord] = []
    quarantined: List[SpotRecord] = []

    for raw in records:
        spot = SpotRecord(**raw)  # validation runs here; never raises on bad duration
        (compliant if spot.compliance_status == "COMPLIANT" else quarantined).append(spot)

    logger.info("batch validated | spot_id=- compliant=%d quarantined=%d",
                len(compliant), len(quarantined))
    await sync_to_avstar(compliant, avstar_endpoint)
    persist_quarantine(quarantined)


async def sync_to_avstar(records: List[SpotRecord], endpoint: str) -> None:
    if not records:
        logger.info("sync skipped | spot_id=- reason=no_compliant_records")
        return

    payload = [r.model_dump() for r in records]
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(
                endpoint, json=payload, timeout=aiohttp.ClientTimeout(total=10)
            ) as resp:
                resp.raise_for_status()
                logger.info("sync ok | spot_id=- transmitted=%d", len(payload))
        except aiohttp.ClientError as exc:
            # Fail safe: never lose compliant records on a transient endpoint error.
            Path("avstar_retry_batch.json").write_text(json.dumps(payload, indent=2))
            logger.critical("sync failed | spot_id=- error=%s dumped=retry_batch", exc)


def persist_quarantine(records: List[SpotRecord]) -> None:
    if not records:
        return
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    target = QUARANTINE_DIR / f"quarantine_{stamp}.json"
    target.write_text(json.dumps([r.model_dump() for r in records], indent=2), encoding="utf-8")
    logger.warning("quarantine written | spot_id=- count=%d path=%s", len(records), target)

Verification & Testing

Confirm three invariants: a clean 29.97-rounded export snaps to its standard length, a genuine off-length buy is flagged rather than snapped, and malformed input fails closed. Drive SpotRecord directly against a fixture of the exact string shapes an Avion export produces.

python 
def _record(raw: str) -> SpotRecord:
    return SpotRecord(spot_id="AV-100291", raw_duration=raw,
                      client_code="ACME", creative_title="Q3 Motors")

def test_duration_compliance() -> None:
    # 29.97-frame drift canonicalizes to the 30s standard.
    r = _record("00:00:29.97")
    assert r.compliance_status == "COMPLIANT"
    assert r.parsed_seconds == 30.0

    # Localized decimal comma is normalized before parsing.
    assert _record("29,97").compliance_status == "COMPLIANT"

    # A real :45 buy is flagged, not snapped to 30 or 60.
    off = _record("45")
    assert off.compliance_status == "NON_STANDARD"
    assert off.parsed_seconds == 45.0

    # Drop-frame semicolon timecode fails closed.
    assert _record("00:00:29;29").compliance_status == "REJECTED"

The 29.97 -> 30.0 canonicalization is the load-bearing assertion: it proves the frame tolerance absorbs rounding drift without admitting a true short-clear, which is what keeps billed seconds honest for reconciliation.

Edge Cases & Failure Handling

Drop-frame timecode ambiguity. NTSC drop-frame strings (00:00:29;29) surface in legacy Avion exports and use a ; delimiter. parse_duration_string rejects the semicolon on purpose — silently mapping ; to . would misread a frame count as fractional seconds and corrupt the duration. When drop-frame appears at volume, fix it upstream by reconfiguring the export profile to emit HH:MM:SS.ff or raw seconds; only if that is impossible add an explicit drop-frame translation layer that converts frames to seconds at the true frame rate before this validator runs.

A malformed-duration spike. A sudden burst of REJECTED records with MALFORMED_DURATION almost always signals an encoding shift, not bad data — Windows-1252 replacing UTF-8, or a new BOM on the export. The BOM and decimal-comma normalization in Step 2 covers the common cases; beyond that, confirm the file encoding (file -i) before ingestion rather than loosening the parser, and treat a rejection spike as an upstream export alarm.

A contractually approved off-length buy. Occasionally a :45 or :90 spot is a legitimate, cleared placement, not an error. Do not widen ALLOWED_DURATIONS_SEC globally — that reopens the door to accidental off-lengths. Instead route NON_STANDARD records to a review queue where a traffic manager can approve a specific spot_id, then re-ingest it against a per-buy allowlist. The same quarantine-and-approve pattern governs the exceptions handled in Async Batch Processing for High-Volume Logs.

FAQ

Why snap 29.97 up to 30 instead of storing the exact parsed value?

Because 29.97 is rounding drift from a 29.97 fps export, not a real 29.97-second spot — no one sells a 29.97-second buy. Snapping to the standard length within the one-frame tolerance keeps billed seconds, break math, and as-run reconciliation aligned to the contracted length. The raw value is still preserved in the audit log if you ever need to trace the original string.

Should duration validation run before or after CSV-to-JSON conversion?

After. Structural parsing comes first — the Avion CSV-to-JSON conversion lifts fields out of the raw file — and duration validation is a semantic check on an already-extracted raw_duration. Running it earlier means re-parsing file structure inside the validator, which couples two concerns that fail for different reasons.

A record is COMPLIANT here but Avstar still rejects it. Why?

This validator only certifies the duration. Avstar can still reject on an unresolved spot schema field, an unnormalized billing code, or a session/rate-limit failure at the transport layer. Inspect the Avstar response body, and confirm the upstream normalization passes ran before this one.

How do I retry the batch after an Avstar sync failure?

sync_to_avstar writes every compliant record to avstar_retry_batch.json on a ClientError, so nothing is lost. Re-POST that file once the endpoint recovers. For sustained outages, wrap the retry in the same backoff-and-session-rotation logic used for Avstar session timeouts rather than hot-looping the request.

Related content

Up one levelSchema Validation with Pydantic for Traffic Data