Parsing Avion Export Formats

Broadcast traffic automation lives or dies on the first deterministic gate after a schedule leaves the traffic system: turning a raw Avion export into a validated, scheduler-ready record set. When parsing is sloppy, the failure is silent — a trailing delimiter shifts a column, a cp1252 smart-quote truncates a read, or a mid-day revision overwrites a live spot — and the corruption only surfaces later as a missed airing, a double-booked break, or a reconciliation gap that costs real revenue. This guide sits inside the broader Avion & Avstar Ingestion Pipelines architecture and owns the extraction boundary: reading historically inconsistent Avion files, tokenizing them safely, normalizing time to UTC, and reconciling delta records before anything reaches validation. It is written for broadcast traffic managers who need to trust the ingested log, and for the Python automation engineers who build and operate the parser under legacy-format constraints.

Concept & Data Model #

The parsing layer is the untrusted-input firewall of the pipeline. Everything upstream of it is an opaque file on disk; everything downstream assumes clean, typed, timezone-aware records. Its single responsibility is structural extraction — delimiter resolution, encoding normalization, header mapping, tokenization, and time normalization — while deferring semantic business rules to the dedicated schema validation with Pydantic for traffic data layer that runs immediately after. Keeping these concerns separate means a malformed row (truncated file, BOM artifact, ragged column count) is isolated from a business-rule failure (invalid spot length, unknown billing code), and each class of error routes to a different remediation queue.

Avion emits three physical shapes in the wild, and a production parser must detect and handle all of them:

Regardless of shape, the parser must produce one canonical intermediate record per airing before validation. The minimum fields it is responsible for extracting and normalizing map directly onto the canonical spot schema and metadata:

Figure — Parsing flow that converts a raw Avion export into structured, UTC-normalized records via dialect/encoding detection, tokenization with source-line hashing, and reconciliation of delta records and preemptions before handoff to validation.

Implementation Approach #

Three design decisions dominate a robust Avion parser, and each is a trade-off worth making deliberately.

Detect the dialect; do not hard-code it. Avion export profiles drift between station groups and software versions — the same “CSV” export may switch from comma to pipe after a vendor upgrade. Sniffing the dialect from a sample of the file is more resilient than a pinned delimiter, but the sniffer must be constrained to a candidate delimiter set (,, |, \t) so it cannot guess a spurious separator from a data value. When a station’s profile is known and stable, an explicit dialect override wins on determinism; treat sniffing as the fallback, not the default.

Decode defensively with an ordered codec chain. Legacy Avion dumps ship with mixed Windows/Unix line endings and inconsistent character sets. Attempting a strict utf-8 read first and falling back to cp1252 (never latin-1, which never raises and therefore hides corruption) surfaces encoding problems loudly instead of silently mangling a billing code. Always record which codec succeeded — an unexpected fallback is an early warning of an upstream export-configuration change.

Decouple retrieval from parsing. Exports are polled or pushed through the Avstar control plane; the credential lifecycle, token scoping, and per-minute request ceilings are documented in Avstar API authentication and rate limits. The parser must never call that API inline. Route raw exports to a staging bucket or durable queue first, so a rate-limit burst upstream cannot corrupt the parser’s state machine, and so extraction and normalization workers scale independently. For daily logs that routinely exceed 10 GB, stream rows through the chunked async batch processing for high-volume logs pattern rather than loading the file into memory. Teams standing up a flat-file-to-JSON migration should follow the step-by-step Avstar CSV to JSON conversion walkthrough, which preserves hierarchical spot metadata while flattening legacy columnar structures.

Production Python Implementation #

The parser below is deployable rather than illustrative. It detects dialect and encoding, tokenizes each row while capturing an immutable source_hash, normalizes the source local time to UTC through DST transitions, classifies delta records, and routes unparseable rows to a dead-letter path instead of dropping them. Semantic validation is intentionally left to the Pydantic layer downstream; this module only guarantees structural integrity. Every reject emits a log line in the traffic-ops format timestamp | level | module | spot_id.

from __future__ import annotations

import csv
import hashlib
import io
import logging
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Iterator, Optional
from zoneinfo import ZoneInfo

# Traffic-ops structured log format: timestamp | level | module | spot_id
logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(spot_id)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("avion.parser")


def _log(level: int, msg: str, spot_id: str = "-") -> None:
    """Emit a log line carrying the spot_id, the pivot key for every traffic audit."""
    logger.log(level, msg, extra={"spot_id": spot_id})


