Syncing FCC ASR filings with an internal compliance ledger

An FCC Antenna Structure Registration record and an internal compliance ledger describe the same physical tower, but they are almost never the same document. The ASR record lives in a federal system with its own field names, its own casing conventions, and its own update cadence; the ledger is the thing your automation actually queries when a lease renewal, a zoning check, or an inspection needs to know a structure’s registered height and owner of record. This page is the concrete synchronization job for that gap, and it belongs to the parent Regulatory Filing Synchronization subsystem: fetch a filing keyed by its fcc_asr_number, normalize it to the canonical schema, upsert it into the ledger without ever double-writing an unchanged record, and seal the result of that sync with a tamper-evident hash so an auditor can later prove exactly what was pulled and when. The worked example below tracks a single real-shaped filing — ASR 1290123, mapped to internal site TWR-8842 — end to end.

Prerequisites & Context #

You need Python 3.9 or later; nothing here depends on third-party packages, because the fetch step is deliberately mocked rather than wired to a live FCC endpoint — production code swaps that one function for an authenticated pull against the licensed ASR data source and keeps everything downstream unchanged. You should already have the canonical schema in front of you: fcc_asr_number (string, e.g. 1290123), site_id (string, e.g. TWR-8842), tower_height_ft (float), and filed_owner (string), all defined in the Regulatory Source Crosswalk, which is also where you’ll find the format rules an ASR number has to satisfy before this sync should trust it at all.

Two things are worth confirming before you wire this job into a schedule. First, that your ledger’s primary key really is fcc_asr_number and not an internal surrogate — keying on the federal identifier is what makes the upsert idempotent, because refetching the same filing twice always resolves to the same row instead of creating a duplicate. Second, that you have a plan for the case this page does not solve: a filing that syncs cleanly but no longer matches what is standing on the ground. That reconciliation problem is the subject of the sibling guide on Detecting drift between filed and as-built tower records, and it assumes the ledger this page builds is already populated and current. Both pages sit under the same parent architecture, Telecom Tower Compliance Architecture & Data Mapping, which defines the broader ingestion-to-ledger data flow this sync job is one instance of.

Step-by-Step Implementation #

Build the sync in five steps. Each one closes a specific way a compliance ledger silently goes stale or gets corrupted.

Step 1 — Fetch and parse the filing, keyed by fcc_asr_number. A real integration calls the FCC’s ASR data feed and gets back a payload keyed by the registration number; the mock source here reproduces that shape for one filing, 1290123, so the rest of the pipeline never has to know the fetch is simulated. fetch_mock_asr_filing takes the ASR number as its only argument and raises immediately if the number is not found — a filing that does not exist upstream must never silently fall through as an empty record.

Step 2 — Normalize the raw payload to canonical fields. The federal payload uses its own field names (ASRNumber, StructureHeight, Registrant, SiteRef), and normalize_filing maps each one onto the ledger’s canonical vocabulary: fcc_asr_number, tower_height_ft, filed_owner, site_id. Height arrives as a string and is coerced to float here, once, so every consumer downstream can assume a numeric type without re-parsing.

Step 3 — Upsert into the ledger, keyed by fcc_asr_number, idempotently. ComplianceLedger.upsert looks up the existing row by the ASR number before writing anything. If a row already exists with an identical content hash, the sync is a no-op — logged as UNCHANGED, not rewritten — which is what makes it safe to run this job on a tight schedule without inflating the ledger’s change history every time nothing actually changed upstream.

Step 4 — Record last_synced and a content hash. Before writing, the normalized record is serialized canonically with json.dumps(..., sort_keys=True, separators=(",", ":")) and hashed with hashlib.sha256 to produce a content_hash. That hash is the comparison key Step 3 relies on, and last_synced is stamped with the wall-clock time of the write, not the time of the fetch, so the ledger always answers “when did we last confirm this” truthfully even if the fetch step retried.

Step 5 — Seal the sync record. A second hashlib.sha256 digest — the sync_seal — is computed over the ASR number, the content hash, and the sync timestamp together. This seal is what an audit actually checks: reproducing it from the three inputs proves the ledger row was written by this exact sync event and has not been hand-edited since, independent of whether the underlying filing content itself later changes.

Figure: an ASR filing moves from external fetch to normalized fields to an idempotent ledger upsert, ending in a sealed sync record.

Complete Runnable Example #

The module below is self-contained standard-library Python — no network call is made; fetch_mock_asr_filing stands in for the licensed FCC feed so the example runs anywhere. Run it directly to sync ASR 1290123 into a ledger keyed to TWR-8842 twice in a row and observe the second pass resolve as unchanged.

import hashlib
import json
import logging
import time
from typing import Any, Dict

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s")
logger = logging.getLogger("telecom.compliance.asr_sync")

MOCK_SOURCE = {"1290123": {"ASRNumber": "1290123", "StructureHeight": "312.5",
                           "Registrant": "Summit Tower Holdings LLC", "SiteRef": "TWR-8842"}}


class FilingSyncError(Exception):
    """Raised when a filing cannot be fetched, normalized, or upserted."""
    def __init__(self, message: str, fcc_asr_number: str):
        super().__init__(message)
        self.fcc_asr_number = fcc_asr_number


