From 52aa3fff78ac98dfded8a68289727848b2695fd9 Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Wed, 9 Sep 2026 18:47:46 +0100 Subject: [PATCH 1/7] 20260909 - Give the tracker a control surface of its own 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 the detections and the controls, and that is exactly the arrangement being unwound. blah2_api is taking over the detection socket so retina-gui stops being the transport between blah2 and here; run_tcp_server accepts one connection at a time, so the auto-calibration search cannot simply open its own alongside it. So control gets its own door: POST /reset on a loopback HTTP surface, plus GET /health, which reports frames seen and tracks held and is the one question worth asking a node while the feeder is being switched over. Nothing here knows who is calling. The auto-calibration search and the Tracker page are equal consumers of a tracker that does not distinguish between them. Loopback by default, for the same reason the ingest socket is: the sidecar runs with network_mode host, where 0.0.0.0 would publish this on the LAN. Stdlib only. This is a handful of routes with one or two clients, and a framework would be a dependency and an image layer for nothing. A reset from a request thread can now land in the middle of a frame, so both paths take one lock. It is held for the duration of the reset, so a 200 means the tracker is already clear rather than scheduled to be: a caller resetting between candidate geometries waits on the next frame, and that frame must not be able to associate into pre-reset state. Deferring to a flag the frame loop reads would have made the response a promise, and would never have applied at all on a node whose detections had stopped. The socket RESET keeps working. It goes when retina-gui stops feeding this socket, not before. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/cli.py | 8 +++ retina_tracker/control.py | 116 ++++++++++++++++++++++++++++++ retina_tracker/server.py | 30 +++++++- tests/test_control.py | 146 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 297 insertions(+), 3 deletions(-) create mode 100644 retina_tracker/control.py create mode 100644 tests/test_control.py diff --git a/retina_tracker/cli.py b/retina_tracker/cli.py index 3b54208..d3766bf 100644 --- a/retina_tracker/cli.py +++ b/retina_tracker/cli.py @@ -205,6 +205,12 @@ 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.") args = parser.parse_args() @@ -239,6 +245,8 @@ def main(): event_writer=event_writer, detection_window=args.detection_window, config=get_config(), + control_host=args.control_host, + control_port=args.control_port, ) 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..46bb44a --- /dev/null +++ b/retina_tracker/control.py @@ -0,0 +1,116 @@ +"""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 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 30101 + + +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. + def log_message(self, fmt, *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() == "/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): + if self._route() in ("/health", ""): + with self.server.tracker_lock: + payload = { + "ok": True, + "frames": self.server.tracker.frame_count, + "tracks": len(self.server.tracker.tracks), + } + self._send(200, payload) + return + self._send(404, {"error": "not found"}) + + 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): + super().__init__((host, port), _Handler) + self.tracker = tracker + self.tracker_lock = tracker_lock + + @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 start_control_server(tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT): + """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) + thread = threading.Thread(target=server.serve_forever, daemon=True, + name="tracker-control") + thread.start() + server.thread = thread + return server diff --git a/retina_tracker/server.py b/retina_tracker/server.py index 4c897d2..5fe034d 100644 --- a/retina_tracker/server.py +++ b/retina_tracker/server.py @@ -3,8 +3,12 @@ import json 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 .tracker import Tracker @@ -35,7 +39,8 @@ 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): +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): """Run tracker as TCP server receiving detection frames from blah2. Args: @@ -44,6 +49,8 @@ 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 """ tracker = Tracker( event_writer=event_writer, @@ -51,6 +58,16 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo config=config or get_config(), ) + # 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) + 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)) @@ -78,10 +95,17 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo # A real detection frame never carries a "type" key, # so this can never misfire on genuine data. if frame.get("type") == "RESET": - tracker.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) continue - process_streaming_frame(tracker, frame) + with tracker_lock: + process_streaming_frame(tracker, frame) except (ConnectionResetError, BrokenPipeError): print("blah2 disconnected", file=sys.stderr) diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..4f7bb53 --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,146 @@ +"""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 urllib.error +import urllib.request + +import pytest + +from retina_tracker.config import get_config +from retina_tracker.control import start_control_server +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 From 3a0e150d2c1dd7d4e34ec2264cc56e3fa6bd4dc0 Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 09:44:35 +0100 Subject: [PATCH 2/7] 20260910 - Let the newest detection feed take the socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ingest loop accepted one connection and then read it to EOF, with a backlog of one behind it. 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 sat there for as long as the kernel took to notice while the next peer waited unserved. That is not a hypothetical here. 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 getting it wrong means a tracker that silently receives nothing. So the loop selects on the listener as well as the connection, and the newest connection wins: accepting closes the one it replaces and drops whatever was half-read, since a partial line belongs to the peer that was sending it. Replacing rather than multiplexing is deliberate. The tracker does not deduplicate frames, so two live feeders would hand it every frame twice at dt=0 and give every track twice the evidence it earned. Two smaller things the rewrite fixes on the way past. The buffer holds bytes and decodes per line, so a multi-byte character split across a recv boundary no longer raises out of the loop. And one malformed frame can no longer end detection ingest until the container is restarted: a line that is not JSON, not an object, or missing what process_frame needs is logged and dropped. Both were reachable from anything with access to the port. serve_detections is split out from run_tcp_server and takes a stop_event, so the tests drive it over real sockets on an ephemeral port. Nothing here is observable against a stub: every case is about what the socket layer does. Reverting to the old "stop watching the listener once connected" behaviour fails three of them. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/server.py | 164 +++++++++++++++++++++-------- tests/test_ingest.py | 217 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 339 insertions(+), 42 deletions(-) create mode 100644 tests/test_ingest.py diff --git a/retina_tracker/server.py b/retina_tracker/server.py index 5fe034d..575ea53 100644 --- a/retina_tracker/server.py +++ b/retina_tracker/server.py @@ -1,6 +1,7 @@ """TCP server for receiving detection frames from blah2.""" import json +import select import socket import sys import threading @@ -39,6 +40,125 @@ def process_streaming_frame(tracker, frame): tracker.process_frame(detections, timestamp) +# 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): """Run tracker as TCP server receiving detection frames from blah2. @@ -71,48 +191,8 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo 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": - # 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) - continue - with tracker_lock: - 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/tests/test_ingest.py b/tests/test_ingest.py new file mode 100644 index 0000000..e56c856 --- /dev/null +++ b/tests/test_ingest.py @@ -0,0 +1,217 @@ +"""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() From ae48f43c4d215ac07e4b71afe3b512797b2a9418 Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 09:54:36 +0100 Subject: [PATCH 3/7] 20260910 - Write each detection to the events file once Every track event carried the track's whole rolling window, up to detection_window points, to communicate the one point that was new. On a 300-frame run with three targets that is 1.38 MB where 277 KB says the same thing: 5x, measured by running the same input through the tracker twice with the filtering off and on. Not the 20x the window size suggests, because at one detection per event the per-event metadata is most of the line. If this file needs to shrink further, that is where the bytes are now, not in the detections. Safe because 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 the tests assert directly by loading both and comparing. A high-water mark per track is enough. A track's history only grows forwards for as long as it can emit: merging is the one thing that splices older points in, 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. The map is an LRU bounded at 512 tracks, since ids are unique for the life of a run and this process runs for weeks. Evicting one costs a repeated window, never a dropped detection, which is the right way round given both consumers dedupe. The event is still written when nothing is new. length, the anomaly flags and shadow_fraction all move over a track's life, and a consumer that missed those updates would be holding a stale opinion of a live track. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/output.py | 42 ++++++++ tests/test_event_deltas.py | 204 +++++++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 tests/test_event_deltas.py 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/tests/test_event_deltas.py b/tests/test_event_deltas.py new file mode 100644 index 0000000..4b09d46 --- /dev/null +++ b/tests/test_event_deltas.py @@ -0,0 +1,204 @@ +"""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)] From 0ef68b220bbc0a97128a436f3a6a3d7c0d299f5b Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 10:03:00 +0100 Subject: [PATCH 4/7] 20260910 - Say what became of every detection, not just the tracked ones MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tracker consumed detections and reported only 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 vanished with it. So nothing downstream could tell "the tracker rejected everything it saw" from "there was nothing to see", which are the two explanations for an empty page and want opposite responses. Every detection now leaves through an optional sink, classified as associated, unassociated, or below_snr. The gate becomes a partition rather than a filter, in one loop rather than two comprehensions so a NaN snr — which fails both comparisons — still lands in exactly one bucket instead of disappearing from the accounting. The hard part is when a verdict is final. "Associated" means the detection ended up in a track that was confirmed, and that is not knowable when the detection arrives: one 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 most. So a frame waits until each of its tracks has settled, promoted or deleted, and is released with a final answer. A detection joining an already-confirmed track settles immediately, so a steady feed is not delayed; on a mixed 200-frame scene the queue sat 6 frames behind, which is N_WINDOW. Frames are released only from the front. Out-of-order release would break the one assumption a consumer buffering these wants to make, and it is why a below-gate detection waits for its frame despite its own verdict being obvious. One call per frame, in order, is worth a few frames of lag. ever_confirmed latches in the state_status setter rather than at the promotion site. Three routes reach ACTIVE — M-of-N promotion, a coasting track re-associating, and tracklet initiation — and latching at one of them was silently wrong for the other two: every track in a steady feed read as never confirmed. Current status cannot stand in for the flag either, since a track that was ACTIVE and has since been deleted reads the same as one that never got there. No sink means no bookkeeping, so the CLI and any node without a consumer pay nothing. With one attached the cost was 3.8% on 200 frames. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/track.py | 26 ++++ retina_tracker/tracker.py | 108 +++++++++++++- tests/test_classification.py | 277 +++++++++++++++++++++++++++++++++++ 3 files changed, 409 insertions(+), 2 deletions(-) create mode 100644 tests/test_classification.py 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..38ccfbd 100644 --- a/retina_tracker/tracker.py +++ b/retina_tracker/tracker.py @@ -29,10 +29,21 @@ 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 +53,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 +75,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 +130,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 +152,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 +168,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 +243,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 +274,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 +286,62 @@ 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..1775089 --- /dev/null +++ b/tests/test_classification.py @@ -0,0 +1,277 @@ +"""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) From d590fa03ce9264ed7dc5764777b74d80709ebc3d Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 10:11:33 +0100 Subject: [PATCH 5/7] 20260910 - Keep the last four hours, and serve them as they change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. In memory, not on disk. It does not need to survive a restart, and the alternative was several hundred megabytes a day onto SD cards the fleet cannot afford to wear out. Stored as array.array columns. 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, and one point-set per class means it is a slice rather than a filter over everything held. Bounded twice. The window is the intent, but a window alone makes the footprint a function of how busy the sky is, which is not something a node can promise. max_points is a hard ceiling underneath it, defaulting to about 10 MB per class, and whichever binds first wins. Tracks are capped too, evicting whichever has the stalest last point so a live one is reached for last. Values are rounded on the way out, to the precision the measurement has. That is not only about wire size: 16.1 held as a float32 reads back as 16.100000381469727, which would be eighteen bytes for four bytes of meaning. GET /events serves it: a snapshot on connect, then only what has been appended. The cursor lives in the connection's own thread and nowhere else, so there is no per-consumer state on the server to expire and nothing to negotiate — a reconnect takes a fresh snapshot, which is always a valid place to start. Sending the snapshot down the same stream is what removes the race between what it contained and where the deltas began. Because pruning drops from the front, a position is a monotonic count rather than an index, and clear() bumps a generation that voids every outstanding cursor at once. /health now reports the footprint, which is the question worth being able to ask of a node whose memory matters. Verified end to end against a running tracker: three detections a frame, one of each class, arriving over a real SSE connection as a snapshot followed by deltas, with the track's own points alongside. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/cli.py | 9 + retina_tracker/control.py | 120 +++++++++++- retina_tracker/history.py | 396 ++++++++++++++++++++++++++++++++++++++ retina_tracker/server.py | 20 +- tests/test_control.py | 135 +++++++++++++ tests/test_history.py | 319 ++++++++++++++++++++++++++++++ 6 files changed, 992 insertions(+), 7 deletions(-) create mode 100644 retina_tracker/history.py create mode 100644 tests/test_history.py diff --git a/retina_tracker/cli.py b/retina_tracker/cli.py index d3766bf..7e6313b 100644 --- a/retina_tracker/cli.py +++ b/retina_tracker/cli.py @@ -211,6 +211,13 @@ def main(): "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() @@ -247,6 +254,8 @@ def main(): 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 index 46bb44a..d610145 100644 --- a/retina_tracker/control.py +++ b/retina_tracker/control.py @@ -23,11 +23,27 @@ 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 @@ -62,17 +78,97 @@ def do_POST(self): self._send(404, {"error": "not found"}) def do_GET(self): - if self._route() in ("/health", ""): + 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") + # No length is knowable, and this never ends of its own accord. + 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.wfile.write(b": keepalive\n\n") + self.wfile.flush() + 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.wfile.write(f"event: {kind}\ndata: {body}\n\n".encode()) + self.wfile.flush() + def _route(self): return self.path.split("?", 1)[0].rstrip("/") @@ -91,10 +187,19 @@ class ControlServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True - def __init__(self, tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT): + 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): @@ -103,12 +208,19 @@ def port(self): return self.server_address[1] -def start_control_server(tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT): +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) + 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() diff --git a/retina_tracker/history.py b/retina_tracker/history.py new file mode 100644 index 0000000..7a6f1eb --- /dev/null +++ b/retina_tracker/history.py @@ -0,0 +1,396 @@ +"""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/server.py b/retina_tracker/server.py index 575ea53..2b56a42 100644 --- a/retina_tracker/server.py +++ b/retina_tracker/server.py @@ -10,6 +10,7 @@ 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 @@ -160,7 +161,8 @@ def serve_detections(server, tracker, tracker_lock, stop_event=None): 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): + 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: @@ -171,12 +173,23 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo 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 @@ -185,7 +198,8 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo if control_port: control = start_control_server(tracker, tracker_lock, - host=control_host, port=control_port) + 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) diff --git a/tests/test_control.py b/tests/test_control.py index 4f7bb53..0180733 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -14,13 +14,16 @@ 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 @@ -144,3 +147,135 @@ def reset_call(): 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 diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..2cc8d1f --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,319 @@ +"""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 From 30f91372820893413c283f6776b0eaa98cfd93da Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 10:49:25 +0100 Subject: [PATCH 6/7] 20260910 - Frame each stream message as its own chunk, and let the record be cleared MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two additions the page needs. POST /history/clear wipes the record without touching the tracker. That is the distinction "Clear buffer" has always carried — clear what I am being shown, keep tracking — and it survives the record moving here from retina-gui, which no longer has a buffer to clear. And the stream is chunked. It was sending "Connection: close" with no length, which is legal and which curl and EventSource both handle, so it looked fine. urllib3 does not: with no framing it reads to EOF to find a message boundary, and on a stream that never ends that means blocking until the read timeout instead of delivering each event as it arrives. retina-gui proxies this with requests, so every message would have arrived only when the connection finally gave up. Framing each message as its own HTTP/1.1 chunk makes it stream to any client rather than only the forgiving ones. Found by proxying it for real rather than by reading the spec, which is the only way this one shows up: every direct client I had tried worked. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/control.py | 30 ++++++++++++++++++++++++++---- tests/test_control.py | 17 +++++++++++++++++ 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/retina_tracker/control.py b/retina_tracker/control.py index d610145..6491172 100644 --- a/retina_tracker/control.py +++ b/retina_tracker/control.py @@ -66,6 +66,18 @@ def _send(self, status, payload): 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 @@ -131,7 +143,14 @@ def _stream(self): self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("X-Accel-Buffering", "no") - # No length is knowable, and this never ends of its own accord. + # 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() @@ -158,15 +177,18 @@ def _stream(self): self._event("delta", delta) last_sent = time.monotonic() elif time.monotonic() - last_sent >= HEARTBEAT_S: - self.wfile.write(b": keepalive\n\n") - self.wfile.flush() + 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.wfile.write(f"event: {kind}\ndata: {body}\n\n".encode()) + 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): diff --git a/tests/test_control.py b/tests/test_control.py index 0180733..7128938 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -279,3 +279,20 @@ def test_the_stream_is_unavailable_without_a_history(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 From 79165bef6682aa6d59f84fd61eb75b0a2a17df17 Mon Sep 17 00:00:00 2001 From: Josh Poole Date: Thu, 10 Sep 2026 13:39:43 +0100 Subject: [PATCH 7/7] 20260910 - Satisfy the formatter and name the control surface's framework hooks Two CI gates, neither of them a code problem. ruff-format reflows the lines this branch added: joined argument lists that fit, an implicit string concatenation collapsed to one literal, trailing commas, and a space after a docstring's opening quotes. Token-for-token identical otherwise. The dead-code gate reported seven names in control.py. Six are called by http.server and socketserver rather than by us: do_GET and do_POST are looked up by name from the request line, log_message is an override the base class calls, and protocol_version, daemon_threads and allow_reuse_address are class attributes the servers read. Those go in the whitelist's contracts section, which had been empty, each with the reason it cannot be deleted. The seventh was real. log_message's fmt parameter went unused because the override discards everything; the base class calls it positionally, so *args covers the format string and the parameter can go. Co-Authored-By: Claude Opus 5 (1M context) --- retina_tracker/cli.py | 37 ++++++++++++++++++---------- retina_tracker/control.py | 14 +++++------ retina_tracker/history.py | 45 +++++++++++++++++----------------- retina_tracker/server.py | 21 ++++++++++------ retina_tracker/tracker.py | 33 ++++++++++++++----------- tests/test_classification.py | 44 ++++++++++++++++++++------------- tests/test_control.py | 16 ++++++------ tests/test_event_deltas.py | 47 ++++++++++++++++++++++-------------- tests/test_history.py | 36 ++++++++++++++++++--------- tests/test_ingest.py | 18 ++++++++------ vulture_whitelist.py | 21 +++++++++++++++- 11 files changed, 204 insertions(+), 128 deletions(-) diff --git a/retina_tracker/cli.py b/retina_tracker/cli.py index 7e6313b..8680e1b 100644 --- a/retina_tracker/cli.py +++ b/retina_tracker/cli.py @@ -205,19 +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).") + 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() diff --git a/retina_tracker/control.py b/retina_tracker/control.py index 6491172..e66a1a9 100644 --- a/retina_tracker/control.py +++ b/retina_tracker/control.py @@ -53,8 +53,9 @@ class _Handler(BaseHTTPRequestHandler): # 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. - def log_message(self, fmt, *args): + # 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): @@ -209,8 +210,7 @@ class ControlServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True - def __init__(self, tracker, tracker_lock, host=DEFAULT_HOST, port=DEFAULT_PORT, - history=None): + 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 @@ -236,15 +236,13 @@ def _has_points(delta): 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): +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 = 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 index 7a6f1eb..8834779 100644 --- a/retina_tracker/history.py +++ b/retina_tracker/history.py @@ -69,8 +69,7 @@ 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")} + return {"t": array("q"), "delay": array("f"), "doppler": array("f"), "snr": array("f")} def _append(cols, timestamp, delay, doppler, snr): @@ -124,10 +123,10 @@ def __init__(self, window_s=WINDOW_S, max_points=MAX_POINTS, max_tracks=MAX_TRAC 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._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 @@ -137,18 +136,25 @@ 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)): + 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)) + _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): + 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 @@ -250,14 +256,8 @@ 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() - }, + "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): @@ -382,6 +382,7 @@ def start_pruner(history, interval_s=60, stop_event=None): 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: diff --git a/retina_tracker/server.py b/retina_tracker/server.py index 2b56a42..45a80e3 100644 --- a/retina_tracker/server.py +++ b/retina_tracker/server.py @@ -160,9 +160,17 @@ def serve_detections(server, tracker, tracker_lock, stop_event=None): _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): +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: @@ -178,8 +186,7 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo """ # 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) + history = DetectionHistory(window_s=history_window_s, max_points=history_max_points) tracker = Tracker( # The file's writer deduplicates detections, so it goes first and the @@ -197,9 +204,7 @@ def run_tcp_server(host="0.0.0.0", port=3012, event_writer=None, detection_windo tracker_lock = threading.Lock() if control_port: - control = start_control_server(tracker, tracker_lock, - host=control_host, port=control_port, - history=history) + 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) diff --git a/retina_tracker/tracker.py b/retina_tracker/tracker.py index 38ccfbd..ba04507 100644 --- a/retina_tracker/tracker.py +++ b/retina_tracker/tracker.py @@ -41,9 +41,15 @@ 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, detection_sink=None, - max_pending_classification_frames=MAX_PENDING_CLASSIFICATION_FRAMES): + 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 = [] @@ -293,12 +299,14 @@ def _classify_frame(self, timestamp, landed_in, detections, below_snr): 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._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): @@ -326,9 +334,7 @@ def _drain_classifications(self): 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"]): + if not aged_out and any(not (track.ever_confirmed or track.retired) for _det, track in entry["pairs"]): break associated = [] @@ -339,8 +345,7 @@ def _drain_classifications(self): (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"]) + self.detection_sink.write_detections(entry["timestamp"], associated, unassociated, entry["below_snr"]) def _associate(self, detections): if not self.tracks or not detections: diff --git a/tests/test_classification.py b/tests/test_classification.py index 1775089..654d8e7 100644 --- a/tests/test_classification.py +++ b/tests/test_classification.py @@ -26,12 +26,14 @@ 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, - }) + self.calls.append( + { + "timestamp": timestamp, + "associated": associated, + "unassociated": unassociated, + "below_snr": below_snr, + } + ) # Convenience views over everything released so far. def snrs(self, bucket): @@ -56,12 +58,15 @@ def make_tracker(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], - }) + 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): @@ -86,6 +91,7 @@ def drain(tracker, start_ts=900000, max_frames=120): # ── 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"] @@ -124,9 +130,11 @@ def test_every_detection_is_accounted_for(sink): 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)] + 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) @@ -137,6 +145,7 @@ def test_every_detection_is_accounted_for(sink): # ── 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) @@ -192,6 +201,7 @@ def test_a_detection_joining_an_established_track_is_not_delayed(sink): # ── 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 @@ -214,8 +224,7 @@ def test_frames_are_released_in_order(sink): 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)]) + 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) @@ -256,6 +265,7 @@ def test_reset_drops_anything_still_pending(sink): # ── 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()) diff --git a/tests/test_control.py b/tests/test_control.py index 7128938..38829e7 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -151,6 +151,7 @@ def reset_call(): # ── The data stream ───────────────────────────────────────────────────────── + def read_events(response, count, timeout=10): """Pull `count` SSE messages off an open stream.""" events = [] @@ -180,8 +181,7 @@ def streaming(monkeypatch): 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) + 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: @@ -226,10 +226,12 @@ def test_all_three_classes_reach_a_consumer(streaming): 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}]) + 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] @@ -282,7 +284,7 @@ def test_the_stream_is_unavailable_without_a_history(served): def test_history_clear_wipes_the_record_without_touching_the_tracker(streaming): - """"Clear buffer" has always meant "clear what I am shown, keep + """ "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}], [], []) diff --git a/tests/test_event_deltas.py b/tests/test_event_deltas.py index 4b09d46..1cee80d 100644 --- a/tests/test_event_deltas.py +++ b/tests/test_event_deltas.py @@ -22,8 +22,13 @@ 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} + 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): @@ -41,8 +46,7 @@ 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) + 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): @@ -51,8 +55,7 @@ def test_the_first_event_for_a_track_carries_everything_it_has(tmp_path): 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] + 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): @@ -76,8 +79,7 @@ def test_an_event_is_still_written_when_nothing_is_new(tmp_path): 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.write_event("T1", BASE_TS + 10, 1, rolling_window(1), is_anomalous=True, anomaly_types=["sustained_orbit"]) writer.close() events = read_events(path) @@ -112,12 +114,23 @@ def test_live_score_reconstructs_the_same_history(tmp_path): 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") + 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)) @@ -156,8 +169,7 @@ def test_the_file_is_dramatically_smaller(tmp_path): writer.close() repeat_bytes = sum( - len(json.dumps({"track_id": "T1", "detections": rolling_window(n)}) + "\n") - for n in range(1, 51) + 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 @@ -200,5 +212,4 @@ def test_reconstruction_holds_at_any_window_size(tmp_path, 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)] + 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 index 2cc8d1f..da718fe 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -31,6 +31,7 @@ def history(): # ── 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() @@ -48,23 +49,32 @@ def test_every_class_is_present_even_when_empty(history): 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) + 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"], + "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) + 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"] @@ -73,14 +83,14 @@ def test_adsb_initialized_is_the_one_field_dropped_on_purpose(history): 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)]) + 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.""" @@ -119,6 +129,7 @@ def test_rounding_keeps_the_payload_compact(history): # ── 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)], [], []) @@ -151,6 +162,7 @@ def test_the_cursor_covers_everything_not_just_the_window(history): # ── deltas ────────────────────────────────────────────────────────────────── + def test_since_returns_only_what_was_appended(history): history.write_detections(BASE, [det(1.0)], [], []) _, cursor = history.snapshot() @@ -177,8 +189,7 @@ 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)]) + 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] @@ -231,6 +242,7 @@ def test_a_fresh_snapshot_works_again_after_a_clear(history): # ── bounds ────────────────────────────────────────────────────────────────── + def test_pruning_drops_what_is_older_than_the_window(): h = DetectionHistory(window_s=10) h.write_detections(BASE, [det(1.0)], [], []) diff --git a/tests/test_ingest.py b/tests/test_ingest.py index e56c856..8cebf04 100644 --- a/tests/test_ingest.py +++ b/tests/test_ingest.py @@ -41,8 +41,7 @@ def feed(): 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 = threading.Thread(target=serve_detections, args=(server, tracker, lock, stop), daemon=True) thread.start() try: yield tracker, port @@ -178,12 +177,15 @@ def test_a_reconnect_after_a_clean_close_is_served(feed): 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 -]) +@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.""" 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