Rerouting Spots Displaced by EAS Activations

This guide solves one exact task: given a finalized Emergency Alert System interrupt window and the commercial schedule it ran over, compute which spots the window displaced, mark each with its precise overlap and disposition, and hand the eligible ones to make-good recovery as clean, idempotent preemption signals. It is the detection-and-hand-off procedure that sits under Handling EAS Emergency Interrupts in Ad Scheduling, itself a subsystem of Spot Scheduling Validation & Rule Engines. Getting it right is not cosmetic. The set of displaced spots is the exact set of contractual debts an activation created; miss one and an advertiser is silently short-changed, double-count one and the station over-credits inventory it never owed. Both errors surface in a billing audit, so the computation must be deterministic, replayable, and defensible line by line.

The recovery itself is not this guide’s job. Once a spot is marked displaced, it is forwarded to make-good routing for preemptions, which owns eligibility scoring and inventory selection. Here we do the upstream work that routing depends on: turn one interrupt window into a correct, per-spot displacement record that carries enough context — overlap, disposition, and the normalized billing identity from billing code normalization — for the recovery to bill against the same account the original spot debited.

Prerequisites #

• Python 3.11+ — required for the X | None union syntax, frozenset typing, and timezone-aware datetime.now(timezone.utc) used throughout.
• Pydantic v2 — pin it exactly (pydantic==2.9.2); the validators below rely on v2’s field_validator and model_validator semantics, which differ from v1.
• A finalized interrupt window — an InterruptWindow with a real started_at and ended_at in UTC, produced when the ENDEC reports end-of-message. This guide does not detect the window; it consumes one.
• A schedule slice in canonical shape — the break-level spot rows that were scheduled during the alert period, each with spot_id, house_number, isci, scheduled_start, and duration_ms, normalized to the single spot schema before they reach the router.
• A make-good routing endpoint — a queue topic or function the hand-off publishes eas.displacement.v1 messages to; the router consumes them asynchronously.

Step-by-Step Implementation #

The router runs as a pure function over one window and one schedule slice: intersect each spot with the window, classify the overlap, drop the non-overlapping rows, and emit a displacement signal per surviving spot. No playout state is touched and nothing is billed here — the output is a list of messages the make-good subsystem consumes.

Step 1 — Model the window, the schedule, and the displacement signal #

Goal: fix the typed models that carry a spot through the computation and emit the traffic-ops audit line. Every field the make-good router needs must survive the hand-off, so the signal carries overlap, disposition, and billing identity, not just an id.

from __future__ import annotations

import hashlib
import logging
from datetime import datetime, timedelta, timezone
from enum import Enum

from pydantic import BaseModel, Field, field_validator, model_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("traffic.eas.reroute")


class Disposition(str, Enum):
    FULLY_DISPLACED = "fully_displaced"
    PARTIALLY_DISPLACED = "partially_displaced"


def _require_utc(v: datetime) -> datetime:
    # Overlap arithmetic in a naive zone drifts by an hour across a DST change.
    if v.tzinfo is None:
        raise ValueError("timestamp must be timezone-aware (UTC)")
    return v.astimezone(timezone.utc)


class InterruptWindow(BaseModel):
    window_id: str
    activation_id: str
    event_code: str                    # EAN, TOR, SVR, RWT, ...
    started_at: datetime
    ended_at: datetime

    _v_start = field_validator("started_at")(classmethod(lambda cls, v: _require_utc(v)))
    _v_end = field_validator("ended_at")(classmethod(lambda cls, v: _require_utc(v)))

    @model_validator(mode="after")
    def _positive_duration(self) -> "InterruptWindow":
        if self.ended_at <= self.started_at:
            raise ValueError("ended_at must be after started_at")
        return self


class ScheduledSpot(BaseModel):
    spot_id: str
    house_number: str
    isci: str
    scheduled_start: datetime
    duration_ms: int = Field(gt=0)     # Zero-duration rows are malformed

    _v_start = field_validator("scheduled_start")(
        classmethod(lambda cls, v: _require_utc(v))
    )

    @property
    def scheduled_end(self) -> datetime:
        return self.scheduled_start + timedelta(milliseconds=self.duration_ms)


class DisplacementSignal(BaseModel):
    entry_id: str                      # Deterministic: a replay yields the same key
    activation_id: str
    event_code: str
    spot_id: str
    house_number: str
    isci: str
    overlap_ms: int = Field(gt=0)
    disposition: Disposition
    makegood_eligible: bool

Step 2 — Compute the interrupt-window overlap #

