Broadcast Traffic Scheduling Automation

How to Migrate Avion Exports to Avstar Schemas

This guide solves one exact task: taking a batch of legacy Avion export rows and transforming them, field by field, into the richer Avstar-aligned canonical schema — applying the station timezone, converting integer cents to decimal dollars, resolving single-letter daypart codes, and quarantining any row that cannot be mapped without guessing. It is the migration procedure under Avion vs Avstar Export Formats: A Side-by-Side Comparison, itself part of Avion & Avstar Ingestion Pipelines. Getting it right is an audit obligation, not a convenience: a spot that aired under Avion and bills after an Avstar cutover must reconcile to the same revenue and the same instant, so a lossy transform fractures billing lineage and leaves an FCC public-inspection reconstruction impossible to defend.

The migration’s governing rule is that Avstar carries strictly more information than Avion — an explicit UTC offset, an explicit currency, a distinguishable null — so the transform widens every Avion row rather than copying it. Where Avion under-specifies, the migration supplies the missing fact from configuration and records that it did so, so an auditor can tell an inferred value from a sourced one. Field structure is assumed already clean; this guide validates identity and semantics, leaning on the boundary models from Schema Validation with Pydantic for Traffic Data.

Prerequisites

  • Python 3.11+ — required for the X | None union syntax and the standard-library zoneinfo module used to localize Avion station times.
  • Pinned dependencies — pydantic==2.7.1 for the canonical models. The Avion parser itself needs only the standard library (csv, decimal, datetime, zoneinfo, logging, pathlib).
  • A station timezone table — one IANA zone key per originating station (e.g. America/New_York), held in config. Avion exports carry no offset, so this table is the only correct source of the airing instant.
  • A default currency per station — Avion rate is implicit; production stations are almost always USD, but the value must be explicit config, never hard-coded in the transform.
  • Read-only access to the Avion export — a mounted file or object-store prefix scoped to read; the migration must never mutate the system it is migrating from.
  • A writable quarantine directory — a path where unmappable rows are isolated for traffic-desk review, per the field contract in the canonical traffic field data dictionary.

Step-by-Step Implementation

The migration runs as a pure boundary function over each Avion row: parse and validate it into an AvionRecord, widen it into a CanonicalSpot using the station timezone and default currency, and either emit the canonical record or quarantine the row with an actionable reason. No partial state is committed — a row either fully migrates or is isolated.

Step 1 — Structured logging and the migration report model

Goal: emit machine-parseable audit lines in the traffic-ops timestamp | level | module | spot_id shape, and fix the typed model that records the outcome of every row so the batch result is itself an audit artifact.

python 
from __future__ import annotations

import csv
import json
import logging
from datetime import datetime, time, timedelta, timezone
from decimal import Decimal, InvalidOperation
from enum import Enum
from pathlib import Path
from typing import Optional
from zoneinfo import ZoneInfo

from pydantic import BaseModel, Field, field_validator

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


def _log(level: int, spot_id: str, msg: str) -> None:
    logger.log(level, "%s | %s", spot_id, msg)


class MigrationStatus(str, Enum):
    MIGRATED = "migrated"
    QUARANTINED = "quarantined"


class MigrationOutcome(BaseModel):
    house_no: str
    status: MigrationStatus
    canonical: Optional[dict] = None       # populated on success
    reason: Optional[str] = None           # populated on quarantine
    decided_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

A migrated row logs … | INFO | avion_avstar.migrate | 00088213 | migrated -> PRIME 2026-07-14T23:30:00+00:00, and a quarantined one logs at WARNING with the reason, so a single grep over the run reconstructs every decision.

Step 2 — Define the Avion source model and the Avstar-aligned target

Goal: validate each Avion row at the boundary (8-digit house number, positive integer cents, blank-or-* billing code coerced to null) and declare the canonical target the migration widens into.

