Implementing fallback routing for offline tower inspections

A technician finishes a structural check at a rural mast, the mobile controller has full data to file, and the carrier backhaul is simply not there — an RF dead zone, a municipal portal outage, or a satellite link that degraded to uselessness under cloud cover. The record legally happened, but if it lives only in RAM on a battery-constrained device, a reboot erases an inspection that an auditor will later read as one that never occurred. This page is the field-device implementation of the Fallback Routing Protocols transport layer: a concrete Python offline router that validates each inspection payload against the canonical lease taxonomy, assigns it a priority, seals it with a tamper-evident hash, and persists it to a durable local queue so it drains — provably unchanged — the moment a network path returns.

Prerequisites & Context #

You need Python 3.10 or later (the code uses dataclass ordering and structural typing), a hardened field tablet with a writable non-volatile volume for the local queue, and a cached copy of the lease registry the device will validate against. That registry is not arbitrary: it is the same canonical asset map produced under the parent Telecom Tower Compliance Architecture & Data Mapping reference model, so the field router rejects exactly what the upstream ingestion service would reject. Before wiring this router in, make sure three neighbouring subsystems are understood, because the offline path enforces their rules at the edge rather than deferring them to sync:

• Field identifiers and lease codes must resolve against Lease Taxonomy Standardization — an inspection whose asset_id has no canonical lease mapping cannot be filed and must fail fast, not queue.
• Setback and height thresholds come from the Zoning Rule Engine Design; the offline router runs a trimmed subset of those checks against cached municipal layers to flag a violation while the technician is still on site.
• Payload sealing and device-identity binding follow Security Boundary Configuration, so nothing enters the local queue without a hash anchored to the capturing device.

If you are mapping raw contract terms into the registry this router reads, the companion guide on How to map FCC tower lease terms to JSON schemas covers the schema construction that produces lease_expiry and the other fields validated below.

Step-by-Step Implementation #

The router is small on purpose — the value is in the ordering of its guarantees, not in line count. Build it in five steps, each anchored to a specific compliance failure it prevents.

Step 1 — Define the schema gate and the error taxonomy. Declare the required fields (asset_id, inspection_type, gps_coords, zoning_code, lease_expiry) as a set, and an ErrorCategory enum that distinguishes a schema violation from a lease expiry, a zoning-critical flag, a crypto failure, and a transient network fault. Categorisation is what lets sync-time logic pick automated retry versus manual escalation instead of treating every failure as fatal.

Step 2 — Reject anything that cannot map to a lease. In validate_schema, subtract the payload keys from the required set; a non-empty difference is a SCHEMA_VIOLATION. Then confirm asset_id exists in the cached registry — an unregistered asset has no lease term to map to, so it fails taxonomy validation at the source rather than queuing a record the destination will bounce. Only when the record is well-formed do you compare lease_expiry against the wall clock to detect an expired lease.

Step 3 — Assign priority deterministically. Start every payload at STANDARD. An expired or near-expiry lease escalates to HIGH because it carries a renewal deadline; a zoning check that trips a cached threshold (a residential R- code over its height cap, for example) escalates to CRITICAL. Priority orders the drain sequence only — it never lets a record skip validation or the audit seal.

Step 4 — Seal the payload before it touches storage. Serialise the record canonically with json.dumps(..., sort_keys=True, separators=(",", ":")), prefix it with the device identity, and hash it with hashlib.sha256. Writing this digest into the record and the audit log is what makes a deferred filing admissible: at sync the receiver regenerates the hash and proves the queued record is byte-for-byte the one captured in the field.

Step 5 — Enqueue, then drain on reconnect. Push sealed payloads into a PriorityQueue so CRITICAL records leave first. On reconnect, drain_queue pops in priority order and yields plain dicts ready for transmission — enum fields are stored as strings so no serialisation conversion is needed at the worst possible moment.

Figure: offline payload validation and priority queuing before deferred sync.

Complete Runnable Example #

The module below is self-contained. Drop in a cached lease_registry, feed it inspection dicts, and run it directly — the __main__ block exercises a compliant capture, an expired lease, and a rejected payload against realistic site IDs (TWR-8842) and zoning codes (MUN-4A-RES style R- codes).

import hashlib
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from enum import Enum
from queue import PriorityQueue
from typing import Any, Dict, List, Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("telecom.fallback.offline_router")


class OfflineRoutingError(Exception):
    """Raised when a payload cannot be admitted to the offline queue."""

    def __init__(self, message: str, category: "ErrorCategory"):
        super().__init__(message)
        self.category = category