# Candidate delimiters we allow the sniffer to consider — nothing else.
_ALLOWED_DELIMITERS = ",|\t"
# Ordered codec chain: strict first so corruption is loud, never latin-1 (never raises).
_CODEC_CHAIN: tuple[str, ...] = ("utf-8", "cp1252")

# Frame rate per market; frame-accurate durations prevent break overrun.
_FRAME_RATE = 30


@dataclass(slots=True)
class ParsedSpotRecord:
    """Structural output of the parser — typed, but not yet business-validated."""

    spot_id: str
    air_datetime: datetime          # timezone-aware, normalized to UTC
    duration_frames: int
    break_id: str
    isci: str
    billing_code: str
    record_kind: str                # "original" | "delta"
    source_hash: str
    source_line: int
    codec_used: str = "utf-8"
    warnings: list[str] = field(default_factory=list)


class AvionParseError(Exception):
    """Raised for a structurally unrecoverable row; carries the raw line for the DLQ."""

    def __init__(self, message: str, raw_line: str, line_no: int) -> None:
        super().__init__(message)
        self.raw_line = raw_line
        self.line_no = line_no


class AvionExportParser:
    """Deterministic reader for delimited Avion traffic exports.

    Retrieval is out of scope by design: point this at a file already staged from
    the Avstar control plane. The parser never mutates the source file.
    """

    def __init__(self, source_tz: str, dialect_override: Optional[str] = None) -> None:
        # Source timezone is explicit — Avion date/time columns are wall-clock local.
        self._source_tz = ZoneInfo(source_tz)
        self._dialect_override = dialect_override

    # ---- decoding -----------------------------------------------------------
    def _decode(self, path: Path) -> tuple[str, str]:
        raw = path.read_bytes().lstrip(b"\xef\xbb\xbf")  # drop UTF-8 BOM if present
        for codec in _CODEC_CHAIN:
            try:
                text = raw.decode(codec)
                if codec != _CODEC_CHAIN[0]:
                    _log(logging.WARNING, f"Fell back to {codec}; check upstream export config")
                return text, codec
            except UnicodeDecodeError:
                continue
        raise AvionParseError("No codec in chain could decode file", "", 0)

    # ---- dialect ------------------------------------------------------------
    def _sniff(self, sample: str) -> csv.Dialect:
        if self._dialect_override:
            csv.register_dialect("_avion", delimiter=self._dialect_override)
            return csv.get_dialect("_avion")  # type: ignore[return-value]
        # Constrain the sniffer so it cannot infer a delimiter from a data value.
        return csv.Sniffer().sniff(sample, delimiters=_ALLOWED_DELIMITERS)

    # ---- duration parsing ---------------------------------------------------
    @staticmethod
    def _to_frames(token: str) -> int:
        token = token.strip().lstrip(":")
        if token.endswith("s"):
            token = token[:-1]
        seconds = int(token)
        if seconds <= 0:
            raise ValueError(f"non-positive duration: {token!r}")
        return seconds * _FRAME_RATE

    # ---- main entry ---------------------------------------------------------
    def parse(self, path: Path) -> Iterator[ParsedSpotRecord]:
        text, codec = self._decode(path)
        dialect = self._sniff(text[:8192])
        reader = csv.DictReader(io.StringIO(text), dialect=dialect)

        for line_no, row in enumerate(reader, start=2):  # line 1 is the header
            # Trailing delimiter yields a stray None key — drop it before use.
            row.pop(None, None)  # type: ignore[arg-type]
            raw_line = dialect.delimiter.join(v or "" for v in row.values())
            source_hash = hashlib.sha256(raw_line.encode("utf-8")).hexdigest()
            spot_id = (row.get("spot_id") or "").strip()

            try:
                air_local = datetime.strptime(
                    f"{row['air_date'].strip()} {row['air_time'].strip()}",
                    "%Y-%m-%d %H:%M:%S",
                ).replace(tzinfo=self._source_tz)
                record = ParsedSpotRecord(
                    spot_id=spot_id,
                    air_datetime=air_local.astimezone(ZoneInfo("UTC")),
                    duration_frames=self._to_frames(row["length"]),
                    break_id=(row.get("break_id") or "").strip(),
                    isci=(row.get("isci") or "").strip().upper(),
                    billing_code=(row.get("billing_code") or "").strip(),
                    record_kind="delta" if (row.get("revision") or "").strip() else "original",
                    source_hash=source_hash,
                    source_line=line_no,
                    codec_used=codec,
                )
            except (KeyError, ValueError) as exc:
                _log(logging.ERROR, f"Row {line_no} unparseable: {exc}", spot_id or "-")
                raise AvionParseError(str(exc), raw_line, line_no) from exc

            _log(logging.INFO, f"Parsed {record.record_kind} spot @ line {line_no}", spot_id)
            yield record

