Reduce reputation cost and expose empirical coverage freshness

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
@@ -362,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
@@ -375,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]
@@ -393,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
@@ -426,6 +432,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 +851,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

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):
@@ -517,13 +518,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 +539,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 +556,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,
@@ -615,6 +622,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")),
             }

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

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)

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))