class ErrorCategory(Enum):
    SCHEMA_VIOLATION = "SCHEMA_VIOLATION"
    LEASE_EXPIRY = "LEASE_EXPIRY"
    ZONING_CRITICAL = "ZONING_CRITICAL"
    CRYPTO_FAILURE = "CRYPTO_FAILURE"
    TRANSIENT_NETWORK = "TRANSIENT_NETWORK"


class RoutingPriority(Enum):
    CRITICAL = 1
    HIGH = 2
    STANDARD = 3


@dataclass(order=True)
class InspectionPayload:
    priority: int
    payload: Dict[str, Any] = field(compare=False)
    audit_hash: str = field(compare=False)
    timestamp: float = field(default_factory=time.time, compare=False)
    error_category: Optional[str] = field(default=None, compare=False)  # string for JSON-safety
    retry_count: int = field(default=0, compare=False)


class OfflineFallbackRouter:
    REQUIRED_FIELDS = {"asset_id", "inspection_type", "gps_coords", "zoning_code", "lease_expiry"}

    def __init__(self, device_id: str, lease_registry: Dict[str, Any]):
        self.device_id = device_id
        self.lease_registry = lease_registry
        self.queue: PriorityQueue = PriorityQueue()

    def compute_audit_hash(self, payload: Dict[str, Any]) -> str:
        """Deterministic SHA-256 over the canonical body, bound to device identity."""
        try:
            canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
            return hashlib.sha256(f"{self.device_id}:{canonical}".encode("utf-8")).hexdigest()
        except (TypeError, ValueError) as exc:
            raise OfflineRoutingError(f"hash generation failed: {exc}", ErrorCategory.CRYPTO_FAILURE)

    def validate_schema(self, data: Dict[str, Any]) -> Optional[ErrorCategory]:
        """Enforce required fields and canonical lease mapping before queuing."""
        missing = self.REQUIRED_FIELDS - data.keys()
        if missing:
            logger.warning("REJECT | missing fields %s", sorted(missing))
            return ErrorCategory.SCHEMA_VIOLATION

        asset_id = data["asset_id"]
        if asset_id not in self.lease_registry:
            logger.warning("REJECT | %s has no canonical lease mapping", asset_id)
            return ErrorCategory.SCHEMA_VIOLATION

        expiry = self.lease_registry[asset_id].get("lease_expiry")
        if expiry and time.time() > expiry:
            return ErrorCategory.LEASE_EXPIRY
        return None

    def evaluate_zoning_risk(self, data: Dict[str, Any]) -> bool:
        """Cached subset of the zoning rules: residential codes cap height at 45 m."""
        zoning_code = data.get("zoning_code", "")
        height_m = data.get("structure_height_m", 0)
        return zoning_code.startswith("R-") and height_m > 45

    def route_payload(self, inspection_data: Dict[str, Any]) -> InspectionPayload:
        """Validate, prioritise, seal, and enqueue one inspection record."""
        error = self.validate_schema(inspection_data)
        if error == ErrorCategory.SCHEMA_VIOLATION:
            raise OfflineRoutingError("payload rejected: schema validation failed", error)

        priority = RoutingPriority.STANDARD.value
        error_str: Optional[str] = None
        if error == ErrorCategory.LEASE_EXPIRY:
            priority = RoutingPriority.HIGH.value
            error_str = ErrorCategory.LEASE_EXPIRY.value
            logger.warning("LEASE_EXPIRY | %s flagged for compliance review", inspection_data["asset_id"])

        if self.evaluate_zoning_risk(inspection_data):
            priority = RoutingPriority.CRITICAL.value
            error_str = ErrorCategory.ZONING_CRITICAL.value
            logger.critical("ZONING_CRITICAL | %s routed to critical queue", inspection_data["asset_id"])

        audit_hash = self.compute_audit_hash(inspection_data)
        record = InspectionPayload(priority=priority, payload=inspection_data,
                                   audit_hash=audit_hash, error_category=error_str)
        self.queue.put(record)
        logger.info("QUEUED | %s | priority=%d | hash=%s", inspection_data["asset_id"], priority, audit_hash[:12])
        return record

    def drain_queue(self) -> List[Dict[str, Any]]:
        """Pop records in priority order for upstream sync; enums already stringified."""
        synced = []
        while not self.queue.empty():
            item = self.queue.get()
            synced.append(asdict(item))
            self.queue.task_done()
        return synced


