diff --git a/retina_tracker/cli.py b/retina_tracker/cli.py index 3b54208..8680e1b 100644 --- a/retina_tracker/cli.py +++ b/retina_tracker/cli.py @@ -205,6 +205,30 @@ def main(): parser.add_argument("--tcp", action="store_true", help="Run as TCP server for streaming input from blah2") parser.add_argument("--tcp-host", default="0.0.0.0", help="TCP bind address (default: 0.0.0.0)") parser.add_argument("--tcp-port", type=int, default=3012, help="TCP port to listen on (default: 3012)") + parser.add_argument( + "--control-host", + default="127.0.0.1", + help="HTTP control surface bind address (default: 127.0.0.1). " + "Loopback by default because the container runs with " + "network_mode host, where 0.0.0.0 would publish it on the LAN.", + ) + parser.add_argument( + "--control-port", type=int, default=30101, help="HTTP control surface port (default: 30101). 0 disables it." + ) + parser.add_argument( + "--history-window", + type=int, + default=4 * 3600, + help="Seconds of detections and tracks kept in memory for GET /events (default: 14400, four hours).", + ) + parser.add_argument( + "--history-max-points", + type=int, + default=500_000, + help="Hard ceiling on points held per detection class, so the " + "footprint stays predictable whatever the detection rate " + "(default: 500000, about 10 MB per class).", + ) args = parser.parse_args() @@ -239,6 +263,10 @@ def main(): event_writer=event_writer, detection_window=args.detection_window, config=get_config(), + control_host=args.control_host, + control_port=args.control_port, + history_window_s=args.history_window, + history_max_points=args.history_max_points, ) else: tracker = process_detections(args.file, event_writer=event_writer, detection_window=args.detection_window) diff --git a/retina_tracker/control.py b/retina_tracker/control.py new file mode 100644 index 0000000..e66a1a9 --- /dev/null +++ b/retina_tracker/control.py @@ -0,0 +1,248 @@ +"""Loopback HTTP control surface for the streaming tracker. + +The tracker's only inbound channel has been the detection socket, which means +the one control operation it supports — clearing state between search +geometries — has had to travel as a `{"type": "RESET"}` message mixed into the +detection stream. That works only while the sender of detections and the sender +of controls are the same process. They are about to stop being: blah2_api +forwards detections directly, and the auto-calibration search lives in +retina-gui. `run_tcp_server` accepts one connection at a time, so a second +consumer cannot simply open its own. + +So control moves to its own door. Nothing here is specific to a caller: the +auto-calibration search and the Tracker page are equal consumers of a tracker +that does not know which is which. + +Bound to loopback by default for the same reason the ingest socket is (see the +sidecar's compose command): the container runs with network_mode host, so +0.0.0.0 would publish this on the LAN. + +Stdlib only, deliberately. This is a handful of routes on a link with one or +two clients; a framework would be a dependency and an image layer for nothing. +""" + +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 30101 + +# How often a stream looks for new points. The tracker appends about once a +# second, so this is the resolution of the feed rather than a poll of anything +# expensive: since() on an unchanged history is a few length comparisons. +STREAM_INTERVAL_S = 1.0 + +# A comment line keeps an idle connection open through anything that times out +# silent sockets, and is how a stream notices the client has gone: the write +# fails. +HEARTBEAT_S = 15.0 + +# Clamped rather than rejected. A window is a display preference, and must +# never let a query string ask for more than is held. +MIN_WINDOW_S = 60 + + +class _Handler(BaseHTTPRequestHandler): + """Routes are matched on the path with any trailing slash removed, so + /reset and /reset/ are the same endpoint.""" + + protocol_version = "HTTP/1.1" + + # The default handler logs every request to stderr. A health check on a + # short interval would bury the tracker's own output, which is the only + # thing anyone reads that stream for. The base class calls this + # positionally, so *args covers the format string too. + def log_message(self, *args): + pass + + def _send(self, status, payload): + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + if self._route() == "/history/clear": + # Wipes the record without touching the tracker. The page's + # "Clear buffer" has always meant "clear what I am being shown, + # keep tracking", and that distinction survives the record moving + # here from retina-gui. An active track repopulates on its own + # within a few events. + if self.server.history is None: + self._send(503, {"error": "history not enabled"}) + return + self.server.history.clear() + self._send(200, {"ok": True}) + return + if self._route() == "/reset": + # Held for the duration, so a 200 means the tracker is already + # clear rather than scheduled to be. A caller that resets between + # candidate geometries needs that: the next frame it waits on must + # not be able to associate into pre-reset state. + with self.server.tracker_lock: + self.server.tracker.reset() + self._send(200, {"ok": True}) + return + self._send(404, {"error": "not found"}) + + def do_GET(self): + route = self._route() + if route in ("/health", ""): + with self.server.tracker_lock: + payload = { + "ok": True, + "frames": self.server.tracker.frame_count, + "tracks": len(self.server.tracker.tracks), + } + if self.server.history is not None: + payload["history"] = self.server.history.stats() + self._send(200, payload) + return + if route == "/events": + self._stream() + return + self._send(404, {"error": "not found"}) + + # ── The data stream ──────────────────────────────────────── + + def _window(self): + raw = parse_qs(urlparse(self.path).query).get("window", [None])[0] + if raw is None: + return None + try: + seconds = int(raw) + except ValueError: + return None + if seconds <= 0: + return None + return max(MIN_WINDOW_S, min(seconds, self.server.history.window_s)) + + def _stream(self): + """Server-sent events: a snapshot, then only what has been appended. + + The connection is the session. Its cursor lives in this thread and + nowhere else, so there is no per-consumer state on the server to + expire, and no negotiation: a reconnect simply takes a fresh + snapshot, which is always a valid thing to start from. + + Sending the snapshot down this same stream rather than having the + consumer fetch it separately is what removes the race between what + the snapshot contained and where the delta stream began. There is one + ordering, and this generator owns it. + """ + if self.server.history is None: + self._send(503, {"error": "history not enabled"}) + return + + window_s = self._window() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("X-Accel-Buffering", "no") + # Chunked, not "read until the connection closes". No length is + # knowable and this never ends of its own accord, so a client told + # only "Connection: close" has to read to EOF to find a message + # boundary — which for urllib3, and therefore for retina-gui's + # requests-based proxy, means blocking until the read timeout rather + # than delivering each event as it arrives. Framing every message as + # its own chunk is what makes it stream to any client. + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Connection", "close") + self.end_headers() + + history = self.server.history + try: + payload, cursor = history.snapshot(window_s=window_s) + self._event("snapshot", payload) + + last_sent = time.monotonic() + while not self.server.stopping.is_set(): + time.sleep(STREAM_INTERVAL_S) + + delta, new_cursor = history.since(cursor) + if delta is None: + # clear() ran, so every outstanding cursor is void. + # Re-seed rather than trying to reconcile. + payload, cursor = history.snapshot(window_s=window_s) + self._event("snapshot", payload) + last_sent = time.monotonic() + continue + + cursor = new_cursor + if _has_points(delta): + self._event("delta", delta) + last_sent = time.monotonic() + elif time.monotonic() - last_sent >= HEARTBEAT_S: + self._chunk(b": keepalive\n\n") + last_sent = time.monotonic() + except (BrokenPipeError, ConnectionResetError, OSError): + pass # the consumer went away, which is how a stream ends + + def _event(self, kind, payload): + body = json.dumps(payload, separators=(",", ":")) + self._chunk(f"event: {kind}\ndata: {body}\n\n".encode()) + + def _chunk(self, data): + """One HTTP/1.1 chunk, so each message is its own frame on the wire.""" + self.wfile.write(b"%X\r\n" % len(data) + data + b"\r\n") + self.wfile.flush() + + def _route(self): + return self.path.split("?", 1)[0].rstrip("/") + + +class ControlServer(ThreadingHTTPServer): + """HTTP control surface over a running Tracker. + + `tracker_lock` is the same lock the frame path takes, so a reset can never + land in the middle of one. Frames arrive about once a second and take + milliseconds, so the contention this introduces is not measurable; the + alternative, deferring the reset to a flag the frame loop reads, would make + a 200 mean "queued" and would never apply at all on a node whose detections + have stopped. + """ + + daemon_threads = True + allow_reuse_address = True + + def __init__(self, tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT, history=None): + super().__init__((host, port), _Handler) + self.tracker = tracker + self.tracker_lock = tracker_lock + self.history = history + # Lets an open stream wind down on shutdown instead of holding the + # process up for a full interval. + self.stopping = threading.Event() + + def shutdown(self): + self.stopping.set() + super().shutdown() + + @property + def port(self): + """The bound port, which is what was asked for unless 0 was, in which + case it is whatever the OS chose.""" + return self.server_address[1] + + +def _has_points(delta): + if delta["tracks"]: + return True + return any(cols["t"] for cols in delta["detections"].values()) + + +def start_control_server(tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT, history=None): + """Serve the control surface on a daemon thread and return the server. + + The thread is a daemon so it never holds up interpreter shutdown: the + tracker process is killed by its supervisor, not asked to wind down.""" + server = ControlServer(tracker, tracker_lock, host=host, port=port, history=history) + thread = threading.Thread(target=server.serve_forever, daemon=True, name="tracker-control") + thread.start() + server.thread = thread + return server diff --git a/retina_tracker/history.py b/retina_tracker/history.py new file mode 100644 index 0000000..8834779 --- /dev/null +++ b/retina_tracker/history.py @@ -0,0 +1,397 @@ +"""The rolling record of what this node has seen, and the shape it is served in. + +The tracker's own Track objects keep a bounded per-track ring and drop +completed tracks after a short merge window, so nothing in it can answer "what +was seen three hours ago". This is that record: every detection the tracker was +given, classified, plus the points and metadata of every track, held for a +rolling window. + +In memory, deliberately. It does not need to survive a restart, and the +alternative was writing several hundred megabytes a day to an SD card that the +fleet cannot afford to wear out. + +Storage +------- +`array.array` columns rather than lists of tuples or dicts. A CPython tuple of +four floats costs about 192 bytes once the list slot, the tuple header and four +boxed floats are counted; the same point here is 20, so a buffer that would +have been 83 MB is 8.7. Columns are also what goes on the wire, so serving a +snapshot is a slice rather than a transposition. + +Bounded twice over. The time window is the point of the thing, but a window +alone makes memory a function of how busy the sky is, which is not something a +node can promise. `max_points` is a hard ceiling underneath it: whichever binds +first, wins. That is what makes the footprint predictable on hardware that has +no headroom to spare. + +Deltas +------ +Same contract as the page already speaks: a snapshot on connect and only what +has been appended since, per consumer. Columns are append-only between prunes, +so a consumer's position is a count. Because pruning drops from the *front* +that count is monotonic rather than an index — hence the `_dropped` counters, +which record how many points have fallen off ahead of the ones still held. +clear() bumps `_gen`, which voids every outstanding cursor at once and makes +consumers take a fresh snapshot rather than trying to reconcile. +""" + +import bisect +import threading +import time +from array import array + +WINDOW_S = 4 * 3600 + +# A hard ceiling on each class, independent of the window. 500,000 points is +# about 10 MB per class at 20 bytes each, and covers four hours at rates well +# above anything measured. A node that starts seeing far more loses the oldest +# points rather than its memory. +MAX_POINTS = 500_000 + +# Tracks are far fewer than detections and each is small, but a run that +# accumulates thousands still has to stop somewhere. +MAX_TRACKS = 2000 + +ASSOCIATED = "associated" +UNASSOCIATED = "unassociated" +BELOW_SNR = "below_snr" +CLASSES = (ASSOCIATED, UNASSOCIATED, BELOW_SNR) + +# Rounded on the way out, to the precision the measurement actually has: 2 dp +# of bistatic range is 10 m, 2 dp of Doppler is 0.01 Hz, 1 dp of SNR. Also +# undoes float32's repr — 16.1 stored as a float32 reads back as +# 16.100000381469727, which would be 18 bytes on the wire for 4 bytes of +# meaning. +DELAY_DP = 2 +DOPPLER_DP = 2 +SNR_DP = 1 + + +def _columns(): + """One point-set. 'q' is 8 bytes, 'f' is 4: 20 bytes a point.""" + return {"t": array("q"), "delay": array("f"), "doppler": array("f"), "snr": array("f")} + + +def _append(cols, timestamp, delay, doppler, snr): + cols["t"].append(int(timestamp)) + cols["delay"].append(delay) + cols["doppler"].append(doppler) + cols["snr"].append(snr) + + +def _drop_front(cols, n): + if n <= 0: + return 0 + for key in cols: + del cols[key][:n] + return n + + +def _slice(cols, start): + """The wire form of cols[start:], rounded.""" + return { + "t": list(cols["t"][start:]), + "delay": [round(v, DELAY_DP) for v in cols["delay"][start:]], + "doppler": [round(v, DOPPLER_DP) for v in cols["doppler"][start:]], + "snr": [round(v, SNR_DP) for v in cols["snr"][start:]], + } + + +def _first_at_or_after(times, cutoff_ms): + """Points are appended in timestamp order, so the cutoff is a bisect + rather than a scan. That is what makes a view window a saving.""" + return bisect.bisect_left(times, cutoff_ms) + + +class DetectionHistory: + """Everything the node has seen recently, and what became of it. + + Written from the tracker's frame thread and read from HTTP request + threads, so every method takes the lock. Reads are slices of contiguous + arrays, so they are short. + """ + + def __init__(self, window_s=WINDOW_S, max_points=MAX_POINTS, max_tracks=MAX_TRACKS): + self._lock = threading.Lock() + self.window_s = window_s + self.max_points = max_points + self.max_tracks = max_tracks + + # One point-set per class rather than one with a class column: the + # page asks for these separately, and keeping them apart makes a + # snapshot a slice instead of a filter over everything held. + self._det = {name: _columns() for name in CLASSES} + self._det_dropped = dict.fromkeys(CLASSES, 0) + + self._tracks = {} # id -> columns + self._track_meta = {} # id -> what the tracker thinks of it + self._track_dropped = {} # id -> points dropped from the front + self._track_last_ts = {} # id -> newest timestamp held + + self._gen = 0 + + # ── Writing ──────────────────────────────────────────────── + + def write_detections(self, timestamp, associated, unassociated, below_snr): + """The tracker's detection sink. One call per frame, in frame order, + with each detection's classification already final.""" + with self._lock: + for name, dets in ((ASSOCIATED, associated), (UNASSOCIATED, unassociated), (BELOW_SNR, below_snr)): + cols = self._det[name] + for det in dets: + _append(cols, timestamp, det["delay"], det["doppler"], det.get("snr", 0.0)) + self._enforce_ceiling(name) + + def write_event( + self, + track_id, + timestamp, + length, + detections, + adsb_hex=None, + is_anomalous=False, + anomaly_types=None, + max_velocity_ms=0.0, + shadow_fraction=0.0, + **_unused, + ): + """The tracker's event-writer duck type. + + Named keywords rather than **kwargs so it is visible which of the + sidecar's fields are kept and which are dropped on purpose: + adsb_initialized is the only deliberate omission. Anything added + later lands in **_unused rather than raising. + + Detections arrive already deduplicated by TrackEventWriter, but the + timestamp guard stays: this is also reachable from a consumer that + replays, and appending an older point would break the ordering + every read here relies on. + """ + with self._lock: + cols = self._tracks.get(track_id) + if cols is None: + if len(self._tracks) >= self.max_tracks: + self._evict_oldest_track() + cols = self._tracks[track_id] = _columns() + self._track_dropped[track_id] = 0 + + last = self._track_last_ts.get(track_id) + for det in detections: + ts = det["timestamp"] + if last is not None and ts <= last: + continue + _append(cols, ts, det["delay"], det["doppler"], det.get("snr", 0.0)) + last = ts if last is None or ts > last else last + if last is not None: + self._track_last_ts[track_id] = last + + self._track_meta[track_id] = { + "adsb_hex": adsb_hex, + "length": length, + "max_velocity_ms": round(max_velocity_ms or 0.0, 1), + "is_anomalous": bool(is_anomalous), + "anomaly_types": sorted(anomaly_types or []), + "shadow_fraction": round(shadow_fraction or 0.0, 3), + } + + def prune(self, now_ms): + """Drop everything older than the window. Runs on its own cadence, + independent of whether anyone is watching: the ceiling below is a + backstop, but this is what keeps the window a window.""" + cutoff = now_ms - self.window_s * 1000 + with self._lock: + for name in CLASSES: + cols = self._det[name] + dropped = _drop_front(cols, _first_at_or_after(cols["t"], cutoff)) + self._det_dropped[name] += dropped + + for track_id in list(self._tracks): + cols = self._tracks[track_id] + cut = _first_at_or_after(cols["t"], cutoff) + if cut >= len(cols["t"]): + self._forget_track(track_id) + elif cut: + self._track_dropped[track_id] += _drop_front(cols, cut) + + def clear(self): + """Wipe everything and void every outstanding cursor.""" + with self._lock: + self._det = {name: _columns() for name in CLASSES} + self._det_dropped = dict.fromkeys(CLASSES, 0) + self._tracks = {} + self._track_meta = {} + self._track_dropped = {} + self._track_last_ts = {} + self._gen += 1 + + # ── Bounds ───────────────────────────────────────────────── + + def _enforce_ceiling(self, name): + """Caller holds the lock. The window is the intent; this is the + promise about memory when the sky is busier than expected.""" + cols = self._det[name] + excess = len(cols["t"]) - self.max_points + if excess > 0: + self._det_dropped[name] += _drop_front(cols, excess) + + def _evict_oldest_track(self): + """Caller holds the lock. Whichever track has the oldest last point: + a live one is still being written to, so this reaches for a dead one + first without needing to be told which are dead.""" + oldest = min(self._track_last_ts, key=self._track_last_ts.get, default=None) + if oldest is None: + oldest = next(iter(self._tracks)) + self._forget_track(oldest) + + def _forget_track(self, track_id): + """Caller holds the lock.""" + self._tracks.pop(track_id, None) + self._track_meta.pop(track_id, None) + self._track_dropped.pop(track_id, None) + self._track_last_ts.pop(track_id, None) + + # ── Reading ──────────────────────────────────────────────── + + def _cursor(self): + """Caller holds the lock.""" + return { + "gen": self._gen, + "detections": {name: self._det_dropped[name] + len(self._det[name]["t"]) for name in CLASSES}, + "tracks": {tid: self._track_dropped.get(tid, 0) + len(cols["t"]) for tid, cols in self._tracks.items()}, + } + + def cursor(self): + with self._lock: + return self._cursor() + + def snapshot(self, window_s=None, now_ms=None): + """Everything a consumer needs to draw from cold, plus the cursor to + continue from. + + `window_s` narrows what is served without touching what is held. The + cursor is always the end of everything, not the end of the window, so + deltas continue from now however much history was asked for. + """ + if window_s is not None and now_ms is None: + now_ms = int(time.time() * 1000) + with self._lock: + cutoff = None if window_s is None else now_ms - window_s * 1000 + + detections = {} + for name in CLASSES: + cols = self._det[name] + start = 0 if cutoff is None else _first_at_or_after(cols["t"], cutoff) + detections[name] = _slice(cols, start) + + tracks = {} + for tid, cols in self._tracks.items(): + start = 0 if cutoff is None else _first_at_or_after(cols["t"], cutoff) + if start >= len(cols["t"]): + continue + tracks[tid] = dict(_slice(cols, start), meta=self._track_meta.get(tid, {})) + + payload = { + "gen": self._gen, + "window_s": window_s, + "detections": detections, + "tracks": tracks, + } + return payload, self._cursor() + + def since(self, cursor): + """Whatever has been appended since `cursor`. + + Returns (payload, new_cursor). payload is None when the cursor cannot + be honoured — a different generation, meaning clear() ran — and the + caller should send a fresh snapshot rather than try to reconcile. + + A track the cursor has never seen comes back whole, which is what + makes a newly promoted track arrive with the history the tracker + backfilled on promotion rather than truncated at the moment the + consumer happened to connect. + """ + if not cursor or "gen" not in cursor: + return None, None + with self._lock: + if cursor["gen"] != self._gen: + return None, None + + seen_det = cursor.get("detections") or {} + detections = {} + for name in CLASSES: + cols = self._det[name] + start = max(0, seen_det.get(name, 0) - self._det_dropped[name]) + detections[name] = _slice(cols, start) + + seen_tracks = cursor.get("tracks") or {} + tracks = {} + for tid, cols in self._tracks.items(): + start = max(0, seen_tracks.get(tid, 0) - self._track_dropped.get(tid, 0)) + if start >= len(cols["t"]): + continue + tracks[tid] = dict(_slice(cols, start), meta=self._track_meta.get(tid, {})) + + payload = {"gen": self._gen, "detections": detections, "tracks": tracks} + return payload, self._cursor() + + # ── Introspection ────────────────────────────────────────── + + def stats(self): + """Point counts and the memory they occupy, for /health.""" + with self._lock: + per_class = {name: len(self._det[name]["t"]) for name in CLASSES} + track_points = sum(len(cols["t"]) for cols in self._tracks.values()) + points = sum(per_class.values()) + track_points + return { + "detections": per_class, + "tracks": len(self._tracks), + "track_points": track_points, + "points": points, + # 20 bytes a point: an int64 and three float32s. + "approx_bytes": points * 20, + } + + +class TeeEventWriter: + """Fans track events out to several writers. + + The tracker takes one event_writer, and there are now two things that want + the events: the JSONL file, which live_score reads, and the in-memory + history the page is served from. Ordering matters — the file's writer is + what deduplicates detections, so it goes first and the history sees what + was actually written. + """ + + def __init__(self, *writers): + self._writers = [w for w in writers if w is not None] + + def write_event(self, *args, **kwargs): + for writer in self._writers: + writer.write_event(*args, **kwargs) + + def close(self): + for writer in self._writers: + if hasattr(writer, "close"): + writer.close() + + +def start_pruner(history, interval_s=60, stop_event=None): + """Drop what has aged out, on its own cadence. + + Independent of whether anyone is watching, and of whether frames are + arriving: a node that stops receiving should still let its window empty + rather than holding four hours of stale points indefinitely. + """ + + def loop(): + while True: + if stop_event is not None: + if stop_event.wait(interval_s): + return + else: + time.sleep(interval_s) + history.prune(int(time.time() * 1000)) + + thread = threading.Thread(target=loop, daemon=True, name="history-pruner") + thread.start() + return thread diff --git a/retina_tracker/output.py b/retina_tracker/output.py index d314643..d99f508 100644 --- a/retina_tracker/output.py +++ b/retina_tracker/output.py @@ -3,10 +3,17 @@ import json import os import sys +from collections import OrderedDict DEFAULT_MAX_BYTES = 64 * 1024 * 1024 DEFAULT_BACKUP_COUNT = 1 +# How many tracks to remember having written detections for. Only live tracks +# emit, so this is an LRU over "recently emitting" rather than over every track +# of the run. Evicting one costs a repeated window, not a lost detection, so a +# generous cap is cheap: a track id and an integer apiece. +EMITTED_MEMORY = 512 + class TrackEventWriter: """Writes track lifecycle events in JSONL (JSON Lines) format. @@ -26,6 +33,8 @@ def __init__(self, output_file, max_bytes=DEFAULT_MAX_BYTES, backup_count=DEFAUL self.max_bytes = max_bytes self.backup_count = backup_count self.bytes_written = 0 + # track_id -> newest detection timestamp already written for it. + self._emitted_through = OrderedDict() if output_file == "-": self.path = None @@ -37,6 +46,33 @@ def __init__(self, output_file, max_bytes=DEFAULT_MAX_BYTES, backup_count=DEFAUL self.output = open(output_file, "w") # noqa: SIM115 self._is_stdout = False + def _new_detections(self, track_id, detections): + """The detections of this event not already written for this track. + + Each event carries a rolling window of the track's most recent points + (Track.get_recent_detections), of which typically one is new. Repeating + the other nineteen every time multiplied this file by roughly twenty + for no consumer's benefit: live_score.load_tracks unions its + detections by timestamp, and so does retina-gui's buffer, so neither + can tell a delta stream from a repeating one. + + A high-water mark is sufficient because a track's history only grows + forwards for as long as it can emit. Merging is the one thing that + splices older points into a track, and it operates on all_tracks, + the post-mortem archive, after the track has been deleted from + self.tracks and can no longer produce an event. + """ + through = self._emitted_through.get(track_id) + if through is not None: + self._emitted_through.move_to_end(track_id) + detections = [d for d in detections if d["timestamp"] > through] + if detections: + self._emitted_through[track_id] = max(d["timestamp"] for d in detections) + self._emitted_through.move_to_end(track_id) + while len(self._emitted_through) > EMITTED_MEMORY: + self._emitted_through.popitem(last=False) + return detections + def write_event( self, track_id, @@ -50,6 +86,12 @@ def write_event( anomaly_types=None, shadow_fraction=0.0, ): + # Only what is new. The event is still written when nothing is — + # length, the anomaly flags and shadow_fraction all move over a + # track's life, and a consumer that missed those updates would be + # reading a stale opinion of a live track. + detections = self._new_detections(track_id, detections) + event = { "track_id": track_id, "adsb_hex": adsb_hex, diff --git a/retina_tracker/server.py b/retina_tracker/server.py index 4c897d2..45a80e3 100644 --- a/retina_tracker/server.py +++ b/retina_tracker/server.py @@ -1,10 +1,16 @@ """TCP server for receiving detection frames from blah2.""" import json +import select import socket import sys +import threading from .config import get_config +from .control import DEFAULT_HOST as CONTROL_HOST +from .control import DEFAULT_PORT as CONTROL_PORT +from .control import start_control_server +from .history import MAX_POINTS, WINDOW_S, DetectionHistory, TeeEventWriter, start_pruner from .tracker import Tracker @@ -35,7 +41,136 @@ def process_streaming_frame(tracker, frame): tracker.process_frame(detections, timestamp) -def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_window=20, config=None): +# A new peer must never queue behind a dead one, so the backlog is bigger than +# the single slot it used to be. +LISTEN_BACKLOG = 8 + +# Bounds how long the loop sits in select() with nothing happening, so a +# stop_event is noticed promptly. Nothing else depends on it. +SELECT_TIMEOUT_S = 0.5 + + +def _close(sock): + if sock is None: + return + try: + sock.close() + except OSError: + pass + + +def _handle_line(line, tracker, tracker_lock): + """One newline-delimited frame from the detection feed. + + Nothing a peer can send may take the feed down. A frame that is not JSON, + not an object, or missing the fields process_frame needs is logged and + dropped: the alternative is one malformed line ending detection ingest + until the container is restarted. + """ + line = line.strip() + if not line: + return + try: + frame = json.loads(line.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as e: + print(f"JSON parse error: {e}", file=sys.stderr) + return + if not isinstance(frame, dict): + print("Ignoring non-object frame", file=sys.stderr) + return + # A real detection frame never carries a "type" key, so this can never + # misfire on genuine data. + if frame.get("type") == "RESET": + # Kept alongside POST /reset while retina-gui is still the process + # feeding this socket. It goes when that does: a control message + # riding in the data stream only works while one process sends both, + # which is the arrangement being unwound. + with tracker_lock: + tracker.reset() + print("Tracker state reset", file=sys.stderr) + return + try: + with tracker_lock: + process_streaming_frame(tracker, frame) + except Exception as e: + print(f"Dropping unusable frame: {e!r}", file=sys.stderr) + + +def serve_detections(server, tracker, tracker_lock, stop_event=None): + """Read detection frames from `server`, newest connection winning. + + Exactly one peer feeds the tracker at a time, and it is whichever + connected most recently. Two things follow, both of which the old + accept-one-then-read-to-EOF loop got wrong. + + A peer that vanishes without closing — a killed container, a dropped + link — leaves a half-open socket that never returns from recv and never + reaches EOF, so the loop blocked there for as long as the kernel took to + notice while a new peer sat unserved in a backlog of one. Selecting on + the listener as well means a new connection is heard immediately, + whatever state the old one is in. That is the case this exists for: the + handover from retina-gui to blah2_api is exactly a new peer arriving + while the old one may still be holding the slot. + + And only one peer should ever be feeding. The tracker does not + deduplicate, so two senders of the same detections would hand it every + frame twice at dt=0, and every track twice the evidence it earned. + Replacing rather than multiplexing makes that unrepresentable. + """ + conn = None + buffer = b"" + + while stop_event is None or not stop_event.is_set(): + watching = [server] if conn is None else [server, conn] + try: + ready, _, _ = select.select(watching, [], [], SELECT_TIMEOUT_S) + except OSError: + break # listener closed underneath us: shutting down + + if server in ready: + try: + new_conn, addr = server.accept() + except OSError: + continue + if conn is not None: + print("Detection feed replaced by a newer connection", file=sys.stderr) + _close(conn) + # Whatever is half-read belongs to the peer being replaced. + buffer = b"" + conn = new_conn + print(f"Detections connected from {addr}", file=sys.stderr) + + if conn is not None and conn in ready: + try: + data = conn.recv(4096) + except OSError: + data = b"" + if not data: + print("Detection feed disconnected", file=sys.stderr) + _close(conn) + conn = None + buffer = b"" + continue + + buffer += data + while b"\n" in buffer: + line, buffer = buffer.split(b"\n", 1) + _handle_line(line, tracker, tracker_lock) + + _close(conn) + + +def run_tcp_server( + host="0.0.0.0", + port=3012, + event_writer=None, + detection_window=20, + config=None, + control_host=CONTROL_HOST, + control_port=CONTROL_PORT, + history_window_s=WINDOW_S, + history_max_points=MAX_POINTS, +): """Run tracker as TCP server receiving detection frames from blah2. Args: @@ -44,51 +179,39 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo event_writer: TrackEventWriter for streaming output detection_window: Number of detections in sliding window config: Configuration dict + control_host: Bind address for the HTTP control surface + control_port: Port for the HTTP control surface; 0 disables it + history_window_s: How much of the recent past to keep in memory + history_max_points: Hard ceiling per detection class, whatever the rate """ + # In memory rather than on disk: it does not need to survive a restart, + # and the alternative was several hundred megabytes a day onto an SD card. + history = DetectionHistory(window_s=history_window_s, max_points=history_max_points) + tracker = Tracker( - event_writer=event_writer, + # The file's writer deduplicates detections, so it goes first and the + # history records what was actually written. + event_writer=TeeEventWriter(event_writer, history), detection_window=detection_window, config=config or get_config(), + detection_sink=history, ) + start_pruner(history) + + # Guards every mutation of `tracker`. The frame path has always been + # single-threaded, so this is uncontended right up until the control + # surface below can reset from a request thread. + tracker_lock = threading.Lock() + + if control_port: + control = start_control_server(tracker, tracker_lock, host=control_host, port=control_port, history=history) + print(f"Tracker control on {control_host}:{control.port}", file=sys.stderr) server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) - server.listen(1) + server.listen(LISTEN_BACKLOG) print(f"Tracker listening on {host}:{port}", file=sys.stderr) - while True: - conn, addr = server.accept() - print(f"blah2 connected from {addr}", file=sys.stderr) - - buffer = "" - while True: - try: - data = conn.recv(4096).decode("utf-8") - if not data: - break - - buffer += data - - while "\n" in buffer: - line, buffer = buffer.split("\n", 1) - if line.strip(): - frame = json.loads(line) - # A real detection frame never carries a "type" key, - # so this can never misfire on genuine data. - if frame.get("type") == "RESET": - tracker.reset() - print("Tracker state reset", file=sys.stderr) - continue - process_streaming_frame(tracker, frame) - - except (ConnectionResetError, BrokenPipeError): - print("blah2 disconnected", file=sys.stderr) - break - except json.JSONDecodeError as e: - print(f"JSON parse error: {e}", file=sys.stderr) - continue - - conn.close() - print("Waiting for blah2 reconnection...", file=sys.stderr) + serve_detections(server, tracker, tracker_lock) diff --git a/retina_tracker/track.py b/retina_tracker/track.py index f26e537..283a50c 100644 --- a/retina_tracker/track.py +++ b/retina_tracker/track.py @@ -56,8 +56,34 @@ class Track: _daily_counter = 0 _last_date = None + @property + def state_status(self): + return self._state_status + + @state_status.setter + def state_status(self, value): + """Latches ever_confirmed on the way through ACTIVE. + + A track reaches ACTIVE by three routes: promotion on M-of-N, a + coasting track re-associating, and tracklet initiation. Latching at + the assignment catches all of them, and any fourth added later, + which matters because the flag is load-bearing for detection + classification and a missed route reads as "this detection was never + in a confirmed track". + + Its current status cannot stand in for the flag. A track that was + ACTIVE and has since coasted or been deleted reads the same as one + that never got there. + """ + self._state_status = value + if value is TrackState.ACTIVE: + self.ever_confirmed = True + def __init__(self, detection, timestamp, kf, frame=0, config=None): self.id = None + # Before state_status: its setter latches ever_confirmed. + self.ever_confirmed = False + self.retired = False self.state_status = TrackState.TENTATIVE self.kf = kf self.adsb_hex = None diff --git a/retina_tracker/tracker.py b/retina_tracker/tracker.py index 8795e24..ba04507 100644 --- a/retina_tracker/tracker.py +++ b/retina_tracker/tracker.py @@ -29,10 +29,27 @@ BACKWARDS_RUN_BEFORE_RESYNC = 3 +# How long a frame's detections may wait for their tracks to resolve before +# the answer is forced. Nearly every detection resolves far sooner: one joining +# an already-confirmed track is settled on arrival, and a new track either +# promotes within N_WINDOW frames or is deleted. This exists for the case that +# does neither, a tentative track that keeps associating but never clears +# promotion, which would otherwise hold up the whole queue behind it. +MAX_PENDING_CLASSIFICATION_FRAMES = 60 + + class Tracker: """Multi-target tracker using Kalman filtering and GNN data association.""" - def __init__(self, event_writer=None, detection_window=20, config=None, max_completed_tracks=MAX_COMPLETED_TRACKS): + def __init__( + self, + event_writer=None, + detection_window=20, + config=None, + max_completed_tracks=MAX_COMPLETED_TRACKS, + detection_sink=None, + max_pending_classification_frames=MAX_PENDING_CLASSIFICATION_FRAMES, + ): self.kf = KalmanFilter() self.tracks = [] self.all_tracks = [] @@ -42,6 +59,12 @@ def __init__(self, event_writer=None, detection_window=20, config=None, max_comp self._reset_counters() self.event_writer = event_writer self.config = config if config else get_config() + # Optional. Receives every detection this tracker was given, classified + # (see _drain_classifications). None costs nothing: the bookkeeping is + # skipped entirely rather than computed and dropped. + self.detection_sink = detection_sink + self._max_pending_frames = max_pending_classification_frames + self._pending_classification = deque() def reset(self): """Clear in-progress and completed-track state in place, as if @@ -58,6 +81,7 @@ def reset(self): self.all_tracks = [] self.completed_tracks.clear() self.last_timestamp = None + self._pending_classification.clear() self._reset_counters() def _reset_counters(self): @@ -112,7 +136,19 @@ def process_frame(self, detections, timestamp): else: dt = 0.5 - detections = [d for d in detections if d["snr"] >= MIN_SNR()] + # A partition, not a filter. Everything below the gate used to be + # dropped here without trace, which is why nothing downstream could + # tell "the tracker rejected what it saw" from "there was nothing to + # see". A single loop rather than two comprehensions so that every + # detection lands in exactly one bucket even when snr is NaN, which + # fails both comparisons. + min_snr = MIN_SNR() + below_snr = [] + kept = [] + for d in detections: + (kept if d["snr"] >= min_snr else below_snr).append(d) + detections = kept + self._mark_shadows(detections) for track in self.tracks: @@ -122,6 +158,11 @@ def process_frame(self, detections, timestamp): associated_tracks = set() associated_detections = set() + # detection index -> the Track it went into, for classification. + # Every surviving detection ends up in some track: either an existing + # one associates it, or it starts a tentative one below. The question + # a consumer actually has is whether that track was ever confirmed. + landed_in = {} if self.detection_sink is not None else None _lazy_write = hasattr(self.event_writer, "write_event_lazy") if self.event_writer else False @@ -133,6 +174,8 @@ def process_frame(self, detections, timestamp): track.state_status = TrackState.ACTIVE associated_tracks.add(track_idx) associated_detections.add(det_idx) + if landed_in is not None: + landed_in[det_idx] = track if track.id and self.event_writer: _det_n = min(track.n_associated, self.detection_window) @@ -206,11 +249,17 @@ def process_frame(self, detections, timestamp): if i not in associated_detections: new_track = Track(det, timestamp, self.kf, frame=self.frame_count, config=self.config) self.tracks.append(new_track) + if landed_in is not None: + landed_in[i] = new_track self._initiate_tracklets(timestamp) deleted_tracks = [t for t in self.tracks if t.should_delete()] for track in deleted_tracks: + # Latched here rather than inferred from absence later: this is + # what makes "never confirmed" a settled answer instead of a + # not-yet. + track.retired = True if track.state_status == TrackState.ACTIVE or track.n_associated >= M_THRESHOLD(): self.all_tracks.append(track) self.tracks = [t for t in self.tracks if not t.should_delete()] @@ -231,6 +280,11 @@ def process_frame(self, detections, timestamp): if len(self.all_tracks) > 1: self._merge_tracks() + # After deletions and promotions, so this frame's own tracks may + # already have settled. + if self.detection_sink is not None: + self._classify_frame(timestamp, landed_in, detections, below_snr) + resync = self.n_backwards >= BACKWARDS_RUN_BEFORE_RESYNC if self.last_timestamp is None or timestamp > self.last_timestamp or resync: if resync: @@ -238,6 +292,61 @@ def process_frame(self, detections, timestamp): self.n_backwards = 0 self.last_timestamp = timestamp + def _classify_frame(self, timestamp, landed_in, detections, below_snr): + """Queue one frame's detections for classification, then drain. + + Splitting queue from drain is what keeps the stream in frame order: + entries are only ever released from the front, so a consumer can + append what it receives and rely on it being ordered by timestamp. + """ + self._pending_classification.append( + { + "timestamp": timestamp, + "frame": self.frame_count, + "pairs": [(det, landed_in[i]) for i, det in enumerate(detections) if i in landed_in], + "below_snr": below_snr, + } + ) + self._drain_classifications() + + def _drain_classifications(self): + """Release every frame at the front whose verdict is settled. + + "Associated" means the detection ended up in a track that was + confirmed, which is not knowable when the detection arrives: a + detection that starts a tentative track is unassociated at that + moment and becomes associated a few frames later if the track + promotes. Classifying on arrival would therefore be wrong for + exactly the detections that matter most. + + So a frame waits until each of its tracks has settled — promoted + (ever_confirmed) or deleted without promoting (retired) — and is then + released with a final answer. A detection joining an already-confirmed + track settles immediately, so a steady-state feed is not delayed at + all; it is only the opening frames of a new track that wait. + + Stopping at the first unsettled frame rather than skipping past it is + deliberate. Out-of-order release would break the one assumption a + consumer buffering these wants to make. + """ + if self.detection_sink is None: + return + while self._pending_classification: + entry = self._pending_classification[0] + aged_out = (self.frame_count - entry["frame"]) >= self._max_pending_frames + if not aged_out and any(not (track.ever_confirmed or track.retired) for _det, track in entry["pairs"]): + break + + associated = [] + unassociated = [] + for det, track in entry["pairs"]: + # On an aged-out frame a still-tentative track reads as + # unassociated, which is the true answer so far. + (associated if track.ever_confirmed else unassociated).append(det) + + self._pending_classification.popleft() + self.detection_sink.write_detections(entry["timestamp"], associated, unassociated, entry["below_snr"]) + def _associate(self, detections): if not self.tracks or not detections: return [] diff --git a/tests/test_classification.py b/tests/test_classification.py new file mode 100644 index 0000000..654d8e7 --- /dev/null +++ b/tests/test_classification.py @@ -0,0 +1,287 @@ +"""Every detection the tracker is given comes back out, classified. + +The tracker consumed detections and only ever reported the ones that ended up +inside a confirmed track. Two exits were silent: anything below MIN_SNR was +dropped at the gate before the tracker looked at it, and anything that started +a tentative track which never promoted disappeared with it. Nothing +downstream could tell "the tracker rejected what it saw" from "there was +nothing to see". + +The subtlety these pin down is *when* a verdict is final. "Associated" means +the detection ended up in a track that was confirmed, and that is not knowable +at the moment the detection arrives. +""" + +import pytest + +from retina_tracker.config import get_config, set_config +from retina_tracker.server import process_streaming_frame +from retina_tracker.tracker import Tracker + + +class RecordingSink: + """Stands in for whatever consumes classified detections.""" + + def __init__(self): + self.calls = [] + + def write_detections(self, timestamp, associated, unassociated, below_snr): + self.calls.append( + { + "timestamp": timestamp, + "associated": associated, + "unassociated": unassociated, + "below_snr": below_snr, + } + ) + + # Convenience views over everything released so far. + def snrs(self, bucket): + return [d["snr"] for call in self.calls for d in call[bucket]] + + def count(self, bucket): + return sum(len(call[bucket]) for call in self.calls) + + @property + def timestamps(self): + return [call["timestamp"] for call in self.calls] + + +@pytest.fixture +def sink(): + return RecordingSink() + + +def make_tracker(sink, **kwargs): + return Tracker(config=get_config(), detection_sink=sink, **kwargs) + + +def frame(tracker, timestamp, points): + """points: list of (delay, doppler, snr).""" + process_streaming_frame( + tracker, + { + "timestamp": timestamp, + "delay": [p[0] for p in points], + "doppler": [p[1] for p in points], + "snr": [p[2] for p in points], + }, + ) + + +def steady_target(tracker, frames, delay=10.0, doppler=50.0, snr=15.0, start=1000, step=500): + for i in range(frames): + frame(tracker, start + i * step, [(delay + i * 0.05, doppler - i * 0.2, snr)]) + + +def drain(tracker, start_ts=900000, max_frames=120): + """Run empty frames until nothing is pending. + + A frame is held until its tracks settle, so a test that sends a handful + and asserts immediately is measuring the queue rather than the + classification. Empty frames let tentative tracks miss their way to + deletion, which settles them.""" + ts = start_ts + for _ in range(max_frames): + if not tracker._pending_classification: + return + ts += 500 + frame(tracker, ts, []) + + +# ── the SNR gate ──────────────────────────────────────────────────────────── + + +def test_detections_below_the_gate_are_reported_not_discarded(sink): + """The blind spot this exists to close.""" + min_snr = get_config()["tracker"]["min_snr"] + tracker = make_tracker(sink) + frame(tracker, 1000, [(10.0, 50.0, min_snr + 5), (20.0, -30.0, min_snr - 1)]) + drain(tracker) + + assert sink.count("below_snr") == 1 + assert sink.snrs("below_snr") == [min_snr - 1] + + +def test_a_detection_exactly_on_the_gate_is_kept(sink): + """`>=`, matching the filter this replaced.""" + min_snr = get_config()["tracker"]["min_snr"] + tracker = make_tracker(sink) + frame(tracker, 1000, [(10.0, 50.0, min_snr)]) + drain(tracker) + + assert sink.count("below_snr") == 0 + assert sink.count("associated") + sink.count("unassociated") == 1 + + +def test_a_nan_snr_lands_in_exactly_one_bucket(sink): + """NaN fails both comparisons, so two comprehensions would have dropped + it from the accounting entirely.""" + tracker = make_tracker(sink) + frame(tracker, 1000, [(10.0, 50.0, float("nan"))]) + drain(tracker) + + total = sink.count("associated") + sink.count("unassociated") + sink.count("below_snr") + assert total == 1 + + +def test_every_detection_is_accounted_for(sink): + min_snr = get_config()["tracker"]["min_snr"] + tracker = make_tracker(sink) + given = 0 + for i in range(12): + points = [ + (10.0 + i * 0.05, 50.0, min_snr + 8), + (200.0, -100.0, min_snr - 2), + (55.0 + i * 3.0, 20.0, min_snr + 1), + ] + given += len(points) + frame(tracker, 1000 + i * 500, points) + drain(tracker) + + seen = sink.count("associated") + sink.count("unassociated") + sink.count("below_snr") + assert seen == given, (seen, given) + + +# ── when a verdict becomes final ──────────────────────────────────────────── + + +def test_a_detection_that_ends_up_in_a_confirmed_track_reads_associated(sink): + tracker = make_tracker(sink) + steady_target(tracker, 12) + + assert any(t.ever_confirmed for t in tracker.tracks), "no track confirmed; test is vacuous" + assert sink.count("associated") > 0 + + +def test_a_detection_is_not_called_unassociated_before_its_track_can_promote(sink): + """The trap. A detection that starts a tentative track is unassociated at + that instant and becomes associated a few frames later if the track + promotes. Answering on arrival would be wrong for exactly the detections + that matter.""" + tracker = make_tracker(sink) + # One frame: the detection starts a tentative track that cannot possibly + # have promoted yet, so nothing may be released about it. + frame(tracker, 1000, [(10.0, 50.0, 15.0)]) + + assert sink.calls == [], "a verdict was published before it could be known" + + +def test_the_held_frame_is_released_once_its_track_confirms(sink): + tracker = make_tracker(sink) + steady_target(tracker, 12) + + assert sink.calls, "the frame was never released" + assert sink.timestamps[0] == 1000 + assert sink.count("associated") > 0 + + +def test_clutter_that_never_confirms_reads_unassociated(sink): + """Each detection is somewhere new, so every one starts a track that dies + without promoting.""" + tracker = make_tracker(sink) + for i in range(40): + frame(tracker, 1000 + i * 500, [(20.0 + i * 25.0, -200.0 + i * 9.0, 15.0)]) + drain(tracker) + + assert sink.count("unassociated") > 0 + assert sink.count("associated") == 0 + + +def test_a_detection_joining_an_established_track_is_not_delayed(sink): + """Once a track is confirmed its verdict is already settled, so a + steady feed is classified without lag.""" + tracker = make_tracker(sink) + steady_target(tracker, 12) + released_before = len(sink.calls) + + steady_target(tracker, 1, start=1000 + 12 * 500) + assert len(sink.calls) == released_before + 1 + + +# ── ordering and bounds ───────────────────────────────────────────────────── + + +def test_a_rejected_detection_waits_for_its_frame(sink): + """Deliberate. A below-gate detection's verdict is known the moment it + arrives, but releasing it ahead of the rest of its frame would mean + emitting the same frame twice and out of order. One call per frame, in + order, is worth a few frames of lag to the consumer that buffers these. + """ + min_snr = get_config()["tracker"]["min_snr"] + tracker = make_tracker(sink) + frame(tracker, 1000, [(10.0, 50.0, min_snr + 5), (20.0, -30.0, min_snr - 1)]) + + assert sink.calls == [], "the rejected detection outran its frame" + + drain(tracker) + assert sink.calls[0]["timestamp"] == 1000 + assert len(sink.calls[0]["below_snr"]) == 1 + + +def test_frames_are_released_in_order(sink): + """A consumer buffers these and wants to rely on append order being + timestamp order, so an unsettled frame blocks rather than being skipped.""" + tracker = make_tracker(sink) + for i in range(30): + frame(tracker, 1000 + i * 500, [(10.0 + i * 0.05, 50.0, 15.0), (300.0 - i * 7.0, -150.0, 15.0)]) + + assert sink.timestamps == sorted(sink.timestamps) + + +def test_a_frame_that_never_settles_is_forced_out(sink): + """A tentative track that keeps associating but never clears promotion + would otherwise hold the whole queue behind it forever.""" + tracker = make_tracker(sink, max_pending_classification_frames=5) + frame(tracker, 1000, [(10.0, 50.0, 15.0)]) + assert sink.calls == [] + + # Frames elsewhere, so the first frame's track neither promotes nor is + # touched by anything that would resolve it quickly. + for i in range(1, 8): + frame(tracker, 1000 + i * 500, [(400.0 + i * 20.0, 300.0, 15.0)]) + + assert sink.calls, "the queue never drained" + assert sink.timestamps[0] == 1000 + + +def test_the_pending_queue_does_not_grow_without_bound(sink): + tracker = make_tracker(sink, max_pending_classification_frames=10) + for i in range(200): + frame(tracker, 1000 + i * 500, [(10.0 + i * 0.05, 50.0, 15.0)]) + + assert len(tracker._pending_classification) <= 10 + + +def test_reset_drops_anything_still_pending(sink): + tracker = make_tracker(sink) + frame(tracker, 1000, [(10.0, 50.0, 15.0)]) + assert tracker._pending_classification + + tracker.reset() + + assert len(tracker._pending_classification) == 0 + + +# ── cost when nobody is listening ─────────────────────────────────────────── + + +def test_no_sink_means_no_bookkeeping(): + """The CLI and any node without a consumer must not pay for this.""" + tracker = Tracker(config=get_config()) + steady_target(tracker, 12) + assert len(tracker._pending_classification) == 0 + + +def test_the_gate_still_filters_what_the_tracker_sees(sink): + """Reporting the rejects must not mean tracking them.""" + original = get_config() + try: + set_config({**original, "tracker": {**original["tracker"], "min_snr": 10.0}}) + tracker = make_tracker(sink) + for i in range(12): + frame(tracker, 1000 + i * 500, [(10.0 + i * 0.05, 50.0, 2.0)]) + assert tracker.tracks == [], "a sub-threshold detection was tracked" + assert sink.count("below_snr") == 12 + finally: + set_config(original) diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..38829e7 --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,300 @@ +"""Tests for the HTTP control surface. + +The one control operation the tracker supports, clearing state between search +geometries, has only ever been reachable as a `{"type": "RESET"}` message mixed +into the detection socket. That works while one process sends both detections +and controls, which is the arrangement being unwound: blah2_api is taking over +the detection socket, and `run_tcp_server` accepts one connection at a time. + +These go over real HTTP against a real server on an ephemeral port rather than +calling the handler directly, because the things worth pinning are what a +client actually gets back: the status code, the body, and whether a 200 means +the reset has happened or merely been scheduled. +""" + +import json +import threading +import time +import urllib.error +import urllib.request + +import pytest + +from retina_tracker import control +from retina_tracker.config import get_config +from retina_tracker.control import start_control_server +from retina_tracker.history import CLASSES, DetectionHistory +from retina_tracker.server import process_streaming_frame +from retina_tracker.tracker import Tracker + + +def make_frame(timestamp, delay, doppler, snr=15.0): + return {"timestamp": timestamp, "delay": [delay], "doppler": [doppler], "snr": [snr]} + + +@pytest.fixture +def served(): + """A tracker with some state, and a control server in front of it.""" + tracker = Tracker(config=get_config()) + lock = threading.Lock() + # Port 0: the OS picks a free one, so these never collide with a real + # sidecar or with each other under parallel test runs. + server = start_control_server(tracker, lock, host="127.0.0.1", port=0) + try: + yield tracker, lock, f"http://127.0.0.1:{server.port}" + finally: + server.shutdown() + server.server_close() + + +def request(url, method="GET"): + req = urllib.request.Request(url, method=method) + try: + with urllib.request.urlopen(req, timeout=5) as response: + return response.status, json.loads(response.read().decode()) + except urllib.error.HTTPError as e: + return e.code, json.loads(e.read().decode()) + + +def build_state(tracker, frames=3): + for i in range(frames): + process_streaming_frame(tracker, make_frame(i * 500, 10.0, 50.0)) + + +def test_reset_clears_tracker_state(served): + tracker, _lock, base = served + build_state(tracker) + assert tracker.tracks or tracker.all_tracks or tracker.last_timestamp is not None + + status, body = request(base + "/reset", method="POST") + + assert status == 200 + assert body == {"ok": True} + assert tracker.tracks == [] + assert tracker.all_tracks == [] + assert tracker.last_timestamp is None + assert tracker.frame_count == 0 + + +def test_reset_has_already_happened_when_the_response_arrives(served): + """Not "scheduled": a caller resetting between candidate geometries waits + on the next frame, and that frame must not be able to associate into + pre-reset state.""" + tracker, _lock, base = served + build_state(tracker) + + request(base + "/reset", method="POST") + assert tracker.frame_count == 0 + + # A detection at a completely different geometry starts fresh rather than + # associating into anything that existed before the reset. + process_streaming_frame(tracker, make_frame(0, 300.0, -200.0)) + assert tracker.frame_count == 1 + + +def test_reset_tolerates_a_trailing_slash(served): + tracker, _lock, base = served + build_state(tracker) + status, _ = request(base + "/reset/", method="POST") + assert status == 200 + assert tracker.frame_count == 0 + + +def test_health_reports_what_the_tracker_has_seen(served): + """The one thing worth asking a node during the switchover: are frames + arriving at all.""" + tracker, _lock, base = served + status, body = request(base + "/health") + assert status == 200 + assert body["ok"] is True + assert body["frames"] == 0 + + build_state(tracker, frames=3) + + _status, body = request(base + "/health") + assert body["frames"] == 3 + assert isinstance(body["tracks"], int) + + +def test_unknown_paths_are_404_not_a_silent_success(served): + _tracker, _lock, base = served + assert request(base + "/nope")[0] == 404 + assert request(base + "/nope", method="POST")[0] == 404 + # /reset is a POST; a GET must not quietly do nothing and return 200. + assert request(base + "/reset")[0] == 404 + + +def test_reset_waits_for_an_in_flight_frame(served): + """The lock is the whole point of taking it: a reset landing mid-frame + would clear state the frame path is part way through mutating.""" + tracker, lock, base = served + build_state(tracker) + + done = threading.Event() + result = {} + + def reset_call(): + result["status"] = request(base + "/reset", method="POST")[0] + done.set() + + with lock: + thread = threading.Thread(target=reset_call, daemon=True) + thread.start() + # Held: the request cannot have completed, so state survives. + assert not done.wait(timeout=0.3) + assert tracker.frame_count == 3 + + assert done.wait(timeout=5) + assert result["status"] == 200 + assert tracker.frame_count == 0 + + +# ── The data stream ───────────────────────────────────────────────────────── + + +def read_events(response, count, timeout=10): + """Pull `count` SSE messages off an open stream.""" + events = [] + kind, buf = None, [] + deadline = time.monotonic() + timeout + for raw in response: + line = raw.decode().rstrip("\n") + if line.startswith("event: "): + kind = line[7:] + elif line.startswith("data: "): + buf.append(line[6:]) + elif line == "": + if kind and buf: + events.append((kind, json.loads("".join(buf)))) + if len(events) >= count: + return events + kind, buf = None, [] + if time.monotonic() > deadline: + break + return events + + +@pytest.fixture +def streaming(monkeypatch): + """A control server with a history behind it, streaming quickly.""" + monkeypatch.setattr(control, "STREAM_INTERVAL_S", 0.05) + tracker = Tracker(config=get_config()) + history = DetectionHistory() + lock = threading.Lock() + server = start_control_server(tracker, lock, host="127.0.0.1", port=0, history=history) + try: + yield history, f"http://127.0.0.1:{server.port}" + finally: + server.shutdown() + server.server_close() + + +def open_stream(base, query=""): + return urllib.request.urlopen(base + "/events" + query, timeout=10) + + +def test_the_stream_opens_with_a_snapshot(streaming): + """One ordering, owned by the connection. Fetching the snapshot + separately would race the start of the delta stream.""" + history, base = streaming + history.write_detections(1000, [{"delay": 10.0, "doppler": 50.0, "snr": 15.0}], [], []) + + with open_stream(base) as response: + kind, payload = read_events(response, 1)[0] + + assert kind == "snapshot" + assert payload["detections"]["associated"]["delay"] == [10.0] + assert set(payload["detections"]) == set(CLASSES) + + +def test_deltas_carry_only_what_was_appended(streaming): + history, base = streaming + history.write_detections(1000, [{"delay": 10.0, "doppler": 50.0, "snr": 15.0}], [], []) + + with open_stream(base) as response: + assert read_events(response, 1)[0][0] == "snapshot" + history.write_detections(2000, [], [{"delay": 20.0, "doppler": -30.0, "snr": 9.0}], []) + kind, payload = read_events(response, 1)[0] + + assert kind == "delta" + assert payload["detections"]["unassociated"]["delay"] == [20.0] + assert payload["detections"]["associated"]["t"] == [], "the snapshot's point came again" + + +def test_all_three_classes_reach_a_consumer(streaming): + """below_snr is the one nothing could see before.""" + history, base = streaming + with open_stream(base) as response: + read_events(response, 1) + history.write_detections( + 1000, + [{"delay": 1.0, "doppler": 0.0, "snr": 15.0}], + [{"delay": 2.0, "doppler": 0.0, "snr": 9.0}], + [{"delay": 3.0, "doppler": 0.0, "snr": 2.0}], + ) + _kind, payload = read_events(response, 1)[0] + + assert payload["detections"]["associated"]["delay"] == [1.0] + assert payload["detections"]["unassociated"]["delay"] == [2.0] + assert payload["detections"]["below_snr"]["delay"] == [3.0] + + +def test_a_clear_reseeds_the_stream_rather_than_reconciling(streaming): + history, base = streaming + history.write_detections(1000, [{"delay": 10.0, "doppler": 50.0, "snr": 15.0}], [], []) + + with open_stream(base) as response: + read_events(response, 1) + history.clear() + history.write_detections(2000, [{"delay": 99.0, "doppler": 0.0, "snr": 15.0}], [], []) + kind, payload = read_events(response, 1)[0] + + assert kind == "snapshot", "a voided cursor produced a delta" + assert payload["detections"]["associated"]["delay"] == [99.0] + + +def test_a_window_is_clamped_to_what_is_held(streaming): + history, base = streaming + with open_stream(base, "?window=999999") as response: + _kind, payload = read_events(response, 1)[0] + assert payload["window_s"] == history.window_s + + with open_stream(base, "?window=1") as response: + _kind, payload = read_events(response, 1)[0] + assert payload["window_s"] == control.MIN_WINDOW_S + + +def test_health_reports_the_history_footprint(streaming): + history, base = streaming + history.write_detections(1000, [{"delay": 10.0, "doppler": 50.0, "snr": 15.0}], [], []) + + status, body = request(base + "/health") + + assert status == 200 + assert body["history"]["detections"]["associated"] == 1 + assert body["history"]["approx_bytes"] == 20 + + +def test_the_stream_is_unavailable_without_a_history(served): + """The CLI runs without one, and must say so rather than pretend.""" + _tracker, _lock, base = served + status, body = request(base + "/events") + assert status == 503 + assert "error" in body + + +def test_history_clear_wipes_the_record_without_touching_the_tracker(streaming): + """ "Clear buffer" has always meant "clear what I am shown, keep + tracking". That distinction survives the record moving into the tracker.""" + history, base = streaming + history.write_detections(1000, [{"delay": 10.0, "doppler": 50.0, "snr": 15.0}], [], []) + + status, body = request(base + "/history/clear", method="POST") + + assert status == 200 and body == {"ok": True} + assert history.stats()["points"] == 0 + + +def test_history_clear_is_unavailable_without_a_history(served): + _tracker, _lock, base = served + assert request(base + "/history/clear", method="POST")[0] == 503 diff --git a/tests/test_event_deltas.py b/tests/test_event_deltas.py new file mode 100644 index 0000000..1cee80d --- /dev/null +++ b/tests/test_event_deltas.py @@ -0,0 +1,215 @@ +"""The events file carries only detections it has not already written. + +Each track event used to repeat the track's whole rolling window +(Track.get_recent_detections, up to detection_window points) to communicate +the one point that was new, multiplying the file by roughly the window size. + +The reason that was safe to change is that neither consumer reads an event as +"the track's state now". live_score.load_tracks unions detections by +timestamp across every event mentioning a track, and retina-gui's buffer +appends only timestamps it has not seen. Both reconstruct the same history +from a delta stream as from a repeating one, which is what these pin down. +""" + +import json + +import pytest + +from retina_tracker.live_score import load_tracks +from retina_tracker.output import EMITTED_MEMORY, TrackEventWriter + +BASE_TS = 1718747745000 + + +def detection(i): + return { + "timestamp": BASE_TS + i * 500, + "delay": 16.1 + i * 0.01, + "doppler": 134.5 - i * 0.2, + "snr": 16.2, + "adsb": None, + } + + +def rolling_window(upto, size=20): + """What get_recent_detections(n=size) returns after `upto` points.""" + start = max(0, upto - size) + return [detection(i) for i in range(start, upto)] + + +def read_events(path): + with open(path) as f: + return [json.loads(line) for line in f if line.strip()] + + +def emit_track_life(writer, track_id, points, window=20, **meta): + """One event per point, each carrying the rolling window, as the tracker + does for a track that is associated on every frame.""" + for n in range(1, points + 1): + writer.write_event(track_id, BASE_TS + n * 500, n, rolling_window(n, window), **meta) + + +def test_the_first_event_for_a_track_carries_everything_it_has(tmp_path): + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + writer.write_event("T1", BASE_TS, 3, rolling_window(3)) + writer.close() + + assert [d["timestamp"] for d in read_events(path)[0]["detections"]] == [BASE_TS, BASE_TS + 500, BASE_TS + 1000] + + +def test_later_events_carry_only_what_is_new(tmp_path): + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + emit_track_life(writer, "T1", 5) + writer.close() + + events = read_events(path) + counts = [len(e["detections"]) for e in events] + assert counts == [1, 1, 1, 1, 1], counts + # ...and between them they still describe every point, in order. + seen = [d["timestamp"] for e in events for d in e["detections"]] + assert seen == [BASE_TS + i * 500 for i in range(5)] + + +def test_an_event_is_still_written_when_nothing_is_new(tmp_path): + """length, the anomaly flags and shadow_fraction move over a track's + life. A consumer that missed those updates would hold a stale opinion of + a live track, so the event goes out with an empty detections list.""" + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + writer.write_event("T1", BASE_TS, 1, rolling_window(1), is_anomalous=False) + writer.write_event("T1", BASE_TS + 10, 1, rolling_window(1), is_anomalous=True, anomaly_types=["sustained_orbit"]) + writer.close() + + events = read_events(path) + assert len(events) == 2 + assert events[1]["detections"] == [] + assert events[1]["is_anomalous"] is True + assert events[1]["anomaly_types"] == ["sustained_orbit"] + + +def test_tracks_are_independent_of_each_other(tmp_path): + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + writer.write_event("T1", BASE_TS, 3, rolling_window(3)) + writer.write_event("T2", BASE_TS, 3, rolling_window(3)) + writer.close() + + events = read_events(path) + assert len(events[0]["detections"]) == 3 + assert len(events[1]["detections"]) == 3, "T2 was filtered against T1's history" + + +def test_live_score_reconstructs_the_same_history(tmp_path): + """The compatibility case. A delta stream and a repeating stream must + collapse to identical per-track detections, because load_tracks unions + by timestamp rather than trusting any single event.""" + delta_path = tmp_path / "delta.jsonl" + writer = TrackEventWriter(str(delta_path), max_bytes=0) + emit_track_life(writer, "T1", 50, adsb_hex="4CA2D1", shadow_fraction=0.25) + writer.close() + + # The same life, written the old way: every event repeating its window. + repeat_path = tmp_path / "repeat.jsonl" + with open(repeat_path, "w") as f: + for n in range(1, 51): + f.write( + json.dumps( + { + "track_id": "T1", + "adsb_hex": "4CA2D1", + "adsb_initialized": False, + "timestamp": BASE_TS + n * 500, + "length": n, + "detections": rolling_window(n), + "is_anomalous": False, + "max_velocity_ms": 0.0, + "anomaly_types": [], + "shadow_fraction": 0.25, + } + ) + + "\n" + ) + + delta = load_tracks(str(delta_path)) + repeat = load_tracks(str(repeat_path)) + + assert delta == repeat + assert len(delta["T1"]["detections"]) == 50 + assert delta["T1"]["adsb_hex"] == "4CA2D1" + assert delta["T1"]["shadow_fraction"] == 0.25 + + +def test_a_retina_gui_style_union_also_reconstructs(tmp_path): + """retina-gui appends only timestamps it has not seen, which is the same + reconstruction by a different route.""" + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + emit_track_life(writer, "T1", 30) + writer.close() + + held = [] + last_seen = None + for event in read_events(path): + for det in event["detections"]: + if last_seen is not None and det["timestamp"] <= last_seen: + continue + held.append(det["timestamp"]) + last_seen = det["timestamp"] + + assert held == [BASE_TS + i * 500 for i in range(30)] + + +def test_the_file_is_dramatically_smaller(tmp_path): + """The whole point. A 50-point track through a 20-point window.""" + delta_path = tmp_path / "delta.jsonl" + writer = TrackEventWriter(str(delta_path), max_bytes=0) + emit_track_life(writer, "T1", 50) + writer.close() + + repeat_bytes = sum( + len(json.dumps({"track_id": "T1", "detections": rolling_window(n)}) + "\n") for n in range(1, 51) + ) + delta_bytes = delta_path.stat().st_size + # Conservative: the delta file still carries per-event metadata the + # comparison above leaves out, and still wins by a wide margin. + assert delta_bytes < repeat_bytes / 3, (delta_bytes, repeat_bytes) + + +def test_the_high_water_map_is_bounded(tmp_path): + """Track ids are unique for the life of a run, so remembering every one + would be an unbounded dict on a process that runs for weeks.""" + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + for i in range(EMITTED_MEMORY + 50): + writer.write_event(f"T{i}", BASE_TS, 1, rolling_window(1)) + writer.close() + + assert len(writer._emitted_through) == EMITTED_MEMORY + + +def test_an_evicted_track_repeats_rather_than_loses(tmp_path): + """Eviction costs a repeated window, never a dropped detection. Both + consumers dedupe, so a repeat is free and an omission would not be.""" + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + writer.write_event("OLD", BASE_TS, 2, rolling_window(2)) + for i in range(EMITTED_MEMORY + 10): + writer.write_event(f"T{i}", BASE_TS, 1, rolling_window(1)) + writer.write_event("OLD", BASE_TS, 3, rolling_window(3)) + writer.close() + + revisit = [e for e in read_events(path) if e["track_id"] == "OLD"][-1] + assert len(revisit["detections"]) == 3, "an evicted track must resend, not skip" + + +@pytest.mark.parametrize("window", [1, 5, 20]) +def test_reconstruction_holds_at_any_window_size(tmp_path, window): + path = tmp_path / "events.jsonl" + writer = TrackEventWriter(str(path), max_bytes=0) + emit_track_life(writer, "T1", 25, window=window) + writer.close() + + tracks = load_tracks(str(path)) + assert [d["timestamp"] for d in tracks["T1"]["detections"]] == [BASE_TS + i * 500 for i in range(25)] diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..da718fe --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,331 @@ +"""The rolling record: what it holds, what it costs, and what it serves. + +Two things here are load-bearing beyond the obvious. Memory is bounded twice, +by the window and by a hard point ceiling, because a window alone makes the +footprint a function of how busy the sky is and a node cannot promise that. +And a consumer's position is a monotonic count rather than an index, because +pruning drops from the front. +""" + +import json + +import pytest + +from retina_tracker.history import CLASSES, DetectionHistory + +BASE = 1789030000000 + + +def det(delay=10.0, doppler=50.0, snr=15.0): + return {"delay": delay, "doppler": doppler, "snr": snr} + + +def track_det(ts, delay=10.0, doppler=50.0, snr=15.0): + return {"timestamp": ts, "delay": delay, "doppler": doppler, "snr": snr} + + +@pytest.fixture +def history(): + return DetectionHistory() + + +# ── what it holds ─────────────────────────────────────────────────────────── + + +def test_detections_land_in_the_class_they_were_given(history): + history.write_detections(BASE, [det(1.0)], [det(2.0)], [det(3.0)]) + payload, _ = history.snapshot() + + assert payload["detections"]["associated"]["delay"] == [1.0] + assert payload["detections"]["unassociated"]["delay"] == [2.0] + assert payload["detections"]["below_snr"]["delay"] == [3.0] + + +def test_every_class_is_present_even_when_empty(history): + """A consumer should not have to special-case a quiet node.""" + payload, _ = history.snapshot() + for name in CLASSES: + assert payload["detections"][name] == {"t": [], "delay": [], "doppler": [], "snr": []} + + +def test_track_points_and_metadata_are_kept(history): + history.write_event( + "T1", + BASE, + 3, + [track_det(BASE), track_det(BASE + 500)], + adsb_hex="4CA2D1", + is_anomalous=True, + anomaly_types=["sustained_orbit"], + max_velocity_ms=231.55, + shadow_fraction=0.6432, + ) + payload, _ = history.snapshot() + + assert payload["tracks"]["T1"]["t"] == [BASE, BASE + 500] + assert payload["tracks"]["T1"]["meta"] == { + "adsb_hex": "4CA2D1", + "length": 3, + "max_velocity_ms": 231.6, + "is_anomalous": True, + "anomaly_types": ["sustained_orbit"], + "shadow_fraction": 0.643, + } + + +def test_adsb_initialized_is_the_one_field_dropped_on_purpose(history): + history.write_event("T1", BASE, 1, [track_det(BASE)], adsb_initialized=True, something_new=7) + payload, _ = history.snapshot() + assert "adsb_initialized" not in payload["tracks"]["T1"]["meta"] + assert "something_new" not in payload["tracks"]["T1"]["meta"] + + +def test_a_track_never_goes_backwards(history): + """Every read relies on points being in timestamp order.""" + history.write_event("T1", BASE, 2, [track_det(BASE), track_det(BASE + 500)]) + history.write_event("T1", BASE, 2, [track_det(BASE), track_det(BASE + 500), track_det(BASE + 1000)]) + payload, _ = history.snapshot() + assert payload["tracks"]["T1"]["t"] == [BASE, BASE + 500, BASE + 1000] + + +# ── the wire form ─────────────────────────────────────────────────────────── + + +def test_values_are_rounded_to_the_precision_they_have(history): + """float32 storage reads 16.1 back as 16.100000381469727, which is 18 + bytes on the wire for four bytes of meaning.""" + history.write_detections(BASE, [det(16.1, -45.678912, 12.3456)], [], []) + payload, _ = history.snapshot() + assoc = payload["detections"]["associated"] + + assert assoc["delay"] == [16.1] + assert assoc["doppler"] == [-45.68] + assert assoc["snr"] == [12.3] + + +def test_the_payload_is_json_serialisable(history): + """array.array is not, so the slice has to materialise lists.""" + history.write_detections(BASE, [det()], [det()], [det()]) + history.write_event("T1", BASE, 1, [track_det(BASE)]) + payload, _ = history.snapshot() + json.dumps(payload) # raises if not + + +def test_rounding_keeps_the_payload_compact(history): + """The reason rounding happens here rather than being left to the + consumer: a float32 read back unrounded serialises to 17 significant + digits, and there are three of them per point.""" + for i in range(200): + history.write_detections(BASE + i, [det(16.1 + i * 0.01, -45.6, 12.3)], [], []) + payload, _ = history.snapshot() + rounded = payload["detections"]["associated"] + + cols = history._det["associated"] + raw = {k: list(cols[k]) for k in cols} + + encode = lambda obj: len(json.dumps(obj, separators=(",", ":"))) # noqa: E731 + assert encode(rounded) < encode(raw) / 2, (encode(rounded), encode(raw)) + + +# ── the view window ───────────────────────────────────────────────────────── + + +def test_a_window_narrows_what_is_served_without_touching_what_is_held(history): + history.write_detections(BASE, [det(1.0)], [], []) + history.write_detections(BASE + 100_000, [det(2.0)], [], []) + history.write_event("OLD", BASE, 1, [track_det(BASE)]) + history.write_event("NEW", BASE + 100_000, 1, [track_det(BASE + 100_000)]) + + payload, _ = history.snapshot(window_s=60, now_ms=BASE + 100_000) + + assert payload["detections"]["associated"]["delay"] == [2.0] + assert list(payload["tracks"]) == ["NEW"] + # Retention untouched: the window is a display concern. + assert len(history._det["associated"]["t"]) == 2 + assert set(history._tracks) == {"OLD", "NEW"} + + +def test_the_cursor_covers_everything_not_just_the_window(history): + """Otherwise a windowed consumer's first delta would re-send the history + the window had just excluded.""" + history.write_detections(BASE, [det(1.0)], [], []) + history.write_detections(BASE + 100_000, [det(2.0)], [], []) + + payload, cursor = history.snapshot(window_s=60, now_ms=BASE + 100_000) + + assert payload["detections"]["associated"]["delay"] == [2.0] + assert cursor["detections"]["associated"] == 2 + + delta, _ = history.since(cursor) + assert delta["detections"]["associated"]["t"] == [] + + +# ── deltas ────────────────────────────────────────────────────────────────── + + +def test_since_returns_only_what_was_appended(history): + history.write_detections(BASE, [det(1.0)], [], []) + _, cursor = history.snapshot() + + history.write_detections(BASE + 500, [det(2.0)], [det(3.0)], []) + delta, cursor2 = history.since(cursor) + + assert delta["detections"]["associated"]["delay"] == [2.0] + assert delta["detections"]["unassociated"]["delay"] == [3.0] + assert cursor2["detections"]["associated"] == 2 + + +def test_since_is_empty_when_nothing_changed(history): + history.write_detections(BASE, [det()], [], []) + _, cursor = history.snapshot() + + delta, _ = history.since(cursor) + + assert all(delta["detections"][name]["t"] == [] for name in CLASSES) + assert delta["tracks"] == {} + + +def test_since_sends_a_track_the_cursor_has_never_seen_whole(history): + """A track promoted after the consumer connected arrives with the + history the tracker backfilled, not truncated at the join.""" + _, cursor = history.snapshot() + history.write_event("T1", BASE, 3, [track_det(BASE), track_det(BASE + 500), track_det(BASE + 1000)]) + + delta, _ = history.since(cursor) + assert delta["tracks"]["T1"]["t"] == [BASE, BASE + 500, BASE + 1000] + + +def test_since_omits_tracks_with_no_new_points(history): + history.write_event("T1", BASE, 1, [track_det(BASE)]) + _, cursor = history.snapshot() + + history.write_event("T2", BASE + 500, 1, [track_det(BASE + 500)]) + delta, _ = history.since(cursor) + + assert list(delta["tracks"]) == ["T2"] + + +def test_since_survives_a_prune_that_dropped_unseen_points(history): + """A position is a monotonic count, not an index, because pruning drops + from the front.""" + h = DetectionHistory(window_s=10) + h.write_detections(BASE, [det(1.0)], [], []) + _, cursor = h.snapshot() + + h.write_detections(BASE + 50_000, [det(2.0)], [], []) + h.prune(now_ms=BASE + 51_000) + + delta, _ = h.since(cursor) + assert delta["detections"]["associated"]["delay"] == [2.0] + + +def test_since_refuses_a_cursor_from_before_a_clear(history): + history.write_detections(BASE, [det()], [], []) + _, cursor = history.snapshot() + + history.clear() + + delta, new_cursor = history.since(cursor) + assert delta is None and new_cursor is None + + +def test_a_fresh_snapshot_works_again_after_a_clear(history): + history.write_detections(BASE, [det()], [], []) + history.clear() + + _, cursor = history.snapshot() + history.write_detections(BASE + 500, [det(9.0)], [], []) + + delta, _ = history.since(cursor) + assert delta["detections"]["associated"]["delay"] == [9.0] + + +# ── bounds ────────────────────────────────────────────────────────────────── + + +def test_pruning_drops_what_is_older_than_the_window(): + h = DetectionHistory(window_s=10) + h.write_detections(BASE, [det(1.0)], [], []) + h.write_detections(BASE + 50_000, [det(2.0)], [], []) + h.write_event("OLD", BASE, 1, [track_det(BASE)]) + h.write_event("NEW", BASE + 50_000, 1, [track_det(BASE + 50_000)]) + + h.prune(now_ms=BASE + 51_000) + + payload, _ = h.snapshot() + assert payload["detections"]["associated"]["delay"] == [2.0] + assert list(payload["tracks"]) == ["NEW"] + + +def test_the_point_ceiling_bounds_memory_whatever_the_rate(): + """A window alone makes the footprint a function of how busy the sky is. + This is the promise underneath it.""" + h = DetectionHistory(window_s=4 * 3600, max_points=100) + for i in range(1000): + h.write_detections(BASE + i, [det(float(i))], [], []) + + assert len(h._det["associated"]["t"]) == 100 + # The newest are kept, the oldest dropped. + payload, _ = h.snapshot() + assert payload["detections"]["associated"]["delay"][-1] == 999.0 + + +def test_the_ceiling_does_not_break_a_cursor(): + h = DetectionHistory(max_points=100) + h.write_detections(BASE, [det(1.0)], [], []) + _, cursor = h.snapshot() + + for i in range(1, 500): + h.write_detections(BASE + i, [det(float(i))], [], []) + + delta, _ = h.since(cursor) + # Everything still held that the cursor had not seen, and nothing twice. + assert delta["detections"]["associated"]["delay"] == [float(i) for i in range(400, 500)] + + +def test_tracks_are_bounded_too(): + h = DetectionHistory(max_tracks=5) + for i in range(20): + h.write_event(f"T{i}", BASE + i * 1000, 1, [track_det(BASE + i * 1000)]) + + assert len(h._tracks) == 5 + + +def test_track_eviction_takes_the_stalest_first(): + h = DetectionHistory(max_tracks=2) + h.write_event("OLD", BASE, 1, [track_det(BASE)]) + h.write_event("LIVE", BASE + 10_000, 1, [track_det(BASE + 10_000)]) + h.write_event("LIVE", BASE + 20_000, 2, [track_det(BASE + 20_000)]) + + h.write_event("NEW", BASE + 30_000, 1, [track_det(BASE + 30_000)]) + + assert "OLD" not in h._tracks + assert set(h._tracks) == {"LIVE", "NEW"} + + +def test_stats_reports_what_is_held_and_what_it_costs(history): + for i in range(100): + history.write_detections(BASE + i, [det()], [det()], [det()]) + history.write_event("T1", BASE, 1, [track_det(BASE)]) + + stats = history.stats() + assert stats["detections"]["associated"] == 100 + assert stats["tracks"] == 1 + assert stats["points"] == 301 + assert stats["approx_bytes"] == 301 * 20 + + +def test_the_footprint_is_what_it_claims(): + """20 bytes a point, against the ~192 a list of 4-float tuples costs.""" + import sys + + h = DetectionHistory() + n = 20_000 + for i in range(n): + h.write_detections(BASE + i, [det()], [], []) + + cols = h._det["associated"] + actual = sum(sys.getsizeof(cols[k]) for k in cols) + per_point = actual / n + assert per_point < 32, per_point + assert h.stats()["approx_bytes"] == n * 20 diff --git a/tests/test_ingest.py b/tests/test_ingest.py new file mode 100644 index 0000000..8cebf04 --- /dev/null +++ b/tests/test_ingest.py @@ -0,0 +1,219 @@ +"""Tests for the detection feed's connection handling. + +Over real sockets rather than a fake, because every case here is about what +the socket layer does: a peer that goes silent without closing, a second peer +arriving while the first still holds the slot, a partial line stranded when +one is replaced. None of that is observable against a stub. + +The case this exists for is the handover from retina-gui to blah2_api, which +is precisely a new peer connecting while the old one may still be holding on. +""" + +import json +import socket +import threading +import time + +import pytest + +from retina_tracker.config import get_config +from retina_tracker.server import serve_detections +from retina_tracker.tracker import Tracker + + +def frame(timestamp, delay=10.0, doppler=50.0, snr=15.0): + return {"timestamp": timestamp, "delay": [delay], "doppler": [doppler], "snr": [snr]} + + +def line(obj): + return (json.dumps(obj) + "\n").encode() + + +@pytest.fixture +def feed(): + """A listening detection feed on an ephemeral port, served on a thread.""" + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(("127.0.0.1", 0)) + server.listen(8) + port = server.getsockname()[1] + + tracker = Tracker(config=get_config()) + lock = threading.Lock() + stop = threading.Event() + thread = threading.Thread(target=serve_detections, args=(server, tracker, lock, stop), daemon=True) + thread.start() + try: + yield tracker, port + finally: + stop.set() + thread.join(timeout=3) + server.close() + + +def connect(port): + return socket.create_connection(("127.0.0.1", port), timeout=3) + + +def wait_for_frames(tracker, n, timeout=3.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if tracker.frame_count >= n: + return True + time.sleep(0.02) + return tracker.frame_count >= n + + +def test_frames_on_a_single_connection_are_processed(feed): + tracker, port = feed + conn = connect(port) + try: + for i in range(3): + conn.sendall(line(frame(1000 + i * 500))) + assert wait_for_frames(tracker, 3) + finally: + conn.close() + + +def test_frames_split_across_packets_are_reassembled(feed): + """A frame is not a packet. The buffer has to survive a boundary landing + mid-JSON.""" + tracker, port = feed + conn = connect(port) + try: + payload = line(frame(1000)) + conn.sendall(payload[:9]) + time.sleep(0.15) + conn.sendall(payload[9:]) + assert wait_for_frames(tracker, 1) + finally: + conn.close() + + +def test_a_new_connection_is_served_while_the_old_one_sits_silent(feed): + """The regression. A peer that stops sending without closing used to + leave the loop blocked in recv, with the next peer unserved behind a + backlog of one. That is the shape of the retina-gui to blah2_api + handover, so it has to work.""" + tracker, port = feed + stale = connect(port) + try: + # Never sends, never closes: exactly a killed container or a dropped link. + time.sleep(0.2) + + fresh = connect(port) + try: + for i in range(3): + fresh.sendall(line(frame(5000 + i * 500))) + assert wait_for_frames(tracker, 3), "new peer was not served" + finally: + fresh.close() + finally: + stale.close() + + +def test_the_replaced_connection_stops_being_read(feed): + """Newest wins. The tracker does not deduplicate, so two live feeders + would give it every frame twice.""" + tracker, port = feed + first = connect(port) + try: + first.sendall(line(frame(1000))) + assert wait_for_frames(tracker, 1) + + second = connect(port) + try: + time.sleep(0.3) # let the replacement land + # The displaced peer keeps writing into a socket nobody reads. + for i in range(5): + try: + first.sendall(line(frame(2000 + i * 500))) + except OSError: + break # closed on us, which is the same outcome + time.sleep(0.4) + assert tracker.frame_count == 1, "frames from the replaced peer were read" + + second.sendall(line(frame(9000))) + assert wait_for_frames(tracker, 2) + finally: + second.close() + finally: + first.close() + + +def test_a_partial_line_from_a_replaced_peer_does_not_corrupt_the_next(feed): + """Half a frame left in the buffer would otherwise be prefixed onto the + new peer's first line and take both of them out.""" + tracker, port = feed + first = connect(port) + try: + first.sendall(b'{"timestamp": 1000, "delay": [10.0], "dop') # cut mid-key + time.sleep(0.2) + + second = connect(port) + try: + time.sleep(0.3) + second.sendall(line(frame(7000))) + assert wait_for_frames(tracker, 1), "the new peer's first frame was lost" + finally: + second.close() + finally: + first.close() + + +def test_a_reconnect_after_a_clean_close_is_served(feed): + tracker, port = feed + conn = connect(port) + conn.sendall(line(frame(1000))) + assert wait_for_frames(tracker, 1) + conn.close() + time.sleep(0.3) + + again = connect(port) + try: + again.sendall(line(frame(2000))) + assert wait_for_frames(tracker, 2) + finally: + again.close() + + +@pytest.mark.parametrize( + "junk", + [ + b"not json at all\n", + b"[1, 2, 3]\n", # valid JSON, not an object + b'{"delay": [1.0]}\n', # object, but no timestamp + b'{"timestamp": "nonsense"}\n', # timestamp of the wrong type + ], +) +def test_a_malformed_frame_does_not_take_the_feed_down(feed, junk): + """One bad line used to be able to end detection ingest until the + container was restarted.""" + tracker, port = feed + conn = connect(port) + try: + conn.sendall(junk) + time.sleep(0.2) + conn.sendall(line(frame(4000))) + assert wait_for_frames(tracker, 1), "the feed stopped after a malformed frame" + finally: + conn.close() + + +def test_reset_in_the_stream_still_works(feed): + """Kept until retina-gui stops feeding this socket.""" + tracker, port = feed + conn = connect(port) + try: + for i in range(3): + conn.sendall(line(frame(1000 + i * 500))) + assert wait_for_frames(tracker, 3) + + conn.sendall(line({"type": "RESET"})) + deadline = time.monotonic() + 3 + while time.monotonic() < deadline and tracker.frame_count != 0: + time.sleep(0.02) + assert tracker.frame_count == 0 + assert tracker.tracks == [] + finally: + conn.close() diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 8428435..7589189 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -20,7 +20,26 @@ _ = type("_", (), {})() # ── Contracts: referenced by something vulture cannot see ───────────────────── -# (none) +# http.server dispatches request handlers by name: BaseHTTPRequestHandler +# builds the method name from the verb on the request line and getattr()s it. +# Nothing in this repo calls either of these, and nothing should. +# retina_tracker/control.py (_Handler) +_.do_GET +_.do_POST + +# An override the base class calls, to keep per-request logging off stderr. +# Deleting it restores the noisy default rather than changing nothing. +# retina_tracker/control.py (_Handler) +_.log_message + +# Class attributes read by http.server and socketserver, never by us. +# protocol_version selects HTTP/1.1, which is what makes chunked framing +# available; without it every SSE message would arrive as one unframed body +# and a urllib3 client would block until its read timeout. +# retina_tracker/control.py +_.protocol_version +_.daemon_threads +_.allow_reuse_address # ── UNREVIEWED: appears dead, needs a decision (delete, or finish wiring) ────── # TODO: no reference found anywhere in the estate