From 6df7c3df2e42271b5bccfda053c01835b0f41006 Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Sun, 13 Sep 2026 20:57:33 +0000 Subject: [PATCH] Publish the declared wedge as a synthetic node's detection area MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_node_summary publishes to_polygon(evidence_only=True) for every node. That is right for real hardware — a real node's beam_azimuth_deg and beam_width_deg are unsurveyed configuration, so drawing them would claim coverage nobody measured — but it is wrong for a simulator node. The simulator emits a detection only for an aircraft inside the node's declared cone (retina_simulation/world.py::_aircraft_in_detection_cone), so for those nodes the cone IS the detection area by definition, and the evidence is the unreliable half: the calibration points come from ADS-B hexes bound to tracks and roughly a third of those binds are to the wrong aircraft. Measured on test 2026-09-13, synth-GVL-SCAT-0032 (42 deg beam) held 3,037 calibration points of which 47% lay outside its wedge (34% ignoring the two edge bins), 55 out-of-wedge bins had opened, and the published polygon covered 71 of 72 bearings. Every synthetic node on the test map draws as a disc. So: EmpiricalCoverageState.declared_wedge_polygon() draws the prior azimuth and width at _reach_at on each bearing — already the bistatic ellipse when a differential limit and the TX are known, else the circle, which is exactly the simulator's own range rule — with no clamp, no bins and no min-points gate. An omni prior gets a full ring; a directional one an apex-closed wedge with both edges exact. NodeAnalyticsManager.register_node grows a keyword-only declared_geometry_is_truth flag (default False, asserted per registration, so a re-registration without it drops the node again), and get_node_summary publishes the declared wedge for a flagged node and the evidence-only shape for everyone else. The FOV diagnostics block is skipped for a flagged node — it describes the learned wedge, which such a node does not publish — while n_points/n_filled_bins stay in the payload: the evidence is still accumulated and still worth reporting, it just is not drawn. Every summary now names its "polygon_source" (declared / evidence / learned) so the map can say which it is showing rather than guess. A flag change invalidates the 60 s summaries cache the same way a rebuilt detection area does; retire_node, _reset_for_tests and the "cannot place this node" path all clear it. Co-Authored-By: Claude Fable 5.1 --- src/retina_analytics/empirical_coverage.py | 77 +++++ src/retina_analytics/manager.py | 60 +++- tests/test_declared_wedge.py | 351 +++++++++++++++++++++ 3 files changed, 484 insertions(+), 4 deletions(-) create mode 100644 tests/test_declared_wedge.py diff --git a/src/retina_analytics/empirical_coverage.py b/src/retina_analytics/empirical_coverage.py index d7c8686..29d52c2 100644 --- a/src/retina_analytics/empirical_coverage.py +++ b/src/retina_analytics/empirical_coverage.py @@ -30,6 +30,14 @@ neither opens a bin nor clips one, and unobserved bearings collapse to the RX apex instead of being interpolated across. +Declared publication (declared_wedge_polygon) +--------------------------------------------- +The exception, and only for SYNTHETIC nodes: the simulator emits a detection +only inside the node's declared cone, so for those nodes the cone is the +detection area by definition and the bins are the unreliable half (about a +third of the ADS-B binds that feed them are to the wrong aircraft). Real +nodes never take this path. See declared_wedge_polygon. + Learned FOV (FOV_MODE, schema 3) --------------------------------- The methods above (observed_limit_km / constraint_digest / to_polygon without @@ -904,6 +912,75 @@ def _evidence_only_polygon(self) -> list[list[float]] | None: return None return polygon + def declared_wedge_polygon(self, step_deg: float = 2.5) -> list[list[float]] | None: + """The declared cone, drawn as-is — for nodes whose geometry IS truth. + + Only ever published for SYNTHETIC (simulator-fleet) nodes; the caller + decides, and for a real node this method is never called. A real + node's beam_azimuth_deg / beam_width_deg are unsurveyed configuration, + so drawing them would claim coverage nobody measured — that is what + _evidence_only_polygon exists to avoid, and it stays the published + shape for real hardware. + + A simulated node is the other way round. The simulator emits a + detection only for an aircraft inside the node's declared cone + (retina_simulation/world.py::_aircraft_in_detection_cone: bearing + within azimuth ± width/2, range within the declared differential limit + when there is one, else within max_range_km on the RX distance), so + the cone is the node's detection area *by definition* and any evidence + outside it is an error in the evidence. It is: the calibration points + come from ADS-B hexes bound to tracks and roughly a third of those + binds are to the wrong aircraft. Measured on test 2026-09-13, + synth-GVL-SCAT-0032 (42° beam) held 3,037 points of which 47 % lay + outside its wedge (34 % ignoring the two edge bins), 55 out-of-wedge + bins had opened, and the published polygon covered 71 of 72 bearings — + a node with a 42° beam drawn as a disc. + + So this method reads no bins at all. Radius on each bearing is + _reach_at — already the bistatic ellipse when the TX and a + differential limit are known, else the monostatic circle, which is + exactly the simulator's own range rule — with no range_clamp_mult, + because there is no mis-attributed far detection to defend against. + + Vertices: for a directional prior, the RX apex, then both wedge edges + exactly plus every *step_deg* between them, then the apex again to + close. For an omni prior (prior_azimuth_deg None) a full ring every + *step_deg*, closed on its first vertex and with no apex — an omni node + has no direction to exclude. None only when the reach is not positive. + """ + half = (self.prior_width_deg if self.prior_width_deg is not None else YAGI_BEAM_WIDTH_DEG) / 2.0 + az = self.prior_azimuth_deg + + if az is None: + n_steps = max(3, int(round(360.0 / step_deg))) + bearings = [i * (360.0 / n_steps) for i in range(n_steps)] + else: + start, end = az - half, az + half + bearings = [start] + b = start + step_deg + while b < end - 1e-9: + bearings.append(b) + b += step_deg + bearings.append(end) + + if not any(self._reach_at(b) > 0.0 for b in bearings): + return None + + apex = [round(self.rx_lat, 5), round(self.rx_lon, 5)] + polygon: list[list[float]] = [] if az is None else [apex] + for bearing in bearings: + r_km = max(0.0, self._reach_at(bearing)) + bearing_rad = math.radians(bearing) + lat, lon = offset_latlon( + self.rx_lat, + self.rx_lon, + east_km=r_km * math.sin(bearing_rad), + north_km=r_km * math.cos(bearing_rad), + ) + polygon.append([round(lat, 5), round(lon, 5)]) + polygon.append(apex if az is not None else polygon[0]) + return polygon + # ── Shrink-only prior ──────────────────────────────────────────────────── def observed_limit_km(self, bearing_deg_: float) -> float | None: diff --git a/src/retina_analytics/manager.py b/src/retina_analytics/manager.py index ba3b541..7d5a8c1 100644 --- a/src/retina_analytics/manager.py +++ b/src/retina_analytics/manager.py @@ -72,6 +72,14 @@ def __init__(self, storage_dir: str = "", fov_mode: str = "off"): self.reputations: dict[str, NodeReputation] = {} self.coverage_maps: dict[str, HistoricalCoverageMap] = {} self.empirical_coverages: dict[str, EmpiricalCoverageState] = {} + # Nodes whose DECLARED geometry is ground truth rather than a config + # guess — synthetic/simulator nodes, whose detection cone is what the + # simulator enforces. Set per registration by the backend (see + # register_node's declared_geometry_is_truth); read only by + # get_node_summary, to publish the declared wedge instead of the + # evidence. Membership is not persisted: it is a property of who + # registered the node, re-asserted on every registration. + self._declared_truth: set[str] = set() self._storage_dir = storage_dir self._last_save_time = 0.0 self._save_interval_s = 300.0 @@ -100,18 +108,41 @@ def _reset_for_tests(self) -> None: self.empirical_coverages, ): store.clear() + self._declared_truth.clear() self._cross_node_cache = None self._cross_node_cache_ts = 0.0 self._summaries_cache = None self._summaries_cache_ts = 0.0 self._last_save_time = 0.0 - def register_node(self, node_id: str, config: dict): + def register_node(self, node_id: str, config: dict, *, declared_geometry_is_truth: bool = False): + """Register or re-register a node. + + declared_geometry_is_truth marks a node whose declared cone IS its + detection area — a simulator node, where the cone is what the + simulator enforces — so get_node_summary publishes + EmpiricalCoverageState.declared_wedge_polygon rather than the + evidence-only shape. Default False: a real receiver's declared aim is + unsurveyed configuration. Asserted per registration, so a node + re-registered without the flag loses it. + """ # Locked: save_coverage_maps / _load_coverage_maps iterate these dicts # from other threads, and an unlocked insert mid-iteration raises # "dictionary changed size during iteration". with self._save_lock: + was_declared = node_id in self._declared_truth + if declared_geometry_is_truth: + self._declared_truth.add(node_id) + else: + self._declared_truth.discard(node_id) self._register_node_locked(node_id, config) + # A flag change swaps the published polygon just as a rebuilt + # detection area does, so it has to drop the 60 s summaries cache + # the same way. Compared against the FINAL membership, not the + # requested one: _register_node_locked drops the flag again for a + # node it cannot place. + if (node_id in self._declared_truth) != was_declared: + self._invalidate_analysis_caches() def _register_node_locked(self, node_id: str, config: dict): # Whether this call adds anything a summary reads, which is what the @@ -162,6 +193,11 @@ def _register_node_locked(self, node_id: str, config: dict): # dropped here, or one of trust/metrics/reputation/coverage_map # was just populated above. had_area = self.detection_areas.pop(node_id, None) is not None + # A node we cannot place has no declared wedge to publish either: + # declared_wedge_polygon is anchored on the RX this registration + # did not supply. register_node's own comparison covers the cache + # when this drops a flag it had just set. + self._declared_truth.discard(node_id) if added_to_summary or had_area: self._invalidate_analysis_caches() return @@ -387,6 +423,7 @@ def retire_node(self, node_id: str) -> dict: self.empirical_coverages, ): store.pop(node_id, None) + self._declared_truth.discard(node_id) files = [] if self._storage_dir: @@ -517,7 +554,7 @@ def get_node_summary(self, node_id: str) -> dict: # polygon anchored on the node's stale receiver. if ec is not None and da is not None: fov_mode_active = self.fov_mode != "off" - # FOV off publishes what the node has been SEEN to detect, with no + # A REAL node publishes what it has been SEEN to detect, with no # theoretical clip. It used to pass the detection area's # beam_azimuth_deg / beam_width_deg / max_range_km, which are # declared configuration — most nodes never had their aim surveyed @@ -528,13 +565,28 @@ def get_node_summary(self, node_id: str) -> dict: # publishes the learned wedge instead, which is itself derived # from evidence. The detection area is still required (above) — # it is the geometry check, not the clip. + # + # A DECLARED-TRUTH node (synthetic: see register_node) inverts + # that. The simulator only ever emits a detection inside the + # node's declared cone, so the cone is its detection area by + # definition and the accumulated bins are the unreliable half — + # about a third of the ADS-B binds feeding them are to the wrong + # aircraft (synth-GVL-SCAT-0032, 2026-09-13: 47 % of 3,037 points + # outside a 42° beam, 71 of 72 bearings published). So it + # publishes the declared wedge, independent of evidence — which is + # also why the min_points gate and the FOV diagnostics below, + # both evidence-derived, do not apply to it. n_points / + # n_filled_bins stay in the payload: the evidence is still + # accumulated and still worth reporting, it just is not drawn. + declared = node_id in self._declared_truth poly_kwargs = {"use_learned_wedge": True} if fov_mode_active else {"evidence_only": True} result["empirical_coverage"] = { "n_points": ec.n_points, "n_filled_bins": ec.n_filled_bins, - "polygon": ec.to_polygon(**poly_kwargs), + "polygon": (ec.declared_wedge_polygon() if declared else ec.to_polygon(**poly_kwargs)), + "polygon_source": ("declared" if declared else ("learned" if fov_mode_active else "evidence")), } - if fov_mode_active: + if fov_mode_active and not declared: # Flows through /api/radar/analytics automatically — no new # dashboard endpoint needed for shadow verification. n_neg_events = sum(len(evs) for evs in ec._neg_events) diff --git a/tests/test_declared_wedge.py b/tests/test_declared_wedge.py new file mode 100644 index 0000000..d2429cf --- /dev/null +++ b/tests/test_declared_wedge.py @@ -0,0 +1,351 @@ +"""The declared wedge: what a SYNTHETIC node publishes as its detection area. + +A real node publishes evidence only (test_learned_fov.py's +TestFovOffPublishesEvidenceOnly) because its declared azimuth and width are +unsurveyed configuration. A simulator node is the opposite case: the +simulator emits a detection only inside the declared cone, so the cone IS the +detection area and the accumulated bins — fed by ADS-B hexes bound to tracks, +about a third of them to the wrong aircraft — are the unreliable half. See +EmpiricalCoverageState.declared_wedge_polygon and +NodeAnalyticsManager.register_node(declared_geometry_is_truth=True). +""" + +import math + +import pytest + +from retina_analytics.constants import KM_PER_DEG_LAT, YAGI_BEAM_WIDTH_DEG, bearing_deg, haversine_km +from retina_analytics.empirical_coverage import EmpiricalCoverageState +from retina_analytics.manager import NodeAnalyticsManager + +_RX_LAT, _RX_LON = 34.85, -82.40 +_TX_LAT, _TX_LON = 35.236, -82.40 # ~43 km due north + + +def _at_bearing(bearing, range_km, rx_lat=_RX_LAT, rx_lon=_RX_LON): + rad = math.radians(bearing) + return ( + rx_lat + range_km * math.cos(rad) / KM_PER_DEG_LAT, + rx_lon + range_km * math.sin(rad) / (KM_PER_DEG_LAT * math.cos(math.radians(rx_lat))), + ) + + +def _apex(ec): + return [round(ec.rx_lat, 5), round(ec.rx_lon, 5)] + + +def _angle_from(az, bearing): + """Signed angular distance from *az* to *bearing*, in (-180, 180].""" + return (bearing - az + 180.0) % 360.0 - 180.0 + + +# Vertices are placed with offset_latlon (the flat east/north step every +# polygon in empirical_coverage.py is built from), so reading a vertex back +# with the spherical bearing_deg returns the intended bearing plus a +# convergence term — about 0.2 deg at 20 deg off axis and 50 km out, growing +# with latitude and range. The tests below measure spherically on purpose, as +# an external check, and carry this slack rather than re-deriving the same +# flat-earth arithmetic they are trying to verify. +_CONVERGENCE_SLACK_DEG = 0.25 + + +def _bearings_of(poly, rx_lat=_RX_LAT, rx_lon=_RX_LON): + return [bearing_deg(rx_lat, rx_lon, lat, lon) for lat, lon in poly] + + +# ── The polygon itself ─────────────────────────────────────────────────────── + + +class TestDeclaredWedgePolygon: + def _directional(self, width=40.0, az=90.0, max_range_km=50.0): + return EmpiricalCoverageState( + rx_lat=_RX_LAT, + rx_lon=_RX_LON, + max_range_km=max_range_km, + prior_azimuth_deg=az, + prior_width_deg=width, + ) + + def test_it_is_a_closed_ring_with_the_rx_as_apex(self): + ec = self._directional() + poly = ec.declared_wedge_polygon() + assert poly is not None + assert poly[0] == poly[-1] == _apex(ec) + + def test_no_vertex_falls_outside_the_declared_half_width(self): + ec = self._directional(width=40.0, az=90.0) + poly = ec.declared_wedge_polygon() + for lat, lon in poly[1:-1]: + bearing = bearing_deg(_RX_LAT, _RX_LON, lat, lon) + assert abs(_angle_from(90.0, bearing)) <= 20.0 + _CONVERGENCE_SLACK_DEG + + def test_every_vertex_sits_at_the_reach_for_its_bearing(self): + ec = self._directional(width=40.0, az=90.0) + poly = ec.declared_wedge_polygon() + for lat, lon in poly[1:-1]: + bearing = bearing_deg(_RX_LAT, _RX_LON, lat, lon) + r = haversine_km(_RX_LAT, _RX_LON, lat, lon) + assert r == pytest.approx(ec._reach_at(bearing), rel=0.01) + + def test_both_wedge_edges_are_drawn_exactly(self): + """The edges are the whole point of the shape, so they are vertices in + their own right rather than whatever the last step_deg landed on.""" + ec = self._directional(width=41.0, az=90.0) # not a multiple of 2.5 + poly = ec.declared_wedge_polygon() + edges = sorted(_angle_from(90.0, b) for b in _bearings_of(poly[1:-1])) + assert edges[0] == pytest.approx(-20.5, abs=_CONVERGENCE_SLACK_DEG) + assert edges[-1] == pytest.approx(20.5, abs=_CONVERGENCE_SLACK_DEG) + + def test_a_wedge_is_sampled_every_step_deg(self): + ec = self._directional(width=40.0, az=90.0) + fine = ec.declared_wedge_polygon(step_deg=2.5) + coarse = ec.declared_wedge_polygon(step_deg=10.0) + assert len(fine) == 2 + (40 // 2.5 + 1) # apex, edge..edge inclusive, apex + assert len(coarse) == 2 + (40 // 10 + 1) + + def test_a_missing_width_falls_back_to_the_yagi_default(self): + """Same fallback _in_theoretical_wedge uses, so the two cannot drift.""" + ec = self._directional(width=None, az=90.0) + poly = ec.declared_wedge_polygon() + spans = [abs(_angle_from(90.0, b)) for b in _bearings_of(poly[1:-1])] + assert max(spans) == pytest.approx(YAGI_BEAM_WIDTH_DEG / 2.0, abs=_CONVERGENCE_SLACK_DEG) + + def test_evidence_does_not_change_it(self): + """The whole point: the bins are not consulted at all.""" + ec = self._directional(width=40.0, az=90.0) + before = ec.declared_wedge_polygon() + for i in range(40): + ec.add_point(*_at_bearing(270.0, 20.0 + i * 0.01)) # a lobe behind the node + assert ec.n_points == 40 + assert ec.declared_wedge_polygon() == before + + def test_it_is_published_with_no_evidence_at_all(self): + ec = self._directional() + assert ec.n_points == 0 + assert ec.to_polygon(evidence_only=True) is None + assert ec.declared_wedge_polygon() is not None + + def test_a_non_positive_reach_has_no_polygon(self): + ec = EmpiricalCoverageState( + rx_lat=_RX_LAT, + rx_lon=_RX_LON, + max_range_km=0.0, + prior_azimuth_deg=90.0, + prior_width_deg=40.0, + ) + assert ec.declared_wedge_polygon() is None + + +class TestOmniPrior: + def _omni(self): + return EmpiricalCoverageState( + rx_lat=_RX_LAT, + rx_lon=_RX_LON, + max_range_km=50.0, + prior_azimuth_deg=None, + prior_width_deg=None, + ) + + def test_it_is_a_full_ring_with_no_apex(self): + """An omni node has no direction to exclude, so there is no apex to + collapse to and the ring closes on its own first vertex.""" + ec = self._omni() + poly = ec.declared_wedge_polygon() + assert poly[0] == poly[-1] + assert _apex(ec) not in poly + bearings = sorted(_bearings_of(poly[:-1])) + assert len(bearings) == 144 # 360 / 2.5 + assert min(bearings) < 2.5 + assert max(bearings) > 357.5 + gaps = [b - a for a, b in zip(bearings, bearings[1:])] + assert max(gaps) == pytest.approx(2.5, abs=_CONVERGENCE_SLACK_DEG) + + def test_the_ring_sits_at_the_reach_all_round(self): + ec = self._omni() + for lat, lon in ec.declared_wedge_polygon()[:-1]: + r = haversine_km(_RX_LAT, _RX_LON, lat, lon) + assert r == pytest.approx(50.0, rel=0.01) + + +class TestReachRule: + """_reach_at is the simulator's own range rule: the bistatic ellipse when a + differential limit and the TX are declared, else the circle on the RX.""" + + def _state(self, bistatic): + ec = EmpiricalCoverageState( + rx_lat=_RX_LAT, + rx_lon=_RX_LON, + max_range_km=50.0, + tx_lat=_TX_LAT, + tx_lon=_TX_LON, + prior_azimuth_deg=None, + prior_width_deg=None, + ) + ec.max_bistatic_range_km = bistatic + return ec + + def _reach_toward_and_away(self, ec): + """Vertex range at the bearing nearest the TX and nearest away from it.""" + poly = ec.declared_wedge_polygon(step_deg=5.0)[:-1] + to_tx = bearing_deg(_RX_LAT, _RX_LON, _TX_LAT, _TX_LON) + + def _nearest(target): + lat, lon = min(poly, key=lambda v: abs(_angle_from(target, bearing_deg(_RX_LAT, _RX_LON, *v)))) + return haversine_km(_RX_LAT, _RX_LON, lat, lon) + + return _nearest(to_tx), _nearest((to_tx + 180.0) % 360.0) + + def test_a_declared_differential_limit_gives_an_ellipse(self): + """Long axis along the baseline: a target toward the TX is nearly on + the line between the two foci, where R_tx + R_rx - L barely grows, + while directly away from the TX the differential is 2r.""" + toward, away = self._reach_toward_and_away(self._state(60.0)) + assert away == pytest.approx(30.0, rel=0.01) # 2r <= 60 + assert toward > away * 1.5 + + def test_without_one_the_reach_is_a_circle(self): + toward, away = self._reach_toward_and_away(self._state(None)) + assert toward == pytest.approx(away, rel=0.01) + assert toward == pytest.approx(50.0, rel=0.01) + + +# ── What the manager publishes ─────────────────────────────────────────────── + + +def _manager_with_a_narrow_wedge(declared_truth): + """A node aimed due north with a 20 deg wedge, evidence on both sides. + + Mirrors test_learned_fov.TestFovOffPublishesEvidenceOnly's fixture — the + same node, registered the two different ways. + """ + m = NodeAnalyticsManager() # fov_mode defaults to "off" + m.register_node( + "N", + dict( + rx_lat=_RX_LAT, + rx_lon=_RX_LON, + tx_lat=_TX_LAT, + tx_lon=_TX_LON, + max_range_km=50, + beam_azimuth_deg=0.0, + beam_width_deg=20.0, + ), + declared_geometry_is_truth=declared_truth, + ) + ec = m.empirical_coverages["N"] + for bearing in (2.5, 182.5): # bin 0 (in wedge) and bin 36 (opposite) + for i in range(12): + ec.add_point(*_at_bearing(bearing, 20.0 + i * 0.01)) + return m + + +class TestManagerPublishesTheDeclaredWedge: + def test_a_flagged_node_publishes_only_in_wedge_vertices(self): + m = _manager_with_a_narrow_wedge(True) + cov = m.get_node_summary("N")["empirical_coverage"] + assert cov["polygon_source"] == "declared" + out_of_wedge = [ + (lat, lon) + for lat, lon in cov["polygon"] + if haversine_km(_RX_LAT, _RX_LON, lat, lon) > 1.0 + and abs(_angle_from(0.0, bearing_deg(_RX_LAT, _RX_LON, lat, lon))) > 10.0 + 1e-6 + ] + assert not out_of_wedge, "the southern lobe is mis-attributed evidence, not coverage" + + def test_it_is_the_declared_wedge_shape(self): + """Not merely "clipped" — the same list declared_wedge_polygon returns, + so the two cannot drift apart unnoticed.""" + m = _manager_with_a_narrow_wedge(True) + cov = m.get_node_summary("N")["empirical_coverage"] + assert cov["polygon"] == m.empirical_coverages["N"].declared_wedge_polygon() + + def test_the_evidence_counts_are_still_reported(self): + m = _manager_with_a_narrow_wedge(True) + cov = m.get_node_summary("N")["empirical_coverage"] + assert cov["n_points"] == 24 + assert cov["n_filled_bins"] == 2 + + def test_the_fov_diagnostics_are_omitted_for_a_flagged_node(self): + """They describe the learned wedge, which a declared-truth node does + not publish.""" + m = NodeAnalyticsManager(fov_mode="shadow") + m.register_node( + "N", + dict(rx_lat=_RX_LAT, rx_lon=_RX_LON, tx_lat=_TX_LAT, tx_lon=_TX_LON, max_range_km=50), + declared_geometry_is_truth=True, + ) + cov = m.get_node_summary("N")["empirical_coverage"] + assert "fov" not in cov + assert cov["polygon_source"] == "declared" + + def test_an_unflagged_node_still_publishes_the_evidence_only_shape(self): + m = _manager_with_a_narrow_wedge(False) + cov = m.get_node_summary("N")["empirical_coverage"] + assert cov["polygon_source"] == "evidence" + assert cov["polygon"] == m.empirical_coverages["N"].to_polygon(evidence_only=True) + + def test_a_flagged_node_with_no_calibration_points_still_publishes(self): + """to_polygon's MIN_POINTS gate is an evidence gate; the declared + wedge is known from registration, before any traffic flies.""" + m = NodeAnalyticsManager() + m.register_node( + "N", + dict(rx_lat=_RX_LAT, rx_lon=_RX_LON, tx_lat=_TX_LAT, tx_lon=_TX_LON, max_range_km=50), + declared_geometry_is_truth=True, + ) + cov = m.get_node_summary("N")["empirical_coverage"] + assert cov["n_points"] == 0 + assert cov["polygon"] + assert cov["polygon_source"] == "declared" + + def test_the_learned_source_is_named_under_fov_active(self): + m = NodeAnalyticsManager(fov_mode="active") + m.register_node("N", dict(rx_lat=_RX_LAT, rx_lon=_RX_LON, tx_lat=_TX_LAT, tx_lon=_TX_LON, max_range_km=50)) + assert m.get_node_summary("N")["empirical_coverage"]["polygon_source"] == "learned" + + +class TestFlagLifecycle: + def _cfg(self): + return dict(rx_lat=_RX_LAT, rx_lon=_RX_LON, tx_lat=_TX_LAT, tx_lon=_TX_LON, max_range_km=50) + + def test_re_registering_without_the_flag_drops_it(self): + m = NodeAnalyticsManager() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + m.register_node("N", self._cfg()) + assert "N" not in m._declared_truth + assert m.get_node_summary("N")["empirical_coverage"]["polygon_source"] == "evidence" + + def test_losing_geometry_drops_it(self): + m = NodeAnalyticsManager() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + m.register_node("N", {"rx_lat": _RX_LAT, "rx_lon": _RX_LON}, declared_geometry_is_truth=True) + assert "N" not in m._declared_truth + + def test_retire_node_drops_it(self): + m = NodeAnalyticsManager() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + m.retire_node("N") + assert "N" not in m._declared_truth + + def test_reset_for_tests_drops_it(self): + m = NodeAnalyticsManager() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + m._reset_for_tests() + assert "N" not in m._declared_truth + + def test_a_flag_change_invalidates_the_summaries_cache(self): + """get_all_summaries memoises for 60 s, so a flag flip that did not + also rebuild the detection area would otherwise be invisible until the + TTL expired.""" + m = NodeAnalyticsManager() + m.register_node("N", self._cfg()) + assert m.get_all_summaries()["N"]["empirical_coverage"]["polygon_source"] == "evidence" + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + assert m.get_all_summaries()["N"]["empirical_coverage"]["polygon_source"] == "declared" + + def test_an_unchanged_flag_leaves_the_cache_alone(self): + m = NodeAnalyticsManager() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + first = m.get_all_summaries() + m.register_node("N", self._cfg(), declared_geometry_is_truth=True) + assert m.get_all_summaries() is first