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.
134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
"""Run one bounded causal experiment against an immutable saved reference.
|
|
|
|
Use the repository virtual environment. This CLI never connects to hardware or
|
|
the application ingress. Inputs and reports belong in private runtime storage.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import platform
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from k1link.device_plugins.xgrids_k1.planning_replay import iter_planning_events
|
|
from k1link.missions.causal_replay import digest, replay
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--reference-run", required=True, type=Path)
|
|
parser.add_argument("--query-raw", required=True, type=Path)
|
|
parser.add_argument("--query-planning", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
parser.add_argument(
|
|
"--mode", choices=["baseline", "tracking", "acquisition"], default="baseline"
|
|
)
|
|
args = parser.parse_args()
|
|
started = time.monotonic()
|
|
repository = Path(__file__).resolve().parents[1]
|
|
code_paths = [Path(__file__).resolve()] + [
|
|
repository / name
|
|
for name in (
|
|
"src/k1link/missions/entry_acquisition.py",
|
|
"src/k1link/missions/entry_acquisition_worker.py",
|
|
"src/k1link/missions/causal_replay.py",
|
|
"src/k1link/missions/causal_tracking.py",
|
|
"src/k1link/missions/live_buffer.py",
|
|
"src/k1link/missions/registration.py",
|
|
"src/k1link/missions/registration_worker.py",
|
|
"src/k1link/device_plugins/xgrids_k1/planning_replay.py",
|
|
"src/k1link/device_plugins/xgrids_k1/planning_live.py",
|
|
)
|
|
]
|
|
code_hashes = {str(p.relative_to(repository)): digest(p) for p in code_paths}
|
|
original = json.loads((args.reference_run / "report.json").read_text())
|
|
query = json.loads(args.query_planning.read_text())
|
|
if query["session_id"] == original["reference"]["session_id"]:
|
|
raise ValueError("Independent replay requires a different query recording.")
|
|
files = {
|
|
args.reference_run / "report.json": digest(args.reference_run / "report.json"),
|
|
args.reference_run / "clouds.npz": original["artifacts"]["clouds.npz"],
|
|
args.query_raw: query["source_digests"]["raw-transport-primary"],
|
|
args.query_raw.with_name("mqtt.metadata.jsonl"): query["source_digests"][
|
|
"raw-transport-index"
|
|
],
|
|
args.query_planning: digest(args.query_planning),
|
|
}
|
|
for path, expected in files.items():
|
|
if digest(path) != expected:
|
|
raise ValueError(f"Source digest mismatch: {path.name}")
|
|
with np.load(args.reference_run / "clouds.npz", allow_pickle=False) as archive:
|
|
# Deliberately do not load the fitted query, its path, or the final transform.
|
|
reference = archive["reference"]
|
|
reference_path = archive["reference_path"]
|
|
prep = time.monotonic() - started
|
|
report = replay(
|
|
iter_planning_events(args.query_raw, query["session_id"]),
|
|
reference,
|
|
reference_path,
|
|
args.output,
|
|
mode=args.mode,
|
|
)
|
|
for path, expected in files.items():
|
|
if digest(path) != expected:
|
|
report.update(state="invalid", source_integrity_verified=False)
|
|
(args.output / "report.json").write_text(json.dumps(report, allow_nan=False))
|
|
raise ValueError("Source changed during replay.")
|
|
report.update(
|
|
reference_run_id=original["id"],
|
|
reference=original["reference"],
|
|
query={k: query[k] for k in ["session_id", "generation", "source_digests", "label"]},
|
|
input_digests={str(path): value for path, value in files.items()},
|
|
source_integrity_verified=True,
|
|
implementation_sha256=code_hashes,
|
|
reference_preparation_s=prep,
|
|
runtime=dict(
|
|
system=platform.system(), machine=platform.machine(), python=platform.python_version()
|
|
),
|
|
)
|
|
(args.output / "report.json").write_text(json.dumps(report, allow_nan=False))
|
|
print(
|
|
json.dumps(
|
|
{
|
|
k: report[k]
|
|
for k in [
|
|
"mode",
|
|
"state",
|
|
"elapsed_s",
|
|
"distance_m",
|
|
"first_heading_s",
|
|
"first_candidate_s",
|
|
"first_tracking_s",
|
|
"source_integrity_verified",
|
|
]
|
|
}
|
|
),
|
|
flush=True,
|
|
)
|
|
print(
|
|
json.dumps(
|
|
[
|
|
dict(
|
|
step=s["step"],
|
|
at=s["requested_s"],
|
|
distance=s["distance_m"],
|
|
seed=s["seed"],
|
|
status=s["result"]["status"],
|
|
temporal=s["temporal"],
|
|
overlap=s["result"].get("overlap"),
|
|
rmse=s["result"].get("inlier_rmse_m"),
|
|
fit_s=s["result"].get("registration_seconds"),
|
|
state=s["tracking_state"],
|
|
)
|
|
for s in report["steps"]
|
|
]
|
|
),
|
|
flush=True,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|