python 
class AvionRecord(BaseModel):
    """A parsed Avion row: pipe-delimited, station-local, integer cents."""

    house_no: str
    air_date: str                          # MMDDYY
    air_time: str                          # HHMM, 24-hour, may exceed 2400
    daypart_code: str                      # single letter
    rate_cents: int = Field(ge=0)
    billing_code: Optional[str] = None

    @field_validator("house_no")
    @classmethod
    def _padded(cls, v: str) -> str:
        d = v.strip()
        if not d.isdigit() or len(d) != 8:
            raise ValueError(f"house_no '{v}' is not an 8-digit Avion identifier")
        return d

    @field_validator("billing_code", mode="before")
    @classmethod
    def _blank_is_none(cls, v: Optional[str]) -> Optional[str]:
        # Avion cannot separate blank from absent; both collapse to None here,
        # and Step 4 decides whether that ambiguity is safe to migrate.
        if v is None:
            return None
        s = v.strip()
        return None if s in ("", "*") else s


AVION_DAYPART_CODES: dict[str, str] = {
    "O": "OVERNIGHT", "M": "MORNING", "D": "DAYTIME",
    "E": "EARLY_FRINGE", "P": "PRIME", "L": "LATE",
}


class CanonicalSpot(BaseModel):
    """Avstar-aligned canonical target: UTC instant, decimal rate, explicit null."""

    house_number: str
    aired_at: datetime
    daypart: str
    rate: Decimal
    currency: str
    billing_code: Optional[str]
    source_format: str = "avion"
    source_local_time: str                 # lineage: pre-transform local instant

Step 3 — Widen one Avion row into the canonical schema

Goal: apply the field-by-field transform. Each canonical field is produced by exactly one rule, and the mapping is the reviewable heart of the migration.

The transform implements this mapping table, which restates the format comparison as concrete operations:

Canonical field Avion source Transform
house_number house_no Copy the 8-digit identifier verbatim
aired_at air_date + air_time Combine, apply broadcast-day rollover, localize to station zone, convert to UTC
daypart daypart_code Look up the single letter in the canonical daypart table
rate rate_cents Divide integer cents by 100 as Decimal
currency — (implicit) Supply the station default, flagged as inferred
billing_code billing_code Copy, or quarantine if blank (ambiguous null)
python 
def widen_avion_row(
    rec: AvionRecord,
    station_tz: str,
    default_currency: str = "USD",
) -> CanonicalSpot:
    """Transform one validated Avion row into the canonical Avstar-aligned shape."""
    sid = rec.house_no

    # Broadcast-day rollover: Avion may export 01:15 next-day as "2515".
    hh, mm = int(rec.air_time[:-2]), int(rec.air_time[-2:])
    day_offset, hh = divmod(hh, 24)     # 25 -> (1 day, 01:00)
    base_date = datetime.strptime(rec.air_date, "%m%d%y").date()
    # Push onto the following calendar day, then localize to the station zone.
    naive = datetime.combine(base_date, time(hh, mm)) + timedelta(days=day_offset)
    local = naive.replace(tzinfo=ZoneInfo(station_tz))
    aired_utc = local.astimezone(timezone.utc)

    daypart = AVION_DAYPART_CODES.get(rec.daypart_code.strip().upper())
    if daypart is None:
        raise ValueError(f"ERR_DAYPART_CODE: unknown Avion code '{rec.daypart_code}'")

    canonical = CanonicalSpot(
        house_number=rec.house_no,
        aired_at=aired_utc,
        daypart=daypart,
        rate=(Decimal(rec.rate_cents) / Decimal(100)).quantize(Decimal("0.01")),
        currency=default_currency,                       # implicit -> explicit
        billing_code=rec.billing_code,
        source_local_time=local.isoformat(),             # lineage annotation
    )
    _log(logging.INFO, sid,
         f"widened -> {daypart} {aired_utc.isoformat()} rate={canonical.rate}")
    return canonical

Step 4 — Quarantine unmappable rows instead of guessing

Goal: any row that would require a guess — an unknown daypart code, an unresolvable time, or an ambiguous blank billing code — is written to quarantine with an actionable reason rather than migrated with a fabricated value.

