From feabebe9bb7ec5317048af6c94ac224e28502da2 Mon Sep 17 00:00:00 2001 From: Babissimo Date: Thu, 17 Sep 2026 07:28:13 +0100 Subject: [PATCH] Delete the modules, endpoints and aliases nothing calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Seven pieces of the backend had no callers left. Each was verified by enumerating references across every repo under the workspace, on all branches, rather than by reading the code that defines them: - services/background.py, a re-export shim whose docstring says it exists so main.py and tests keep importing the old path. No test imported it and services/tasks/__init__.py already exported the same sixteen names, so main.py now imports the package directly. - core/types.py, four TypedDicts with zero importers, justified in the vulture whitelist as "used as type annotations" when they are not. NodeState also shadowed an unrelated NodeState in routes/node_schemas.py. - scripts/smoke_parquet_archive.py, whose two behaviours are covered by test_parquet_writer.py and test_storage.py. - /api/test/anomalies, superseded by /api/radar/anomalies, which the dashboard actually calls. The test version dumped anomaly_log raw and unauthenticated where the radar one redacts, so this also closes a small disclosure. - the two _get_stale_tasks aliases, left behind when the duplicated rule was collapsed into core/task_registry. - /api/radar/load-file and pipeline.passive_radar's batch __main__, the two entry points to the offline .detection path. Neither is reachable from a runbook, justfile recipe, deploy script or CI step. load-file was also the codebase's only path-traversal check, security surface maintained for no traffic. Deleting those last two orphaned PassiveRadarPipeline.process_file, their only caller, so it goes with them and the batch .detection path leaves the tree entirely. The live geolocation path is _run_geolocation driven by the frame loop, which is untouched and well covered. generate_aircraft_json stays: its tar1090 batch schema is superseded by the flush task's, but test_pipeline.py still pins it. core/types.py's descriptions were worth keeping, but not verbatim: checked against the code, two of the four had drifted. AircraftPosition listed callsign, seen, squawk and rssi, where tcp_handler builds flight, last_seen_ms, recv_ms and world, and omitted last_seen_ms altogether — the field adsb_store orders writes on and feed_gc evicts on. NodeState gave status as "active" | "disconnected", but _handle_heartbeat assigns msg.get("status") straight from the wire, so it is not a closed set. connected_nodes and adsb_aircraft therefore carry corrected comments rather than copies, and the adsb one names its writer instead of restating a field list that had already drifted once. TaskHealth was already stated verbatim beside task_last_success and task_error_counts, and GeoAircraft describes a payload one readable function builds, so neither was carried across. Nothing branches on either description today, so this fixes no live fault; it stops the file's claims being re-inherited as fact. Co-Authored-By: Claude Opus 5 --- backend/core/state.py | 12 +++- backend/core/types.py | 73 ------------------------ backend/main.py | 8 +-- backend/pipeline/passive_radar.py | 59 ------------------- backend/pyproject.toml | 2 - backend/routes/admin.py | 7 +-- backend/routes/radar.py | 30 +--------- backend/routes/test.py | 22 +------ backend/scripts/smoke_parquet_archive.py | 39 ------------- backend/services/background.py | 28 --------- backend/services/tasks/__init__.py | 10 +++- backend/tests/test_admin_routes.py | 8 +-- backend/vulture_whitelist.py | 20 ------- 13 files changed, 29 insertions(+), 289 deletions(-) delete mode 100644 backend/core/types.py delete mode 100644 backend/scripts/smoke_parquet_archive.py delete mode 100644 backend/services/background.py diff --git a/backend/core/state.py b/backend/core/state.py index c8d3ce096..87f60fd6c 100644 --- a/backend/core/state.py +++ b/backend/core/state.py @@ -37,7 +37,13 @@ # ── Connected node state tracking ───────────────────────────────────────────── connected_nodes: dict[str, dict] = {} -# node_id → {config_hash, config, status, last_heartbeat, peer, is_synthetic, capabilities} +# node_id → {config_hash: str, config: dict, status: str, +# last_heartbeat: ISO-8601 str, peer: str, is_synthetic: bool, +# capabilities: dict, node_ref: str} +# node_ref is present on mirrored nodes only; see services/node_refs._mirrored_ref. +# status is set locally to "active" on ingest and "disconnected" when the TCP +# peer drops, but a heartbeat overwrites it with whatever the node reported, so +# it is not a closed set — do not branch on it as if it were. # Empirical FOV (see retina_analytics.empirical_coverage / manager / # association). off/shadow/active, ASSOC_CLAIM_MODE precedent — an @@ -396,6 +402,10 @@ def _adsb_for_seeding() -> dict[str, dict]: # ── ADS-B positions reported inside detection frames ────────────────────────── adsb_aircraft: dict[str, dict] = {} +# normalize_hex_key'd hex → one record, built in services/tcp_handler and +# extended by adsb_derived_fields; see services/feed_helpers.adsb_store for the +# staleness rule that orders writes. last_seen_ms is the node's clock and +# recv_ms is ours, so the two are not interchangeable. # ── Known-target claims registry (KNOWN_LANE_MODE) ──────────────────────────── # Written by services/known_claiming.py once per frame; the interface between diff --git a/backend/core/types.py b/backend/core/types.py deleted file mode 100644 index d812720cb..000000000 --- a/backend/core/types.py +++ /dev/null @@ -1,73 +0,0 @@ -"""Typed definitions for core data structures. - -These TypedDicts describe the shapes of dicts stored in state.py. They are -documentation-first — not enforced at runtime — so introducing them is -backwards-compatible with existing code that constructs plain dicts. -""" - -from __future__ import annotations - -from typing import TypedDict - - -class NodeState(TypedDict, total=False): - """Shape of entries in ``state.connected_nodes``.""" - - config_hash: str - config: dict - status: str # "active" | "disconnected" - last_heartbeat: str # ISO-8601 - peer: str - is_synthetic: bool - capabilities: dict - node_ref: str # mirrored nodes only; see services/node_refs._mirrored_ref - - -class AircraftPosition(TypedDict, total=False): - """Shape of entries in ``state.adsb_aircraft``.""" - - hex: str - callsign: str - lat: float - lon: float - alt_baro: int - gs: float # ground speed (knots) - track: float # heading (degrees) - baro_rate: int - squawk: str - category: str - seen: float # epoch of last update - rssi: float - node_id: str - - -class GeoAircraft(TypedDict, total=False): - """Shape of entries in the aircraft.json ``aircraft`` list.""" - - hex: str - flight: str - lat: float - lon: float - alt_baro: int - alt_geom: int - gs: float - track: float - baro_rate: int - squawk: str - category: str - seen: float - rssi: float - type: str # "adsb_icao" | "radar" | ... - node_id: str - multi_node: bool - anomaly: bool - anomaly_types: list[str] - trail: list[list[float]] - arc: list[dict] - - -class TaskHealth(TypedDict, total=False): - """Shape of entries exposed in ``/api/test/dashboard`` → ``task_health``.""" - - last_success: dict[str, float] # task_name → epoch - error_counts: dict[str, int] # task_name → cumulative diff --git a/backend/main.py b/backend/main.py index a24696216..480dc9da6 100644 --- a/backend/main.py +++ b/backend/main.py @@ -58,7 +58,10 @@ from routes.test import router as test_router from services import detection_mirror, publication from services.alerting import log_destination -from services.background import ( +from services.runtime_coverage import start as _start_coverage +from services.runtime_coverage import stop as _stop_coverage +from services.state_snapshot import SAVE_INTERVAL_S, restore_snapshot, save_snapshot +from services.tasks import ( adsb_truth_fetcher, aircraft_flush_task, analytics_refresh_task, @@ -77,9 +80,6 @@ track_flush_task, users_backup_task, ) -from services.runtime_coverage import start as _start_coverage -from services.runtime_coverage import stop as _stop_coverage -from services.state_snapshot import SAVE_INTERVAL_S, restore_snapshot, save_snapshot from services.tasks.executor import task_executor, unfinished_task_executors from services.tasks.solver import solver_workers_stopping from services.tcp_handler import handle_tcp_client diff --git a/backend/pipeline/passive_radar.py b/backend/pipeline/passive_radar.py index d95bba0ff..43db45fff 100644 --- a/backend/pipeline/passive_radar.py +++ b/backend/pipeline/passive_radar.py @@ -6,8 +6,6 @@ Pipeline: detection data → retina-tracker → retina-geolocator → tar1090 JSON """ -import glob -import json import logging import math import os @@ -819,28 +817,6 @@ def process_frame(self, frame: dict): self._last_prune_time = _now_mono self._prune_stale_tracks() - def process_file(self, filepath: str) -> list: - """Process an entire .detection file. Returns geolocated tracks.""" - with open(filepath) as f: - content = f.read().strip() - if not content.startswith("["): - content = "[" + content + "]" - frames = json.loads(content) - - # Feed all frames to tracker only (skip per-frame geolocation for batch) - for frame in frames: - ts = frame["timestamp"] - delays = frame.get("delay", []) - dopplers = frame.get("doppler", []) - snrs = frame.get("snr", []) - detections = [{"delay": d, "doppler": f, "snr": s} for d, f, s in zip(delays, dopplers, snrs)] - self.tracker.process_frame(detections, ts) - - # Run geolocation once after all frames are processed - self._run_geolocation() - - return list(self.geolocated_tracks.values()) - def generate_aircraft_json(self) -> dict: """Generate tar1090-compatible aircraft.json from geolocated tracks.""" now = time.time() @@ -903,38 +879,3 @@ def generate_receiver_json(self) -> dict: "lat": lat, "lon": lon, } - - -def process_detection_folder(folder: str, output_dir: str, node_config: dict = None): - """Process all .detection files in a folder and write tar1090 JSON to output_dir.""" - pipeline = PassiveRadarPipeline(node_config) - - detection_files = sorted(glob.glob(os.path.join(folder, "*.detection"))) - if not detection_files: - logger.info("No .detection files found in %s", folder) - return - - os.makedirs(output_dir, exist_ok=True) - - receiver = pipeline.generate_receiver_json() - with open(os.path.join(output_dir, "receiver.json"), "w") as f: - json.dump(receiver, f) - - for filepath in detection_files: - logger.info("Processing: %s", os.path.basename(filepath)) - pipeline.process_file(filepath) - - aircraft_data = pipeline.generate_aircraft_json() - with open(os.path.join(output_dir, "aircraft.json"), "w") as f: - json.dump(aircraft_data, f) - - logger.info("Output: %d geolocated targets", len(aircraft_data["aircraft"])) - return aircraft_data - - -if __name__ == "__main__": - import sys - - folder = sys.argv[1] if len(sys.argv) > 1 else "." - output = sys.argv[2] if len(sys.argv) > 2 else "./tar1090_data" - process_detection_folder(folder, output) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 2fa86a78e..f14c8b0fd 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -33,8 +33,6 @@ omit = [ "routes/auth.py", # Simulation/test-only endpoints — not production code paths: "routes/test.py", - # Pure TypedDict definitions — no runtime code to cover: - "core/types.py", ] [tool.coverage.report] diff --git a/backend/routes/admin.py b/backend/routes/admin.py index 1a9fefe26..1a76285e0 100644 --- a/backend/routes/admin.py +++ b/backend/routes/admin.py @@ -51,11 +51,6 @@ logger = logging.getLogger(__name__) -# Was a byte-identical copy of routes/test.py's; the rule now lives beside the -# interval table it reads. -_get_stale_tasks = get_stale_tasks - - def _mn_pos_history_size() -> int: """Size of the solver's per-hex smoothing buffer (soak observability).""" from services.tasks import solver as _solver @@ -769,7 +764,7 @@ async def system_metrics(_user=Depends(require_admin)): "ws_clients": len(state.ws_clients), "ws_send_timeouts": state.ws_send_timeouts, "ws_live_clients": len(state.ws_live_clients), - "stale_tasks": _get_stale_tasks(), + "stale_tasks": get_stale_tasks(), "process_rss_mb": round(rss_mb, 1), "load_avg": list(os.getloadavg()), "disk_total_gb": round(disk.total / (1024**3), 1), diff --git a/backend/routes/radar.py b/backend/routes/radar.py index f875fc2c7..7de980365 100644 --- a/backend/routes/radar.py +++ b/backend/routes/radar.py @@ -6,13 +6,12 @@ import time from datetime import datetime, timezone -from fastapi import APIRouter, Depends, Header, HTTPException, Request +from fastapi import APIRouter, Header, HTTPException, Request from fastapi.responses import Response from pydantic import BaseModel, Field from config.constants import RATE_BUCKETS_MAX_IPS from core import state -from core.users import require_admin from pipeline.passive_radar import PassiveRadarPipeline from routes.node_schemas import NodeRef from services import node_refs, node_registration @@ -51,10 +50,6 @@ class BulkDetectionRequest(BaseModel): nodes: list[BulkNodeEntry] = Field(..., max_length=500) -class LoadFileRequest(BaseModel): - path: str = Field(..., min_length=1) - - RADAR_API_KEY = os.getenv("RADAR_API_KEY", "") _RETINA_ENV = os.getenv("RETINA_ENV", "").lower() if not RADAR_API_KEY: @@ -65,9 +60,6 @@ class LoadFileRequest(BaseModel): logging.warning(_msg) _RATE_LIMIT = int(os.getenv("RADAR_RATE_LIMIT", "60")) _RATE_WINDOW = int(os.getenv("RADAR_RATE_WINDOW", "60")) -_ALLOWED_DETECTION_DIR = os.path.realpath( - os.path.join(os.path.dirname(os.path.dirname(__file__)), "coverage_data", "archive") -) # Module-level reference to default pipeline; set from main.py at startup _default_pipeline: PassiveRadarPipeline | None = None @@ -272,26 +264,6 @@ async def ingest_detections_bulk( return {"status": "ok", "nodes_registered": registered, "frames_queued": queued} -@router.post("/api/radar/load-file") -async def load_detection_file(body: LoadFileRequest, _admin=Depends(require_admin)): - filepath = os.path.realpath(body.path) - if not filepath.startswith(_ALLOWED_DETECTION_DIR + os.sep): - raise HTTPException(status_code=400, detail="Path must be inside the coverage archive directory") - if not os.path.isfile(filepath): - raise HTTPException(status_code=400, detail="File not found") - if not filepath.endswith(".detection"): - raise HTTPException(status_code=400, detail="Only .detection files accepted") - - tracks = _default_pipeline.process_file(filepath) - aircraft_data = _default_pipeline.generate_aircraft_json() - # Do NOT write aircraft.json here. The flush task is the file's single - # writer (a different schema was being raced in from this request - # thread); marking dirty makes the next 1 Hz flush pick the load up. - state.aircraft_dirty = True - - return {"status": "ok", "tracks": len(tracks), "aircraft": aircraft_data["aircraft"]} - - @router.get("/api/radar/status") async def radar_status(): # Unauthenticated, and the config block quotes the default pipeline's diff --git a/backend/routes/test.py b/backend/routes/test.py index 8fd34963a..827295ac0 100644 --- a/backend/routes/test.py +++ b/backend/routes/test.py @@ -57,11 +57,6 @@ def _verify_sim_key(x_api_key: str = Header(default="", alias="X-API-Key")): raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key") -# Was a byte-identical copy of routes/admin.py's; the rule now lives beside the -# interval table it reads. -_get_stale_tasks = get_stale_tasks - - # Module-level reference set from main.py at startup _default_pipeline = None @@ -300,7 +295,7 @@ def _build_dashboard_data() -> bytes: "task_health": { "last_success": dict(state.task_last_success), "error_counts": state.task_error_snapshot(), - "stale_tasks": _get_stale_tasks(), + "stale_tasks": get_stale_tasks(), }, } ) @@ -471,21 +466,6 @@ async def get_ground_truth_trail(hex_code: str): } -@router.get("/api/test/anomalies") -async def get_anomaly_log(): - """Return the anomaly event log and currently flagged hex codes.""" - return Response( - content=orjson.dumps( - { - "flagged_count": len(state.anomaly_hexes), - "flagged_hexes": sorted(state.anomaly_hexes), - "events": state.anomaly_log[-100:], - } - ), - media_type="application/json", - ) - - # ── Known-track hold (path H) ───────────────────────────────────────────────── diff --git a/backend/scripts/smoke_parquet_archive.py b/backend/scripts/smoke_parquet_archive.py deleted file mode 100644 index b2d8399a8..000000000 --- a/backend/scripts/smoke_parquet_archive.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Smoke test: drive a frame through archive_detections and read it back.""" - -import shutil -from pathlib import Path - -import services.storage as st -from services.storage import archive_detections, read_archived_file - - -def main(): - base = Path("/tmp/parquet_smoke") - if base.exists(): - shutil.rmtree(base) - base.mkdir() - - st._LOCAL_ARCHIVE_DIR = str(base) - - frame = { - "timestamp": 1700000000000, - "delay": [12.34, 56.78], - "doppler": [-100.5, 33.3], - "snr": [15.0, 22.0], - "adsb": [ - {"hex": "abcdef", "lat": 40.71, "lon": -74.0, "alt_baro": 35000, "gs": 480, "track": 270, "flight": "UAL1"}, - None, - ], - "_signing_mode": "unknown", - "_signature_valid": False, - } - key = archive_detections("smoke-node", [frame]) - print("wrote:", key) - decoded = read_archived_file(key) - print("decoded count:", decoded["count"]) - print("first detection delay:", decoded["detections"][0]["delay"]) - print("adsb match:", decoded["detections"][0]["adsb"][0]) - - -if __name__ == "__main__": - main() diff --git a/backend/services/background.py b/backend/services/background.py deleted file mode 100644 index d205d7255..000000000 --- a/backend/services/background.py +++ /dev/null @@ -1,28 +0,0 @@ -"""Background async tasks — thin re-export shim. - -The actual implementations live in ``services.tasks.*`` sub-modules. -This file keeps the original import surface so ``main.py`` and tests -continue to work without changes:: - - from services.background import frame_processor_loop, start_solver_workers -""" - -from services.tasks import ( # noqa: F401 - adsb_truth_fetcher, - aircraft_flush_task, - analytics_refresh_task, - archive_flush_task, - archive_lifecycle_task, - coverage_constraints_task, - feed_gc_task, - frame_processor_loop, - health_monitor_task, - heartbeat_task, - prune_synthetic_nodes, - reputation_evaluator, - start_solver_workers, - stop_solver_workers, - storage_refresh_task, - track_flush_task, - users_backup_task, -) diff --git a/backend/services/tasks/__init__.py b/backend/services/tasks/__init__.py index ded5d8639..61cf330a6 100644 --- a/backend/services/tasks/__init__.py +++ b/backend/services/tasks/__init__.py @@ -1,7 +1,11 @@ -"""Background task sub-modules split from services/background.py. +"""Background async tasks, one sub-module per task. -All public names previously exported by background.py are re-exported here -so ``from services.tasks import frame_processor_loop`` works. +The task entry points are re-exported here so schedulers can name them +without knowing which sub-module each lives in:: + + from services.tasks import frame_processor_loop, start_solver_workers + +Anything else a sub-module exposes is imported from that sub-module directly. """ from services.tasks.aircraft_flush import aircraft_flush_task diff --git a/backend/tests/test_admin_routes.py b/backend/tests/test_admin_routes.py index ca1d6cfbb..f93ec55e5 100644 --- a/backend/tests/test_admin_routes.py +++ b/backend/tests/test_admin_routes.py @@ -463,17 +463,17 @@ def test_check_detects_offline(self): class TestStaleTasks: def test_no_stale_when_recent(self): - from routes.admin import _get_stale_tasks + from core.task_registry import get_stale_tasks state.task_last_success["frame_processor"] = time.time() - result = _get_stale_tasks() + result = get_stale_tasks() assert "frame_processor" not in result def test_stale_when_old(self): - from routes.admin import _get_stale_tasks + from core.task_registry import get_stale_tasks state.task_last_success["frame_processor"] = time.time() - 9999 - result = _get_stale_tasks() + result = get_stale_tasks() assert "frame_processor" in result diff --git a/backend/vulture_whitelist.py b/backend/vulture_whitelist.py index fe7aa82bf..5ed2f006e 100644 --- a/backend/vulture_whitelist.py +++ b/backend/vulture_whitelist.py @@ -20,26 +20,6 @@ # file's own preamble. ANALYTICS_REFRESH_INTERVAL_S is now wired; # ASSOC_MIN_INTERVAL_S is passed by core/state.) -# ── core/types.py ───────────────────────────────────────────────────────────── - -# TypedDict definitions — used as type annotations; fields are accessed via -# dict keys at runtime, not attribute access, so vulture misses the usage. -NodeState -_.capabilities -AircraftPosition -_.hex -_.baro_rate -_.squawk -_.rssi -GeoAircraft -_.flight -_.alt_geom -_.multi_node -_.anomaly -TaskHealth -_.last_success -_.error_counts - # ── pipeline/passive_radar.py ───────────────────────────────────────────────── # EventWriter public API — tested in tests/test_pipeline.py