DCCONSTRUCTIONS
e515ab1b8c
Preserve the completed teach-and-repeat laboratory stage: reference preparation, cascaded acquisition, local tracking and recovery, recording lifecycle, replay qualification, and persistent Rerun scene controls. Document the open grid-picking regression and Rerun upgrade contract. No autonomous driving or loop-closure optimization is claimed.
161 lines
6.1 KiB
Python
161 lines
6.1 KiB
Python
"""Tiny deterministic fixtures: temporal authority and causality, not load."""
|
|
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.missions.causal_replay import replay
|
|
from k1link.missions.causal_tracking import CausalTracking
|
|
from k1link.sessions.live_planning import PlanningLiveEvent
|
|
|
|
|
|
def sample(t, segment=0):
|
|
return dict(monotonic_ns=int(t * 1e9), segment=segment, path=np.array([[0.0, 0, 0], [3, 0, 0]]))
|
|
|
|
|
|
def result(x=0, status="candidate"):
|
|
matrix = np.eye(4)
|
|
matrix[0, 3] = x
|
|
return dict(status=status, T_reference_query=matrix.tolist())
|
|
|
|
|
|
def test_three_consistent_candidates_and_timeout():
|
|
gate = CausalTracking()
|
|
for t in (1, 6, 11):
|
|
assert gate.accept(result(), sample(t), int((t + 1) * 1e9), 0)["accepted"]
|
|
assert gate.state == "tracking" and gate.streak == 3
|
|
gate.tick(20_000_000_000, 0)
|
|
assert gate.state == "lost" and gate.matrix is None and gate.reason == "stale"
|
|
|
|
|
|
def test_inconsistent_candidate_is_not_seeded_or_counted():
|
|
gate = CausalTracking()
|
|
gate.accept(result(), sample(1), 2_000_000_000, 0)
|
|
rejection = gate.accept(result(0.6), sample(6), 7_000_000_000, 0)
|
|
assert not rejection["accepted"] and rejection["reason"] == "inconsistent-candidate"
|
|
assert gate.matrix is None and gate.streak == 0
|
|
gate.accept(result(0.6), sample(11), 12_000_000_000, 0)
|
|
assert gate.streak == 1 and gate.state == "acquiring"
|
|
|
|
|
|
def test_old_segment_result_does_not_override_current_candidate():
|
|
gate = CausalTracking()
|
|
gate.accept(result(), sample(1), 2_000_000_000, 0)
|
|
gate.tick(3_000_000_000, 1)
|
|
assert gate.matrix is None
|
|
gate.accept(result(0.1), sample(4, 1), 5_000_000_000, 1)
|
|
old = gate.accept(result(10), sample(2), 6_000_000_000, 1)
|
|
assert old["reason"] == "old-segment" and not old["accepted"]
|
|
assert gate.streak == 1 and gate.matrix[0, 3] == pytest.approx(0.1)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"status,stamp,now,reason",
|
|
[
|
|
("rejected", 1, 2, "registration-rejected"),
|
|
("candidate", 1, 10, "stale-result"),
|
|
("candidate", 3, 2, "stale-result"),
|
|
],
|
|
)
|
|
def test_bad_or_stale_fit_cannot_establish_tracking(status, stamp, now, reason):
|
|
gate = CausalTracking()
|
|
assert not gate.accept(result(status=status), sample(stamp), int(now * 1e9), 0)["accepted"]
|
|
assert gate.reason == reason and gate.matrix is None
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["baseline", "acquisition"])
|
|
def test_replay_fit_never_contains_future_points(tmp_path, mode):
|
|
points = np.random.default_rng(11).uniform([0, -3, -1], [8, 3, 3], (1500, 3))
|
|
start = PlanningLiveEvent("B", 1, 1, 1_000_000_000, 1, "pose", position=(0, 0, 0))
|
|
events = [
|
|
start,
|
|
replace(start, sequence=2, monotonic_ns=1_001_000_000, position=(3, 0, 0)),
|
|
replace(start, kind="points", sequence=3, monotonic_ns=1_002_000_000, points=points),
|
|
replace(start, sequence=4, monotonic_ns=1_100_000_000, position=(3.1, 0, 0)),
|
|
replace(start, kind="points", sequence=5, monotonic_ns=1_110_000_000, points=points + 100),
|
|
]
|
|
calls = []
|
|
|
|
def calculate(directory, ref, query, hint):
|
|
calls.append(query.copy())
|
|
return result()
|
|
|
|
report = replay(
|
|
iter(events),
|
|
points,
|
|
np.array([[0, 0, 0], [30, 0, 0]]),
|
|
tmp_path / "replay",
|
|
calculate=calculate,
|
|
initialize=lambda directory, ref, query, hint, anchor, forward: calculate(
|
|
directory, ref, query, hint
|
|
),
|
|
mode=mode,
|
|
max_seconds=0.2,
|
|
)
|
|
assert len(calls) == 1 and calls[0].max() < 20
|
|
assert report["steps"][0]["sequence"] == 3
|
|
assert max(e["sequence"] for e in report["steps"][0]["source_events"]) == 3
|
|
assert report["transitions"][-1]["reason"] == "input-ended"
|
|
assert json.loads((tmp_path / "replay/report.json").read_text())["vehicle_control"] is False
|
|
|
|
|
|
def test_empty_or_oversized_replay_fails(tmp_path):
|
|
with pytest.raises(ValueError, match="bounds"):
|
|
replay([], [], [], tmp_path / "too-long", max_seconds=121)
|
|
with pytest.raises(ValueError, match="Empty"):
|
|
replay([], [], [], tmp_path / "empty")
|
|
|
|
|
|
def test_job_finishing_after_input_end_is_historical_only(tmp_path):
|
|
import time
|
|
|
|
points = np.random.default_rng(21).uniform([0, -3, -1], [8, 3, 3], (1500, 3))
|
|
start = PlanningLiveEvent("B", 1, 1, 1_000_000_000, 1, "pose", position=(0, 0, 0))
|
|
events = [
|
|
start,
|
|
replace(start, sequence=2, monotonic_ns=1_001_000_000, position=(3, 0, 0)),
|
|
replace(start, kind="points", sequence=3, monotonic_ns=1_002_000_000, points=points),
|
|
]
|
|
|
|
def delayed(directory, reference, query, hint):
|
|
time.sleep(0.05)
|
|
return result()
|
|
|
|
report = replay(
|
|
iter(events),
|
|
points,
|
|
np.array([[0, 0, 0], [30, 0, 0]]),
|
|
tmp_path / "ended",
|
|
calculate=delayed,
|
|
max_seconds=0.2,
|
|
)
|
|
assert report["first_candidate_s"] is None
|
|
assert report["steps"][0]["temporal"]["reason"] == "input-ended"
|
|
assert not report["steps"][0]["temporal"]["accepted"]
|
|
assert report["transitions"][-1]["time_s"] < report["steps"][0]["completed_s"]
|
|
|
|
|
|
def test_archived_adapter_requires_real_clock_and_reuses_live_decoder(tmp_path):
|
|
from test_stream_summary import _pose_payload
|
|
from test_viewer_replay import _write_native
|
|
|
|
from k1link.device_plugins.xgrids_k1.planning_replay import iter_planning_events
|
|
|
|
source = tmp_path / "mqtt.raw.k1mqtt"
|
|
_write_native(source, "x/lio_pose", _pose_payload((5, 1, 2)))
|
|
with pytest.raises(ValueError, match="metadata"):
|
|
list(iter_planning_events(source, "B"))
|
|
meta = dict(
|
|
record_type="message", sequence=1, received_at_epoch_ns=10, received_monotonic_ns=20
|
|
)
|
|
source.with_name("mqtt.metadata.jsonl").write_text(json.dumps(meta) + "\n")
|
|
events = list(iter_planning_events(source, "B"))
|
|
assert len(events) == 1 and events[0].position == (5.0, 1.0, 2.0)
|
|
assert events[0].monotonic_ns == 20 and events[0].epoch_ns == 10
|
|
del meta["received_monotonic_ns"]
|
|
source.with_name("mqtt.metadata.jsonl").write_text(json.dumps(meta) + "\n")
|
|
with pytest.raises(ValueError, match="clock"):
|
|
list(iter_planning_events(source, "B"))
|