python 
class AvionMigrator:
    def __init__(
        self,
        station_tz: str,
        default_currency: str = "USD",
        quarantine_path: Path = Path("quarantine/"),
    ) -> None:
        self.station_tz = station_tz
        self.default_currency = default_currency
        self.quarantine_path = quarantine_path
        self.quarantine_path.mkdir(parents=True, exist_ok=True)

    def migrate_row(self, raw: dict[str, str]) -> MigrationOutcome:
        house = raw.get("house_no", "<missing>")
        try:
            rec = AvionRecord(**raw)
            # A blank billing code is ambiguous from Avion: refuse to assume absent.
            if rec.billing_code is None and raw.get("billing_code", "").strip() == "":
                raise ValueError("ERR_NULL_AMBIGUOUS: blank Avion billing_code")
            canonical = widen_avion_row(rec, self.station_tz, self.default_currency)
            return MigrationOutcome(
                house_no=rec.house_no,
                status=MigrationStatus.MIGRATED,
                canonical=canonical.model_dump(mode="json"),
            )
        except (ValueError, InvalidOperation) as exc:
            self._quarantine(house, raw, str(exc))
            _log(logging.WARNING, house, f"quarantined | {exc}")
            return MigrationOutcome(
                house_no=house, status=MigrationStatus.QUARANTINED, reason=str(exc)
            )

    def _quarantine(self, house: str, raw: dict[str, str], error: str) -> None:
        target = self.quarantine_path / f"{house}.json"
        target.write_text(json.dumps(
            {"avion_row": raw, "error": error,
             "quarantined_at": datetime.now(timezone.utc).isoformat()},
            indent=2,
        ))

    def migrate_batch(self, rows: list[dict[str, str]]) -> list[MigrationOutcome]:
        return [self.migrate_row(r) for r in rows]
Avion-to-Avstar migration transformation flow A raw Avion row enters the migrator and is validated into an AvionRecord, checking the eight-digit house number, non-negative integer cents, and coercing a blank or asterisk billing code. Validated rows enter the widening transform, which combines the split MMDDYY date and HHMM time, applies the broadcast-day rollover for times past 2400, localizes to the station timezone and converts to UTC, looks up the single-letter daypart code in the canonical table, divides integer cents by one hundred into a decimal rate, and supplies the implicit currency from station config as an explicit value. A decision gate then asks whether every field mapped without a guess. If yes, the row becomes a CanonicalSpot with status MIGRATED carrying a UTC instant, decimal rate, and source-local-time lineage. If a daypart code is unknown, a time is unresolvable, or a blank billing code is ambiguous, the row is written to quarantine with status QUARANTINED and an actionable reason for traffic-desk review. Raw Avion row 00088213|071426| 1930|P|425000|* Validate AvionRecord house · cents · code Widen transform • date+time → UTC instant • broadcast-day rollover • localize station tz • daypart letter → name • cents ÷ 100 → decimal • implicit → explicit USD + source_local_time lineage mapped clean? CanonicalSpot status = MIGRATED PRIME · 4250.00 USD Quarantine status = QUARANTINED <house>.json + reason yes no

Figure — The migration flow: a raw Avion row is validated, widened field by field into the Avstar-aligned canonical shape, and either emitted as a CanonicalSpot or quarantined with an actionable reason when a field cannot be mapped without guessing.

Step 5 — Orchestrate the batch and read the export

Goal: stream the Avion export (decoded as CP1252, never UTF-8), map each row, and summarize the run so the migrated and quarantined counts are visible at a glance.

python 
def run_migration(export_path: Path, station_tz: str) -> list[MigrationOutcome]:
    migrator = AvionMigrator(station_tz=station_tz)
    outcomes: list[MigrationOutcome] = []
    # Avion is CP1252; decoding as UTF-8 corrupts advertiser names with mojibake.
    with export_path.open(encoding="cp1252", newline="") as fh:
        # Skip the leading #HDR control record; the rest is pipe-delimited.
        reader = csv.reader((ln for ln in fh if not ln.startswith("#HDR")),
                            delimiter="|")
        for cols in reader:
            if len(cols) < 6:
                continue  # short row; the parser guarantees column count upstream
            raw = {
                "house_no": cols[0], "air_date": cols[1], "air_time": cols[2],
                "daypart_code": cols[3], "rate_cents": cols[4], "billing_code": cols[5],
            }
            outcomes.append(migrator.migrate_row(raw))
    migrated = sum(o.status is MigrationStatus.MIGRATED for o in outcomes)
    _log(logging.INFO, "BATCH",
         f"migrated={migrated} quarantined={len(outcomes) - migrated}")
    return outcomes

Verification & Testing

Correct behaviour rests on two properties: the transform is lossless-widening (every canonical value is either sourced or explicitly inferred) and it fails closed (an unmappable row never produces a MIGRATED outcome). Both are assertable against fixture rows drawn from a real Avion export.

python 
from decimal import Decimal