Goal: measure exactly how many milliseconds of each spot the window covered. The overlap is the intersection of two half-open intervals; anything else is arithmetic waiting to double-count a boundary.

def overlap_ms(window: InterruptWindow, spot: ScheduledSpot) -> int:
    # Intersect [spot_start, spot_end) with [window_start, window_end).
    # Half-open intervals mean a spot ending exactly at window_start counts as 0.
    latest_start = max(window.started_at, spot.scheduled_start)
    earliest_end = min(window.ended_at, spot.scheduled_end)
    delta = (earliest_end - latest_start).total_seconds() * 1000
    return int(delta) if delta > 0 else 0

Running this on a 30-second spot that started five seconds before a tornado warning seized the air returns 25000 — the spot delivered its first five seconds, then lost the remaining twenty-five. A spot that finished before the window opened returns 0 and will be dropped in Step 4.

Step 3 — Classify the disposition #

Goal: decide whether the window took the whole spot or clipped it. The distinction drives whether billing owes a full or pro-rated make-good, so it must be explicit rather than inferred later.

# Codes that always owe a make-good on displaced commercial inventory.
_ALWAYS_ELIGIBLE: frozenset[str] = frozenset({"EAN", "TOR", "SVR", "SVA", "FFW"})


def classify(overlap: int, spot_duration_ms: int) -> Disposition:
    # >=99% coverage is "full": a sub-frame remainder is not a real airing.
    if overlap >= spot_duration_ms * 0.99:
        return Disposition.FULLY_DISPLACED
    return Disposition.PARTIALLY_DISPLACED

The 99% threshold absorbs frame-boundary rounding: a spot the window covered to within a single frame did not meaningfully air, so treating it as fully displaced avoids a spurious pro-rated credit that no advertiser would accept and no auditor would expect.

Step 4 — Mark displaced spots and build signals #

Goal: walk the schedule slice, keep only the spots the window actually overlapped, and build one DisplacementSignal per survivor with a deterministic key. The key makes the whole computation idempotent — reran on the same window, it produces byte-identical signals.

def _entry_id(activation_id: str, spot_id: str) -> str:
    # Deterministic key: a replayed activation reroutes idempotently.
    seed = f"{activation_id}:{spot_id}"
    return hashlib.sha256(seed.encode()).hexdigest()[:24]


def mark_displaced(
    window: InterruptWindow, schedule: list[ScheduledSpot]
) -> list[DisplacementSignal]:
    eligible = window.event_code in _ALWAYS_ELIGIBLE
    signals: list[DisplacementSignal] = []
    for spot in schedule:
        covered = overlap_ms(window, spot)
        if covered <= 0:
            continue  # Window never touched this spot; no debt, no signal.
        signal = DisplacementSignal(
            entry_id=_entry_id(window.activation_id, spot.spot_id),
            activation_id=window.activation_id,
            event_code=window.event_code,
            spot_id=spot.spot_id,
            house_number=spot.house_number,
            isci=spot.isci,
            overlap_ms=covered,
            disposition=classify(covered, spot.duration_ms),
            makegood_eligible=eligible,
        )
        signals.append(signal)
        logger.info(
            "%s | displaced by %s (%s) | overlap=%dms | %s | eligible=%s",
            spot.spot_id, window.event_code, window.activation_id,
            covered, signal.disposition.value, eligible,
        )
    return signals

A representative log line reads 2026-07-14T21:03:11+00:00 | INFO | traffic.eas.reroute | SP-2026-0714-0192 | displaced by TOR (EAS-2026-0714-TOR-0007) | overlap=30000ms | fully_displaced | eligible=True.

Step 5 — Hand off to make-good routing #

Goal: forward each eligible signal to the recovery subsystem as an eas.displacement.v1 message, and let ineligible ones (a scheduled test under order terms that owe nothing) fall through to the ledger without a route. The hand-off is publish-only, so the router stays the single owner of inventory decisions.

def route_displacements(
    window: InterruptWindow,
    schedule: list[ScheduledSpot],
    publish,                            # callable(topic: str, payload: dict) -> None
) -> list[DisplacementSignal]:
    signals = mark_displaced(window, schedule)
    for signal in signals:
        if not signal.makegood_eligible:
            logger.info(
                "%s | test-code displacement, no auto-route | deferred to terms",
                signal.spot_id,
            )
            continue
        publish("makegood.preemptions", signal.model_dump(mode="json"))
        logger.info(
            "%s | handed to make-good routing | entry=%s",
            signal.spot_id, signal.entry_id,
        )
    return signals

