From 598c4afe4d35ca5b7b64ca0661087622f82555bb Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Mon, 16 Feb 2026 00:52:29 +0000 Subject: [PATCH 01/12] feat: add ServerConfig for HTTP receive endpoint Co-Authored-By: Claude Opus 4.6 --- tests/test_config.py | 15 ++++++++++++++- tracker_host/config.py | 16 ++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/tests/test_config.py b/tests/test_config.py index dcff3ec..fe1f03e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,7 +2,7 @@ import yaml -from tracker_host.config import Config, load_config, _parse_config +from tracker_host.config import Config, ServerConfig, load_config, _parse_config class TestParseConfig: @@ -79,3 +79,16 @@ def test_load_config_from_file(self, tmp_path): assert config.poll_interval_sec == 1.0 assert config.startup_delay_sec == 2.5 assert len(config.trackers) == 1 + + +class TestServerConfig: + def test_server_defaults(self): + config = _parse_config({}) + assert config.server.host == "0.0.0.0" + assert config.server.port == 8080 + + def test_server_custom(self): + raw = {"server": {"host": "127.0.0.1", "port": 9090}} + config = _parse_config(raw) + assert config.server.host == "127.0.0.1" + assert config.server.port == 9090 diff --git a/tracker_host/config.py b/tracker_host/config.py index c9e59ff..151a7be 100644 --- a/tracker_host/config.py +++ b/tracker_host/config.py @@ -7,6 +7,14 @@ import yaml +@dataclass +class ServerConfig: + """HTTP server configuration for receiving track events from nodes.""" + + host: str = "0.0.0.0" + port: int = 8080 + + @dataclass class RetryConfig: """Retry and resilience settings.""" @@ -60,6 +68,7 @@ class Config: tcp_connect_retries: int = 10 tcp_retry_delay_sec: float = 0.5 config_poll_interval_sec: float = 60.0 + server: ServerConfig = field(default_factory=ServerConfig) retry: RetryConfig = field(default_factory=RetryConfig) api_forward: ApiForwardConfig = field(default_factory=ApiForwardConfig) trackers: list[TrackerConfig] = field(default_factory=list) @@ -82,6 +91,12 @@ def load_config(config_path: str | Path) -> Config: def _parse_config(raw: dict) -> Config: """Parse raw YAML dict into Config dataclass.""" + server_raw = raw.get("server", {}) + server = ServerConfig( + host=server_raw.get("host", "0.0.0.0"), + port=server_raw.get("port", 8080), + ) + retry_raw = raw.get("retry", {}) retry = RetryConfig( max_attempts=retry_raw.get("max_attempts", 5), @@ -130,6 +145,7 @@ def _parse_config(raw: dict) -> Config: ) return Config( + server=server, output_dir=raw.get("output_dir", "./output"), poll_interval_sec=raw.get("poll_interval_sec", 0.5), status_interval_sec=raw.get("status_interval_sec", 30.0), From 4a282625d9ddade9c13905dd787508871ad8ff3c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 00:56:22 +0000 Subject: [PATCH 02/12] feat: add NodeRegistry for dynamic node management Co-Authored-By: Claude Opus 4.6 --- ...2026-02-16-node-to-server-track-posting.md | 1265 +++++++++++++++++ tests/test_node_registry.py | 52 + tracker_host/node_registry.py | 82 ++ 3 files changed, 1399 insertions(+) create mode 100644 docs/plans/2026-02-16-node-to-server-track-posting.md create mode 100644 tests/test_node_registry.py create mode 100644 tracker_host/node_registry.py diff --git a/docs/plans/2026-02-16-node-to-server-track-posting.md b/docs/plans/2026-02-16-node-to-server-track-posting.md new file mode 100644 index 0000000..dfb8b09 --- /dev/null +++ b/docs/plans/2026-02-16-node-to-server-track-posting.md @@ -0,0 +1,1265 @@ +# Node-to-Server Track Posting Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Replace the current server-polls-nodes architecture with nodes running retina-tracker locally and POSTing track events to a central server via HTTP. + +**Architecture:** A thin `node_agent` package runs on each Raspberry Pi node alongside retina-tracker. It starts retina-tracker as a subprocess (which receives detections from blah2 via TCP), reads track events from stdout, and POSTs them to the central `tracker-host` server. The server receives tracks via new aiohttp endpoints, writes them to JSONL files (reusing the existing `OutputHandler`), and runs the geolocator centrally for better multi-radar fusion accuracy. Nodes register dynamically by pushing their blah2 config on startup. + +**Tech Stack:** Python 3.10+, aiohttp (server-side HTTP), urllib.request (node-side HTTP, stdlib only), existing retina-tracker and tracker-host infrastructure. + +**Key architectural decision:** Dynamic node registration (no pre-configuration needed on the server). Nodes push their radar config to the server, and the server creates per-node output handlers on the fly. + +--- + +## Phase 1: Server-Side HTTP Receive Endpoint + +### Task 1: Add ServerConfig to config parsing + +**Files:** +- Modify: `tracker_host/config.py` +- Modify: `tests/test_config.py` + +**Step 1: Write the failing test** + +Add to `tests/test_config.py`: + +```python +class TestServerConfig: + def test_server_defaults(self): + config = _parse_config({}) + assert config.server.host == "0.0.0.0" + assert config.server.port == 8080 + + def test_server_custom(self): + raw = {"server": {"host": "127.0.0.1", "port": 9090}} + config = _parse_config(raw) + assert config.server.host == "127.0.0.1" + assert config.server.port == 9090 +``` + +**Step 2: Run test to verify it fails** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_config.py::TestServerConfig -v` +Expected: FAIL with `AttributeError: 'Config' object has no attribute 'server'` + +**Step 3: Write minimal implementation** + +Add to `tracker_host/config.py`: + +```python +@dataclass +class ServerConfig: + """HTTP server configuration for receiving track events from nodes.""" + host: str = "0.0.0.0" + port: int = 8080 +``` + +Add `server` field to `Config` dataclass: +```python +server: ServerConfig = field(default_factory=ServerConfig) +``` + +Add parsing in `_parse_config()` before the `return Config(...)`: +```python +server_raw = raw.get("server", {}) +server = ServerConfig( + host=server_raw.get("host", "0.0.0.0"), + port=server_raw.get("port", 8080), +) +``` + +And add `server=server` to the `return Config(...)` call. + +**Step 4: Run test to verify it passes** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_config.py -v` +Expected: ALL PASS + +**Step 5: Commit** + +```bash +git add tracker_host/config.py tests/test_config.py +git commit -m "feat: add ServerConfig for HTTP receive endpoint" +``` + +--- + +### Task 2: Create NodeRegistry for dynamic node management + +**Files:** +- Create: `tracker_host/node_registry.py` +- Create: `tests/test_node_registry.py` + +**Step 1: Write the failing tests** + +Create `tests/test_node_registry.py`: + +```python +"""Tests for NodeRegistry.""" + +import json +import pytest +from unittest.mock import AsyncMock, patch, MagicMock +from tracker_host.node_registry import NodeRegistry + + +@pytest.fixture +def registry(tmp_path): + return NodeRegistry(output_dir=str(tmp_path)) + + +class TestNodeRegistry: + @pytest.mark.asyncio + async def test_register_node(self, registry): + config_data = {"location": {"rx": {"latitude": 33.9}}} + await registry.register_node("radar3", config_data) + nodes = registry.list_nodes() + assert "radar3" in nodes + assert nodes["radar3"]["has_config"] is True + + @pytest.mark.asyncio + async def test_handle_track_event_writes_jsonl(self, registry, tmp_path): + await registry.register_node("radar3", {"location": {}}) + event = {"track_id": "test-001", "length": 5, "timestamp": 1000} + await registry.handle_track_event("radar3", json.dumps(event)) + + # Check output file was created + import glob + files = glob.glob(str(tmp_path / "radar3_*.jsonl")) + assert len(files) == 1 + with open(files[0]) as f: + line = f.readline() + assert json.loads(line)["track_id"] == "test-001" + + @pytest.mark.asyncio + async def test_handle_track_auto_registers(self, registry): + event = {"track_id": "auto-001", "length": 1, "timestamp": 1000} + await registry.handle_track_event("newnode", json.dumps(event)) + nodes = registry.list_nodes() + assert "newnode" in nodes + + @pytest.mark.asyncio + async def test_list_nodes_empty(self, registry): + assert registry.list_nodes() == {} + + @pytest.mark.asyncio + async def test_get_node_config(self, registry): + config_data = {"capture": {"fc": 195000000}} + await registry.register_node("radar3", config_data) + assert registry.get_node_config("radar3") == config_data + assert registry.get_node_config("missing") is None +``` + +**Step 2: Run test to verify it fails** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_registry.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'tracker_host.node_registry'` + +Note: You may need to install `pytest-asyncio`. Check `requirements.txt` and add if needed: +Run: `pip install pytest-asyncio` (if not already installed) + +**Step 3: Write minimal implementation** + +Create `tracker_host/node_registry.py`: + +```python +"""NodeRegistry: dynamic registration and management of radar nodes.""" + +import logging +from typing import Any, Optional + +from .output_handler import OutputHandler + +logger = logging.getLogger(__name__) + + +class NodeRegistry: + """Manages dynamically registered radar nodes. + + Nodes register by POSTing their config. Each registered node gets + an OutputHandler for writing track events to daily JSONL files. + """ + + def __init__(self, output_dir: str): + self.output_dir = output_dir + self._nodes: dict[str, dict[str, Any]] = {} + + async def register_node(self, name: str, config_data: dict) -> None: + """Register a node or update its config.""" + if name not in self._nodes: + output_handler = OutputHandler( + name=name, + output_dir=self.output_dir, + ) + self._nodes[name] = { + "config": config_data, + "output_handler": output_handler, + } + logger.info(f"Registered new node: {name}") + else: + self._nodes[name]["config"] = config_data + logger.info(f"Updated config for node: {name}") + + async def handle_track_event(self, name: str, event_line: str) -> None: + """Handle a track event from a node.""" + if name not in self._nodes: + await self.register_node(name, {}) + + node = self._nodes[name] + await node["output_handler"].handle_event(event_line) + + def get_node_config(self, name: str) -> Optional[dict]: + """Get stored config for a node, or None if not registered.""" + if name in self._nodes: + return self._nodes[name]["config"] + return None + + def list_nodes(self) -> dict[str, Any]: + """List all registered nodes with summary info.""" + return { + name: { + "has_config": bool(info["config"]), + "active_tracks": info["output_handler"].metrics.count, + } + for name, info in self._nodes.items() + } + + async def close(self) -> None: + """Close all output handlers.""" + for info in self._nodes.values(): + await info["output_handler"].close() + self._nodes.clear() +``` + +**Step 4: Run test to verify it passes** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_registry.py -v` +Expected: ALL PASS + +**Step 5: Commit** + +```bash +git add tracker_host/node_registry.py tests/test_node_registry.py +git commit -m "feat: add NodeRegistry for dynamic node management" +``` + +--- + +### Task 3: Create HTTP receive server + +**Files:** +- Create: `tracker_host/server.py` +- Create: `tests/test_server.py` + +**Step 1: Write the failing tests** + +Create `tests/test_server.py`: + +```python +"""Tests for the HTTP receive server.""" + +import json +import pytest +from aiohttp import web +from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop + +from tracker_host.server import create_app +from tracker_host.node_registry import NodeRegistry + + +@pytest.fixture +def registry(tmp_path): + return NodeRegistry(output_dir=str(tmp_path)) + + +@pytest.fixture +def app(registry): + return create_app(registry) + + +@pytest.fixture +async def client(aiohttp_client, app): + return await aiohttp_client(app) + + +class TestReceiveServer: + @pytest.mark.asyncio + async def test_post_config(self, client): + config = {"location": {"rx": {"latitude": 33.9}}} + resp = await client.post( + "/api/node/radar3/config", + json=config, + ) + assert resp.status == 200 + body = await resp.json() + assert body["status"] == "registered" + assert body["node"] == "radar3" + + @pytest.mark.asyncio + async def test_post_track(self, client): + event = {"track_id": "t-001", "length": 5, "timestamp": 1000} + resp = await client.post( + "/api/node/radar3/tracks", + data=json.dumps(event), + headers={"Content-Type": "application/x-ndjson"}, + ) + assert resp.status == 200 + + @pytest.mark.asyncio + async def test_list_nodes_empty(self, client): + resp = await client.get("/api/nodes") + assert resp.status == 200 + body = await resp.json() + assert body["nodes"] == {} + + @pytest.mark.asyncio + async def test_list_nodes_after_registration(self, client): + await client.post("/api/node/radar3/config", json={"location": {}}) + resp = await client.get("/api/nodes") + body = await resp.json() + assert "radar3" in body["nodes"] + + @pytest.mark.asyncio + async def test_post_track_batch(self, client): + """Test posting multiple events as newline-delimited JSON.""" + events = [ + {"track_id": "t-001", "length": 3, "timestamp": 1000}, + {"track_id": "t-002", "length": 5, "timestamp": 1001}, + ] + body = "\n".join(json.dumps(e) for e in events) + resp = await client.post( + "/api/node/radar3/tracks", + data=body, + headers={"Content-Type": "application/x-ndjson"}, + ) + assert resp.status == 200 +``` + +**Step 2: Run test to verify it fails** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_server.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'tracker_host.server'` + +Note: may need `pip install pytest-aiohttp` for the `aiohttp_client` fixture. + +**Step 3: Write minimal implementation** + +Create `tracker_host/server.py`: + +```python +"""HTTP receive server for accepting track events from remote nodes.""" + +import json +import logging + +from aiohttp import web + +from .node_registry import NodeRegistry + +logger = logging.getLogger(__name__) + + +def create_app(registry: NodeRegistry) -> web.Application: + """Create the aiohttp web application.""" + app = web.Application() + app["registry"] = registry + + app.router.add_post("/api/node/{name}/config", handle_config) + app.router.add_post("/api/node/{name}/tracks", handle_tracks) + app.router.add_get("/api/nodes", handle_list_nodes) + + return app + + +async def handle_config(request: web.Request) -> web.Response: + """Receive and store radar config from a node.""" + name = request.match_info["name"] + registry: NodeRegistry = request.app["registry"] + + try: + config_data = await request.json() + except json.JSONDecodeError: + return web.json_response( + {"status": "error", "message": "invalid JSON"}, status=400 + ) + + await registry.register_node(name, config_data) + + logger.info(f"Received config from node: {name}") + return web.json_response({"status": "registered", "node": name}) + + +async def handle_tracks(request: web.Request) -> web.Response: + """Receive track events from a node (JSONL: one event per line).""" + name = request.match_info["name"] + registry: NodeRegistry = request.app["registry"] + + body = await request.text() + + for line in body.strip().split("\n"): + line = line.strip() + if line: + await registry.handle_track_event(name, line) + + return web.json_response({"status": "ok"}) + + +async def handle_list_nodes(request: web.Request) -> web.Response: + """List all registered nodes (for debugging).""" + registry: NodeRegistry = request.app["registry"] + return web.json_response({"nodes": registry.list_nodes()}) +``` + +**Step 4: Run test to verify it passes** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_server.py -v` +Expected: ALL PASS + +**Step 5: Commit** + +```bash +git add tracker_host/server.py tests/test_server.py +git commit -m "feat: add HTTP receive server for node track events" +``` + +--- + +### Task 4: Wire server mode into __main__.py and config.yaml + +**Files:** +- Modify: `tracker_host/__main__.py` +- Modify: `tracker_host/manager.py` +- Modify: `config.yaml` + +**Step 1: Update __main__.py to support server mode** + +Add `--server` flag. When set, run the HTTP server instead of the polling manager. + +In `tracker_host/__main__.py`, update imports and add to the argument parser: + +```python +parser.add_argument( + "--server", + action="store_true", + help="Run in server mode (receive tracks from nodes via HTTP)", +) +``` + +Update the run section: + +```python +if args.server: + from .server_mode import run_server + asyncio.run(run_server(str(config_path), verbose=args.verbose)) +else: + asyncio.run(run_manager(str(config_path), verbose=args.verbose)) +``` + +**Step 2: Create server_mode.py** + +Create `tracker_host/server_mode.py`: + +```python +"""Server mode: receive track events from remote nodes via HTTP.""" + +import asyncio +import logging +import signal + +from aiohttp import web + +from .config import load_config +from .node_registry import NodeRegistry +from .server import create_app + +logger = logging.getLogger(__name__) + + +async def run_server(config_path: str, verbose: bool = False) -> None: + """Run tracker-host in server mode.""" + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + config = load_config(config_path) + registry = NodeRegistry(output_dir=config.output_dir) + app = create_app(registry) + + runner = web.AppRunner(app) + await runner.setup() + + site = web.TCPSite(runner, config.server.host, config.server.port) + await site.start() + + logger.info( + f"Server listening on http://{config.server.host}:{config.server.port}" + ) + + # Wait for shutdown signal + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + + def handle_signal(): + logger.info("Received shutdown signal") + stop_event.set() + + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, handle_signal) + + try: + await stop_event.wait() + finally: + await registry.close() + await runner.cleanup() + logger.info("Server stopped") +``` + +**Step 3: Update config.yaml with server section** + +Add to the top of `config.yaml`: + +```yaml +# Server mode settings (used with --server flag) +server: + host: "0.0.0.0" + port: 8080 +``` + +**Step 4: Test manually** + +```bash +cd /opt/apps/tracker-host +python -m tracker_host --server -v & +# In another terminal: +curl -s http://localhost:8080/api/nodes | python3 -m json.tool +curl -s -X POST http://localhost:8080/api/node/test/config \ + -H 'Content-Type: application/json' \ + -d '{"location": {"rx": {"lat": 33.9}}}' | python3 -m json.tool +curl -s http://localhost:8080/api/nodes | python3 -m json.tool +# Kill the server +kill %1 +``` + +Expected: First curl returns `{"nodes": {}}`. After POST, second curl shows `test` registered. + +**Step 5: Commit** + +```bash +git add tracker_host/__main__.py tracker_host/server_mode.py config.yaml +git commit -m "feat: add --server mode for receiving node track POSTs" +``` + +--- + +## Phase 2: Node-Side Agent + +### Task 5: Create node_agent package + +**Files:** +- Create: `node_agent/__init__.py` +- Create: `node_agent/__main__.py` +- Create: `tests/test_node_agent.py` + +The node agent uses only stdlib (`urllib.request`, `subprocess`, `json`) so it has no extra dependencies beyond Python itself. It can be deployed to a Pi by copying just the `node_agent/` directory. + +**Step 1: Write the failing tests** + +Create `tests/test_node_agent.py`: + +```python +"""Tests for node_agent.""" + +import json +import pytest +from unittest.mock import patch, MagicMock, mock_open +from http.server import HTTPServer, BaseHTTPRequestHandler +import threading + +from node_agent import push_config, post_track_event, fetch_blah2_config + + +class MockHandler(BaseHTTPRequestHandler): + """Simple HTTP handler that records requests.""" + received = [] + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length).decode() + MockHandler.received.append({"path": self.path, "body": body}) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"status": "ok"}).encode()) + + def do_GET(self): + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + config = {"location": {"rx": {"latitude": 33.9}}, "capture": {"fc": 195000000}} + self.wfile.write(json.dumps(config).encode()) + + def log_message(self, format, *args): + pass # Suppress output + + +@pytest.fixture +def mock_server(): + MockHandler.received = [] + server = HTTPServer(("127.0.0.1", 0), MockHandler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever) + thread.daemon = True + thread.start() + yield f"http://127.0.0.1:{port}" + server.shutdown() + + +class TestNodeAgent: + def test_fetch_blah2_config(self, mock_server): + config = fetch_blah2_config(mock_server) + assert config["location"]["rx"]["latitude"] == 33.9 + + def test_push_config(self, mock_server): + config_data = {"location": {"rx": {"latitude": 33.9}}} + push_config(mock_server, "radar3", config_data) + assert len(MockHandler.received) == 1 + assert MockHandler.received[0]["path"] == "/api/node/radar3/config" + assert json.loads(MockHandler.received[0]["body"]) == config_data + + def test_post_track_event(self, mock_server): + event = {"track_id": "t-001", "length": 5} + post_track_event(mock_server, "radar3", json.dumps(event)) + assert len(MockHandler.received) == 1 + assert MockHandler.received[0]["path"] == "/api/node/radar3/tracks" + + def test_post_track_event_failure_does_not_raise(self): + """POST to unreachable server should log but not crash.""" + event = {"track_id": "t-001", "length": 5} + # Should not raise + post_track_event("http://127.0.0.1:1", "radar3", json.dumps(event)) +``` + +**Step 2: Run test to verify it fails** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_agent.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'node_agent'` + +**Step 3: Write minimal implementation** + +Create `node_agent/__init__.py`: + +```python +"""Node agent: runs retina-tracker locally and POSTs track events to central server.""" + +from .agent import fetch_blah2_config, push_config, post_track_event +``` + +Create `node_agent/agent.py`: + +```python +"""Core node agent logic.""" + +import json +import sys +import urllib.error +import urllib.request + + +def fetch_blah2_config(blah2_url: str) -> dict: + """Fetch radar config from local blah2 API.""" + url = f"{blah2_url.rstrip('/')}/api/config" + req = urllib.request.Request(url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=10) as resp: + return json.loads(resp.read()) + + +def push_config(server_url: str, node_name: str, config_data: dict) -> None: + """POST radar config to the central server.""" + url = f"{server_url.rstrip('/')}/api/node/{node_name}/config" + data = json.dumps(config_data).encode() + req = urllib.request.Request( + url, data=data, headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=10) as resp: + resp.read() + + +def post_track_event(server_url: str, node_name: str, event_line: str) -> None: + """POST a single track event to the central server.""" + url = f"{server_url.rstrip('/')}/api/node/{node_name}/tracks" + data = event_line.encode() + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/x-ndjson"}, + ) + try: + with urllib.request.urlopen(req, timeout=5) as resp: + resp.read() + except (urllib.error.URLError, OSError) as e: + print(f"POST failed: {e}", file=sys.stderr) +``` + +**Step 4: Run test to verify it passes** + +Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_agent.py -v` +Expected: ALL PASS + +**Step 5: Commit** + +```bash +git add node_agent/__init__.py node_agent/agent.py tests/test_node_agent.py +git commit -m "feat: add node_agent core HTTP functions" +``` + +--- + +### Task 6: Add node agent CLI and subprocess management + +**Files:** +- Create: `node_agent/__main__.py` + +**Step 1: Write the CLI entry point** + +Create `node_agent/__main__.py`: + +```python +"""CLI entry point: python -m node_agent""" + +import argparse +import signal +import subprocess +import sys +import time + +from .agent import fetch_blah2_config, post_track_event, push_config + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Node agent: run retina-tracker and POST tracks to central server" + ) + parser.add_argument("--node-name", required=True, help="Unique name for this node") + parser.add_argument("--server-url", required=True, help="Central server URL") + parser.add_argument( + "--blah2-url", + default="http://localhost:3000", + help="Local blah2 API URL (default: http://localhost:3000)", + ) + parser.add_argument( + "--tracker-path", + default="../retina-tracker", + help="Path to retina-tracker directory (default: ../retina-tracker)", + ) + parser.add_argument( + "--tcp-port", + type=int, + default=3012, + help="TCP port for retina-tracker (default: 3012)", + ) + parser.add_argument( + "--tracker-config", + help="Optional path to retina-tracker config.yaml", + ) + + args = parser.parse_args() + + # Step 1: Fetch config from blah2 and push to server + print(f"Fetching radar config from {args.blah2_url}...") + try: + config = fetch_blah2_config(args.blah2_url) + print(f"Pushing config to {args.server_url}...") + push_config(args.server_url, args.node_name, config) + print("Config registered with server.") + except Exception as e: + print(f"Warning: Config push failed: {e}", file=sys.stderr) + print("Continuing without config push...", file=sys.stderr) + + # Step 2: Start retina-tracker subprocess + cmd = [ + sys.executable, + "-m", + "tracker.track_detections", + "--tcp", + "--tcp-host", + "0.0.0.0", + "--tcp-port", + str(args.tcp_port), + "-s", + "-", # Stream events to stdout + ] + + if args.tracker_config: + cmd.extend(["-c", args.tracker_config]) + + print(f"Starting retina-tracker on TCP port {args.tcp_port}...") + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=sys.stderr, + cwd=args.tracker_path, + ) + + # Handle shutdown gracefully + def shutdown(signum, frame): + print("\nShutting down...") + proc.terminate() + proc.wait(timeout=5) + sys.exit(0) + + signal.signal(signal.SIGINT, shutdown) + signal.signal(signal.SIGTERM, shutdown) + + print(f"Reading track events, POSTing to {args.server_url}...") + event_count = 0 + + # Step 3: Read stdout and POST events + try: + for line in proc.stdout: + decoded = line.decode().strip() + if decoded: + post_track_event(args.server_url, args.node_name, decoded) + event_count += 1 + if event_count % 50 == 0: + print(f" Posted {event_count} track events") + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + finally: + proc.terminate() + proc.wait(timeout=5) + print(f"Done. Posted {event_count} track events total.") + + +if __name__ == "__main__": + main() +``` + +**Step 2: Test manually (quick sanity check)** + +```bash +cd /opt/apps/tracker-host +python -m node_agent --help +``` + +Expected: Help text printed showing all arguments. + +**Step 3: Commit** + +```bash +git add node_agent/__main__.py +git commit -m "feat: add node_agent CLI with subprocess management" +``` + +--- + +## Phase 3: Testing Infrastructure + +### Task 7: Create detection relay for staging tests + +On this staging machine, blah2 isn't running locally. This relay script polls a remote detection endpoint (e.g. `radar3.retnode.com/api/detection`) and forwards the data to retina-tracker's TCP port, simulating what blah2 would do on a real node. + +**Files:** +- Create: `node_agent/relay.py` + +**Step 1: Write the relay** + +Create `node_agent/relay.py`: + +```python +"""Detection relay: polls HTTP detection endpoint and forwards to retina-tracker TCP. + +Used for testing when blah2 isn't running locally. Simulates the blah2→retina-tracker +TCP connection by polling a remote detection API. + +Usage: + python -m node_agent.relay \ + --detection-url https://radar3.retnode.com/api/detection \ + --tcp-port 3012 +""" + +import argparse +import json +import socket +import sys +import time +import urllib.error +import urllib.request + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Relay: poll detection HTTP endpoint and forward to retina-tracker TCP" + ) + parser.add_argument( + "--detection-url", + default="https://radar3.retnode.com/api/detection", + help="HTTP detection endpoint to poll", + ) + parser.add_argument("--tcp-host", default="127.0.0.1", help="Tracker TCP host") + parser.add_argument("--tcp-port", type=int, default=3012, help="Tracker TCP port") + parser.add_argument( + "--interval", type=float, default=0.5, help="Poll interval in seconds" + ) + + args = parser.parse_args() + + # Connect to retina-tracker TCP port + print(f"Connecting to retina-tracker at {args.tcp_host}:{args.tcp_port}...") + max_retries = 30 + sock = None + for attempt in range(max_retries): + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((args.tcp_host, args.tcp_port)) + break + except ConnectionRefusedError: + sock.close() + sock = None + if attempt < max_retries - 1: + print( + f" Connection refused, retrying ({attempt + 1}/{max_retries})..." + ) + time.sleep(1) + + if sock is None: + print("Failed to connect to retina-tracker", file=sys.stderr) + sys.exit(1) + + print(f"Connected! Polling {args.detection_url} every {args.interval}s") + frame_count = 0 + + try: + while True: + try: + req = urllib.request.Request( + args.detection_url, + headers={"Accept": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=5) as resp: + data = resp.read() + + # Validate it's JSON, then forward as a single line + json.loads(data) # Validate + sock.sendall(data + b"\n") + frame_count += 1 + + if frame_count % 100 == 0: + print(f" Forwarded {frame_count} detection frames") + + except (urllib.error.URLError, OSError, json.JSONDecodeError) as e: + print(f"Error: {e}", file=sys.stderr) + + time.sleep(args.interval) + + except KeyboardInterrupt: + print(f"\nDone. Forwarded {frame_count} detection frames.") + finally: + sock.close() + + +if __name__ == "__main__": + main() +``` + +**Step 2: Test the relay connects (quick sanity check)** + +Don't run the full relay yet (needs retina-tracker running). Just verify it starts: + +```bash +cd /opt/apps/tracker-host +python -m node_agent.relay --help +``` + +Expected: Help text printed. + +**Step 3: Commit** + +```bash +git add node_agent/relay.py +git commit -m "feat: add detection relay for staging tests" +``` + +--- + +### Task 8: End-to-end integration test + +This is a manual test that validates the full pipeline. Run each command in a separate terminal. + +**Terminal 1: Start the server** + +```bash +cd /opt/apps/tracker-host +python -m tracker_host --server -v +``` + +Expected: `Server listening on http://0.0.0.0:8080` + +**Terminal 2: Start the node agent** + +```bash +cd /opt/apps/tracker-host +python -m node_agent \ + --node-name radar3 \ + --server-url http://localhost:8080 \ + --blah2-url https://radar3.retnode.com \ + --tracker-path /opt/apps/retina-tracker \ + --tcp-port 3012 +``` + +Expected: Config pushed, retina-tracker started, waiting for TCP connection. + +**Terminal 3: Start the detection relay** + +```bash +cd /opt/apps/tracker-host +python -m node_agent.relay \ + --detection-url https://radar3.retnode.com/api/detection \ + --tcp-port 3012 +``` + +Expected: Connected, forwarding detection frames. + +**Verify output:** + +```bash +# After 30-60 seconds, check for JSONL output +ls -la output/radar3_*.jsonl +head -1 output/radar3_*.jsonl | python3 -m json.tool +``` + +Expected: JSONL file exists with track events containing `track_id`, `detections`, `timestamp`, etc. + +```bash +# Check server sees the node +curl -s http://localhost:8080/api/nodes | python3 -m json.tool +``` + +Expected: `radar3` listed with `has_config: true` and `active_tracks > 0`. + +**After successful verification, commit a note:** + +```bash +git add -A +git commit -m "feat: complete node-to-server track posting pipeline" +``` + +--- + +## Phase 4: Server-Side Geolocator Integration + +### Task 9: Wire geolocator to received track events + +Once tracks are flowing from nodes to server, the server can run geolocator instances. This reuses the existing `GeolocatorInstance` class. + +**Files:** +- Modify: `tracker_host/node_registry.py` +- Modify: `tracker_host/config.py` +- Create: `tests/test_node_registry_geolocator.py` + +**Step 1: Add geolocator config to server settings** + +In `tracker_host/config.py`, add to `ServerConfig`: + +```python +@dataclass +class ServerConfig: + """HTTP server configuration for receiving track events from nodes.""" + host: str = "0.0.0.0" + port: int = 8080 + geolocator_enabled: bool = False + geolocator_tcp_port_base: int = 31000 +``` + +Update parsing in `_parse_config()`: + +```python +server = ServerConfig( + host=server_raw.get("host", "0.0.0.0"), + port=server_raw.get("port", 8080), + geolocator_enabled=server_raw.get("geolocator_enabled", False), + geolocator_tcp_port_base=server_raw.get("geolocator_tcp_port_base", 31000), +) +``` + +**Step 2: Extend NodeRegistry to manage geolocators** + +When a node registers with config (containing `location.rx` and `location.tx`), and geolocator is enabled, the registry: +1. Saves the radar config to a temp YAML file (same as `GeolocatorInstance._fetch_radar_config` does) +2. Creates a `GeolocatorInstance` for that node +3. Forwards track events to the geolocator + +In `tracker_host/node_registry.py`, add geolocator support: + +```python +import yaml +from pathlib import Path +from .config import Config, GeolocatorConfig +from .geolocator import GeolocatorInstance + +class NodeRegistry: + def __init__(self, output_dir: str, global_config: Optional[Config] = None): + self.output_dir = output_dir + self.global_config = global_config + self._nodes: dict[str, dict[str, Any]] = {} + self._next_geo_port: int = ( + global_config.server.geolocator_tcp_port_base + if global_config + else 31000 + ) + + async def register_node(self, name: str, config_data: dict) -> None: + """Register a node or update its config. Start geolocator if enabled.""" + if name not in self._nodes: + output_handler = OutputHandler(name=name, output_dir=self.output_dir) + self._nodes[name] = { + "config": config_data, + "output_handler": output_handler, + "geolocator": None, + } + logger.info(f"Registered new node: {name}") + else: + self._nodes[name]["config"] = config_data + logger.info(f"Updated config for node: {name}") + + # Start geolocator if enabled and config has location data + if ( + self.global_config + and self.global_config.server.geolocator_enabled + and config_data.get("location") + and self._nodes[name]["geolocator"] is None + ): + await self._start_geolocator(name, config_data) + + async def _start_geolocator(self, name: str, config_data: dict) -> None: + """Start a geolocator instance for a node.""" + port = self._next_geo_port + self._next_geo_port += 1 + + # Save radar config to temp file + config_path = Path(self.output_dir) / f".{name}_config.yml" + with open(config_path, "w") as f: + yaml.dump(config_data, f, default_flow_style=False) + + geo_config = GeolocatorConfig(enabled=True, tcp_port=port) + geo = GeolocatorInstance( + name=name, + geo_config=geo_config, + global_config=self.global_config, + config_url=None, # Config already saved locally + session=None, + ) + # Override the config path since we saved it ourselves + geo._radar_config_path = str(config_path.resolve()) + + try: + # Skip the fetch step (we already have the config) + await geo._start_process() + await geo._connect_tcp() + geo._running = True + geo._output_task = asyncio.create_task( + geo._output_loop(), name=f"{name}-geo-output" + ) + geo._stderr_task = asyncio.create_task( + geo._stderr_loop(), name=f"{name}-geo-stderr" + ) + self._nodes[name]["geolocator"] = geo + logger.info(f"Started geolocator for {name} on port {port}") + except Exception as e: + logger.error(f"Failed to start geolocator for {name}: {e}") + + async def handle_track_event(self, name: str, event_line: str) -> None: + """Handle a track event from a node.""" + if name not in self._nodes: + await self.register_node(name, {}) + + node = self._nodes[name] + await node["output_handler"].handle_event(event_line) + + # Forward to geolocator if running + if node["geolocator"] is not None: + await node["geolocator"].send_track_event(event_line) +``` + +**Step 3: Update server_mode.py to pass global_config** + +```python +registry = NodeRegistry(output_dir=config.output_dir, global_config=config) +``` + +**Step 4: Update config.yaml** + +```yaml +server: + host: "0.0.0.0" + port: 8080 + geolocator_enabled: true + geolocator_tcp_port_base: 31000 +``` + +**Step 5: Test** + +Run the same end-to-end test as Task 8. After tracks flow for 60+ seconds, check for solution output: + +```bash +ls -la output/*solutions*.jsonl +``` + +Expected: Solution JSONL files appear for each node with geolocator enabled. + +**Step 6: Commit** + +```bash +git add tracker_host/config.py tracker_host/node_registry.py tracker_host/server_mode.py config.yaml +git commit -m "feat: add server-side geolocator for received node tracks" +``` + +--- + +## Phase 5: Documentation + +### Task 10: Document architecture and update config.yaml + +**Files:** +- Modify: `config.yaml` (add documentation comments) +- Modify: `README.md` (add server mode and node agent sections) + +**Step 1: Update config.yaml with documentation** + +Add clear comments to `config.yaml` explaining both modes and the dynamic node registration decision. Include example commands. + +**Step 2: Update README.md** + +Add sections for: +- **Server Mode**: How to run with `--server`, what endpoints are available +- **Node Agent**: How to deploy to a Raspberry Pi, example commands +- **Detection Relay**: How to use for staging/testing +- **Architecture**: Brief description of the node→server data flow +- **Architecture Decision: Dynamic Node Registration**: Nodes register by POSTing config, no pre-configuration needed. This simplifies deployment and allows adding/removing nodes without touching the server config. + +**Step 3: Commit** + +```bash +git add config.yaml README.md +git commit -m "docs: add server mode and node agent documentation" +``` + +--- + +## Summary + +| Phase | What | Files | Dependencies | +|-------|------|-------|-------------| +| 1 (Tasks 1-4) | Server HTTP endpoint | `config.py`, `node_registry.py`, `server.py`, `server_mode.py`, `__main__.py` | aiohttp (existing) | +| 2 (Tasks 5-6) | Node agent | `node_agent/agent.py`, `node_agent/__main__.py` | stdlib only | +| 3 (Tasks 7-8) | Testing infrastructure | `node_agent/relay.py` + manual E2E test | stdlib only | +| 4 (Task 9) | Server geolocator | `node_registry.py` (extend), `config.py` | existing GeolocatorInstance | +| 5 (Task 10) | Documentation | `config.yaml`, `README.md` | — | + +**Deployment to a real Pi node requires only:** +1. `retina-tracker/` (already needed) +2. `node_agent/` directory (3 small Python files, no extra deps) +3. One command: `python -m node_agent --node-name radar3 --server-url https://server.example.com` diff --git a/tests/test_node_registry.py b/tests/test_node_registry.py new file mode 100644 index 0000000..e7e1bf1 --- /dev/null +++ b/tests/test_node_registry.py @@ -0,0 +1,52 @@ +"""Tests for NodeRegistry.""" + +import glob +import json +import pytest +from tracker_host.node_registry import NodeRegistry + + +@pytest.fixture +def registry(tmp_path): + return NodeRegistry(output_dir=str(tmp_path)) + + +class TestNodeRegistry: + @pytest.mark.asyncio + async def test_register_node(self, registry): + config_data = {"location": {"rx": {"latitude": 33.9}}} + await registry.register_node("radar3", config_data) + nodes = registry.list_nodes() + assert "radar3" in nodes + assert nodes["radar3"]["has_config"] is True + + @pytest.mark.asyncio + async def test_handle_track_event_writes_jsonl(self, registry, tmp_path): + await registry.register_node("radar3", {"location": {}}) + event = {"track_id": "test-001", "length": 5, "timestamp": 1000} + await registry.handle_track_event("radar3", json.dumps(event)) + + # Check output file was created + files = glob.glob(str(tmp_path / "radar3_*.jsonl")) + assert len(files) == 1 + with open(files[0]) as f: + line = f.readline() + assert json.loads(line)["track_id"] == "test-001" + + @pytest.mark.asyncio + async def test_handle_track_auto_registers(self, registry): + event = {"track_id": "auto-001", "length": 1, "timestamp": 1000} + await registry.handle_track_event("newnode", json.dumps(event)) + nodes = registry.list_nodes() + assert "newnode" in nodes + + @pytest.mark.asyncio + async def test_list_nodes_empty(self, registry): + assert registry.list_nodes() == {} + + @pytest.mark.asyncio + async def test_get_node_config(self, registry): + config_data = {"capture": {"fc": 195000000}} + await registry.register_node("radar3", config_data) + assert registry.get_node_config("radar3") == config_data + assert registry.get_node_config("missing") is None diff --git a/tracker_host/node_registry.py b/tracker_host/node_registry.py new file mode 100644 index 0000000..a144878 --- /dev/null +++ b/tracker_host/node_registry.py @@ -0,0 +1,82 @@ +"""NodeRegistry: dynamic registration and management of radar nodes.""" + +import logging +import re +from dataclasses import dataclass +from typing import Any, Optional + +from .output_handler import OutputHandler + +logger = logging.getLogger(__name__) + +_VALID_NODE_NAME = re.compile(r"^[a-zA-Z0-9_-]+$") + + +@dataclass +class RegisteredNode: + """A registered radar node with its config and output handler.""" + + config: dict[str, Any] + output_handler: OutputHandler + + +class NodeRegistry: + """Manages dynamically registered radar nodes. + + Nodes register by POSTing their config. Each registered node gets + an OutputHandler for writing track events to daily JSONL files. + """ + + def __init__(self, output_dir: str): + self.output_dir = output_dir + self._nodes: dict[str, RegisteredNode] = {} + + async def register_node(self, name: str, config_data: dict[str, Any]) -> None: + """Register a node or update its config.""" + if not _VALID_NODE_NAME.match(name): + raise ValueError( + f"Invalid node name {name!r}: must match [a-zA-Z0-9_-]+" + ) + if name not in self._nodes: + output_handler = OutputHandler( + name=name, + output_dir=self.output_dir, + ) + self._nodes[name] = RegisteredNode( + config=config_data, + output_handler=output_handler, + ) + logger.info(f"Registered new node: {name}") + else: + self._nodes[name].config = config_data + logger.info(f"Updated config for node: {name}") + + async def handle_track_event(self, name: str, event_line: str) -> None: + """Handle a track event from a node.""" + if name not in self._nodes: + await self.register_node(name, {}) + + node = self._nodes[name] + await node.output_handler.handle_event(event_line) + + def get_node_config(self, name: str) -> Optional[dict]: + """Get stored config for a node, or None if not registered.""" + if name in self._nodes: + return self._nodes[name].config + return None + + def list_nodes(self) -> dict[str, Any]: + """List all registered nodes with summary info.""" + return { + name: { + "has_config": bool(node.config), + "active_tracks": node.output_handler.metrics.count, + } + for name, node in self._nodes.items() + } + + async def close(self) -> None: + """Close all output handlers.""" + for node in self._nodes.values(): + await node.output_handler.close() + self._nodes.clear() From 17a35defa6da58dba2197740b0add637758859e1 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:02:25 +0000 Subject: [PATCH 03/12] feat: add HTTP receive server for node track events Co-Authored-By: Claude Opus 4.6 --- tests/test_server.py | 112 +++++++++++++++++++++++++++++++++++++++++ tracker_host/server.py | 71 ++++++++++++++++++++++++++ 2 files changed, 183 insertions(+) create mode 100644 tests/test_server.py create mode 100644 tracker_host/server.py diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..bed480c --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,112 @@ +"""Tests for the HTTP receive server.""" + +import json +import pytest +import pytest_asyncio + +from tracker_host.server import create_app +from tracker_host.node_registry import NodeRegistry + + +@pytest.fixture +def registry(tmp_path): + return NodeRegistry(output_dir=str(tmp_path)) + + +@pytest.fixture +def app(registry): + return create_app(registry) + + +@pytest_asyncio.fixture +async def client(aiohttp_client, app): + return await aiohttp_client(app) + + +class TestReceiveServer: + @pytest.mark.asyncio + async def test_post_config(self, client): + config = {"location": {"rx": {"latitude": 33.9}}} + resp = await client.post( + "/api/node/radar3/config", + json=config, + ) + assert resp.status == 200 + body = await resp.json() + assert body["status"] == "registered" + assert body["node"] == "radar3" + + @pytest.mark.asyncio + async def test_post_track(self, client): + event = {"track_id": "t-001", "length": 5, "timestamp": 1000} + resp = await client.post( + "/api/node/radar3/tracks", + data=json.dumps(event), + headers={"Content-Type": "application/x-ndjson"}, + ) + assert resp.status == 200 + + @pytest.mark.asyncio + async def test_list_nodes_empty(self, client): + resp = await client.get("/api/nodes") + assert resp.status == 200 + body = await resp.json() + assert body["nodes"] == {} + + @pytest.mark.asyncio + async def test_list_nodes_after_registration(self, client): + await client.post("/api/node/radar3/config", json={"location": {}}) + resp = await client.get("/api/nodes") + body = await resp.json() + assert "radar3" in body["nodes"] + + @pytest.mark.asyncio + async def test_post_track_batch(self, client): + """Test posting multiple events as newline-delimited JSON.""" + events = [ + {"track_id": "t-001", "length": 3, "timestamp": 1000}, + {"track_id": "t-002", "length": 5, "timestamp": 1001}, + ] + body = "\n".join(json.dumps(e) for e in events) + resp = await client.post( + "/api/node/radar3/tracks", + data=body, + headers={"Content-Type": "application/x-ndjson"}, + ) + assert resp.status == 200 + + @pytest.mark.asyncio + async def test_post_config_invalid_name(self, client): + """Test that an invalid node name returns 400.""" + resp = await client.post( + "/api/node/bad%20name!/config", + json={"location": {}}, + ) + assert resp.status == 400 + body = await resp.json() + assert body["status"] == "error" + + @pytest.mark.asyncio + async def test_post_track_invalid_name(self, client): + """Test that an invalid node name returns 400 for track events.""" + event = {"track_id": "t-001", "length": 5, "timestamp": 1000} + resp = await client.post( + "/api/node/bad%20name!/tracks", + data=json.dumps(event), + headers={"Content-Type": "application/x-ndjson"}, + ) + assert resp.status == 400 + body = await resp.json() + assert body["status"] == "error" + + @pytest.mark.asyncio + async def test_post_config_invalid_json(self, client): + """Test that invalid JSON body returns 400.""" + resp = await client.post( + "/api/node/radar3/config", + data="not valid json", + headers={"Content-Type": "application/json"}, + ) + assert resp.status == 400 + body = await resp.json() + assert body["status"] == "error" diff --git a/tracker_host/server.py b/tracker_host/server.py new file mode 100644 index 0000000..603a307 --- /dev/null +++ b/tracker_host/server.py @@ -0,0 +1,71 @@ +"""HTTP receive server for accepting track events from remote nodes.""" + +import json +import logging + +from aiohttp import web + +from .node_registry import NodeRegistry + +logger = logging.getLogger(__name__) + + +def create_app(registry: NodeRegistry) -> web.Application: + """Create the aiohttp web application.""" + app = web.Application() + app["registry"] = registry + + app.router.add_post("/api/node/{name}/config", handle_config) + app.router.add_post("/api/node/{name}/tracks", handle_tracks) + app.router.add_get("/api/nodes", handle_list_nodes) + + return app + + +async def handle_config(request: web.Request) -> web.Response: + """Receive and store radar config from a node.""" + name = request.match_info["name"] + registry: NodeRegistry = request.app["registry"] + + try: + config_data = await request.json() + except json.JSONDecodeError: + return web.json_response( + {"status": "error", "message": "invalid JSON"}, status=400 + ) + + try: + await registry.register_node(name, config_data) + except ValueError as exc: + return web.json_response( + {"status": "error", "message": str(exc)}, status=400 + ) + + logger.info(f"Received config from node: {name}") + return web.json_response({"status": "registered", "node": name}) + + +async def handle_tracks(request: web.Request) -> web.Response: + """Receive track events from a node (JSONL: one event per line).""" + name = request.match_info["name"] + registry: NodeRegistry = request.app["registry"] + + body = await request.text() + + try: + for line in body.strip().split("\n"): + line = line.strip() + if line: + await registry.handle_track_event(name, line) + except ValueError as exc: + return web.json_response( + {"status": "error", "message": str(exc)}, status=400 + ) + + return web.json_response({"status": "ok"}) + + +async def handle_list_nodes(request: web.Request) -> web.Response: + """List all registered nodes (for debugging).""" + registry: NodeRegistry = request.app["registry"] + return web.json_response({"nodes": registry.list_nodes()}) From 28c0322141765c54b834d189e79a15aeee372df7 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:07:37 +0000 Subject: [PATCH 04/12] feat: add --server mode for receiving node track POSTs Co-Authored-By: Claude Opus 4.6 --- config.yaml | 5 ++++ tracker_host/__main__.py | 16 +++++++++-- tracker_host/server_mode.py | 55 +++++++++++++++++++++++++++++++++++++ 3 files changed, 74 insertions(+), 2 deletions(-) create mode 100644 tracker_host/server_mode.py diff --git a/config.yaml b/config.yaml index 05a82bf..5f6b029 100644 --- a/config.yaml +++ b/config.yaml @@ -1,5 +1,10 @@ # tracker-host configuration +# Server mode settings (used with --server flag) +server: + host: "0.0.0.0" + port: 8080 + # Global settings output_dir: "./output" poll_interval_sec: 0.5 diff --git a/tracker_host/__main__.py b/tracker_host/__main__.py index 9580149..650221c 100644 --- a/tracker_host/__main__.py +++ b/tracker_host/__main__.py @@ -18,6 +18,8 @@ def main() -> None: python -m tracker_host # Use default config.yaml python -m tracker_host -c myconfig.yaml # Use custom config python -m tracker_host -v # Verbose logging + python -m tracker_host --server # Run in server mode + python -m tracker_host --server -v # Server mode with verbose logging """, ) @@ -35,6 +37,12 @@ def main() -> None: help="Enable verbose (debug) logging", ) + parser.add_argument( + "--server", + action="store_true", + help="Run in server mode (receive tracks from nodes via HTTP)", + ) + args = parser.parse_args() # Check config exists @@ -44,9 +52,13 @@ def main() -> None: print("Create a config.yaml or specify a different path with -c", file=sys.stderr) sys.exit(1) - # Run the manager + # Run the appropriate mode try: - asyncio.run(run_manager(str(config_path), verbose=args.verbose)) + if args.server: + from .server_mode import run_server + asyncio.run(run_server(str(config_path), verbose=args.verbose)) + else: + asyncio.run(run_manager(str(config_path), verbose=args.verbose)) except KeyboardInterrupt: pass diff --git a/tracker_host/server_mode.py b/tracker_host/server_mode.py new file mode 100644 index 0000000..2830707 --- /dev/null +++ b/tracker_host/server_mode.py @@ -0,0 +1,55 @@ +"""Server mode: receive track events from remote nodes via HTTP.""" + +import asyncio +import logging +import signal + +from aiohttp import web + +from .config import load_config +from .node_registry import NodeRegistry +from .server import create_app + +logger = logging.getLogger(__name__) + + +async def run_server(config_path: str, verbose: bool = False) -> None: + """Run tracker-host in server mode.""" + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + config = load_config(config_path) + registry = NodeRegistry(output_dir=config.output_dir) + app = create_app(registry) + + runner = web.AppRunner(app) + await runner.setup() + + site = web.TCPSite(runner, config.server.host, config.server.port) + await site.start() + + logger.info( + f"Server listening on http://{config.server.host}:{config.server.port}" + ) + + # Wait for shutdown signal + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + + def handle_signal(): + logger.info("Received shutdown signal") + stop_event.set() + + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, handle_signal) + + try: + await stop_event.wait() + finally: + await registry.close() + await runner.cleanup() + logger.info("Server stopped") From 3a93df20eab0691863f5dfc8df6e478d039a6ee3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:09:46 +0000 Subject: [PATCH 05/12] feat: add node_agent core HTTP functions Co-Authored-By: Claude Opus 4.6 --- node_agent/__init__.py | 3 ++ node_agent/agent.py | 41 ++++++++++++++++++++++++ tests/test_node_agent.py | 69 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 113 insertions(+) create mode 100644 node_agent/__init__.py create mode 100644 node_agent/agent.py create mode 100644 tests/test_node_agent.py diff --git a/node_agent/__init__.py b/node_agent/__init__.py new file mode 100644 index 0000000..0e952cf --- /dev/null +++ b/node_agent/__init__.py @@ -0,0 +1,3 @@ +"""Node agent: runs retina-tracker locally and POSTs track events to central server.""" + +from .agent import fetch_blah2_config, push_config, post_track_event diff --git a/node_agent/agent.py b/node_agent/agent.py new file mode 100644 index 0000000..f6a8a91 --- /dev/null +++ b/node_agent/agent.py @@ -0,0 +1,41 @@ +"""Core node agent logic.""" + +import json +import sys +import urllib.error +import urllib.request + + +def fetch_blah2_config(blah2_url: str) -> dict: + """Fetch radar config from local blah2 API.""" + url = f"{blah2_url.rstrip('/')}/api/config" + req = urllib.request.Request(url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=10) as resp: + return json.loads(resp.read()) + + +def push_config(server_url: str, node_name: str, config_data: dict) -> None: + """POST radar config to the central server.""" + url = f"{server_url.rstrip('/')}/api/node/{node_name}/config" + data = json.dumps(config_data).encode() + req = urllib.request.Request( + url, data=data, headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=10) as resp: + resp.read() + + +def post_track_event(server_url: str, node_name: str, event_line: str) -> None: + """POST a single track event to the central server.""" + url = f"{server_url.rstrip('/')}/api/node/{node_name}/tracks" + data = event_line.encode() + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/x-ndjson"}, + ) + try: + with urllib.request.urlopen(req, timeout=5) as resp: + resp.read() + except (urllib.error.URLError, OSError) as e: + print(f"POST failed: {e}", file=sys.stderr) diff --git a/tests/test_node_agent.py b/tests/test_node_agent.py new file mode 100644 index 0000000..334ec2b --- /dev/null +++ b/tests/test_node_agent.py @@ -0,0 +1,69 @@ +"""Tests for node_agent.""" + +import json +import pytest +from http.server import HTTPServer, BaseHTTPRequestHandler +import threading + +from node_agent import fetch_blah2_config, push_config, post_track_event + + +class MockHandler(BaseHTTPRequestHandler): + """Simple HTTP handler that records requests.""" + received = [] + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length).decode() + MockHandler.received.append({"path": self.path, "body": body}) + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"status": "ok"}).encode()) + + def do_GET(self): + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + config = {"location": {"rx": {"latitude": 33.9}}, "capture": {"fc": 195000000}} + self.wfile.write(json.dumps(config).encode()) + + def log_message(self, format, *args): + pass # Suppress output + + +@pytest.fixture(autouse=False) +def mock_server(): + MockHandler.received = [] + server = HTTPServer(("127.0.0.1", 0), MockHandler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever) + thread.daemon = True + thread.start() + yield f"http://127.0.0.1:{port}" + server.shutdown() + + +class TestNodeAgent: + def test_fetch_blah2_config(self, mock_server): + config = fetch_blah2_config(mock_server) + assert config["location"]["rx"]["latitude"] == 33.9 + + def test_push_config(self, mock_server): + config_data = {"location": {"rx": {"latitude": 33.9}}} + push_config(mock_server, "radar3", config_data) + assert len(MockHandler.received) == 1 + assert MockHandler.received[0]["path"] == "/api/node/radar3/config" + assert json.loads(MockHandler.received[0]["body"]) == config_data + + def test_post_track_event(self, mock_server): + event = {"track_id": "t-001", "length": 5} + post_track_event(mock_server, "radar3", json.dumps(event)) + assert len(MockHandler.received) == 1 + assert MockHandler.received[0]["path"] == "/api/node/radar3/tracks" + + def test_post_track_event_failure_does_not_raise(self): + """POST to unreachable server should log but not crash.""" + event = {"track_id": "t-001", "length": 5} + # Should not raise + post_track_event("http://127.0.0.1:1", "radar3", json.dumps(event)) From 2eec873ba42ff6e20b242ac4fb604edc5883a865 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:11:46 +0000 Subject: [PATCH 06/12] feat: add node_agent CLI with subprocess management Co-Authored-By: Claude Opus 4.6 --- node_agent/__main__.py | 108 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 node_agent/__main__.py diff --git a/node_agent/__main__.py b/node_agent/__main__.py new file mode 100644 index 0000000..1ff1414 --- /dev/null +++ b/node_agent/__main__.py @@ -0,0 +1,108 @@ +"""CLI entry point: python -m node_agent""" + +import argparse +import signal +import subprocess +import sys +import time + +from .agent import fetch_blah2_config, post_track_event, push_config + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Node agent: run retina-tracker and POST tracks to central server" + ) + parser.add_argument("--node-name", required=True, help="Unique name for this node") + parser.add_argument("--server-url", required=True, help="Central server URL") + parser.add_argument( + "--blah2-url", + default="http://localhost:3000", + help="Local blah2 API URL (default: http://localhost:3000)", + ) + parser.add_argument( + "--tracker-path", + default="../retina-tracker", + help="Path to retina-tracker directory (default: ../retina-tracker)", + ) + parser.add_argument( + "--tcp-port", + type=int, + default=3012, + help="TCP port for retina-tracker (default: 3012)", + ) + parser.add_argument( + "--tracker-config", + help="Optional path to retina-tracker config.yaml", + ) + + args = parser.parse_args() + + # Step 1: Fetch config from blah2 and push to server + print(f"Fetching radar config from {args.blah2_url}...") + try: + config = fetch_blah2_config(args.blah2_url) + print(f"Pushing config to {args.server_url}...") + push_config(args.server_url, args.node_name, config) + print("Config registered with server.") + except Exception as e: + print(f"Warning: Config push failed: {e}", file=sys.stderr) + print("Continuing without config push...", file=sys.stderr) + + # Step 2: Start retina-tracker subprocess + cmd = [ + sys.executable, + "-m", + "tracker.track_detections", + "--tcp", + "--tcp-host", + "0.0.0.0", + "--tcp-port", + str(args.tcp_port), + "-s", + "-", # Stream events to stdout + ] + + if args.tracker_config: + cmd.extend(["-c", args.tracker_config]) + + print(f"Starting retina-tracker on TCP port {args.tcp_port}...") + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=sys.stderr, + cwd=args.tracker_path, + ) + + # Handle shutdown gracefully + def shutdown(signum, frame): + print("\nShutting down...") + proc.terminate() + proc.wait(timeout=5) + sys.exit(0) + + signal.signal(signal.SIGINT, shutdown) + signal.signal(signal.SIGTERM, shutdown) + + print(f"Reading track events, POSTing to {args.server_url}...") + event_count = 0 + + # Step 3: Read stdout and POST events + try: + for line in proc.stdout: + decoded = line.decode().strip() + if decoded: + post_track_event(args.server_url, args.node_name, decoded) + event_count += 1 + if event_count % 50 == 0: + print(f" Posted {event_count} track events") + except Exception as e: + print(f"Error: {e}", file=sys.stderr) + finally: + proc.terminate() + proc.wait(timeout=5) + print(f"Done. Posted {event_count} track events total.") + + +if __name__ == "__main__": + main() From c4a90c8a1c45dabf9ca3d296d2fe393371631e32 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:13:29 +0000 Subject: [PATCH 07/12] feat: add detection relay for staging tests Co-Authored-By: Claude Opus 4.6 --- node_agent/relay.py | 93 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) create mode 100644 node_agent/relay.py diff --git a/node_agent/relay.py b/node_agent/relay.py new file mode 100644 index 0000000..033af68 --- /dev/null +++ b/node_agent/relay.py @@ -0,0 +1,93 @@ +"""Detection relay: polls HTTP detection endpoint and forwards to retina-tracker TCP. + +Used for testing when blah2 isn't running locally. Simulates the blah2->retina-tracker +TCP connection by polling a remote detection API. + +Usage: + python -m node_agent.relay \ + --detection-url https://radar3.retnode.com/api/detection \ + --tcp-port 3012 +""" + +import argparse +import json +import socket +import sys +import time +import urllib.error +import urllib.request + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Relay: poll detection HTTP endpoint and forward to retina-tracker TCP" + ) + parser.add_argument( + "--detection-url", + default="https://radar3.retnode.com/api/detection", + help="HTTP detection endpoint to poll", + ) + parser.add_argument("--tcp-host", default="127.0.0.1", help="Tracker TCP host") + parser.add_argument("--tcp-port", type=int, default=3012, help="Tracker TCP port") + parser.add_argument( + "--interval", type=float, default=0.5, help="Poll interval in seconds" + ) + + args = parser.parse_args() + + # Connect to retina-tracker TCP port + print(f"Connecting to retina-tracker at {args.tcp_host}:{args.tcp_port}...") + max_retries = 30 + sock = None + for attempt in range(max_retries): + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((args.tcp_host, args.tcp_port)) + break + except ConnectionRefusedError: + sock.close() + sock = None + if attempt < max_retries - 1: + print( + f" Connection refused, retrying ({attempt + 1}/{max_retries})..." + ) + time.sleep(1) + + if sock is None: + print("Failed to connect to retina-tracker", file=sys.stderr) + sys.exit(1) + + print(f"Connected! Polling {args.detection_url} every {args.interval}s") + frame_count = 0 + + try: + while True: + try: + req = urllib.request.Request( + args.detection_url, + headers={"Accept": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=5) as resp: + data = resp.read() + + # Validate it's JSON, then forward as a single line + json.loads(data) # Validate + sock.sendall(data + b"\n") + frame_count += 1 + + if frame_count % 100 == 0: + print(f" Forwarded {frame_count} detection frames") + + except (urllib.error.URLError, OSError, json.JSONDecodeError) as e: + print(f"Error: {e}", file=sys.stderr) + + time.sleep(args.interval) + + except KeyboardInterrupt: + print(f"\nDone. Forwarded {frame_count} detection frames.") + finally: + sock.close() + + +if __name__ == "__main__": + main() From bcdee30d3ce5745d0b6ffcdbae3445e3c55edb1b Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:37:15 +0000 Subject: [PATCH 08/12] fix: add User-Agent header to avoid Cloudflare 403 blocks Python's default urllib User-Agent is blocked by Cloudflare Zero Trust. Set User-Agent to retina-node-agent/1.0 on all outgoing HTTP requests. Co-Authored-By: Claude Opus 4.6 --- node_agent/agent.py | 8 +++++--- node_agent/relay.py | 4 +++- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/node_agent/agent.py b/node_agent/agent.py index f6a8a91..772d0b2 100644 --- a/node_agent/agent.py +++ b/node_agent/agent.py @@ -5,11 +5,13 @@ import urllib.error import urllib.request +_HEADERS = {"User-Agent": "retina-node-agent/1.0"} + def fetch_blah2_config(blah2_url: str) -> dict: """Fetch radar config from local blah2 API.""" url = f"{blah2_url.rstrip('/')}/api/config" - req = urllib.request.Request(url, headers={"Accept": "application/json"}) + req = urllib.request.Request(url, headers={**_HEADERS, "Accept": "application/json"}) with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read()) @@ -19,7 +21,7 @@ def push_config(server_url: str, node_name: str, config_data: dict) -> None: url = f"{server_url.rstrip('/')}/api/node/{node_name}/config" data = json.dumps(config_data).encode() req = urllib.request.Request( - url, data=data, headers={"Content-Type": "application/json"} + url, data=data, headers={**_HEADERS, "Content-Type": "application/json"} ) with urllib.request.urlopen(req, timeout=10) as resp: resp.read() @@ -32,7 +34,7 @@ def post_track_event(server_url: str, node_name: str, event_line: str) -> None: req = urllib.request.Request( url, data=data, - headers={"Content-Type": "application/x-ndjson"}, + headers={**_HEADERS, "Content-Type": "application/x-ndjson"}, ) try: with urllib.request.urlopen(req, timeout=5) as resp: diff --git a/node_agent/relay.py b/node_agent/relay.py index 033af68..e2f5e5e 100644 --- a/node_agent/relay.py +++ b/node_agent/relay.py @@ -17,6 +17,8 @@ import urllib.error import urllib.request +_HEADERS = {"User-Agent": "retina-node-agent/1.0"} + def main() -> None: parser = argparse.ArgumentParser( @@ -65,7 +67,7 @@ def main() -> None: try: req = urllib.request.Request( args.detection_url, - headers={"Accept": "application/json"}, + headers={**_HEADERS, "Accept": "application/json"}, ) with urllib.request.urlopen(req, timeout=5) as resp: data = resp.read() From d56855ecaef41c3fe8ef2824f61c7b2f6c7d2386 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:41:09 +0000 Subject: [PATCH 09/12] feat: wire geolocator to server-mode node registry Co-Authored-By: Claude Opus 4.6 --- tests/test_config.py | 15 +++ tests/test_node_registry_geolocator.py | 165 +++++++++++++++++++++++++ tracker_host/config.py | 4 + tracker_host/geolocator.py | 3 + tracker_host/node_registry.py | 72 ++++++++++- tracker_host/server_mode.py | 2 +- 6 files changed, 257 insertions(+), 4 deletions(-) create mode 100644 tests/test_node_registry_geolocator.py diff --git a/tests/test_config.py b/tests/test_config.py index fe1f03e..1b645b0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -92,3 +92,18 @@ def test_server_custom(self): config = _parse_config(raw) assert config.server.host == "127.0.0.1" assert config.server.port == 9090 + + def test_server_geolocator_defaults(self): + config = _parse_config({}) + assert config.server.geolocator_enabled is False + assert config.server.geolocator_tcp_port_base == 31000 + + def test_server_geolocator_custom(self): + config = _parse_config({ + "server": { + "geolocator_enabled": True, + "geolocator_tcp_port_base": 32000, + } + }) + assert config.server.geolocator_enabled is True + assert config.server.geolocator_tcp_port_base == 32000 diff --git a/tests/test_node_registry_geolocator.py b/tests/test_node_registry_geolocator.py new file mode 100644 index 0000000..765bbff --- /dev/null +++ b/tests/test_node_registry_geolocator.py @@ -0,0 +1,165 @@ +"""Tests for NodeRegistry geolocator integration.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from tracker_host.config import Config, ServerConfig +from tracker_host.node_registry import NodeRegistry + + +def _make_config(geolocator_enabled=True, port_base=31000): + """Create a Config with geolocator settings.""" + return Config( + output_dir="/tmp/test-geo-output", + server=ServerConfig( + host="0.0.0.0", + port=8080, + geolocator_enabled=geolocator_enabled, + geolocator_tcp_port_base=port_base, + ), + ) + + +class TestNodeRegistryGeolocator: + @pytest.fixture + def registry_no_geo(self, tmp_path): + config = _make_config(geolocator_enabled=False) + config.output_dir = str(tmp_path) + return NodeRegistry(output_dir=str(tmp_path), global_config=config) + + @pytest.fixture + def registry_with_geo(self, tmp_path): + config = _make_config(geolocator_enabled=True) + config.output_dir = str(tmp_path) + return NodeRegistry(output_dir=str(tmp_path), global_config=config) + + @pytest.mark.asyncio + async def test_no_geolocator_when_disabled(self, registry_no_geo): + """No geolocator started when geolocator_enabled is False.""" + await registry_no_geo.register_node( + "node1", {"location": {"rx": {}, "tx": {}}} + ) + assert registry_no_geo._nodes["node1"].geolocator is None + await registry_no_geo.close() + + @pytest.mark.asyncio + async def test_no_geolocator_without_location(self, registry_with_geo): + """No geolocator started when config lacks location data.""" + await registry_with_geo.register_node("node1", {"some": "config"}) + assert registry_with_geo._nodes["node1"].geolocator is None + await registry_with_geo.close() + + @pytest.mark.asyncio + @patch("tracker_host.node_registry.GeolocatorInstance") + async def test_geolocator_started_with_location( + self, MockGeo, registry_with_geo + ): + """Geolocator started when enabled and config has location.""" + mock_geo = AsyncMock() + MockGeo.return_value = mock_geo + mock_geo._radar_config_path = None + + config_data = { + "location": { + "rx": {"lat": 1, "lon": 2}, + "tx": {"lat": 3, "lon": 4}, + } + } + await registry_with_geo.register_node("node1", config_data) + + mock_geo.start.assert_awaited_once() + assert registry_with_geo._nodes["node1"].geolocator is mock_geo + await registry_with_geo.close() + + @pytest.mark.asyncio + @patch("tracker_host.node_registry.GeolocatorInstance") + async def test_track_event_forwarded_to_geolocator( + self, MockGeo, registry_with_geo + ): + """Track events forwarded to geolocator when running.""" + mock_geo = AsyncMock() + MockGeo.return_value = mock_geo + mock_geo._radar_config_path = None + + config_data = { + "location": { + "rx": {"lat": 1, "lon": 2}, + "tx": {"lat": 3, "lon": 4}, + } + } + await registry_with_geo.register_node("node1", config_data) + + event = '{"track_id": "t1"}' + await registry_with_geo.handle_track_event("node1", event) + + mock_geo.send_track_event.assert_awaited_once_with(event) + await registry_with_geo.close() + + @pytest.mark.asyncio + @patch("tracker_host.node_registry.GeolocatorInstance") + async def test_geolocator_stopped_on_close( + self, MockGeo, registry_with_geo + ): + """Geolocator stopped when registry is closed.""" + mock_geo = AsyncMock() + MockGeo.return_value = mock_geo + mock_geo._radar_config_path = None + + config_data = {"location": {"rx": {}, "tx": {}}} + await registry_with_geo.register_node("node1", config_data) + + await registry_with_geo.close() + mock_geo.stop.assert_awaited_once() + + @pytest.mark.asyncio + @patch("tracker_host.node_registry.GeolocatorInstance") + async def test_geolocator_port_increments( + self, MockGeo, registry_with_geo + ): + """Each node gets a unique geolocator port.""" + mock_geo = AsyncMock() + MockGeo.return_value = mock_geo + mock_geo._radar_config_path = None + + await registry_with_geo.register_node( + "node1", {"location": {"rx": {}, "tx": {}}} + ) + await registry_with_geo.register_node( + "node2", {"location": {"rx": {}, "tx": {}}} + ) + + calls = MockGeo.call_args_list + port1 = calls[0].kwargs["geo_config"].tcp_port + port2 = calls[1].kwargs["geo_config"].tcp_port + assert port1 == 31000 + assert port2 == 31001 + await registry_with_geo.close() + + @pytest.mark.asyncio + @patch("tracker_host.node_registry.GeolocatorInstance") + async def test_geolocator_start_failure_logged( + self, MockGeo, registry_with_geo + ): + """Geolocator start failure is logged, not raised.""" + mock_geo = AsyncMock() + MockGeo.return_value = mock_geo + mock_geo._radar_config_path = None + mock_geo.start.side_effect = RuntimeError("port in use") + + config_data = {"location": {"rx": {}, "tx": {}}} + await registry_with_geo.register_node("node1", config_data) + + # Should not raise + assert registry_with_geo._nodes["node1"].geolocator is None + await registry_with_geo.close() + + @pytest.mark.asyncio + async def test_no_geolocator_without_global_config(self, tmp_path): + """No geolocator when no global_config provided.""" + registry = NodeRegistry(output_dir=str(tmp_path)) + await registry.register_node( + "node1", {"location": {"rx": {}, "tx": {}}} + ) + assert registry._nodes["node1"].geolocator is None + await registry.close() diff --git a/tracker_host/config.py b/tracker_host/config.py index 151a7be..b187fcb 100644 --- a/tracker_host/config.py +++ b/tracker_host/config.py @@ -13,6 +13,8 @@ class ServerConfig: host: str = "0.0.0.0" port: int = 8080 + geolocator_enabled: bool = False + geolocator_tcp_port_base: int = 31000 @dataclass @@ -95,6 +97,8 @@ def _parse_config(raw: dict) -> Config: server = ServerConfig( host=server_raw.get("host", "0.0.0.0"), port=server_raw.get("port", 8080), + geolocator_enabled=server_raw.get("geolocator_enabled", False), + geolocator_tcp_port_base=server_raw.get("geolocator_tcp_port_base", 31000), ) retry_raw = raw.get("retry", {}) diff --git a/tracker_host/geolocator.py b/tracker_host/geolocator.py index c070c46..06a00cb 100644 --- a/tracker_host/geolocator.py +++ b/tracker_host/geolocator.py @@ -117,6 +117,9 @@ async def send_track_event(self, line: str) -> None: async def _fetch_radar_config(self) -> None: """Fetch the radar's config from its config_url and save to a temp file.""" + if self._radar_config_path: + return # Config already provided externally + if not self.config_url: raise RuntimeError( f"No config_url for {self.name}, required for geolocator" diff --git a/tracker_host/node_registry.py b/tracker_host/node_registry.py index a144878..a6844c5 100644 --- a/tracker_host/node_registry.py +++ b/tracker_host/node_registry.py @@ -1,10 +1,16 @@ """NodeRegistry: dynamic registration and management of radar nodes.""" +import asyncio import logging import re -from dataclasses import dataclass +from dataclasses import dataclass, field +from pathlib import Path from typing import Any, Optional +import yaml + +from .config import Config, GeolocatorConfig +from .geolocator import GeolocatorInstance from .output_handler import OutputHandler logger = logging.getLogger(__name__) @@ -18,6 +24,7 @@ class RegisteredNode: config: dict[str, Any] output_handler: OutputHandler + geolocator: Optional[GeolocatorInstance] = None class NodeRegistry: @@ -25,11 +32,26 @@ class NodeRegistry: Nodes register by POSTing their config. Each registered node gets an OutputHandler for writing track events to daily JSONL files. + When geolocator is enabled, nodes with location data also get a + GeolocatorInstance for converting tracks to geographic solutions. """ - def __init__(self, output_dir: str): + def __init__(self, output_dir: str, global_config: Optional[Config] = None): self.output_dir = output_dir + self.global_config = global_config self._nodes: dict[str, RegisteredNode] = {} + self._next_geo_port: int = ( + global_config.server.geolocator_tcp_port_base + if global_config and global_config.server + else 31000 + ) + + @property + def _geolocator_enabled(self) -> bool: + return ( + self.global_config is not None + and self.global_config.server.geolocator_enabled + ) async def register_node(self, name: str, config_data: dict[str, Any]) -> None: """Register a node or update its config.""" @@ -51,6 +73,44 @@ async def register_node(self, name: str, config_data: dict[str, Any]) -> None: self._nodes[name].config = config_data logger.info(f"Updated config for node: {name}") + # Start geolocator if enabled and config has location data + if ( + self._geolocator_enabled + and config_data.get("location") + and self._nodes[name].geolocator is None + ): + await self._start_geolocator(name, config_data) + + async def _start_geolocator(self, name: str, config_data: dict) -> None: + """Start a geolocator instance for a node.""" + port = self._next_geo_port + self._next_geo_port += 1 + + # Save radar config to file so geolocator can read it + output_path = Path(self.output_dir) + output_path.mkdir(parents=True, exist_ok=True) + config_path = output_path / f".{name}_radar_config.yml" + with open(config_path, "w") as f: + yaml.dump(config_data, f, default_flow_style=False) + + geo_config = GeolocatorConfig(enabled=True, tcp_port=port) + geo = GeolocatorInstance( + name=name, + geo_config=geo_config, + global_config=self.global_config, + config_url=None, + session=None, + ) + # Provide the saved config path so _fetch_radar_config is skipped + geo._radar_config_path = str(config_path.resolve()) + + try: + await geo.start() + self._nodes[name].geolocator = geo + logger.info(f"Started geolocator for {name} on port {port}") + except Exception as e: + logger.error(f"Failed to start geolocator for {name}: {e}") + async def handle_track_event(self, name: str, event_line: str) -> None: """Handle a track event from a node.""" if name not in self._nodes: @@ -59,6 +119,10 @@ async def handle_track_event(self, name: str, event_line: str) -> None: node = self._nodes[name] await node.output_handler.handle_event(event_line) + # Forward to geolocator if running + if node.geolocator is not None: + await node.geolocator.send_track_event(event_line) + def get_node_config(self, name: str) -> Optional[dict]: """Get stored config for a node, or None if not registered.""" if name in self._nodes: @@ -76,7 +140,9 @@ def list_nodes(self) -> dict[str, Any]: } async def close(self) -> None: - """Close all output handlers.""" + """Close all output handlers and stop geolocators.""" for node in self._nodes.values(): + if node.geolocator is not None: + await node.geolocator.stop() await node.output_handler.close() self._nodes.clear() diff --git a/tracker_host/server_mode.py b/tracker_host/server_mode.py index 2830707..ca635b2 100644 --- a/tracker_host/server_mode.py +++ b/tracker_host/server_mode.py @@ -23,7 +23,7 @@ async def run_server(config_path: str, verbose: bool = False) -> None: ) config = load_config(config_path) - registry = NodeRegistry(output_dir=config.output_dir) + registry = NodeRegistry(output_dir=config.output_dir, global_config=config) app = create_app(registry) runner = web.AppRunner(app) From 58c441a191e20365bd2177af99da25f3a0716644 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:43:41 +0000 Subject: [PATCH 10/12] fix: remove unused imports, add geolocator settings to config.yaml Post-review cleanup for Task 9: remove unused asyncio and field imports from node_registry.py, add geolocator_enabled and geolocator_tcp_port_base to config.yaml for discoverability. Co-Authored-By: Claude Opus 4.6 --- config.yaml | 2 ++ tracker_host/node_registry.py | 3 +-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/config.yaml b/config.yaml index 5f6b029..831f3a0 100644 --- a/config.yaml +++ b/config.yaml @@ -4,6 +4,8 @@ server: host: "0.0.0.0" port: 8080 + geolocator_enabled: false + geolocator_tcp_port_base: 31000 # Global settings output_dir: "./output" diff --git a/tracker_host/node_registry.py b/tracker_host/node_registry.py index a6844c5..263ae61 100644 --- a/tracker_host/node_registry.py +++ b/tracker_host/node_registry.py @@ -1,9 +1,8 @@ """NodeRegistry: dynamic registration and management of radar nodes.""" -import asyncio import logging import re -from dataclasses import dataclass, field +from dataclasses import dataclass from pathlib import Path from typing import Any, Optional From 9090847e502e26d04536833c78987d1ebb31dd53 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 01:45:59 +0000 Subject: [PATCH 11/12] docs: add server mode and node agent documentation Co-Authored-By: Claude Opus 4.6 --- README.md | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 174 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index a046e33..1722d4d 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,28 @@ # tracker-host -Multi-instance tracker manager for [retina-tracker](https://github.com/offworldlabs/retina-tracker). Fetches detection data from multiple radar endpoints and manages tracker subprocesses. +Multi-instance tracker manager for [retina-tracker](https://github.com/offworldlabs/retina-tracker). Supports two operating modes: **manager mode** (polls remote detection endpoints centrally) and **server mode** (receives track events pushed from remote nodes). ## Overview -tracker-host is a Python service that manages multiple retina-tracker instances concurrently. For each configured detection endpoint, it: +tracker-host is a Python service that manages multiple retina-tracker instances. It operates in one of two modes: + +### Manager Mode (default) + +The server polls detection endpoints on remote radar nodes and manages tracker subprocesses locally. For each configured detection endpoint, it: 1. Fetches detection data from the endpoint at ~2 Hz 2. Feeds data to a dedicated retina-tracker subprocess via TCP 3. Reads tracker output (streaming JSONL) and saves to daily files 4. Optionally forwards track events to a configurable API in real-time +### Server Mode (`--server`) + +Remote radar nodes run retina-tracker locally and POST track events to the central server via HTTP. The server receives tracks, writes them to JSONL files, and optionally runs geolocator instances to convert tracks to geographic solutions. Nodes self-register by POSTing their config -- no pre-configuration is needed on the server side. + ## Architecture +### Manager Mode + ``` ┌─────────────────────────────────────────────────────────────┐ │ tracker-host │ @@ -38,6 +48,30 @@ tracker-host is a Python service that manages multiple retina-tracker instances └─────────────────────────────────────────────────────────────┘ ``` +### Server Mode + +``` +Node (Raspberry Pi) Central Server +┌───────────────────────────┐ ┌──────────────────────────────────┐ +│ │ │ tracker-host --server │ +│ blah2 (radar) │ │ │ +│ │ │ │ ┌────────────┐ │ +│ ▼ TCP │ │ │ HTTP Server │ :8080 │ +│ retina-tracker │ │ │ (aiohttp) │ │ +│ │ │ HTTP POST │ └─────┬──────┘ │ +│ ▼ stdout JSONL │ ──────────── │ │ │ +│ node_agent │ /api/node/ │ ▼ │ +│ (POST tracks + config) │ {name}/... │ ┌──────────────┐ │ +│ │ │ │ NodeRegistry │ │ +└───────────────────────────┘ │ └──┬───────┬───┘ │ + │ │ │ │ + │ ▼ ▼ │ + │ Output GeolocatorInstance │ + │ Handler (optional lat/lon/alt) │ + │ (JSONL) │ + └──────────────────────────────────┘ +``` + ## Installation ```bash @@ -53,6 +87,8 @@ pip install -r requirements.txt ## Usage +### Manager Mode (default) + ```bash # Run with default config python -m tracker_host @@ -64,6 +100,132 @@ python -m tracker_host -c /path/to/config.yaml python -m tracker_host -v ``` +### Server Mode + +Start the server to receive track events from remote nodes: + +```bash +# Run in server mode (listens on 0.0.0.0:8080 by default) +python -m tracker_host --server + +# Server mode with verbose logging +python -m tracker_host --server -v + +# Server mode with custom config +python -m tracker_host --server -c /path/to/config.yaml +``` + +Server mode configuration is specified in the `server` section of `config.yaml`: + +```yaml +server: + host: "0.0.0.0" + port: 8080 + geolocator_enabled: false + geolocator_tcp_port_base: 31000 +``` + +| Setting | Default | Description | +|---------|---------|-------------| +| `host` | `0.0.0.0` | Bind address for the HTTP server | +| `port` | `8080` | Port for the HTTP server | +| `geolocator_enabled` | `false` | Start geolocator instances for nodes that provide location data | +| `geolocator_tcp_port_base` | `31000` | Starting port number for geolocator TCP connections (auto-increments per node) | + +#### API Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| `POST` | `/api/node/{name}/config` | Register a node and push its radar config (JSON body) | +| `POST` | `/api/node/{name}/tracks` | Push track events from a node (JSONL body, one event per line) | +| `GET` | `/api/nodes` | List all registered nodes with status info | + +The `{name}` parameter must match `[a-zA-Z0-9_-]+`. + +Example: register a node and push tracks: + +```bash +# Push radar config for a node +curl -X POST http://server:8080/api/node/radar3/config \ + -H "Content-Type: application/json" \ + -d '{"location": {"rx_latitude": -36.8, "rx_longitude": 174.7, "rx_altitude": 50}}' + +# Push a track event +curl -X POST http://server:8080/api/node/radar3/tracks \ + -H "Content-Type: application/x-ndjson" \ + -d '{"track_id":"260120-000000","timestamp":1768932173636,"length":3,"detections":[...]}' + +# List registered nodes +curl http://server:8080/api/nodes +``` + +### Node Agent + +The `node_agent/` package runs on Raspberry Pi 5 nodes (or any machine with blah2 and retina-tracker). It is a **stdlib-only** Python package with no external dependencies, designed for easy deployment to resource-constrained devices. + +The node agent: + +1. Fetches the radar config from the local blah2 instance and pushes it to the central server +2. Starts retina-tracker as a subprocess, listening for detections on a TCP port +3. Reads track events from retina-tracker stdout (streaming JSONL) and POSTs each event to the central server + +#### Deploying to a Raspberry Pi + +Copy the `node_agent/` directory and ensure `retina-tracker` is available on the node: + +```bash +# On the Pi (assuming retina-tracker is at ../retina-tracker) +python3 -m node_agent \ + --node-name radar3 \ + --server-url http://central-server:8080 \ + --blah2-url http://localhost:3000 \ + --tracker-path ../retina-tracker \ + --tcp-port 3012 +``` + +| Flag | Default | Description | +|------|---------|-------------| +| `--node-name` | (required) | Unique name for this node (used in API paths and output filenames) | +| `--server-url` | (required) | URL of the central tracker-host server | +| `--blah2-url` | `http://localhost:3000` | URL of the local blah2 radar API | +| `--tracker-path` | `../retina-tracker` | Path to the retina-tracker directory | +| `--tcp-port` | `3012` | TCP port for the retina-tracker subprocess | +| `--tracker-config` | (none) | Optional path to a retina-tracker config.yaml | + +The node agent uses only the Python standard library (`urllib`, `subprocess`, `socket`, `json`), so no `pip install` step is needed on the Pi. + +### Detection Relay (staging/testing) + +The detection relay (`node_agent/relay.py`) is for staging and testing when blah2 is not running locally on the node. It polls a remote HTTP detection endpoint and forwards the data to retina-tracker over TCP, simulating the blah2-to-retina-tracker connection. + +```bash +# Start retina-tracker first (in another terminal), then run the relay: +python3 -m node_agent.relay \ + --detection-url https://radar3.retnode.com/api/detection \ + --tcp-host 127.0.0.1 \ + --tcp-port 3012 \ + --interval 0.5 +``` + +| Flag | Default | Description | +|------|---------|-------------| +| `--detection-url` | `https://radar3.retnode.com/api/detection` | HTTP detection endpoint to poll | +| `--tcp-host` | `127.0.0.1` | retina-tracker TCP host | +| `--tcp-port` | `3012` | retina-tracker TCP port | +| `--interval` | `0.5` | Poll interval in seconds | + +The relay retries the TCP connection up to 30 times (1 second apart), so you can start it before retina-tracker is ready. + +### Architecture Decision: Dynamic Node Registration + +Nodes self-register with the central server by POSTing their config to `/api/node/{name}/config`. This means: + +- **No pre-configuration needed on the server.** The server does not need a list of nodes in its config file. Nodes appear automatically when they connect. +- **Adding or removing nodes requires no server restart.** Deploy a new Pi, point it at the server, and it registers itself. +- **Each node controls its own identity** via the `--node-name` flag. + +If a node POSTs config again, the existing registration is updated rather than duplicated. + ## Configuration Edit `config.yaml` to configure tracker instances: @@ -141,12 +303,21 @@ python -m tracker_host.plotter output/radar3_2026-01-20.jsonl --min-length 10 ## Features +### Manager Mode - **Multi-instance management**: Run multiple tracker instances from a single process - **Resilience**: Exponential backoff on failures, auto-restart after extended outages - **Port collision detection**: Checks port availability before starting trackers -- **Daily output files**: Automatic file rotation at midnight - **Real-time status**: Periodic console output showing track counts and lengths - **API forwarding**: Optional real-time forwarding of track events to external APIs + +### Server Mode +- **Dynamic node registration**: Nodes self-register by POSTing config; no server-side pre-configuration +- **HTTP track ingestion**: Receives JSONL track events via POST from remote nodes +- **Automatic geolocator**: Optionally starts geolocator instances for nodes with location data +- **Node listing API**: Query registered nodes and their status via REST + +### Shared +- **Daily output files**: Automatic file rotation at midnight - **Graceful shutdown**: Clean termination on SIGINT/SIGTERM ## Requirements From cab3b2356279714eeb66cc42acac14f629094256 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 21 Feb 2026 00:25:26 +0000 Subject: [PATCH 12/12] chore: remove implementation plan from PR Co-Authored-By: Claude Opus 4.6 --- ...2026-02-16-node-to-server-track-posting.md | 1265 ----------------- 1 file changed, 1265 deletions(-) delete mode 100644 docs/plans/2026-02-16-node-to-server-track-posting.md diff --git a/docs/plans/2026-02-16-node-to-server-track-posting.md b/docs/plans/2026-02-16-node-to-server-track-posting.md deleted file mode 100644 index dfb8b09..0000000 --- a/docs/plans/2026-02-16-node-to-server-track-posting.md +++ /dev/null @@ -1,1265 +0,0 @@ -# Node-to-Server Track Posting Implementation Plan - -> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. - -**Goal:** Replace the current server-polls-nodes architecture with nodes running retina-tracker locally and POSTing track events to a central server via HTTP. - -**Architecture:** A thin `node_agent` package runs on each Raspberry Pi node alongside retina-tracker. It starts retina-tracker as a subprocess (which receives detections from blah2 via TCP), reads track events from stdout, and POSTs them to the central `tracker-host` server. The server receives tracks via new aiohttp endpoints, writes them to JSONL files (reusing the existing `OutputHandler`), and runs the geolocator centrally for better multi-radar fusion accuracy. Nodes register dynamically by pushing their blah2 config on startup. - -**Tech Stack:** Python 3.10+, aiohttp (server-side HTTP), urllib.request (node-side HTTP, stdlib only), existing retina-tracker and tracker-host infrastructure. - -**Key architectural decision:** Dynamic node registration (no pre-configuration needed on the server). Nodes push their radar config to the server, and the server creates per-node output handlers on the fly. - ---- - -## Phase 1: Server-Side HTTP Receive Endpoint - -### Task 1: Add ServerConfig to config parsing - -**Files:** -- Modify: `tracker_host/config.py` -- Modify: `tests/test_config.py` - -**Step 1: Write the failing test** - -Add to `tests/test_config.py`: - -```python -class TestServerConfig: - def test_server_defaults(self): - config = _parse_config({}) - assert config.server.host == "0.0.0.0" - assert config.server.port == 8080 - - def test_server_custom(self): - raw = {"server": {"host": "127.0.0.1", "port": 9090}} - config = _parse_config(raw) - assert config.server.host == "127.0.0.1" - assert config.server.port == 9090 -``` - -**Step 2: Run test to verify it fails** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_config.py::TestServerConfig -v` -Expected: FAIL with `AttributeError: 'Config' object has no attribute 'server'` - -**Step 3: Write minimal implementation** - -Add to `tracker_host/config.py`: - -```python -@dataclass -class ServerConfig: - """HTTP server configuration for receiving track events from nodes.""" - host: str = "0.0.0.0" - port: int = 8080 -``` - -Add `server` field to `Config` dataclass: -```python -server: ServerConfig = field(default_factory=ServerConfig) -``` - -Add parsing in `_parse_config()` before the `return Config(...)`: -```python -server_raw = raw.get("server", {}) -server = ServerConfig( - host=server_raw.get("host", "0.0.0.0"), - port=server_raw.get("port", 8080), -) -``` - -And add `server=server` to the `return Config(...)` call. - -**Step 4: Run test to verify it passes** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_config.py -v` -Expected: ALL PASS - -**Step 5: Commit** - -```bash -git add tracker_host/config.py tests/test_config.py -git commit -m "feat: add ServerConfig for HTTP receive endpoint" -``` - ---- - -### Task 2: Create NodeRegistry for dynamic node management - -**Files:** -- Create: `tracker_host/node_registry.py` -- Create: `tests/test_node_registry.py` - -**Step 1: Write the failing tests** - -Create `tests/test_node_registry.py`: - -```python -"""Tests for NodeRegistry.""" - -import json -import pytest -from unittest.mock import AsyncMock, patch, MagicMock -from tracker_host.node_registry import NodeRegistry - - -@pytest.fixture -def registry(tmp_path): - return NodeRegistry(output_dir=str(tmp_path)) - - -class TestNodeRegistry: - @pytest.mark.asyncio - async def test_register_node(self, registry): - config_data = {"location": {"rx": {"latitude": 33.9}}} - await registry.register_node("radar3", config_data) - nodes = registry.list_nodes() - assert "radar3" in nodes - assert nodes["radar3"]["has_config"] is True - - @pytest.mark.asyncio - async def test_handle_track_event_writes_jsonl(self, registry, tmp_path): - await registry.register_node("radar3", {"location": {}}) - event = {"track_id": "test-001", "length": 5, "timestamp": 1000} - await registry.handle_track_event("radar3", json.dumps(event)) - - # Check output file was created - import glob - files = glob.glob(str(tmp_path / "radar3_*.jsonl")) - assert len(files) == 1 - with open(files[0]) as f: - line = f.readline() - assert json.loads(line)["track_id"] == "test-001" - - @pytest.mark.asyncio - async def test_handle_track_auto_registers(self, registry): - event = {"track_id": "auto-001", "length": 1, "timestamp": 1000} - await registry.handle_track_event("newnode", json.dumps(event)) - nodes = registry.list_nodes() - assert "newnode" in nodes - - @pytest.mark.asyncio - async def test_list_nodes_empty(self, registry): - assert registry.list_nodes() == {} - - @pytest.mark.asyncio - async def test_get_node_config(self, registry): - config_data = {"capture": {"fc": 195000000}} - await registry.register_node("radar3", config_data) - assert registry.get_node_config("radar3") == config_data - assert registry.get_node_config("missing") is None -``` - -**Step 2: Run test to verify it fails** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_registry.py -v` -Expected: FAIL with `ModuleNotFoundError: No module named 'tracker_host.node_registry'` - -Note: You may need to install `pytest-asyncio`. Check `requirements.txt` and add if needed: -Run: `pip install pytest-asyncio` (if not already installed) - -**Step 3: Write minimal implementation** - -Create `tracker_host/node_registry.py`: - -```python -"""NodeRegistry: dynamic registration and management of radar nodes.""" - -import logging -from typing import Any, Optional - -from .output_handler import OutputHandler - -logger = logging.getLogger(__name__) - - -class NodeRegistry: - """Manages dynamically registered radar nodes. - - Nodes register by POSTing their config. Each registered node gets - an OutputHandler for writing track events to daily JSONL files. - """ - - def __init__(self, output_dir: str): - self.output_dir = output_dir - self._nodes: dict[str, dict[str, Any]] = {} - - async def register_node(self, name: str, config_data: dict) -> None: - """Register a node or update its config.""" - if name not in self._nodes: - output_handler = OutputHandler( - name=name, - output_dir=self.output_dir, - ) - self._nodes[name] = { - "config": config_data, - "output_handler": output_handler, - } - logger.info(f"Registered new node: {name}") - else: - self._nodes[name]["config"] = config_data - logger.info(f"Updated config for node: {name}") - - async def handle_track_event(self, name: str, event_line: str) -> None: - """Handle a track event from a node.""" - if name not in self._nodes: - await self.register_node(name, {}) - - node = self._nodes[name] - await node["output_handler"].handle_event(event_line) - - def get_node_config(self, name: str) -> Optional[dict]: - """Get stored config for a node, or None if not registered.""" - if name in self._nodes: - return self._nodes[name]["config"] - return None - - def list_nodes(self) -> dict[str, Any]: - """List all registered nodes with summary info.""" - return { - name: { - "has_config": bool(info["config"]), - "active_tracks": info["output_handler"].metrics.count, - } - for name, info in self._nodes.items() - } - - async def close(self) -> None: - """Close all output handlers.""" - for info in self._nodes.values(): - await info["output_handler"].close() - self._nodes.clear() -``` - -**Step 4: Run test to verify it passes** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_registry.py -v` -Expected: ALL PASS - -**Step 5: Commit** - -```bash -git add tracker_host/node_registry.py tests/test_node_registry.py -git commit -m "feat: add NodeRegistry for dynamic node management" -``` - ---- - -### Task 3: Create HTTP receive server - -**Files:** -- Create: `tracker_host/server.py` -- Create: `tests/test_server.py` - -**Step 1: Write the failing tests** - -Create `tests/test_server.py`: - -```python -"""Tests for the HTTP receive server.""" - -import json -import pytest -from aiohttp import web -from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop - -from tracker_host.server import create_app -from tracker_host.node_registry import NodeRegistry - - -@pytest.fixture -def registry(tmp_path): - return NodeRegistry(output_dir=str(tmp_path)) - - -@pytest.fixture -def app(registry): - return create_app(registry) - - -@pytest.fixture -async def client(aiohttp_client, app): - return await aiohttp_client(app) - - -class TestReceiveServer: - @pytest.mark.asyncio - async def test_post_config(self, client): - config = {"location": {"rx": {"latitude": 33.9}}} - resp = await client.post( - "/api/node/radar3/config", - json=config, - ) - assert resp.status == 200 - body = await resp.json() - assert body["status"] == "registered" - assert body["node"] == "radar3" - - @pytest.mark.asyncio - async def test_post_track(self, client): - event = {"track_id": "t-001", "length": 5, "timestamp": 1000} - resp = await client.post( - "/api/node/radar3/tracks", - data=json.dumps(event), - headers={"Content-Type": "application/x-ndjson"}, - ) - assert resp.status == 200 - - @pytest.mark.asyncio - async def test_list_nodes_empty(self, client): - resp = await client.get("/api/nodes") - assert resp.status == 200 - body = await resp.json() - assert body["nodes"] == {} - - @pytest.mark.asyncio - async def test_list_nodes_after_registration(self, client): - await client.post("/api/node/radar3/config", json={"location": {}}) - resp = await client.get("/api/nodes") - body = await resp.json() - assert "radar3" in body["nodes"] - - @pytest.mark.asyncio - async def test_post_track_batch(self, client): - """Test posting multiple events as newline-delimited JSON.""" - events = [ - {"track_id": "t-001", "length": 3, "timestamp": 1000}, - {"track_id": "t-002", "length": 5, "timestamp": 1001}, - ] - body = "\n".join(json.dumps(e) for e in events) - resp = await client.post( - "/api/node/radar3/tracks", - data=body, - headers={"Content-Type": "application/x-ndjson"}, - ) - assert resp.status == 200 -``` - -**Step 2: Run test to verify it fails** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_server.py -v` -Expected: FAIL with `ModuleNotFoundError: No module named 'tracker_host.server'` - -Note: may need `pip install pytest-aiohttp` for the `aiohttp_client` fixture. - -**Step 3: Write minimal implementation** - -Create `tracker_host/server.py`: - -```python -"""HTTP receive server for accepting track events from remote nodes.""" - -import json -import logging - -from aiohttp import web - -from .node_registry import NodeRegistry - -logger = logging.getLogger(__name__) - - -def create_app(registry: NodeRegistry) -> web.Application: - """Create the aiohttp web application.""" - app = web.Application() - app["registry"] = registry - - app.router.add_post("/api/node/{name}/config", handle_config) - app.router.add_post("/api/node/{name}/tracks", handle_tracks) - app.router.add_get("/api/nodes", handle_list_nodes) - - return app - - -async def handle_config(request: web.Request) -> web.Response: - """Receive and store radar config from a node.""" - name = request.match_info["name"] - registry: NodeRegistry = request.app["registry"] - - try: - config_data = await request.json() - except json.JSONDecodeError: - return web.json_response( - {"status": "error", "message": "invalid JSON"}, status=400 - ) - - await registry.register_node(name, config_data) - - logger.info(f"Received config from node: {name}") - return web.json_response({"status": "registered", "node": name}) - - -async def handle_tracks(request: web.Request) -> web.Response: - """Receive track events from a node (JSONL: one event per line).""" - name = request.match_info["name"] - registry: NodeRegistry = request.app["registry"] - - body = await request.text() - - for line in body.strip().split("\n"): - line = line.strip() - if line: - await registry.handle_track_event(name, line) - - return web.json_response({"status": "ok"}) - - -async def handle_list_nodes(request: web.Request) -> web.Response: - """List all registered nodes (for debugging).""" - registry: NodeRegistry = request.app["registry"] - return web.json_response({"nodes": registry.list_nodes()}) -``` - -**Step 4: Run test to verify it passes** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_server.py -v` -Expected: ALL PASS - -**Step 5: Commit** - -```bash -git add tracker_host/server.py tests/test_server.py -git commit -m "feat: add HTTP receive server for node track events" -``` - ---- - -### Task 4: Wire server mode into __main__.py and config.yaml - -**Files:** -- Modify: `tracker_host/__main__.py` -- Modify: `tracker_host/manager.py` -- Modify: `config.yaml` - -**Step 1: Update __main__.py to support server mode** - -Add `--server` flag. When set, run the HTTP server instead of the polling manager. - -In `tracker_host/__main__.py`, update imports and add to the argument parser: - -```python -parser.add_argument( - "--server", - action="store_true", - help="Run in server mode (receive tracks from nodes via HTTP)", -) -``` - -Update the run section: - -```python -if args.server: - from .server_mode import run_server - asyncio.run(run_server(str(config_path), verbose=args.verbose)) -else: - asyncio.run(run_manager(str(config_path), verbose=args.verbose)) -``` - -**Step 2: Create server_mode.py** - -Create `tracker_host/server_mode.py`: - -```python -"""Server mode: receive track events from remote nodes via HTTP.""" - -import asyncio -import logging -import signal - -from aiohttp import web - -from .config import load_config -from .node_registry import NodeRegistry -from .server import create_app - -logger = logging.getLogger(__name__) - - -async def run_server(config_path: str, verbose: bool = False) -> None: - """Run tracker-host in server mode.""" - level = logging.DEBUG if verbose else logging.INFO - logging.basicConfig( - level=level, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) - - config = load_config(config_path) - registry = NodeRegistry(output_dir=config.output_dir) - app = create_app(registry) - - runner = web.AppRunner(app) - await runner.setup() - - site = web.TCPSite(runner, config.server.host, config.server.port) - await site.start() - - logger.info( - f"Server listening on http://{config.server.host}:{config.server.port}" - ) - - # Wait for shutdown signal - stop_event = asyncio.Event() - loop = asyncio.get_running_loop() - - def handle_signal(): - logger.info("Received shutdown signal") - stop_event.set() - - for sig in (signal.SIGINT, signal.SIGTERM): - loop.add_signal_handler(sig, handle_signal) - - try: - await stop_event.wait() - finally: - await registry.close() - await runner.cleanup() - logger.info("Server stopped") -``` - -**Step 3: Update config.yaml with server section** - -Add to the top of `config.yaml`: - -```yaml -# Server mode settings (used with --server flag) -server: - host: "0.0.0.0" - port: 8080 -``` - -**Step 4: Test manually** - -```bash -cd /opt/apps/tracker-host -python -m tracker_host --server -v & -# In another terminal: -curl -s http://localhost:8080/api/nodes | python3 -m json.tool -curl -s -X POST http://localhost:8080/api/node/test/config \ - -H 'Content-Type: application/json' \ - -d '{"location": {"rx": {"lat": 33.9}}}' | python3 -m json.tool -curl -s http://localhost:8080/api/nodes | python3 -m json.tool -# Kill the server -kill %1 -``` - -Expected: First curl returns `{"nodes": {}}`. After POST, second curl shows `test` registered. - -**Step 5: Commit** - -```bash -git add tracker_host/__main__.py tracker_host/server_mode.py config.yaml -git commit -m "feat: add --server mode for receiving node track POSTs" -``` - ---- - -## Phase 2: Node-Side Agent - -### Task 5: Create node_agent package - -**Files:** -- Create: `node_agent/__init__.py` -- Create: `node_agent/__main__.py` -- Create: `tests/test_node_agent.py` - -The node agent uses only stdlib (`urllib.request`, `subprocess`, `json`) so it has no extra dependencies beyond Python itself. It can be deployed to a Pi by copying just the `node_agent/` directory. - -**Step 1: Write the failing tests** - -Create `tests/test_node_agent.py`: - -```python -"""Tests for node_agent.""" - -import json -import pytest -from unittest.mock import patch, MagicMock, mock_open -from http.server import HTTPServer, BaseHTTPRequestHandler -import threading - -from node_agent import push_config, post_track_event, fetch_blah2_config - - -class MockHandler(BaseHTTPRequestHandler): - """Simple HTTP handler that records requests.""" - received = [] - - def do_POST(self): - length = int(self.headers.get("Content-Length", 0)) - body = self.rfile.read(length).decode() - MockHandler.received.append({"path": self.path, "body": body}) - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.end_headers() - self.wfile.write(json.dumps({"status": "ok"}).encode()) - - def do_GET(self): - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.end_headers() - config = {"location": {"rx": {"latitude": 33.9}}, "capture": {"fc": 195000000}} - self.wfile.write(json.dumps(config).encode()) - - def log_message(self, format, *args): - pass # Suppress output - - -@pytest.fixture -def mock_server(): - MockHandler.received = [] - server = HTTPServer(("127.0.0.1", 0), MockHandler) - port = server.server_address[1] - thread = threading.Thread(target=server.serve_forever) - thread.daemon = True - thread.start() - yield f"http://127.0.0.1:{port}" - server.shutdown() - - -class TestNodeAgent: - def test_fetch_blah2_config(self, mock_server): - config = fetch_blah2_config(mock_server) - assert config["location"]["rx"]["latitude"] == 33.9 - - def test_push_config(self, mock_server): - config_data = {"location": {"rx": {"latitude": 33.9}}} - push_config(mock_server, "radar3", config_data) - assert len(MockHandler.received) == 1 - assert MockHandler.received[0]["path"] == "/api/node/radar3/config" - assert json.loads(MockHandler.received[0]["body"]) == config_data - - def test_post_track_event(self, mock_server): - event = {"track_id": "t-001", "length": 5} - post_track_event(mock_server, "radar3", json.dumps(event)) - assert len(MockHandler.received) == 1 - assert MockHandler.received[0]["path"] == "/api/node/radar3/tracks" - - def test_post_track_event_failure_does_not_raise(self): - """POST to unreachable server should log but not crash.""" - event = {"track_id": "t-001", "length": 5} - # Should not raise - post_track_event("http://127.0.0.1:1", "radar3", json.dumps(event)) -``` - -**Step 2: Run test to verify it fails** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_agent.py -v` -Expected: FAIL with `ModuleNotFoundError: No module named 'node_agent'` - -**Step 3: Write minimal implementation** - -Create `node_agent/__init__.py`: - -```python -"""Node agent: runs retina-tracker locally and POSTs track events to central server.""" - -from .agent import fetch_blah2_config, push_config, post_track_event -``` - -Create `node_agent/agent.py`: - -```python -"""Core node agent logic.""" - -import json -import sys -import urllib.error -import urllib.request - - -def fetch_blah2_config(blah2_url: str) -> dict: - """Fetch radar config from local blah2 API.""" - url = f"{blah2_url.rstrip('/')}/api/config" - req = urllib.request.Request(url, headers={"Accept": "application/json"}) - with urllib.request.urlopen(req, timeout=10) as resp: - return json.loads(resp.read()) - - -def push_config(server_url: str, node_name: str, config_data: dict) -> None: - """POST radar config to the central server.""" - url = f"{server_url.rstrip('/')}/api/node/{node_name}/config" - data = json.dumps(config_data).encode() - req = urllib.request.Request( - url, data=data, headers={"Content-Type": "application/json"} - ) - with urllib.request.urlopen(req, timeout=10) as resp: - resp.read() - - -def post_track_event(server_url: str, node_name: str, event_line: str) -> None: - """POST a single track event to the central server.""" - url = f"{server_url.rstrip('/')}/api/node/{node_name}/tracks" - data = event_line.encode() - req = urllib.request.Request( - url, - data=data, - headers={"Content-Type": "application/x-ndjson"}, - ) - try: - with urllib.request.urlopen(req, timeout=5) as resp: - resp.read() - except (urllib.error.URLError, OSError) as e: - print(f"POST failed: {e}", file=sys.stderr) -``` - -**Step 4: Run test to verify it passes** - -Run: `cd /opt/apps/tracker-host && python -m pytest tests/test_node_agent.py -v` -Expected: ALL PASS - -**Step 5: Commit** - -```bash -git add node_agent/__init__.py node_agent/agent.py tests/test_node_agent.py -git commit -m "feat: add node_agent core HTTP functions" -``` - ---- - -### Task 6: Add node agent CLI and subprocess management - -**Files:** -- Create: `node_agent/__main__.py` - -**Step 1: Write the CLI entry point** - -Create `node_agent/__main__.py`: - -```python -"""CLI entry point: python -m node_agent""" - -import argparse -import signal -import subprocess -import sys -import time - -from .agent import fetch_blah2_config, post_track_event, push_config - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Node agent: run retina-tracker and POST tracks to central server" - ) - parser.add_argument("--node-name", required=True, help="Unique name for this node") - parser.add_argument("--server-url", required=True, help="Central server URL") - parser.add_argument( - "--blah2-url", - default="http://localhost:3000", - help="Local blah2 API URL (default: http://localhost:3000)", - ) - parser.add_argument( - "--tracker-path", - default="../retina-tracker", - help="Path to retina-tracker directory (default: ../retina-tracker)", - ) - parser.add_argument( - "--tcp-port", - type=int, - default=3012, - help="TCP port for retina-tracker (default: 3012)", - ) - parser.add_argument( - "--tracker-config", - help="Optional path to retina-tracker config.yaml", - ) - - args = parser.parse_args() - - # Step 1: Fetch config from blah2 and push to server - print(f"Fetching radar config from {args.blah2_url}...") - try: - config = fetch_blah2_config(args.blah2_url) - print(f"Pushing config to {args.server_url}...") - push_config(args.server_url, args.node_name, config) - print("Config registered with server.") - except Exception as e: - print(f"Warning: Config push failed: {e}", file=sys.stderr) - print("Continuing without config push...", file=sys.stderr) - - # Step 2: Start retina-tracker subprocess - cmd = [ - sys.executable, - "-m", - "tracker.track_detections", - "--tcp", - "--tcp-host", - "0.0.0.0", - "--tcp-port", - str(args.tcp_port), - "-s", - "-", # Stream events to stdout - ] - - if args.tracker_config: - cmd.extend(["-c", args.tracker_config]) - - print(f"Starting retina-tracker on TCP port {args.tcp_port}...") - proc = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, - stderr=sys.stderr, - cwd=args.tracker_path, - ) - - # Handle shutdown gracefully - def shutdown(signum, frame): - print("\nShutting down...") - proc.terminate() - proc.wait(timeout=5) - sys.exit(0) - - signal.signal(signal.SIGINT, shutdown) - signal.signal(signal.SIGTERM, shutdown) - - print(f"Reading track events, POSTing to {args.server_url}...") - event_count = 0 - - # Step 3: Read stdout and POST events - try: - for line in proc.stdout: - decoded = line.decode().strip() - if decoded: - post_track_event(args.server_url, args.node_name, decoded) - event_count += 1 - if event_count % 50 == 0: - print(f" Posted {event_count} track events") - except Exception as e: - print(f"Error: {e}", file=sys.stderr) - finally: - proc.terminate() - proc.wait(timeout=5) - print(f"Done. Posted {event_count} track events total.") - - -if __name__ == "__main__": - main() -``` - -**Step 2: Test manually (quick sanity check)** - -```bash -cd /opt/apps/tracker-host -python -m node_agent --help -``` - -Expected: Help text printed showing all arguments. - -**Step 3: Commit** - -```bash -git add node_agent/__main__.py -git commit -m "feat: add node_agent CLI with subprocess management" -``` - ---- - -## Phase 3: Testing Infrastructure - -### Task 7: Create detection relay for staging tests - -On this staging machine, blah2 isn't running locally. This relay script polls a remote detection endpoint (e.g. `radar3.retnode.com/api/detection`) and forwards the data to retina-tracker's TCP port, simulating what blah2 would do on a real node. - -**Files:** -- Create: `node_agent/relay.py` - -**Step 1: Write the relay** - -Create `node_agent/relay.py`: - -```python -"""Detection relay: polls HTTP detection endpoint and forwards to retina-tracker TCP. - -Used for testing when blah2 isn't running locally. Simulates the blah2→retina-tracker -TCP connection by polling a remote detection API. - -Usage: - python -m node_agent.relay \ - --detection-url https://radar3.retnode.com/api/detection \ - --tcp-port 3012 -""" - -import argparse -import json -import socket -import sys -import time -import urllib.error -import urllib.request - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Relay: poll detection HTTP endpoint and forward to retina-tracker TCP" - ) - parser.add_argument( - "--detection-url", - default="https://radar3.retnode.com/api/detection", - help="HTTP detection endpoint to poll", - ) - parser.add_argument("--tcp-host", default="127.0.0.1", help="Tracker TCP host") - parser.add_argument("--tcp-port", type=int, default=3012, help="Tracker TCP port") - parser.add_argument( - "--interval", type=float, default=0.5, help="Poll interval in seconds" - ) - - args = parser.parse_args() - - # Connect to retina-tracker TCP port - print(f"Connecting to retina-tracker at {args.tcp_host}:{args.tcp_port}...") - max_retries = 30 - sock = None - for attempt in range(max_retries): - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.connect((args.tcp_host, args.tcp_port)) - break - except ConnectionRefusedError: - sock.close() - sock = None - if attempt < max_retries - 1: - print( - f" Connection refused, retrying ({attempt + 1}/{max_retries})..." - ) - time.sleep(1) - - if sock is None: - print("Failed to connect to retina-tracker", file=sys.stderr) - sys.exit(1) - - print(f"Connected! Polling {args.detection_url} every {args.interval}s") - frame_count = 0 - - try: - while True: - try: - req = urllib.request.Request( - args.detection_url, - headers={"Accept": "application/json"}, - ) - with urllib.request.urlopen(req, timeout=5) as resp: - data = resp.read() - - # Validate it's JSON, then forward as a single line - json.loads(data) # Validate - sock.sendall(data + b"\n") - frame_count += 1 - - if frame_count % 100 == 0: - print(f" Forwarded {frame_count} detection frames") - - except (urllib.error.URLError, OSError, json.JSONDecodeError) as e: - print(f"Error: {e}", file=sys.stderr) - - time.sleep(args.interval) - - except KeyboardInterrupt: - print(f"\nDone. Forwarded {frame_count} detection frames.") - finally: - sock.close() - - -if __name__ == "__main__": - main() -``` - -**Step 2: Test the relay connects (quick sanity check)** - -Don't run the full relay yet (needs retina-tracker running). Just verify it starts: - -```bash -cd /opt/apps/tracker-host -python -m node_agent.relay --help -``` - -Expected: Help text printed. - -**Step 3: Commit** - -```bash -git add node_agent/relay.py -git commit -m "feat: add detection relay for staging tests" -``` - ---- - -### Task 8: End-to-end integration test - -This is a manual test that validates the full pipeline. Run each command in a separate terminal. - -**Terminal 1: Start the server** - -```bash -cd /opt/apps/tracker-host -python -m tracker_host --server -v -``` - -Expected: `Server listening on http://0.0.0.0:8080` - -**Terminal 2: Start the node agent** - -```bash -cd /opt/apps/tracker-host -python -m node_agent \ - --node-name radar3 \ - --server-url http://localhost:8080 \ - --blah2-url https://radar3.retnode.com \ - --tracker-path /opt/apps/retina-tracker \ - --tcp-port 3012 -``` - -Expected: Config pushed, retina-tracker started, waiting for TCP connection. - -**Terminal 3: Start the detection relay** - -```bash -cd /opt/apps/tracker-host -python -m node_agent.relay \ - --detection-url https://radar3.retnode.com/api/detection \ - --tcp-port 3012 -``` - -Expected: Connected, forwarding detection frames. - -**Verify output:** - -```bash -# After 30-60 seconds, check for JSONL output -ls -la output/radar3_*.jsonl -head -1 output/radar3_*.jsonl | python3 -m json.tool -``` - -Expected: JSONL file exists with track events containing `track_id`, `detections`, `timestamp`, etc. - -```bash -# Check server sees the node -curl -s http://localhost:8080/api/nodes | python3 -m json.tool -``` - -Expected: `radar3` listed with `has_config: true` and `active_tracks > 0`. - -**After successful verification, commit a note:** - -```bash -git add -A -git commit -m "feat: complete node-to-server track posting pipeline" -``` - ---- - -## Phase 4: Server-Side Geolocator Integration - -### Task 9: Wire geolocator to received track events - -Once tracks are flowing from nodes to server, the server can run geolocator instances. This reuses the existing `GeolocatorInstance` class. - -**Files:** -- Modify: `tracker_host/node_registry.py` -- Modify: `tracker_host/config.py` -- Create: `tests/test_node_registry_geolocator.py` - -**Step 1: Add geolocator config to server settings** - -In `tracker_host/config.py`, add to `ServerConfig`: - -```python -@dataclass -class ServerConfig: - """HTTP server configuration for receiving track events from nodes.""" - host: str = "0.0.0.0" - port: int = 8080 - geolocator_enabled: bool = False - geolocator_tcp_port_base: int = 31000 -``` - -Update parsing in `_parse_config()`: - -```python -server = ServerConfig( - host=server_raw.get("host", "0.0.0.0"), - port=server_raw.get("port", 8080), - geolocator_enabled=server_raw.get("geolocator_enabled", False), - geolocator_tcp_port_base=server_raw.get("geolocator_tcp_port_base", 31000), -) -``` - -**Step 2: Extend NodeRegistry to manage geolocators** - -When a node registers with config (containing `location.rx` and `location.tx`), and geolocator is enabled, the registry: -1. Saves the radar config to a temp YAML file (same as `GeolocatorInstance._fetch_radar_config` does) -2. Creates a `GeolocatorInstance` for that node -3. Forwards track events to the geolocator - -In `tracker_host/node_registry.py`, add geolocator support: - -```python -import yaml -from pathlib import Path -from .config import Config, GeolocatorConfig -from .geolocator import GeolocatorInstance - -class NodeRegistry: - def __init__(self, output_dir: str, global_config: Optional[Config] = None): - self.output_dir = output_dir - self.global_config = global_config - self._nodes: dict[str, dict[str, Any]] = {} - self._next_geo_port: int = ( - global_config.server.geolocator_tcp_port_base - if global_config - else 31000 - ) - - async def register_node(self, name: str, config_data: dict) -> None: - """Register a node or update its config. Start geolocator if enabled.""" - if name not in self._nodes: - output_handler = OutputHandler(name=name, output_dir=self.output_dir) - self._nodes[name] = { - "config": config_data, - "output_handler": output_handler, - "geolocator": None, - } - logger.info(f"Registered new node: {name}") - else: - self._nodes[name]["config"] = config_data - logger.info(f"Updated config for node: {name}") - - # Start geolocator if enabled and config has location data - if ( - self.global_config - and self.global_config.server.geolocator_enabled - and config_data.get("location") - and self._nodes[name]["geolocator"] is None - ): - await self._start_geolocator(name, config_data) - - async def _start_geolocator(self, name: str, config_data: dict) -> None: - """Start a geolocator instance for a node.""" - port = self._next_geo_port - self._next_geo_port += 1 - - # Save radar config to temp file - config_path = Path(self.output_dir) / f".{name}_config.yml" - with open(config_path, "w") as f: - yaml.dump(config_data, f, default_flow_style=False) - - geo_config = GeolocatorConfig(enabled=True, tcp_port=port) - geo = GeolocatorInstance( - name=name, - geo_config=geo_config, - global_config=self.global_config, - config_url=None, # Config already saved locally - session=None, - ) - # Override the config path since we saved it ourselves - geo._radar_config_path = str(config_path.resolve()) - - try: - # Skip the fetch step (we already have the config) - await geo._start_process() - await geo._connect_tcp() - geo._running = True - geo._output_task = asyncio.create_task( - geo._output_loop(), name=f"{name}-geo-output" - ) - geo._stderr_task = asyncio.create_task( - geo._stderr_loop(), name=f"{name}-geo-stderr" - ) - self._nodes[name]["geolocator"] = geo - logger.info(f"Started geolocator for {name} on port {port}") - except Exception as e: - logger.error(f"Failed to start geolocator for {name}: {e}") - - async def handle_track_event(self, name: str, event_line: str) -> None: - """Handle a track event from a node.""" - if name not in self._nodes: - await self.register_node(name, {}) - - node = self._nodes[name] - await node["output_handler"].handle_event(event_line) - - # Forward to geolocator if running - if node["geolocator"] is not None: - await node["geolocator"].send_track_event(event_line) -``` - -**Step 3: Update server_mode.py to pass global_config** - -```python -registry = NodeRegistry(output_dir=config.output_dir, global_config=config) -``` - -**Step 4: Update config.yaml** - -```yaml -server: - host: "0.0.0.0" - port: 8080 - geolocator_enabled: true - geolocator_tcp_port_base: 31000 -``` - -**Step 5: Test** - -Run the same end-to-end test as Task 8. After tracks flow for 60+ seconds, check for solution output: - -```bash -ls -la output/*solutions*.jsonl -``` - -Expected: Solution JSONL files appear for each node with geolocator enabled. - -**Step 6: Commit** - -```bash -git add tracker_host/config.py tracker_host/node_registry.py tracker_host/server_mode.py config.yaml -git commit -m "feat: add server-side geolocator for received node tracks" -``` - ---- - -## Phase 5: Documentation - -### Task 10: Document architecture and update config.yaml - -**Files:** -- Modify: `config.yaml` (add documentation comments) -- Modify: `README.md` (add server mode and node agent sections) - -**Step 1: Update config.yaml with documentation** - -Add clear comments to `config.yaml` explaining both modes and the dynamic node registration decision. Include example commands. - -**Step 2: Update README.md** - -Add sections for: -- **Server Mode**: How to run with `--server`, what endpoints are available -- **Node Agent**: How to deploy to a Raspberry Pi, example commands -- **Detection Relay**: How to use for staging/testing -- **Architecture**: Brief description of the node→server data flow -- **Architecture Decision: Dynamic Node Registration**: Nodes register by POSTing config, no pre-configuration needed. This simplifies deployment and allows adding/removing nodes without touching the server config. - -**Step 3: Commit** - -```bash -git add config.yaml README.md -git commit -m "docs: add server mode and node agent documentation" -``` - ---- - -## Summary - -| Phase | What | Files | Dependencies | -|-------|------|-------|-------------| -| 1 (Tasks 1-4) | Server HTTP endpoint | `config.py`, `node_registry.py`, `server.py`, `server_mode.py`, `__main__.py` | aiohttp (existing) | -| 2 (Tasks 5-6) | Node agent | `node_agent/agent.py`, `node_agent/__main__.py` | stdlib only | -| 3 (Tasks 7-8) | Testing infrastructure | `node_agent/relay.py` + manual E2E test | stdlib only | -| 4 (Task 9) | Server geolocator | `node_registry.py` (extend), `config.py` | existing GeolocatorInstance | -| 5 (Task 10) | Documentation | `config.yaml`, `README.md` | — | - -**Deployment to a real Pi node requires only:** -1. `retina-tracker/` (already needed) -2. `node_agent/` directory (3 small Python files, no extra deps) -3. One command: `python -m node_agent --node-name radar3 --server-url https://server.example.com`