if __name__ == "__main__":
    registry = {
        "TWR-8842": {"lease_expiry": time.time() + 86_400 * 90},   # 90 days out
        "TWR-7310": {"lease_expiry": time.time() - 86_400 * 5},    # expired 5 days ago
    }
    router = OfflineFallbackRouter(device_id="FIELD-TAB-014", lease_registry=registry)

    router.route_payload({"asset_id": "TWR-8842", "inspection_type": "structural",
                          "gps_coords": [34.05, -118.24], "zoning_code": "R-1",
                          "lease_expiry": registry["TWR-8842"]["lease_expiry"],
                          "structure_height_m": 52})            # residential over 45 m -> CRITICAL
    router.route_payload({"asset_id": "TWR-7310", "inspection_type": "corrosion",
                          "gps_coords": [40.71, -74.00], "zoning_code": "C-2",
                          "lease_expiry": registry["TWR-7310"]["lease_expiry"]})  # expired -> HIGH
    try:
        router.route_payload({"asset_id": "TWR-9999", "inspection_type": "rf_sweep"})
    except OfflineRoutingError as exc:
        logger.error("dropped at source | %s | %s", exc.category.value, exc)

    for rec in router.drain_queue():
        print(f"sync -> {rec['payload']['asset_id']} | priority={rec['priority']} "
              f"| flag={rec['error_category']} | {rec['audit_hash'][:12]}")

Verification & Expected Output #

Run python offline_router.py. The two valid captures queue and the third is refused at the source; the drain then emits records in priority order — CRITICAL (1) before HIGH (2):

2026-07-03 09:14:01 | CRITICAL | telecom.fallback.offline_router | ZONING_CRITICAL | TWR-8842 routed to critical queue
2026-07-03 09:14:01 | INFO     | telecom.fallback.offline_router | QUEUED | TWR-8842 | priority=1 | hash=1f90a3c7be42
2026-07-03 09:14:01 | WARNING  | telecom.fallback.offline_router | LEASE_EXPIRY | TWR-7310 flagged for compliance review
2026-07-03 09:14:01 | INFO     | telecom.fallback.offline_router | QUEUED | TWR-7310 | priority=2 | hash=7c2d5e08a1bb
2026-07-03 09:14:01 | WARNING  | telecom.fallback.offline_router | REJECT | missing fields ['gps_coords', 'lease_expiry', 'zoning_code']
2026-07-03 09:14:01 | ERROR    | telecom.fallback.offline_router | dropped at source | SCHEMA_VIOLATION | payload rejected: schema validation failed
sync -> TWR-8842 | priority=1 | flag=ZONING_CRITICAL | 1f90a3c7be42
sync -> TWR-7310 | priority=2 | flag=LEASE_EXPIRY | 7c2d5e08a1bb

Two signatures tell you the router is healthy. A REJECT line with no following QUEUED for the same asset means the schema gate fired correctly and nothing was admitted — the expected outcome for malformed input. The failure to watch for is a QUEUED line whose asset_id never appears in the drain output: that means the record entered the queue but was lost before sync, which is the exact data-loss the local persistence is meant to prevent. In production, back the queue with an fsync’d volume and re-read it on restart so a QUEUED record survives a power cycle rather than living only in the in-memory PriorityQueue shown here.

Gotchas & Edge Cases #

A stale cached registry silently rejects valid assets. The router only knows the lease mappings it cached before going offline. If a new site was commissioned or an asset_id re-keyed after the last sync, a perfectly legitimate inspection fails as a SCHEMA_VIOLATION at the source. Timestamp the cached registry and surface its age to the technician; a registry older than the lease-change cadence should trigger a re-sync prompt before fieldwork, not a wave of false rejects.

Wall-clock skew corrupts the expiry decision. Lease-expiry priority compares time.time() on the tablet against a stored epoch, and field devices that have been offline and power-cycled drift. A clock running fast will flag a still-valid lease as expired (harmless, just noisy HIGH traffic); a clock running slow will miss a genuinely expired lease and route it as STANDARD, defeating the deadline the escalation exists to catch. Discipline the device clock from GPS time when a fix is available rather than trusting the system RTC.

Priority alone does not de-duplicate. A controller that persisted a record locally and then also delivered it on a brief connectivity blip will present the same inspection twice. Because the audit hash is deterministic over the canonical body, downstream de-duplication is a hash equality check — but only if the field payload is never mutated after sealing. Any post-seal edit (a re-encoded photo path, a re-ordered coordinate pair) changes the digest and defeats de-duplication, so treat a sealed payload as immutable.

FAQ #

Related #

• Up to the parent transport layer: Fallback Routing Protocols
• Sibling how-to: How to map FCC tower lease terms to JSON schemas
• Up to the parent architecture: Telecom Tower Compliance Architecture & Data Mapping