The publish callable is injected so the same function drives a real queue in production and a list-appending stub in tests. The payload is the DisplacementSignal serialized to JSON — the exact eas.displacement.v1 contract the make-good router consumes, mapping event_code to its EAS_ACTIVATION preemption reason.

Figure — Rerouting flow: each scheduled spot is intersected with the interrupt window; overlapping spots are classified and, if the event code is federally eligible, published to make-good routing while zero-overlap spots are dropped.

Verification & Testing #

Correct behaviour rests on two properties: the overlap is exact at the boundaries, and a non-overlapping spot never produces a signal. Both are assertable against fixture data drawn from a real severe-weather log.

NOOP: list[dict] = []
def _stub_publish(topic: str, payload: dict) -> None:
    NOOP.append(payload)

window = InterruptWindow(
    window_id="W-1", activation_id="EAS-2026-0714-TOR-0007", event_code="TOR",
    started_at=datetime(2026, 7, 14, 21, 3, 0, tzinfo=timezone.utc),
    ended_at=datetime(2026, 7, 14, 21, 5, 0, tzinfo=timezone.utc),
)
schedule = [
    # A: fully inside the window → 30000ms, fully displaced
    ScheduledSpot(spot_id="SP-A", house_number="HN-1", isci="AAAA1111",
                  scheduled_start=datetime(2026, 7, 14, 21, 3, 30, tzinfo=timezone.utc),
                  duration_ms=30000),
    # B: starts 5s before the window → 25000ms, partially displaced
    ScheduledSpot(spot_id="SP-B", house_number="HN-2", isci="BBBB2222",
                  scheduled_start=datetime(2026, 7, 14, 21, 2, 55, tzinfo=timezone.utc),
                  duration_ms=30000),
    # C: finished before the window opened → 0ms, dropped
    ScheduledSpot(spot_id="SP-C", house_number="HN-3", isci="CCCC3333",
                  scheduled_start=datetime(2026, 7, 14, 21, 2, 0, tzinfo=timezone.utc),
                  duration_ms=30000),
]

signals = route_displacements(window, schedule, _stub_publish)

# 1. Exactly two spots displaced; the pre-window spot is dropped.
assert {s.spot_id for s in signals} == {"SP-A", "SP-B"}
assert signals[0].overlap_ms == 30000
assert signals[0].disposition is Disposition.FULLY_DISPLACED
assert signals[1].overlap_ms == 25000
assert signals[1].disposition is Disposition.PARTIALLY_DISPLACED

# 2. Idempotency: rerunning yields identical deterministic keys.
again = mark_displaced(window, schedule)
assert [s.entry_id for s in again] == [s.entry_id for s in signals]

# 3. Both eligible TOR spots reached the make-good topic.
assert len(NOOP) == 2

Because mark_displaced is pure and its keys are a hash of (activation_id, spot_id), replaying an archived window during an audit reproduces byte-identical signals. Run this suite in CI against a golden fixture of window-plus-schedule pairs so a change to the overlap or classification rule that would silently re-credit inventory fails the build instead of shipping.

Edge Cases & Failure Handling #

• Spot straddling the window edge on a frame boundary. A spot whose last frame lands within a millisecond of started_at can score a tiny positive overlap and route a make-good for a spot that effectively aired. Align the spot boundaries to the signalled break edges from SCTE-35 ad signaling before computing overlap, so the arithmetic runs on true splice points rather than nominal log times, and let the 99% classification threshold absorb the residue.
• A window with no end-of-message. If the ENDEC never reports EOM, an upstream fault, the InterruptWindow model rejects an ended_at that is not strictly after started_at, raising a validation error rather than admitting an open-ended interval that would displace the entire remaining log. Bound the window at the next activation or the break boundary upstream, then re-run — the deterministic key means the recovered window reroutes to the same signals it always would have.
• Duplicate delivery of the same window. Under at-least-once queue semantics, the same window can arrive twice. Because every signal’s entry_id is a deterministic hash, the second pass publishes identical messages that the make-good router deduplicates and discards; no distributed lock is needed, and the downstream ledger entry stays single even though the signal was seen twice.

FAQ #

Related #

• Handling EAS Emergency Interrupts in Ad Scheduling — the parent guide that defines the entities, state machine, and window-detection this rerouting step consumes.
• Automating Make-Good Routing for Preemptions — the recovery subsystem that receives each eligible displacement signal and selects compensatory inventory.
• Standardizing Billing Codes Across Traffic Systems — the normalization that keeps a recovered airing billing against the same account as the spot it replaced.