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.
116 lines
5.0 KiB
Python
116 lines
5.0 KiB
Python
"""Intentional STOP is not stale localisation; no sockets or device commands."""
|
|
|
|
import json
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from test_planning_live import event, fixture_service, until
|
|
from test_xgrids_acquisition_lifecycle import (
|
|
_install_real_prepared_stop_dispatch_fixture,
|
|
service_with_fake_runtime,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("late_error", [False, True])
|
|
def test_stop_during_search_finishes_before_late_worker_without_publishing_loss(
|
|
tmp_path, monkeypatch, late_error,
|
|
):
|
|
import k1link.missions.live_tests as module
|
|
|
|
service, source, lock, _ = fixture_service(tmp_path, monkeypatch)
|
|
clock = [100.]
|
|
entered, release = threading.Event(), threading.Event()
|
|
monkeypatch.setattr(module, "time", SimpleNamespace(
|
|
monotonic=lambda: clock[0], monotonic_ns=lambda: int(clock[0] * 1e9)))
|
|
|
|
def search(*args, **kwargs):
|
|
entered.set()
|
|
assert release.wait(5)
|
|
if late_error:
|
|
raise ValueError("late worker failure")
|
|
return dict(status="candidate", T_reference_query=np.eye(4).tolist())
|
|
|
|
monkeypatch.setattr(module, "run_route_relocalization", search)
|
|
run = service.start("draft", 1)
|
|
points = np.random.default_rng(11).uniform([-1, -3, -1], [8, 3, 3], (1500, 3))
|
|
try:
|
|
until(lambda: service.get()["state"] == "waiting")
|
|
source.state.update(active=True, session_id="B", session_generation=2)
|
|
for i in range(20):
|
|
source.queue.put(event("pose", t=100+i*.5, sequence=2*i+1))
|
|
source.queue.put(event("points", t=100+i*.5+.001, sequence=2*i+2, points=points))
|
|
until(source.queue.empty)
|
|
clock[0] = 110.01
|
|
until(entered.is_set)
|
|
until(lambda: service.get()["planning_phase"] == "searching")
|
|
source.state["spatial_stop_requested"] = True
|
|
until(lambda: service.get()["planning_phase"] == "ended")
|
|
assert service.thread.is_alive() # Fit ownership not released prematurely.
|
|
assert service.accepted_sample is None
|
|
release.set()
|
|
until(lambda: not service.thread.is_alive())
|
|
report = service.get()
|
|
assert report["state"] == "completed"
|
|
assert report["termination_reason"] == "spatial-stop-requested"
|
|
assert report["result"] is None and report.get("initialization_result") is None
|
|
assert not any(t["phase"] == "lost" for t in report["phase_transitions"])
|
|
assert source.state["active"] and not lock.locked()
|
|
decision = json.loads((service.directory(run["id"]) / "step-001/decision.json").read_text())
|
|
assert not decision["temporal"]["accepted"]
|
|
assert decision["temporal"]["reason"] == (
|
|
"calculation-unavailable" if late_error else "spatial-stop-requested")
|
|
finally:
|
|
release.set()
|
|
service.close()
|
|
|
|
|
|
def test_stop_before_first_cloud_keeps_capture_running(tmp_path, monkeypatch):
|
|
service, source, lock, _ = fixture_service(tmp_path, monkeypatch)
|
|
service.start("draft", 1)
|
|
try:
|
|
until(lambda: service.get()["state"] == "waiting")
|
|
source.state.update(active=True, session_id="B", session_generation=2,
|
|
spatial_stop_requested=True)
|
|
source.queue.put(event("session-start"))
|
|
until(lambda: not service.thread.is_alive())
|
|
assert service.get()["termination_reason"] == "spatial-stop-requested"
|
|
assert source.state["active"] and not lock.locked()
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
def test_scene_bounds_are_cached_per_immutable_reference(tmp_path, monkeypatch):
|
|
service, _, _, _ = fixture_service(tmp_path, monkeypatch)
|
|
service.reference = np.array([[0., 0., -2.], [1., 1., 4.]])
|
|
first = service.scene_height_bounds()
|
|
assert first == (-2., 80.)
|
|
assert service.scene_height_bounds() is first
|
|
service.scene_reference = np.array([[0., 0., -4.], [1., 1., np.nan]])
|
|
assert service.scene_height_bounds() == (-4., 80.)
|
|
service.scene_reference = np.empty((0, 3))
|
|
assert service.scene_height_bounds() == (None, None)
|
|
service.scene_reference = service.reference = None
|
|
assert service.scene_height_bounds() == (None, None)
|
|
|
|
|
|
@pytest.mark.parametrize("rejected", [False, True])
|
|
def test_only_admitted_device_stop_fences_planning(tmp_path, monkeypatch, rejected):
|
|
device, runtime = service_with_fake_runtime(tmp_path)
|
|
fixture = _install_real_prepared_stop_dispatch_fixture(device, runtime)
|
|
ingress = device.live_perception_ingress
|
|
assert ingress.snapshot()["active"]
|
|
assert not ingress.snapshot()["spatial_stop_requested"]
|
|
if rejected:
|
|
def reject(**kwargs):
|
|
raise ValueError("fixture rejected before dispatch")
|
|
monkeypatch.setattr(fixture.control, "request_stop", reject)
|
|
with pytest.raises(ValueError, match="fixture rejected"):
|
|
device.stop_acquisition(fixture.request)
|
|
else:
|
|
device.stop_acquisition(fixture.request)
|
|
assert ingress.snapshot()["active"] # Recording finalisation is independent.
|
|
assert ingress.snapshot()["spatial_stop_requested"] is not rejected
|
|
assert runtime.stop_calls == 0
|