From 42d897c8a4e26da6ef1cd124dac27f6851d08f6a Mon Sep 17 00:00:00 2001 From: Babissimo Date: Tue, 25 Aug 2026 13:16:36 +0100 Subject: [PATCH] tracker: clamp frame dt and reject non-positive-definite gates (86cb9bq7p) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One chain, three links, no symptom at any of them: an out-of-order frame makes dt negative, a negative dt makes the innovation covariance non-positive-definite, the gate check tested the wrong property so such a track kept gating, and its Mahalanobis distances came out negative -- which wins a minimisation outright. The corrupted track then took detections from every healthy track it could reach, with no exception, no log and no counter. In a 40-frame run it absorbs another target's detections permanently and ends up ACTIVE, positive definite and indistinguishable from healthy, reporting the wrong aircraft. dt is clamped to [0, 60] and clamps are counted in `n_dt_clamped`. Zero rather than a nominal step for the lower bound: at dt=0, F is the identity and Q is zero, so the predict is a bitwise no-op and the update runs from the last good state. An out-of-order frame carries no usable time relationship, and substituting a nominal +0.5 s would advance the prediction in the wrong temporal direction while inflating P, loosening the gate at exactly the wrong moment. Repeated zero-dt cycles decay P harmonically rather than geometrically and one ordinary frame reinflates it, so there is no collapse risk. Nothing divides by this dt; the four anomaly detectors compute their own from raw timestamps and already guard `dt > 0`. Sixty for the upper bound, which is a cost trade rather than a safe point. Q grows as dt cubed, so the gate already spans the whole observed delay axis by about 23 s and no tolerable bound prevents that -- only wall-clock track ageing does, and that is a ticket of its own. What the bound decides is which failure you take: the clamp under-predicts by the truncated interval while injecting only the truncated process noise, so too tight a bound is its own regression. At 10 s tracking dies once a real gap exceeds ~20 s, and since empty frames are dropped upstream, multi-second droughts are ordinary -- that would cause by another route the very behaviour change the empty-frame ticket defers. Sixty keeps drought recovery to ~120 s. The clock now advances as a high-water mark rather than to whatever the last frame happened to carry, because clamping alone closes only half the path. A frame 6 s late is clamped to 0 and counted, but still leaves `last_timestamp` at T-6 s, so the next legitimate frame at T+1 computes 7 s. That is inside the bounds, so it is neither clamped nor counted, and Q[0,0] comes out 343x too large with a gate to match: one out-of-order frame corrupted two frames while the counter owned up to one. Nothing depends on the old semantics -- `last_timestamp` is read only to form this dt and as the clock in tests, and no other repo touches it. A non-finite timestamp is rejected outright, at the top of process_frame, before anything reads it. `min(max(nan, 0.0), 60.0)` returns nan, because Python's min and max yield the first operand whenever the comparison is false and every comparison against a NaN is, so the derived dt needed a guard either way. Guarding the dt and the clock alone was not enough: the raw stamp also reaches track IDs, `death_timestamp` and the merge-window cutoff. A NaN there sets `death_timestamp` on every coasting track, so `get_quality_score` returns nan; it makes the cutoff nan, so every comparison against it is false and `all_tracks` drains into `completed_tracks` in a single frame, emptying the merge working set; and on the frame a tentative track promotes, `datetime.fromtimestamp` raises straight out of `process_frame`, ValueError for a NaN and OverflowError for an infinity. One rejection at entry closes all three, and keeps the guard in one place rather than in each consumer of the stamp. Rejections are counted in `n_frames_rejected` rather than folded into `n_dt_clamped`: an unusable frame and an out-of-order but usable one are different failures, and a single counter for both cannot tell an operator which is happening. The mark yields after MAX_BACKWARDS_RUN consecutive non-advancing frames, because a one-way ratchet is a failure of its own. Barring an infinity from the mark is not sufficient: any implausible but finite future stamp pins it just as well, and every later frame then clamps to dt=0, so the filter never predicts again. Nothing upstream stops one arriving; the TCP ingest path checks only that a timestamp is present, never its value. Measured over 60 ordinary frames after a single stamp a day ahead: with the bare ratchet the clock stays pinned, 61 of 61 frames clamp and one aircraft fragments into 10 tracks, permanently; with the run bound it resyncs once and recovers to a single track. Three rather than one, because at one this is just deleting the mark, and because two interleaved node clocks alternate rather than run, so they keep it. The rationale above lives here rather than beside the code, because `CLAUDE.md` and `.claude/rules/code-style.md` both bar comments outright and ask names and structure to carry the intent. So `MAX_FRAME_DT` gains its unit as `MAX_FRAME_DT_S`, `MAX_BACKWARDS_RUN` says what it triggers as `BACKWARDS_RUN_BEFORE_RESYNC`, and the gate's mask is `pos_def` rather than `valid`. Both constants are new here, so neither rename can reach a consumer. Two things the deleted comments had been carrying are now carried by code instead. The gate omits the `d > 0` term of Sylvester's criterion, which is sound only while S is diagonal, so `TestGateStructuralInvariant` pins that the covariance stays block-diagonal through predict and update rather than leaving the precondition to prose: injecting a 1e-9 cross term into P fails it. And the entry guard is widened to `numbers.Real`, because `math.isfinite` raises off a real number rather than returning False, and `process_streaming_frame` reads the stamp out of an external frame without checking its type, so a node sending an ISO-8601 string crashed the frame rather than having it dropped. Test isolation goes in alongside, the suite having had none. `set_config` writes a module global that every accessor reads at call time, and with no conftest a file that set it supplied the config for every later file in collection order: 74 of the 75 tests ran under a config inherited from somewhere else. The fixture snapshots and restores it, reading the private global rather than `get_config()`, which assigns when the value is None and would corrupt the state it is meant to save. The gate now applies Sylvester's criterion rather than the sign of the determinant alone. The audit prescribed `det_S > 1e-15`, which is necessary but not sufficient: H P Hᵀ is symmetric and neither F nor Q couples delay to Doppler, so b = c = 0 and det_S = a*d exactly. Two negative diagonal entries therefore multiply to a positive determinant, and that -- not an indefinite matrix -- is what a late frame actually produces on a converged track inside the bridge's ±10 s staleness window. The determinant test cannot see it. The last test here fails with `det_S > 1e-15` and passes with `a > 0` added, which is the whole argument in one assertion. The cost lower bound goes in as prescribed. Costs are a Mahalanobis distance times a strictly positive SNR weight, so a negative cost is never legitimate however it arose. It cannot be reached through `_associate` while the gate holds, since a positive-definite S cannot yield a negative distance, and mutation-testing confirms as much: reverting it alone leaves the whole suite green. No test can protect it, so the comment beside it states the constraint instead: a distance times a positive weight is never negative. Association counts will shift slightly: detections that previously received a garbage negative distance are now excluded rather than preferred. That is the intent, not a regression. None of this had test coverage -- every existing frame steps 500 ms -- so the behaviours are pinned here rather than left to be rediscovered. Every production change is mutation-tested except the cost bound, which is the one the gate makes unreachable. Co-Authored-By: Claude Opus 5 --- retina_tracker/tracker.py | 46 ++++- tests/conftest.py | 22 +++ tests/test_frame_dt_and_gating.py | 300 +++++++++++++++++++++++++++++ tests/test_timestamp_validation.py | 278 ++++++++++++++++++++++++++ 4 files changed, 637 insertions(+), 9 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_frame_dt_and_gating.py create mode 100644 tests/test_timestamp_validation.py diff --git a/retina_tracker/tracker.py b/retina_tracker/tracker.py index b0398ae..d586a7f 100644 --- a/retina_tracker/tracker.py +++ b/retina_tracker/tracker.py @@ -1,5 +1,7 @@ """Core Tracker class and GNN data association logic.""" +import math +import numbers from collections import deque import numpy as np @@ -19,6 +21,8 @@ MERGE_WINDOW_MS = 5000 MAX_COMPLETED_TRACKS = 5000 +MAX_FRAME_DT_S = 60.0 +BACKWARDS_RUN_BEFORE_RESYNC = 3 class Tracker: @@ -31,7 +35,7 @@ def __init__(self, event_writer=None, detection_window=20, config=None): self.completed_tracks = deque(maxlen=MAX_COMPLETED_TRACKS) self.last_timestamp = None self.detection_window = detection_window - self.frame_count = 0 + self._reset_counters() self.event_writer = event_writer self.config = config if config else get_config() @@ -50,13 +54,32 @@ def reset(self): self.all_tracks = [] self.completed_tracks.clear() self.last_timestamp = None + self._reset_counters() + + def _reset_counters(self): self.frame_count = 0 + self.n_dt_clamped = 0 + self.n_frames_rejected = 0 + self.n_clock_resyncs = 0 + self.n_backwards = 0 def process_frame(self, detections, timestamp): + """Advance every track by one frame, `timestamp` in milliseconds. + + A timestamp that is not a finite number drops the frame. + """ self.frame_count += 1 + if not isinstance(timestamp, numbers.Real) or not math.isfinite(timestamp): + self.n_frames_rejected += 1 + return + if self.last_timestamp is not None: - dt = (timestamp - self.last_timestamp) / 1000.0 + raw_dt = (timestamp - self.last_timestamp) / 1000.0 + dt = min(max(raw_dt, 0.0), MAX_FRAME_DT_S) + if dt != raw_dt: + self.n_dt_clamped += 1 + self.n_backwards = self.n_backwards + 1 if raw_dt <= 0 else 0 else: dt = 0.5 @@ -176,7 +199,12 @@ def process_frame(self, detections, timestamp): if len(self.all_tracks) > 1: self._merge_tracks() - self.last_timestamp = timestamp + resync = self.n_backwards >= BACKWARDS_RUN_BEFORE_RESYNC + if self.last_timestamp is None or timestamp > self.last_timestamp or resync: + if resync: + self.n_clock_resyncs += 1 + self.n_backwards = 0 + self.last_timestamp = timestamp def _associate(self, detections): if not self.tracks or not detections: @@ -212,8 +240,8 @@ def _associate(self, detections): b = B[0, 1] c = B[1, 0] det_S = a * d - b * c - valid = np.abs(det_S) > 1e-15 - if not np.any(valid): + pos_def = (a > 0) & (det_S > 1e-15) + if not np.any(pos_def): continue gate = base_gate @@ -226,9 +254,9 @@ def _associate(self, detections): nu0 = innovations[:, 0] nu1 = innovations[:, 1] mahal = np.full(len(detections), np.inf) - mahal[valid] = ( - d[valid] * nu0[valid] ** 2 - (b + c) * nu0[valid] * nu1[valid] + a[valid] * nu1[valid] ** 2 - ) / det_S[valid] + mahal[pos_def] = ( + d[pos_def] * nu0[pos_def] ** 2 - (b + c) * nu0[pos_def] * nu1[pos_def] + a[pos_def] * nu1[pos_def] ** 2 + ) / det_S[pos_def] within_gate = mahal < gate if not np.any(within_gate): @@ -243,7 +271,7 @@ def _associate(self, detections): row_ind, col_ind = linear_sum_assignment(cost_matrix) - associations = [(r, c) for r, c in zip(row_ind, col_ind) if cost_matrix[r, c] < 1e6] + associations = [(r, c) for r, c in zip(row_ind, col_ind) if 0 <= cost_matrix[r, c] < 1e6] return associations diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e905818 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,22 @@ +"""Restore the global config between tests. + +`retina_tracker.config` keeps the config in a module global that `set_config` +replaces and every accessor (`MIN_SNR()`, `GATE_THRESHOLD()`, ...) reads at call +time, so without this a file that sets it supplies the config for every later +file in collection order. The snapshot is deep, because the config is nested and +a test that mutates a subsection in place would otherwise write through a +shallow copy into the next test. +""" + +import copy + +import pytest + +from retina_tracker import config as config_module + + +@pytest.fixture(autouse=True) +def _isolate_global_config(): + saved = copy.deepcopy(config_module._config) + yield + config_module.set_config(saved) diff --git a/tests/test_frame_dt_and_gating.py b/tests/test_frame_dt_and_gating.py new file mode 100644 index 0000000..b0ff7fc --- /dev/null +++ b/tests/test_frame_dt_and_gating.py @@ -0,0 +1,300 @@ +"""An out-of-order frame must not let one track out-compete every other. + +The chain: dt is a bare subtraction of two timestamps, so a frame that +arrives late makes it negative; a negative dt makes F P Fᵀ + Q +non-positive-definite; the gate check tested only that S was invertible, so +such a track kept gating; and its Mahalanobis distances came out negative, +which wins a minimisation. The corrupted track then took detections from +healthy tracks with no exception, no log and no counter. + +A large positive dt is the mirror image: Q grows as dt³, so after a node +outage the gate is wider than the delay axis and everything falls inside it. +Tracks are deleted by missed-frame count rather than by elapsed time, so +nothing ages out across the gap to stop it. + +Clamping alone does not close either: the clock has to stop rewinding too, or +the frame after a clamped one computes the whole excursion as its dt, and a +non-finite timestamp passes straight through min and max untouched. +""" + +import numpy as np +import pytest + +from retina_tracker.config import set_config +from retina_tracker.tracker import MAX_FRAME_DT_S, Tracker + + +def build_config(): + return { + "tracker": { + "m_threshold": 4, + "n_window": 20, + "n_delete": 20, + "n_coast": 3, + "min_snr": 7.0, + "gate_threshold": 9.0, + "detection_window": 20, + }, + "process_noise": {"delay": 0.1, "doppler": 0.5}, + "tracklet": {"max_delay_residual": 2.0, "max_doppler_residual": 10.0, "max_time_span": 3.0}, + "adsb": { + "enabled": False, + "priority": True, + "reference_location": None, + "initial_covariance": {"position": 100.0, "velocity": 5.0}, + }, + "radar": {"center_frequency": 200000000}, + } + + +def _det(delay, doppler, snr=15.0): + return {"delay": delay, "doppler": doppler, "snr": snr} + + +def _settled_tracker(): + """A single track carried far enough that its covariance has converged.""" + config = build_config() + set_config(config) + tracker = Tracker(config=config) + for i in range(40): + tracker.process_frame([_det(10.0 + 0.2 * i, -70.0 + 10.0 * i)], i * 1000) + return tracker + + +class TestDtClamp: + def test_backwards_frame_is_clamped_and_counted(self): + tracker = _settled_tracker() + before = tracker.n_dt_clamped + tracker.process_frame([_det(18.0, 320.0)], tracker.last_timestamp - 6000) + assert tracker.n_dt_clamped == before + 1 + + def test_long_gap_is_clamped_and_counted(self): + tracker = _settled_tracker() + before = tracker.n_dt_clamped + tracker.process_frame([_det(18.0, 320.0)], tracker.last_timestamp + 300_000) + assert tracker.n_dt_clamped == before + 1 + + def test_ordinary_cadence_is_not_clamped(self): + tracker = _settled_tracker() + before = tracker.n_dt_clamped + tracker.process_frame([_det(18.0, 320.0)], tracker.last_timestamp + 1000) + assert tracker.n_dt_clamped == before + + def test_covariance_stays_positive_definite_after_a_backwards_frame(self): + """The whole failure chain starts here: without the clamp this goes negative.""" + tracker = _settled_tracker() + tracker.process_frame([_det(18.0, 320.0)], tracker.last_timestamp - 6000) + assert tracker.tracks + for track in tracker.tracks: + assert np.all(np.linalg.eigvalsh(track.covariance) > 0) + + def test_gap_recovery_survives_a_drought_shorter_than_the_clamp(self): + """Empty frames are dropped upstream, so multi-second droughts are normal.""" + tracker = _settled_tracker() + ts = tracker.last_timestamp + int(MAX_FRAME_DT_S * 1000) - 1000 + tracker.process_frame([_det(18.0, 320.0)], ts) + assert tracker.tracks, "the track must survive a gap inside the clamp window" + + def test_a_long_gap_predicts_exactly_the_clamped_interval(self): + """Pins the bound itself, not just that some bound applies. + + Process noise grows as dt cubed, so the covariance after the gap is what + distinguishes a clamp at 60 s from one at 10 s or 600 s. + """ + clamped = _settled_tracker() + clamped.process_frame([], clamped.last_timestamp + 3_600_000) + + reference = _settled_tracker() + reference.process_frame([], reference.last_timestamp + 60_000) + + assert MAX_FRAME_DT_S == 60.0 + assert clamped.tracks[0].covariance == pytest.approx(reference.tracks[0].covariance) + + def test_frame_after_a_clamped_one_gets_the_ordinary_dt(self): + """Clamping the late frame is wasted if it still rewinds the clock. + + A frame 6 s late is clamped to 0 and counted, but leaves the clock at + T-6 s, so the next legitimate frame computes 7 s: inside the bounds, so + neither clamped nor counted, and 343x the process noise. + + Both sides coast the same number of frames on purpose: covariance + growth freezes above N_COAST misses, so an asymmetric coast would + diverge for a reason that has nothing to do with the clock. + """ + late = _settled_tracker() + t = late.last_timestamp + late.process_frame([], t - 6000) + late.process_frame([], t + 1000) + + reference = _settled_tracker() + t_ref = reference.last_timestamp + reference.process_frame([], t_ref) # dt = 0 without needing the clamp + reference.process_frame([], t_ref + 1000) + + assert late.tracks and reference.tracks + assert late.tracks[0].n_missed == reference.tracks[0].n_missed + assert late.tracks[0].covariance == pytest.approx(reference.tracks[0].covariance), ( + "dt = 0 predicts nothing, so the clamped frame must leave the covariance alone" + ) + + def test_counter_resets_with_the_tracker(self): + tracker = _settled_tracker() + tracker.process_frame([_det(18.0, 320.0)], tracker.last_timestamp - 6000) + assert tracker.n_dt_clamped > 0 + tracker.reset() + assert tracker.n_dt_clamped == 0 + + +class TestNonFiniteTimestamp: + """A NaN defeats min/max entirely, so the clamp has to reject it by name. + + `min(max(nan, 0.0), 60.0)` is nan: Python's min and max return the first + operand whenever the comparison is false, which every NaN comparison is. + The nan then reaches F and Q and destroys every track on the node, while + `dt != raw_dt` reads true forever and the counter reports clamping working. + """ + + def test_nan_timestamp_leaves_the_filter_finite(self): + tracker = _settled_tracker() + before = tracker.tracks[0].covariance.copy() + + tracker.process_frame([], float("nan")) + + assert tracker.tracks + assert tracker.tracks[0].covariance == pytest.approx(before) + + def test_infinite_timestamp_leaves_the_filter_finite(self): + tracker = _settled_tracker() + before = tracker.tracks[0].covariance.copy() + + tracker.process_frame([], float("inf")) + + assert tracker.tracks + assert tracker.tracks[0].covariance == pytest.approx(before) + + def test_non_finite_timestamp_does_not_become_the_clock(self): + """Otherwise no later frame can ever pass the mark, and tracking stops.""" + tracker = _settled_tracker() + t = tracker.last_timestamp + + tracker.process_frame([], float("nan")) + tracker.process_frame([], float("inf")) + + assert tracker.last_timestamp == t + + def test_the_counter_stops_once_the_finite_frames_resume(self): + """The counter must report the bad frames, not every frame after them.""" + tracker = _settled_tracker() + tracker.process_frame([], float("nan")) + assert tracker.n_frames_rejected == 1 + + for i in range(1, 4): + tracker.process_frame([_det(18.0 + i, 320.0 + 10.0 * i)], tracker.last_timestamp + 1000) + + assert tracker.n_frames_rejected == 1 + + def test_tracking_recovers_after_a_non_finite_frame(self): + tracker = _settled_tracker() + t = tracker.last_timestamp + tracker.process_frame([], float("nan")) + tracker.process_frame([], t + 1000) + + reference = _settled_tracker() + reference.process_frame([], reference.last_timestamp + 1000) + + assert tracker.tracks and reference.tracks + assert tracker.tracks[0].covariance == pytest.approx(reference.tracks[0].covariance) + + +def _make_negative_definite(track, snr=15.0): + """Drive S negative definite, as an out-of-order frame does. + + S = H P Hᵀ + R·noise_scale, so pushing the two variances below their + measurement-noise terms flips both diagonal entries negative while leaving + det_S = a*d positive (the state a determinant-sign test cannot see). + """ + noise_scale = 1.0 / max(10 ** (snr / 10) / 10, 0.1) + covariance = track.covariance.copy() + covariance[0, 0] = -track.kf.R[0, 0] * noise_scale - 1.0 + covariance[2, 2] = -track.kf.R[1, 1] * noise_scale - 1.0 + track.covariance = covariance + + base = track.get_innovation_base() + a = base[0, 0] + track.kf.R[0, 0] * noise_scale + d = base[1, 1] + track.kf.R[1, 1] * noise_scale + assert a < 0 and d < 0 and a * d - base[0, 1] * base[1, 0] > 1e-15, ( + "fixture guard: holds only while the calling test's detections all share this snr" + ) + return track + + +class TestCorruptTrackTakesNothing: + """The property the gate and the cost lower bound exist to protect. + + They are layered, and the first two tests here show it: a corrupt track on + its own is rejected by either change alone, so that pair only both-fails + when both are reverted. Only the last test separates them: with the + determinant test in place of the definiteness one, the corrupt track still + takes a healthy track's detection. + """ + + def test_negative_definite_covariance_wins_nothing(self): + tracker = _settled_tracker() + _make_negative_definite(tracker.tracks[0]) + + associations = tracker._associate([_det(22.8, 380.0), _det(48.0, -155.0)]) + + assert associations == [] + + def test_healthy_track_still_associates(self): + """The added a > 0 term must not reject legitimate detections.""" + tracker = _settled_tracker() + z = tracker.tracks[0].kf.H @ tracker.tracks[0].state + + associations = tracker._associate([_det(float(z[0]), float(z[1]))]) + + assert associations == [(0, 0)] + + def test_corrupt_track_does_not_steal_a_healthy_track_detection(self): + config = build_config() + set_config(config) + tracker = Tracker(config=config) + + for i in range(40): + tracker.process_frame( + [_det(10.0 + 0.2 * i, -70.0 + 10.0 * i), _det(60.0 - 0.2 * i, 400.0 - 5.0 * i)], + i * 1000, + ) + assert len(tracker.tracks) >= 2 + + healthy = tracker.tracks[0] + _make_negative_definite(tracker.tracks[1]) + z = healthy.kf.H @ healthy.state + wanted = _det(float(z[0]), float(z[1])) + + associations = tracker._associate([wanted]) + + claimed = [track_i for track_i, det_i in associations if det_i == 0] + assert claimed == [tracker.tracks.index(healthy)] + + +class TestGateStructuralInvariant: + """The gate omits the `d > 0` term of Sylvester's criterion, which is sound + only while S is diagonal. F, Q and H never couple the delay and Doppler + subspaces and both init paths are diagonal, so nothing else in the suite + would notice a change that coupled them and made the gate admit an + indefinite S again.""" + + def test_the_covariance_stays_block_diagonal(self): + tracker = _settled_tracker() + for track in tracker.tracks: + for i, j in [(0, 2), (0, 3), (1, 2), (1, 3)]: + assert track.covariance[i, j] == 0.0 + assert track.covariance[j, i] == 0.0 + + def test_the_innovation_base_therefore_has_no_off_diagonal(self): + tracker = _settled_tracker() + for track in tracker.tracks: + base = track.get_innovation_base() + assert base[0, 1] == 0.0 + assert base[1, 0] == 0.0 diff --git a/tests/test_timestamp_validation.py b/tests/test_timestamp_validation.py new file mode 100644 index 0000000..934a896 --- /dev/null +++ b/tests/test_timestamp_validation.py @@ -0,0 +1,278 @@ +"""A frame's timestamp is read by far more than the dt clamp. + +Past the clamp the raw value reaches datetime.fromtimestamp() when a track +promotes and takes an ID, death_timestamp on every track the frame missed, and +the merge-window cutoff. A non-finite stamp defeats all three: the ID +generation raises out of process_frame, the cutoff comparison is false for +every track so the whole merge working set empties in one frame, and a NaN +death_timestamp makes the quality score NaN. Guarding only the two derived +values (dt, and the clock mark) leaves each of those open, so the frame is +rejected at the entry instead. + +The clock mark has the mirror problem. It only ever advances, so one finite but +implausible future timestamp pins it: every later frame is then backwards, +clamps to dt = 0, and the filter stops predicting for good. It resyncs after a +run of frames that never passes it. A run rather than a single frame, because +two node clocks interleaved into one stream alternate above and below the mark +indefinitely without either being wrong. +""" + +import numpy as np +import pytest + +from retina_tracker.config import set_config +from retina_tracker.tracker import BACKWARDS_RUN_BEFORE_RESYNC, Tracker + +NON_FINITE = [float("nan"), float("inf"), float("-inf")] +NON_NUMERIC = [None, "1700000000000", [1], {}] + +N_WINDOW = 20 +CADENCE_OUTSPANNING_TRACKLETS_MS = 2000 + + +def build_config(**overrides): + config = { + "tracker": { + "m_threshold": 4, + "n_window": N_WINDOW, + "n_delete": 20, + "n_coast": 3, + "min_snr": 7.0, + "gate_threshold": 9.0, + "detection_window": 20, + }, + "process_noise": {"delay": 0.1, "doppler": 0.5}, + "tracklet": {"max_delay_residual": 2.0, "max_doppler_residual": 10.0, "max_time_span": 3.0}, + "adsb": { + "enabled": False, + "priority": True, + "reference_location": None, + "initial_covariance": {"position": 100.0, "velocity": 5.0}, + }, + "radar": {"center_frequency": 200000000}, + } + config["tracker"].update(overrides) + return config + + +def _det(delay, doppler, snr=15.0): + return {"delay": delay, "doppler": doppler, "snr": snr} + + +def _tracker(): + config = build_config() + set_config(config) + return Tracker(config=config) + + +def _settled_tracker(n_frames=40, cadence_ms=1000): + tracker = _tracker() + for i in range(n_frames): + tracker.process_frame([_det(10.0 + 0.2 * i, -70.0 + 10.0 * i)], i * cadence_ms) + return tracker + + +def _tentative_tracker(n_frames): + """A track held TENTATIVE until promote_if_ready() takes it at N_WINDOW. + + The cadence puts any three detections 4 s apart, past TRACKLET_MAX_TIME_SPAN, + so tracklet initiation never fires and never promotes the track early. + """ + tracker = _tracker() + for i in range(n_frames): + tracker.process_frame([_det(10.0 + 0.2 * i, -70.0 + 10.0 * i)], i * CADENCE_OUTSPANNING_TRACKLETS_MS) + return tracker + + +class TestNonFiniteFrameCarryingDetections: + """Every existing non-finite test passes an empty detection list, which is + the one shape of frame that never reaches the timestamp's other readers.""" + + @pytest.mark.parametrize("timestamp", NON_FINITE) + def test_a_non_finite_frame_carrying_detections_does_not_raise(self, timestamp): + tracker = _settled_tracker() + + tracker.process_frame([_det(18.0, 320.0), _det(52.0, -140.0)], timestamp) + + assert tracker.tracks + + @pytest.mark.parametrize("timestamp", NON_FINITE) + def test_a_non_finite_frame_on_the_promotion_frame_does_not_raise(self, timestamp): + """Track._generate_id calls datetime.fromtimestamp(t / 1000.0), which + rejects NaN and overflows on an infinity.""" + tracker = _tentative_tracker(N_WINDOW - 1) + track = tracker.tracks[0] + assert track.id is None and track.n_frames == N_WINDOW - 1 + + tracker.process_frame([_det(10.0 + 0.2 * 19, -70.0 + 10.0 * 19)], timestamp) + + assert track.n_frames == N_WINDOW - 1, "the rejected frame must not advance the track" + + @pytest.mark.parametrize("timestamp", NON_FINITE) + def test_the_promotion_still_happens_on_the_next_good_frame(self, timestamp): + """Rejecting the frame defers the promotion rather than losing it.""" + tracker = _tentative_tracker(N_WINDOW - 1) + track = tracker.tracks[0] + + tracker.process_frame([_det(10.0 + 0.2 * 19, -70.0 + 10.0 * 19)], timestamp) + tracker.process_frame([_det(10.0 + 0.2 * 19, -70.0 + 10.0 * 19)], 19 * CADENCE_OUTSPANNING_TRACKLETS_MS) + + assert track.id is not None + assert "NAN" not in track.id.upper() + + +class TestNonFiniteFrameLeavesStateIntact: + def test_a_nan_frame_does_not_drain_the_merge_working_set(self): + """cutoff = timestamp - MERGE_WINDOW_MS is NaN, so `death_timestamp >= + cutoff` is false for every entry and all_tracks empties in one frame.""" + tracker = _settled_tracker() + for i in range(40, 65): + tracker.process_frame([], i * 1000) + assert tracker.all_tracks and not tracker.completed_tracks + + before = list(tracker.all_tracks) + tracker.process_frame([], float("nan")) + + assert tracker.all_tracks == before + assert not tracker.completed_tracks + + @pytest.mark.parametrize("timestamp", NON_FINITE) + def test_a_non_finite_frame_does_not_poison_the_quality_score(self, timestamp): + """mark_missed() sets death_timestamp on every track the frame missed, + and get_quality_score() divides by it.""" + tracker = _settled_tracker() + track = tracker.tracks[0] + death, quality = track.death_timestamp, track.get_quality_score() + assert np.isfinite(death) and np.isfinite(quality) + + tracker.process_frame([], timestamp) + + assert track.death_timestamp == death + assert track.get_quality_score() == quality + + def test_a_rejected_frame_is_still_counted_as_a_frame(self): + tracker = _settled_tracker() + before = tracker.frame_count + + tracker.process_frame([], float("nan")) + + assert tracker.frame_count == before + 1 + + @pytest.mark.parametrize("timestamp", NON_NUMERIC) + def test_a_non_numeric_timestamp_drops_the_frame(self, timestamp): + """math.isfinite raises rather than returning False off a real number, + and server.process_streaming_frame reads the stamp straight out of an + external frame without checking its type.""" + tracker = _settled_tracker() + clock = tracker.last_timestamp + rejected = tracker.n_frames_rejected + + tracker.process_frame([_det(18.0, 330.0)], timestamp) + + assert tracker.n_frames_rejected == rejected + 1 + assert tracker.last_timestamp == clock + + def test_a_rejected_frame_is_counted_apart_from_a_clamped_dt(self): + """An unusable frame and an out-of-order but usable one are different + failures, and one counter for both cannot tell an operator which.""" + tracker = _settled_tracker() + + tracker.process_frame([], tracker.last_timestamp - 6000) + assert (tracker.n_dt_clamped, tracker.n_frames_rejected) == (1, 0) + + tracker.process_frame([], float("nan")) + assert (tracker.n_dt_clamped, tracker.n_frames_rejected) == (1, 1) + + +class TestClockResync: + """The high-water mark stops a clamped frame rewinding the clock, but on + its own it is a one-way ratchet with no way back down.""" + + def test_a_far_future_frame_does_not_pin_the_clock_for_good(self): + tracker = _settled_tracker() + base = tracker.last_timestamp + tracker.process_frame([], base + 86_400_000) # a day ahead of the stream + clamped = tracker.n_dt_clamped + + for i in range(1, 11): + tracker.process_frame([_det(18.0 + 0.2 * i, 320.0 + 10.0 * i)], base + i * 1000) + + assert tracker.last_timestamp == base + 10_000 + assert tracker.n_clock_resyncs == 1 + assert tracker.n_dt_clamped - clamped == BACKWARDS_RUN_BEFORE_RESYNC, ( + "recovery costs the run and nothing after it" + ) + + def test_the_filter_predicts_again_after_a_resync(self): + """While the mark is pinned every dt is 0, so predict is F = I with + Q = 0 and the covariance never moves.""" + tracker = _settled_tracker() + base = tracker.last_timestamp + tracker.process_frame([], base + 86_400_000) + for i in range(1, 11): + tracker.process_frame([_det(18.0 + 0.2 * i, 320.0 + 10.0 * i)], base + i * 1000) + + track = next(t for t in tracker.tracks if t.n_missed == 0) + assert track.n_associated >= 5, "the pinned track loses the target and the stream respawns it" + before = np.diag(track.covariance).copy() + next_frame_on_the_streams_own_clock = base + 11_000 + tracker.process_frame([], next_frame_on_the_streams_own_clock) + + assert np.all(np.diag(track.covariance) > before) + + def test_interleaved_node_clocks_do_not_resync(self): + """Two nodes 30 s apart in one stream alternate either side of the mark + forever. Alternating is not a run, and neither clock is the outlier.""" + tracker = _settled_tracker() + base = tracker.last_timestamp + + for i in range(1, 31): + tracker.process_frame([_det(18.0 + 0.2 * i, 320.0 + 10.0 * i)], base + i * 1000) + assert tracker.n_backwards < BACKWARDS_RUN_BEFORE_RESYNC + tracker.process_frame([_det(60.0 + 0.2 * i, -300.0 + 10.0 * i)], base + 30_000 + i * 1000) + assert tracker.n_backwards < BACKWARDS_RUN_BEFORE_RESYNC + + assert tracker.n_clock_resyncs == 0 + + def test_a_duplicate_timestamp_counts_toward_the_run(self): + """A stream stuck on one stamp advances no more than a backwards one + does, and clamping never fires on it because dt is already 0.""" + tracker = _settled_tracker() + stuck = tracker.last_timestamp + clamped = tracker.n_dt_clamped + + for _ in range(BACKWARDS_RUN_BEFORE_RESYNC): + tracker.process_frame([], stuck) + + assert tracker.n_backwards == 0 + assert tracker.n_clock_resyncs == 1 + assert tracker.n_dt_clamped == clamped + + def test_a_non_finite_frame_does_not_count_toward_the_run(self): + """It is dropped, not late, so it says nothing about the mark.""" + tracker = _settled_tracker() + stuck = tracker.last_timestamp + tracker.process_frame([], stuck) + tracker.process_frame([], stuck) + assert tracker.n_backwards == BACKWARDS_RUN_BEFORE_RESYNC - 1 + + tracker.process_frame([], float("nan")) + + assert tracker.n_backwards == BACKWARDS_RUN_BEFORE_RESYNC - 1 + assert tracker.n_clock_resyncs == 0 + + def test_the_resync_counters_reset_with_the_tracker(self): + tracker = _settled_tracker() + stuck = tracker.last_timestamp + for _ in range(BACKWARDS_RUN_BEFORE_RESYNC - 1): + tracker.process_frame([], stuck) + assert tracker.n_backwards > 0 + + tracker.process_frame([], float("nan")) + assert tracker.n_frames_rejected > 0 + + tracker.reset() + + assert tracker.n_backwards == 0 + assert tracker.n_clock_resyncs == 0 + assert tracker.n_frames_rejected == 0