feat(observatory): admit canonical recorded replay

scripts/publish_canonical_recorded_lab_projection.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Publish one exact sealed RAV004 result into the Session/LAB catalog."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from k1link.laboratory.canonical_recorded_catalog import (
+    publish_canonical_recorded_vegetation_result,
+)
+from k1link.sessions import SessionStore
+from k1link.sessions.store import resolve_missioncore_data_dir
+
+
+def _arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Verify and publish an immutable canonical recorded LAB result. "
+            "This command does not build RRD, run inference or alter source evidence."
+        )
+    )
+    parser.add_argument("--repository-root", required=True, type=Path)
+    parser.add_argument("--result-id", required=True)
+    return parser.parse_args()
+
+
+def main() -> None:
+    arguments = _arguments()
+    repository_root = arguments.repository_root.expanduser().absolute()
+    if repository_root.is_symlink():
+        raise SystemExit("repository root must not be a symlink")
+    try:
+        repository_root = repository_root.resolve(strict=True)
+        runtime_root = (
+            repository_root / ".runtime" / "compute-experiments"
+        ).resolve(strict=True)
+    except OSError as exc:
+        raise SystemExit("canonical repository runtime is unavailable") from exc
+    if not repository_root.is_dir() or not runtime_root.is_dir():
+        raise SystemExit("canonical repository runtime is unavailable")
+    if not runtime_root.is_relative_to(repository_root):
+        raise SystemExit("canonical repository runtime escaped its repository root")
+    data_dir = resolve_missioncore_data_dir(repository_root)
+    database_path = data_dir / "mission-core.sqlite3"
+    if database_path.is_symlink() or not database_path.is_file():
+        raise SystemExit("existing Mission Core catalog is unavailable")
+    result_root = (
+        runtime_root
+        / "lab-v1-vegetation"
+        / "results"
+        / arguments.result_id
+    )
+    binding = publish_canonical_recorded_vegetation_result(
+        store=SessionStore(repository_root, data_dir=data_dir),
+        runtime_root=runtime_root,
+        result_root=result_root,
+    )
+    print(json.dumps(binding.as_dict(), ensure_ascii=False, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()