migrator = AvionMigrator(station_tz="America/New_York", default_currency="USD")

# 1. Happy path: a prime-time spot widens into the canonical UTC instant.
ok = migrator.migrate_row({
    "house_no": "00088213", "air_date": "071426", "air_time": "1930",
    "daypart_code": "P", "rate_cents": "425000", "billing_code": "RCPRIME30",
})
assert ok.status is MigrationStatus.MIGRATED
assert ok.canonical["aired_at"] == "2026-07-14T23:30:00Z"   # 19:30 EDT -> 23:30 UTC
assert ok.canonical["rate"] == "4250.00"                    # cents -> dollars
assert ok.canonical["daypart"] == "PRIME"

# 2. Fail-closed: an ambiguous blank billing code is quarantined, never migrated.
bad = migrator.migrate_row({
    "house_no": "00088999", "air_date": "071426", "air_time": "1930",
    "daypart_code": "P", "rate_cents": "425000", "billing_code": "",
})
assert bad.status is MigrationStatus.QUARANTINED
assert "ERR_NULL_AMBIGUOUS" in bad.reason
assert Path("quarantine/00088999.json").exists()

Because widen_avion_row is a pure function of its row, station timezone, and default currency, replaying an archived Avion export during an audit reproduces byte-identical canonical records. Run the suite in CI against a golden fixture of Avion → canonical pairs so a change to the daypart table or the timezone logic that would silently re-date inventory fails the build instead of shipping. The full field constraints these records satisfy are defined in the canonical traffic field data dictionary.

Edge Cases & Failure Handling

  • Broadcast-day rollover. An Avion air_time of 2515 means 01:15 on the following broadcast day, a station convention for late-night spots. The transform applies divmod(hh, 24) and pushes the date forward before localizing, so the airing lands on the correct calendar instant rather than 24 hours early. Verify with a fixture that spans midnight against a known-good Avstar timestamp.
  • DST-ambiguous local time. A 0130 airing on a fall-back night maps to two possible UTC instants. ZoneInfo resolves it deterministically, and the source_local_time lineage field records the pre-transform value so an auditor can confirm the choice. Where the station’s own records disagree, quarantine rather than migrate.
  • Unknown daypart code. A station that adds a W (weekend) daypart the canonical table does not know raises ERR_DAYPART_CODE and is quarantined. Extend AVION_DAYPART_CODES only after confirming the station’s authorized code table, then re-run — the pure transform maps the recovered rows identically. This mirrors the reconciliation discipline in Reconciling Field Differences Between Avion and Avstar.

FAQ

Why widen toward Avstar instead of just copying Avion fields across?

Because Avstar carries strictly more information than Avion, a straight copy would leave the canonical record missing the offset, the currency, and a distinguishable null. Migrating the other direction — Avstar down to Avion’s shape — is irreversibly lossy. The full information asymmetry is laid out in Avion vs Avstar Export Formats: A Side-by-Side Comparison, and it is the reason the transform takes a station timezone and default currency as required arguments.

Where does the timezone come from if Avion never exports one?

From a per-station configuration table of IANA zone keys, never from the host clock. An Avion time is station-local with no offset, so the airing instant is undefined until you apply the originating station’s zone. The migration records the pre-transform local time as lineage so the applied zone is auditable. The parsing side of this — decoding the raw bytes correctly first — is covered in Parsing Avion Export Formats.

Can I safely re-run a migration after fixing quarantined rows?

Yes. widen_avion_row is a pure function, so a row that migrated once produces a byte-identical canonical record on any re-run — committing it again is idempotent. Fix the source data behind the quarantine reason (an unknown daypart code, a blank billing code), re-run the batch, and only the newly resolvable rows change state. Persist the migrated house numbers if you want to skip already-done work, gating on the canonical identifier rather than arrival order.

What should happen to a blank billing code from Avion?

Quarantine it. Avion cannot distinguish a deliberately empty billing code from an absent one, and a spot that airs without a billing code cannot be reconciled to revenue. Assuming “absent” would let an unbillable spot into the schedule. Resolve the ambiguity against the source system, or apply the field constraints from the spot schema to confirm the intended value before migrating.

Related content

Up one levelAvion vs Avstar Export Formats: A Side-by-Side Comparison RelatedReconciling Field Differences Between Avion and Avstar