"""Regression for 006: coverage, pose freshness and stable accepted-frame display.""" import hashlib import json from types import SimpleNamespace import numpy as np import pytest from test_planning_live import event, fixture_service from k1link.missions.live_presentation import LivePresentation, stored_alignment from k1link.missions.reference_map import build_reference_map, reference_intervals def sample(t=100.0, x=0.0, segment=0): hint = np.eye(4) hint[:2, :2] = [[0, -1], [1, 0]] return dict( points=np.array([[x, 1.0, 0], [x, 2.0, 0]]), path=np.array([[0.0, 0, 0], [x, 0, 0]]), hint=hint, monotonic_ns=int(t * 1e9), sequence=int(t * 10), segment=segment, events=[], ) def fit(): matrix = np.eye(4) matrix[:3, 3] = [3, -2, 1] return dict( status="candidate", T_reference_query=matrix.tolist(), matched_query_indices=[0], overlap=0.9, inlier_rmse_m=0.1, reasons=[], ) def test_fresh_display_advances_without_reusing_correspondence_indices(): view = LivePresentation() accepted = sample() view.accept(fit(), accepted) assert view.result["matched_query_indices"] == [] latest = sample(t=101, x=3) view.advance(latest, accepted, int(101.1e9)) assert view.sample is latest assert view.result["T_reference_query"] == fit()["T_reference_query"] for current, qualifier, now in [ (sample(t=102, x=8), None, int(102.1e9)), (sample(t=102, x=8, segment=1), accepted, int(102.1e9)), (sample(t=99, x=8), accepted, int(100e9)), (sample(t=102, x=8), accepted, int(105e9)), (sample(t=109, x=8), accepted, int(109.1e9)), ]: view.advance(current, qualifier, now) assert view.sample is latest def test_no_hint_before_fit_and_no_rotation_after_loss_or_completion(tmp_path, monkeypatch): import k1link.missions.live_tests as module service, source, _, _ = fixture_service(tmp_path, monkeypatch) service.run = dict( id="test", state="running", query_session_id="B", query_generation=2, tracking_state="tracking", ) service.reference = np.zeros((3, 3)) service.reference_path = np.zeros((2, 3)) service.source = source source.state.update(active=True, session_id="B", session_generation=2) now = [int(100.1e9)] monkeypatch.setattr(module, "time", SimpleNamespace(monotonic_ns=lambda: now[0])) monkeypatch.setattr(service, "persist", lambda: None) monkeypatch.setattr(module, "scene_bytes", lambda *a, **kw: (a[3], a[4], kw.get("evidence"))) service.update_sample(sample(), now[0]) assert service.scene("test") == (None, None, None) accepted = sample() service.commit_result( fit(), accepted, {"accepted": True}, "tracking", phase="tracking", message="tracking", tracking_established=True, ) now[0] = int(101.1e9) latest = sample(t=101, x=2) service.update_sample(latest, now[0]) current, result, evidence = service.scene("test") assert current is latest and evidence[0] is accepted assert result["matched_query_indices"] == [] assert service.get()["presentation_state"] == "live" # Renderer freshness, not green extrapolation, controls the current label. now[0] = int(103.2e9) assert service.get()["presentation_state"] == "historical" assert service.scene("test")[2] is None service.commit_result( {**fit(), "status": "rejected"}, sample(t=104), {"accepted": False}, "lost", phase="lost", message="lost", tracking_established=True, ) service.update_sample(sample(t=105, x=9), int(105.1e9)) assert service.scene("test")[0] is latest assert service.scene("test")[1]["T_reference_query"] == fit()["T_reference_query"] service.run["state"] = "completed" source.state["active"] = False assert service.scene("test")[0] is latest and service.scene("test")[2] is None def test_pose_metadata_retained_independently_of_point_snapshot(tmp_path, monkeypatch): from dataclasses import replace service, _, _, _ = fixture_service(tmp_path, monkeypatch) pose = replace(event("pose", t=101, p=(4, 5, 6)), orientation_xyzw=(0, 0, 0.6, 0.8)) service.observe_pose(pose) assert service.latest_pose["orientation_xyzw"] == [0, 0, 0.6, 0.8] assert service.latest_pose["position"] == [4, 5, 6] assert service.latest_pose["frame_id"] == "session/B" assert service.sample is None def test_renderer_never_applies_hint_or_colors_unchecked_latest_points(monkeypatch): import k1link.missions.live_scene as renderer logged = {} class Recording: def __init__(self, *a, **kw): pass def binary_stream(self): return SimpleNamespace(read=lambda: b"RRF2") def log(self, name, value, **kw): logged[name] = value def flush(self): pass def disconnect(self): pass monkeypatch.setattr(renderer.rr, "RecordingStream", Recording) monkeypatch.setattr(renderer.rr, "Points3D", lambda p, **kw: (np.asarray(p), kw["colors"])) monkeypatch.setattr(renderer.rr, "LineStrips3D", lambda p, **kw: np.asarray(p)) raw = sample(x=5) renderer.scene_bytes("test", raw["points"], raw["path"], raw, None) assert not isinstance(logged["world/query"], tuple) renderer.scene_bytes("test", raw["points"], raw["path"], raw, fit(), evidence=(sample(), fit())) np.testing.assert_allclose(logged["world/query"][0], raw["points"] + [3, -2, 1]) assert not (logged["world/query"][1] == [154, 235, 75]).all(axis=1).any() np.testing.assert_allclose(logged["world/validated_query"][0], [[3, -1, 1]]) assert (logged["world/validated_query"][1] == [154, 235, 75]).all() renderer.scene_bytes("test", raw["points"], raw["path"], raw, fit()) assert not isinstance(logged["world/validated_query"], tuple) def seal(directory): return { str(p.relative_to(directory)): hashlib.sha256(p.read_bytes()).hexdigest() for p in directory.rglob("*") if p.is_file() } def test_saved_display_restores_exact_geometry_and_transform(tmp_path): view = LivePresentation() view.accept(fit(), sample()) view.advance(sample(t=101, x=8), sample(), int(101.1e9)) metadata = view.save(tmp_path) np.save(tmp_path / "reference.npy", np.zeros((3, 3))) doc = dict(presentation=metadata, artifacts=seal(tmp_path)) _, restored, result = stored_alignment(tmp_path, doc) np.testing.assert_equal(restored["points"], view.sample["points"]) assert result["T_reference_query"] == fit()["T_reference_query"] assert result["matched_query_indices"] == [] (tmp_path / "aligned-preview.npz").write_bytes(b"corrupt") with pytest.raises(ValueError, match="целостности"): stored_alignment(tmp_path, doc) def test_legacy_last_rejected_fit_does_not_replace_accepted_alignment(tmp_path): for n, accepted in [(1, True), (2, False)]: d = tmp_path / f"step-{n:03d}" d.mkdir() r = fit() r["status"] = "candidate" if accepted else "rejected" r["T_reference_query"][0][3] = n * 3 (d / "source.json").write_text( json.dumps(dict(sequence=n, query_path=[[0, 0, 0], [1, 0, 0]])) ) (d / "decision.json").write_text(json.dumps(dict(temporal=dict(accepted=accepted)))) (d / "registration-result.json").write_text(json.dumps(r)) np.savez(d / "registration-input.npz", reference=np.zeros((3, 3)), query=np.ones((3, 3))) doc = dict(artifacts=seal(tmp_path), result_source_sequence=2) _, _, result = stored_alignment(tmp_path, doc) assert result["T_reference_query"][0][3] == 3 (tmp_path / "step-001/decision.json").write_text( json.dumps(dict(temporal=dict(accepted=False))) ) doc["artifacts"] = seal(tmp_path) assert stored_alignment(tmp_path, doc) is None def poses(): return [dict(distance_m=i, position=[i, 0, 0]) for i in range(160)] @pytest.mark.parametrize( "start,end,expected", [(0, 30, [(0, 40), (40, 50)]), (50, 90, [(30, 70), (70, 110)]), (140, 159, [(120, 159)])], ) def test_context_is_independent_of_route_with_bounded_tiles(start, end, expected): assert reference_intervals(poses(), start, end) == expected def test_reference_context_rejects_bad_distances_but_not_long_route(): bad = poses() bad[4]["distance_m"] = -1 with pytest.raises(ValueError): reference_intervals(bad, 0, 30) assert reference_intervals(poses(), 0, 101)[-1][1] == 121 def test_context_map_preserves_source_bound_tiles_and_deduplicates(): calls = [] def extract(session, generation, a, b): calls.append((session, generation, a, b)) return np.array([[0, 0, 0], [a, 1, 0], [b, 1, 0]], dtype=float), dict( session_id=session, generation=generation, start_index=a, end_index=b ) sources = SimpleNamespace(bound=lambda s, g: dict(poses=poses()), submap=extract) points, metadata = build_reference_map(sources, "A", "hash", 0, 30) assert calls == [("A", "hash", 0, 40), ("A", "hash", 40, 50)] assert metadata["route_interval"] == [0, 30] and metadata["map_interval"] == [0, 50] assert len(points) == 4 and len(metadata["tiles"]) == 2 def test_context_does_not_hide_source_changes_between_tiles(): calls = [] def extract(*args): calls.append(args) if len(calls) > 1: raise ValueError("source changed") return np.zeros((3, 3)), {} sources = SimpleNamespace(bound=lambda *a: dict(poses=poses()), submap=extract) with pytest.raises(ValueError, match="source changed"): build_reference_map(sources, "A", "hash", 0, 30)