Validation & Edge Cases #

Structural parsing survives production only if it handles the broadcast-specific boundary conditions that generic CSV code ignores.

• DST transition ambiguity. A spot scheduled at 01:30 on a fall-back night maps to two real UTC instants. Because Avion stores wall-clock local time, the parser must resolve the offset through the source ZoneInfo, and — for the ambiguous hour — apply the station’s documented fold convention rather than guessing. Never store a naive datetime; a spot without an offset is unschedulable.
• Sports overruns and delta records. When a live event runs long, Avion emits mid-day delta rows that revise or preempt already-placed spots. The parser tags these as record_kind="delta" but must not itself decide the winner; reconciliation of overlapping air_datetime values and conflicting break positions is a deterministic later step, and the priority logic aligns with detecting time-slot conflicts in traffic logs.
• Zero-duration and malformed length tokens. A :00 or empty length token is a structural error, not a valid spot — it is raised, hashed, and dead-lettered rather than coerced to zero frames, which would silently corrupt break timing.
• Trailing delimiters and ragged rows. Avion exports routinely append a trailing separator, producing a phantom column. The parser strips the stray None key before field access; a row whose real column count still disagrees with the header is rejected, never realigned by position.
• Preemption tiers and competitive separation. Position and break tokens are preserved verbatim so downstream logic can enforce separation windows; the parser must not renumber positions, because ordinal slot is what separation rules depend on.

Integration Points #

The parser is a segment in a longer contract. Upstream, it consumes files that the Avstar polling worker has staged; it must treat the staging location as the boundary and read files read-only. Downstream, it hands its ParsedSpotRecord stream to the Pydantic traffic-data validators, which apply semantic business rules and reject anything the parser could only structurally accept. The source_hash and source_line fields travel with every record so that a spot rejected three stages later can still be traced back to an exact byte range of the original export — the lineage that billing and as-run reconciliation depend on, and that the standardizing billing codes across traffic systems normalization step later keys off. The message the parser publishes to the validation queue is small and versioned:

{
  "schema_version": "avion-parse.v2",
  "spot_id": "AV-2026-183-0442",
  "air_datetime": "2026-07-03T18:30:00Z",
  "duration_frames": 900,
  "break_id": "P07-B03",
  "isci": "ABCD1234H",
  "billing_code": "RAW-77-Q3",
  "record_kind": "original",
  "source_hash": "9f2c…e1",
  "source_line": 4471
}

Compliance & Audit Considerations #

Because the parser is the first point at which an external file becomes internal data, it is also the first point that must be auditable. Three rules are non-negotiable. First, the parser never mutates the source file — the staged export is retained unchanged so an auditor can re-derive every record from the original bytes. Second, the source_hash captured before any transformation is the immutable link between an emitted record and its origin line; storing that hash alongside the record satisfies the lineage requirement that FCC political-file reconstruction and SOC 2 change-traceability both impose. Third, rejected rows are quarantined to a dead-letter path with their raw line and error, never silently discarded — a dropped political or EAS-adjacent spot is a compliance incident, not a data-quality footnote. The access controls governing where staged files and quarantine artifacts may live are defined in security boundaries for traffic database access; the parser process should hold read-only rights to the staging area and append-only rights to the audit log.

Troubleshooting & Common Errors #

Related #

• Schema validation with Pydantic for traffic data — the semantic gate that runs immediately after parsing, enforcing spot-length, daypart, and billing-code rules the parser deliberately defers.
• Async batch processing for high-volume logs — how to stream 10 GB+ daily exports through the parser under bounded memory with explicit backpressure.
• Avstar API authentication and rate limits — the retrieval boundary that stages exports before parsing, covering OAuth2 flows and per-minute request ceilings.
• Step-by-step Avstar CSV to JSON conversion — a concrete flat-file-to-JSON mapping that preserves hierarchical spot metadata.
• Avion & Avstar Ingestion Pipelines — the parent architecture that positions parsing within the full ingestion-to-scheduling path.