def fetch_mock_asr_filing(fcc_asr_number: str) -> Dict[str, Any]:
    """Stand-in for the FCC ASR feed; real callers replace only this call."""
    if fcc_asr_number not in MOCK_SOURCE:
        raise FilingSyncError(f"no filing found for {fcc_asr_number}", fcc_asr_number)
    return MOCK_SOURCE[fcc_asr_number]


def normalize_filing(raw: Dict[str, Any]) -> Dict[str, Any]:
    """Map federal field names onto the ledger's canonical schema."""
    try:
        return {"fcc_asr_number": raw["ASRNumber"], "site_id": raw["SiteRef"],
                "tower_height_ft": float(raw["StructureHeight"]), "filed_owner": raw["Registrant"]}
    except (KeyError, ValueError) as exc:
        raise FilingSyncError(f"normalization failed: {exc}", raw.get("ASRNumber", "unknown"))


def content_hash(record: Dict[str, Any]) -> str:
    canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class ComplianceLedger:
    def __init__(self):
        self._rows: Dict[str, Dict[str, Any]] = {}

    def upsert(self, record: Dict[str, Any]) -> Dict[str, Any]:
        key = record["fcc_asr_number"]
        existing = self._rows.get(key)
        record_hash = content_hash(record)
        if existing and existing["content_hash"] == record_hash:
            logger.info("UNCHANGED | %s | hash=%s", key, record_hash[:12])
            return existing
        sealed = dict(record, last_synced=time.time(), content_hash=record_hash)
        sealed["sync_seal"] = hashlib.sha256(
            f"{key}:{record_hash}:{sealed['last_synced']}".encode("utf-8")).hexdigest()
        self._rows[key] = sealed
        logger.info("UPSERTED | %s | seal=%s", key, sealed["sync_seal"][:12])
        return sealed


if __name__ == "__main__":
    ledger = ComplianceLedger()
    row = {}
    for _ in range(2):  # second pass proves the upsert is idempotent
        row = ledger.upsert(normalize_filing(fetch_mock_asr_filing("1290123")))
    print(f"ledger row -> {row['site_id']} | height_ft={row['tower_height_ft']} "
          f"| owner={row['filed_owner']} | seal={row['sync_seal'][:12]}")
    try:
        fetch_mock_asr_filing("9999999")
    except FilingSyncError as exc:
        logger.error("sync aborted | %s | %s", exc.fcc_asr_number, exc)

Verification & Expected Output #

Run python asr_sync.py. The first pass upserts the row; the second pass, on identical input, must resolve as UNCHANGED rather than writing again:

2026-07-17 09:02:11 | INFO     | telecom.compliance.asr_sync | UPSERTED | 1290123 | seal=4a1e9c02f6b8
2026-07-17 09:02:11 | INFO     | telecom.compliance.asr_sync | UNCHANGED | 1290123 | hash=8d3f21ab90cc
ledger row -> TWR-8842 | height_ft=312.5 | owner=Summit Tower Holdings LLC | seal=4a1e9c02f6b8
2026-07-17 09:02:11 | ERROR    | telecom.compliance.asr_sync | sync aborted | 9999999 | no filing found for 9999999

The signature to watch for is a second UPSERTED line where you expected UNCHANGED — that means the content hash changed between two fetches of what should be the identical filing, which almost always traces back to non-deterministic serialization (an unsorted dict, a float formatted two different ways) rather than an actual upstream change. The failure to watch for in the other direction is an UNCHANGED result when the filing genuinely did change upstream; if that happens, confirm the normalization step is mapping every field the ledger cares about, because a field silently dropped in normalize_filing never reaches the hash and can never trigger a real update.

Gotchas & Edge Cases #

Height precision drift breaks idempotency silently. StructureHeight arrives from the feed as a string, and float("312.50") and float("312.5") are equal in Python but not always equal after a round-trip through a different serializer upstream. If the federal feed varies its own string formatting between pulls, content_hash will register a change that is not a real change. Round height to a fixed number of decimal places during normalization rather than trusting the source string’s formatting to stay stable.

An ASR number that passes format checks may not exist yet. The Regulatory Source Crosswalk validates that fcc_asr_number looks like a real registration number, but a syntactically valid number can still be unfiled — pending, withdrawn, or simply not yet indexed by the feed. fetch_mock_asr_filing treats “not found” as a FilingSyncError, and production code should route that specific failure to a pending-filing queue rather than treating it the same as a malformed request, because the correct remediation is different: wait and retry, not fix the input.

Re-keying a tower changes the upsert target, not the tower. If a site is reassigned to a new ASR filing — a replacement structure, a corrected registration — the old fcc_asr_number key stops receiving updates and a new row appears under the new key. Left alone, the ledger now carries two rows for one physical site_id. Sync jobs should reconcile site_id uniqueness as a separate check after the upsert, not assume that one FCC ASR number per tower holds forever.

FAQ #

Related #

• Up to the parent transport layer: Regulatory Filing Synchronization
• Sibling how-to: Detecting drift between filed and as-built tower records
• Up to the parent architecture: Telecom Tower Compliance Architecture & Data Mapping