From c9c9f603797c12b975b4b2c965c8cc6c7456e37c Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Sat, 19 Sep 2026 12:56:22 +0000 Subject: [PATCH 1/3] Reuse each node trust score throughout a reputation pass --- src/retina_analytics/manager.py | 20 +++++++++++++------- tests/test_analytics.py | 19 +++++++++++++++++++ 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/src/retina_analytics/manager.py b/src/retina_analytics/manager.py index 0314326..2996fd0 100644 --- a/src/retina_analytics/manager.py +++ b/src/retina_analytics/manager.py @@ -517,13 +517,19 @@ def record_node_tracks(self, node_id: str, confirmed_ids, geolocated_ids=()): m.record_tracks(confirmed_ids, geolocated_ids) def evaluate_reputations(self): - for node_id, rep in self.reputations.items(): + reputations = dict(self.reputations) + # A trust score scans up to 500 residuals. Recomputing it for every + # overlapping neighbour turns one pass into O(nodes² * samples). + # Use one per-node snapshot throughout this pass; the next pass sees + # new samples and threshold changes without a persistent cache. + trust_by_node = {node_id: _trust_for_reputation(self.trust_scores.get(node_id)) for node_id in reputations} + for node_id, rep in reputations.items(): # Below TRUST_MIN_SAMPLES there is no score to act on: the neutral # 0.5 prior the backend substitutes there sits between the warn # threshold and the reward threshold, so it would neither penalise # nor reward — skipping is equivalent, and it keeps one claim # residual from starting a block. - trust = _trust_for_reputation(self.trust_scores.get(node_id)) + trust = trust_by_node[node_id] if trust is not None: rep.evaluate_trust(trust) @@ -532,7 +538,7 @@ def evaluate_reputations(self): rep.evaluate_heartbeat(metrics.last_heartbeat) rep.evaluate_detection_rate(metrics.avg_detections_per_frame) - node_ids = sorted(self.reputations.keys()) + node_ids = sorted(reputations) for i, a_id in enumerate(node_ids): for b_id in node_ids[i + 1 :]: area_a = self.detection_areas.get(a_id) @@ -549,14 +555,14 @@ def evaluate_reputations(self): # bar it gets the neutral prior, which is under the 0.7 # trusted threshold, so a one-sample neighbour can no # longer condemn anyone. - trust_a = _trust_for_reputation(self.trust_scores.get(a_id)) - trust_b = _trust_for_reputation(self.trust_scores.get(b_id)) - self.reputations[a_id].evaluate_neighbour_consistency( + trust_a = trust_by_node[a_id] + trust_b = trust_by_node[b_id] + reputations[a_id].evaluate_neighbour_consistency( overlap["overlap_ratio"], _NEUTRAL_TRUST_PRIOR if trust_b is None else trust_b, neighbour_id=b_id, ) - self.reputations[b_id].evaluate_neighbour_consistency( + reputations[b_id].evaluate_neighbour_consistency( overlap["overlap_ratio"], _NEUTRAL_TRUST_PRIOR if trust_a is None else trust_a, neighbour_id=a_id, diff --git a/tests/test_analytics.py b/tests/test_analytics.py index c0c1598..72f0f6c 100644 --- a/tests/test_analytics.py +++ b/tests/test_analytics.py @@ -375,6 +375,25 @@ def test_evaluate_reputations(self, mgr): # live reputation object after a pass. assert all(nid in mgr.reputations for nid in mgr.trust_scores) + def test_trust_is_scanned_once_per_node_per_pass_and_refreshed_next_pass(self, mgr, monkeypatch): + calls = [] + score = TrustScoreState.score.fget + + def counted_score(state): + calls.append(state.node_id) + return score(state) + + monkeypatch.setattr(TrustScoreState, "score", property(counted_score)) + for nid in ("node-A", "node-B"): + mgr.record_detection_frame(nid, {"delay": [50], "doppler": [20], "snr": [15]}) + for i in range(3): + mgr.record_adsb_correlation(nid, self._bad_sample(i)) + mgr.evaluate_reputations() + assert sorted(calls) == ["node-A", "node-B"] + calls.clear() + mgr.evaluate_reputations() + assert sorted(calls) == ["node-A", "node-B"] + def test_cross_node_analysis(self, mgr): cross = mgr.get_cross_node_analysis() assert "blocked_nodes" in cross From 253219ca693f7570af3a2dbb7acc5a2b3811b290 Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Sun, 20 Sep 2026 00:40:08 +0000 Subject: [PATCH 2/3] Expose coverage evidence freshness and align display range admission --- src/retina_analytics/empirical_coverage.py | 17 +++++++++++++++-- src/retina_analytics/manager.py | 1 + tests/test_empirical_coverage.py | 9 +++++++++ tests/test_manager_registration.py | 19 +++++++++++++++++++ 4 files changed, 44 insertions(+), 2 deletions(-) diff --git a/src/retina_analytics/empirical_coverage.py b/src/retina_analytics/empirical_coverage.py index 29d52c2..1c45588 100644 --- a/src/retina_analytics/empirical_coverage.py +++ b/src/retina_analytics/empirical_coverage.py @@ -1,7 +1,7 @@ """Empirical detection-area characterisation built from known-position calibration points. Instead of assuming a fixed Yagi-like antenna lobe, this module accumulates -ground-truth target positions (from ADS-B or multinode-solver solutions) that a +independent target positions (from detection-backed ADS-B references) that a node has positively detected, then derives a smoothed coverage polygon that reflects the node's *actual* detection area as observed over time. @@ -17,6 +17,11 @@ 4. A circular rolling average (window = 3 bins) smooths the resulting vector. 5. Polygon vertices are computed at each bin centre and returned as [[lat, lon]]. +This is a typical observed footprint, not the outer envelope of every accepted +detection and not a hard limit on detectability. Per-bin histories retain only +200 points, so a flat n_points does not imply that recording has stopped; +last_detection_ts reports the newest accepted evidence independently. + The polygon is only returned once at least MIN_POINTS calibration points have been recorded; below that there is no published detection area at all. The map draws nothing rather than a theoretical Yagi sector: a sector drawn from a @@ -426,6 +431,11 @@ def record_disappearance(self, lat: float, lon: float, ts: float | None = None) def n_points(self) -> int: return sum(len(b) for b in self._bins) + @property + def last_detection_ts(self) -> float | None: + """Newest accepted evidence time, even when the bounded counts are full.""" + return max((ts for ts in self._bin_last_pos_ts if math.isfinite(ts) and ts > 0), default=None) + @property def n_filled_bins(self) -> int: return sum(1 for b in self._bins if b) @@ -840,7 +850,10 @@ def _evidence_only_polygon(self) -> list[list[float]] | None: ranges.append(0.0) continue bearing_i = (i + 0.5) * _DEG_PER_BIN - ranges.append(min(_p85(b), self._reach_at(bearing_i) * self.range_clamp_mult)) + # Match the recorder's admissible range. A configured monostatic + # radius is only a guess: add_point admits evidence up to 4x it, + # so a separate 2x display clamp silently hides accepted evidence. + ranges.append(min(_p85(b), self._reach_at(bearing_i) * self._admit_mult())) if not any(r > 0.0 for r in ranges): return None diff --git a/src/retina_analytics/manager.py b/src/retina_analytics/manager.py index 2996fd0..0c704c4 100644 --- a/src/retina_analytics/manager.py +++ b/src/retina_analytics/manager.py @@ -621,6 +621,7 @@ def get_node_summary(self, node_id: str) -> dict: result["empirical_coverage"] = { "n_points": ec.n_points, "n_filled_bins": ec.n_filled_bins, + "last_detection_ts": ec.last_detection_ts, "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")), } diff --git a/tests/test_empirical_coverage.py b/tests/test_empirical_coverage.py index 8a33a31..776c45d 100644 --- a/tests/test_empirical_coverage.py +++ b/tests/test_empirical_coverage.py @@ -269,6 +269,15 @@ def _fill(self, cov, bins, n_points=6, range_km=RANGE_KM): def _is_apex(self, cov, vertex): return vertex == [round(cov.rx_lat, 5), round(cov.rx_lon, 5)] + def test_display_keeps_evidence_admitted_beyond_twice_the_configured_radius(self): + cov = EmpiricalCoverageState(RX_LAT, RX_LON, max_range_km=10.0) + self._fill(cov, range(N_BINS), range_km=30.0) + poly = cov.to_polygon(evidence_only=True) + assert poly is not None + for lat, lon in poly[:-1]: + _, distance = _bearing_and_range(RX_LAT, RX_LON, lat, lon) + assert 29.5 < distance < 30.5 # was silently clipped to 20 km + def test_full_coverage_is_a_ring_with_no_apex(self): cov = EmpiricalCoverageState(RX_LAT, RX_LON) self._fill(cov, range(N_BINS), n_points=FOV_OPEN_MIN_POINTS) diff --git a/tests/test_manager_registration.py b/tests/test_manager_registration.py index 2d1ad52..2b959e7 100644 --- a/tests/test_manager_registration.py +++ b/tests/test_manager_registration.py @@ -23,6 +23,25 @@ def test_manager_honors_explicit_aim(): assert m.detection_areas["N"].beam_azimuth_deg == 123.0 +def test_coverage_freshness_advances_when_retained_point_count_is_full(): + from retina_analytics.empirical_coverage import EmpiricalCoverageState + + m = NodeAnalyticsManager() + m.register_node("N", dict(_CFG)) + assert m.get_node_summary("N")["empirical_coverage"]["last_detection_ts"] is None + for ts in range(1, 202): + m.record_calibration_point("N", _RX_LAT + 0.1, _RX_LON, ts=float(ts)) + summary = m.get_node_summary("N")["empirical_coverage"] + assert summary["n_points"] == 200 + assert summary["last_detection_ts"] == 201.0 + ec = m.empirical_coverages["N"] + restored = EmpiricalCoverageState.from_dict(ec.to_dict()) + assert restored.last_detection_ts == 201.0 + # An implausibly distant sample cannot make a stalled recorder look fresh. + restored.add_point(_RX_LAT + 10, _RX_LON, ts=202.0) + assert restored.last_detection_ts == 201.0 + + def test_manager_defaults_to_broadside_when_unaimed(): m = NodeAnalyticsManager() m.register_node("N", dict(_CFG)) From 73c34ac190f699d5898d918754ae821120c22339 Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Sun, 20 Sep 2026 00:42:37 +0000 Subject: [PATCH 3/3] Report whether coverage evidence was actually retained --- src/retina_analytics/empirical_coverage.py | 7 ++++--- src/retina_analytics/manager.py | 5 +++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/retina_analytics/empirical_coverage.py b/src/retina_analytics/empirical_coverage.py index 1c45588..fe900ff 100644 --- a/src/retina_analytics/empirical_coverage.py +++ b/src/retina_analytics/empirical_coverage.py @@ -367,7 +367,7 @@ def _admit_mult(self) -> float: """ return self.range_clamp_mult if self.max_bistatic_range_km else FOV_CLAMP_MULT_MONOSTATIC - def add_point(self, lat: float, lon: float, ts: float | None = None) -> None: + def add_point(self, lat: float, lon: float, ts: float | None = None) -> bool: """Record one calibration point (known target position). ts defaults to wall-clock time; the learned-FOV shrink logic @@ -380,9 +380,9 @@ def add_point(self, lat: float, lon: float, ts: float | None = None) -> None: """ bearing, range_km = _bearing_and_range(self.rx_lat, self.rx_lon, lat, lon) if range_km < 0.5: - return # too close — not informative + return False # too close — not informative if range_km > self._reach_at(bearing) * self._admit_mult(): - return # implausibly far — mis-attributed detection + return False # implausibly far — mis-attributed detection i = _bin_for_bearing(bearing) t = ts if ts is not None else time.time() b = self._bins[i] @@ -398,6 +398,7 @@ def add_point(self, lat: float, lon: float, ts: float | None = None) -> None: # observed_limit_km's neighbour widening); clearing all N is simpler # and costs nothing measurable against the _p85 it saves. self._observed_limit_cache = [_UNSET] * N_BINS + return True def record_disappearance(self, lat: float, lon: float, ts: float | None = None) -> bool: """Record one negative-evidence event: a target predicted detectable at diff --git a/src/retina_analytics/manager.py b/src/retina_analytics/manager.py index 0c704c4..a8449be 100644 --- a/src/retina_analytics/manager.py +++ b/src/retina_analytics/manager.py @@ -464,7 +464,7 @@ def is_node_blocked(self, node_id: str) -> bool: rep = self.reputations.get(node_id) return rep.blocked if rep else False - def record_calibration_point(self, node_id: str, lat: float, lon: float, ts: float | None = None) -> None: + def record_calibration_point(self, node_id: str, lat: float, lon: float, ts: float | None = None) -> bool: """Record a detection at an independently-known target position. ADS-B only. Callers used to pass solver output here, which made the @@ -480,7 +480,8 @@ def record_calibration_point(self, node_id: str, lat: float, lon: float, ts: flo """ ec = self.empirical_coverages.get(node_id) if ec is not None: - ec.add_point(lat, lon, ts=ts) + return ec.add_point(lat, lon, ts=ts) + return False def record_detection_frame(self, node_id: str, frame: dict): if self.is_node_blocked(node_id):