From af1e530162db3f020d072ea2cc9798fed0b4a868 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 28 Aug 2026 23:13:48 +0300 Subject: [PATCH] feat(perception): add mixed-route vegetation review --- .../src/core/laboratory/vegetationShadow.ts | 129 +++++- .../laboratory/VegetationShadowResult.tsx | 128 ++++++ .../annotation/useL34AnnotationCapability.tsx | 28 +- .../test/vegetationShadow.test.mjs | 3 +- config/laboratory-value-review.json | 2 +- .../perception/mixed-route-tgs-review-v1.json | 68 +++ .../prepare_mixed_route_lidar_pack.py | 418 ++++++++++++++++++ .../run_mixed_route_ddrnet_islands.py | 340 ++++++++++++++ .../build_mixed_route_tgs_evidence.py | 277 ++++++++++++ .../prepare_mixed_route_tgs_inputs.py | 243 ++++++++++ scripts/prepare_mixed_route_review_pack.py | 296 +++++++++++++ .../mixed_route_vegetation_review.py | 417 +++++++++++++++++ 12 files changed, 2332 insertions(+), 17 deletions(-) create mode 100644 config/perception/mixed-route-tgs-review-v1.json create mode 100644 experiments/perception/prepare_mixed_route_lidar_pack.py create mode 100644 experiments/perception/worker/lab_v1_vegetation_goose/run_mixed_route_ddrnet_islands.py create mode 100644 experiments/perception/worker/m49_t3_travel/build_mixed_route_tgs_evidence.py create mode 100644 experiments/perception/worker/m49_t3_travel/prepare_mixed_route_tgs_inputs.py create mode 100644 scripts/prepare_mixed_route_review_pack.py create mode 100644 src/k1link/laboratory/mixed_route_vegetation_review.py diff --git a/apps/control-station/src/core/laboratory/vegetationShadow.ts b/apps/control-station/src/core/laboratory/vegetationShadow.ts index d3c962b..08d8669 100644 --- a/apps/control-station/src/core/laboratory/vegetationShadow.ts +++ b/apps/control-station/src/core/laboratory/vegetationShadow.ts @@ -79,6 +79,33 @@ export interface VegetationRouteVideo { validFovMaskSha256: string | null; } +export interface VegetationMixedRouteCase { + caseId: string; + phase: "rural" | "transition" | "urban"; + sourceSequence: number; + sessionSeconds: number; + assets: Readonly>; + tgs: { + groundCells: number; + occupiedCells: number; + rejectedCells: number; + unobservedCells: number; + }; +} + +export interface VegetationMixedRouteReview { + sourceId: "RAVNOVES004TREE"; + sessionId: string; + packId: string; + frameCount: 10; + models: { + city: { name: string; inferenceFps: number; endToEndP95Ms: number }; + vegetation: { name: string; latencyP95Ms: number }; + tgs: { name: string; latencyP95Ms: number; cellSizeM: number; radiusM: number }; + }; + cases: readonly VegetationMixedRouteCase[]; +} + export interface VegetationShadowResult { resultId: string; createdAtUtc: string; @@ -88,6 +115,7 @@ export interface VegetationShadowResult { routeCases: readonly VegetationVisualCase[]; validationCases: readonly VegetationVisualCase[]; routeVideo: VegetationRouteVideo | null; + routeReview: VegetationMixedRouteReview | null; limitations: readonly string[]; visualShadowReady: true; missionPolicyReadyForConfiguration: true; @@ -423,6 +451,100 @@ function routeVideoValue(value: unknown): VegetationRouteVideo | null { }; } +function mixedRouteReviewValue( + value: unknown, + resultId: string, + endpointRoot: string, +): VegetationMixedRouteReview | null { + if (value === null || value === undefined) return null; + const row = objectValue(value, "vegetation.route_review"); + exact(row.source_id, "RAVNOVES004TREE", "vegetation.route_review.source_id"); + exact(row.frame_count, 10, "vegetation.route_review.frame_count"); + exact(row.ground_truth, false, "vegetation.route_review.ground_truth"); + exact( + row.selection_policy, + "same-scene-camera-lidar-aligned-review-islands/v1", + "vegetation.route_review.selection_policy", + ); + const packId = textValue(row.pack_id, "vegetation.route_review.pack_id"); + if (!/^mixed-route-review-pack-[a-f0-9]{64}$/.test(packId)) { + throw new VegetationShadowContractError("vegetation.route_review.pack_id: identity invalid."); + } + const models = objectValue(row.models, "vegetation.route_review.models"); + const city = objectValue(models.city, "vegetation.route_review.models.city"); + const vegetation = objectValue(models.vegetation, "vegetation.route_review.models.vegetation"); + const tgsModel = objectValue(models.tgs, "vegetation.route_review.models.tgs"); + exact(city.frames, 10, "vegetation.route_review.models.city.frames"); + exact(vegetation.frames, 10, "vegetation.route_review.models.vegetation.frames"); + exact(tgsModel.frames, 10, "vegetation.route_review.models.tgs.frames"); + const cases = arrayValue(row.cases, "vegetation.route_review.cases").map((raw, index) => { + const item = objectValue(raw, `vegetation.route_review.cases[${index}]`); + const caseId = textValue(item.case_id, `vegetation.route_review.cases[${index}].case_id`); + if (caseId !== `route-${String(index + 1).padStart(2, "0")}`) { + throw new VegetationShadowContractError("vegetation.route_review.case order changed."); + } + const phaseValue = item.phase; + if (phaseValue !== "rural" && phaseValue !== "transition" && phaseValue !== "urban") { + throw new VegetationShadowContractError("vegetation.route_review.phase changed."); + } + const phase: VegetationMixedRouteCase["phase"] = phaseValue; + const assets = objectValue(item.assets, `vegetation.route_review.cases[${index}].assets`); + const projected = Object.fromEntries(["source", "city", "vegetation", "tgs"].map((key) => { + const descriptor = objectValue(assets[key], `vegetation.route_review.assets.${key}`); + const path = textValue(descriptor.path, `vegetation.route_review.assets.${key}.path`); + const digest = textValue(descriptor.sha256, `vegetation.route_review.assets.${key}.sha256`); + if (!SHA256.test(digest) || !path.startsWith(`route-review/${caseId}/`)) { + throw new VegetationShadowContractError(`vegetation.route_review.assets.${key}: proof invalid.`); + } + return [key, `${endpointRoot}/${encodeURIComponent(resultId)}/assets/${path + .split("/").map(encodeURIComponent).join("/")}`]; + })) as Record<"source" | "city" | "vegetation" | "tgs", string>; + const tgs = objectValue(item.tgs, `vegetation.route_review.cases[${index}].tgs`); + const groundCells = integerValue(tgs.ground_cells, "vegetation.route_review.tgs.ground"); + const occupiedCells = integerValue(tgs.occupied_cells, "vegetation.route_review.tgs.occupied"); + const rejectedCells = integerValue(tgs.rejected_cells, "vegetation.route_review.tgs.rejected"); + const unobservedCells = integerValue(tgs.unobserved_cells, "vegetation.route_review.tgs.unobserved"); + if (groundCells + occupiedCells + rejectedCells + unobservedCells !== 2244) { + throw new VegetationShadowContractError("vegetation.route_review.tgs cell accounting changed."); + } + return { + caseId, + phase, + sourceSequence: integerValue(item.source_sequence, "vegetation.route_review.source_sequence"), + sessionSeconds: numberValue(item.session_seconds, "vegetation.route_review.session_seconds"), + assets: projected, + tgs: { groundCells, occupiedCells, rejectedCells, unobservedCells }, + }; + }); + if (cases.length !== 10) { + throw new VegetationShadowContractError("vegetation.route_review.cases: expected 10 aligned islands."); + } + return { + sourceId: "RAVNOVES004TREE", + sessionId: textValue(row.session_id, "vegetation.route_review.session_id"), + packId, + frameCount: 10, + models: { + city: { + name: textValue(city.name, "vegetation.route_review.models.city.name"), + inferenceFps: numberValue(city.inference_fps, "vegetation.route_review.models.city.fps"), + endToEndP95Ms: numberValue(city.end_to_end_p95_ms, "vegetation.route_review.models.city.p95"), + }, + vegetation: { + name: textValue(vegetation.name, "vegetation.route_review.models.vegetation.name"), + latencyP95Ms: numberValue(vegetation.latency_p95_ms, "vegetation.route_review.models.vegetation.p95"), + }, + tgs: { + name: textValue(tgsModel.name, "vegetation.route_review.models.tgs.name"), + latencyP95Ms: numberValue(tgsModel.latency_p95_ms, "vegetation.route_review.models.tgs.p95"), + cellSizeM: numberValue(tgsModel.cell_size_m, "vegetation.route_review.models.tgs.cell"), + radiusM: numberValue(tgsModel.radius_m, "vegetation.route_review.models.tgs.radius"), + }, + }, + cases, + }; +} + function parseResult( value: unknown, resultId: string, @@ -466,7 +588,11 @@ function parseResult( .map((item) => visualCaseValue(item, resultId, "ravnoves", endpointRoot)); const validationCases = arrayValue(catalogs.goose, "vegetation.catalogs.goose") .map((item) => visualCaseValue(item, resultId, "goose", endpointRoot)); - if (routeCases.length !== 0 || validationCases.length !== 12) { + const routeReview = mixedRouteReviewValue(payload.route_review, resultId, endpointRoot); + if ( + routeCases.length !== 0 + || (routeReview ? validationCases.length !== 0 : validationCases.length !== 12) + ) { throw new VegetationShadowContractError("vegetation.catalogs: ожидалось 12 truth-backed GOOSE случаев без route viewer."); } return { @@ -478,6 +604,7 @@ function parseResult( routeCases, validationCases, routeVideo: routeVideoValue(payload.route_video), + routeReview, limitations: arrayValue(payload.limitations, "vegetation.limitations") .map((item, index) => textValue(item, `vegetation.limitations[${index}]`)), visualShadowReady: true, diff --git a/apps/control-station/src/workspaces/laboratory/VegetationShadowResult.tsx b/apps/control-station/src/workspaces/laboratory/VegetationShadowResult.tsx index b1b5c93..3243ccb 100644 --- a/apps/control-station/src/workspaces/laboratory/VegetationShadowResult.tsx +++ b/apps/control-station/src/workspaces/laboratory/VegetationShadowResult.tsx @@ -1,5 +1,7 @@ import { useEffect, useState } from "react"; +import { Icon, IconButton, StatusBadge } from "@nodedc/ui-react"; +import { LaboratoryEvidenceViewer } from "../../components/laboratory/LaboratoryEvidenceViewer"; import { LaboratoryEvidence, LaboratoryResultSummary, @@ -8,6 +10,7 @@ import { } from "../../components/laboratory/LaboratoryPresentation"; import { vegetationVideoMaskUrl, + type VegetationMixedRouteReview, type VegetationShadowResult, } from "../../core/laboratory/vegetationShadow"; import { @@ -21,6 +24,128 @@ function decimal(value: number, digits = 1): string { return value.toLocaleString("ru-RU", { maximumFractionDigits: digits }); } +const MIXED_ROUTE_MODES = [ + { value: "source", label: "SOURCE" }, + { value: "city", label: "ГОРОД · EoMT" }, + { value: "vegetation", label: "ПРИРОДА · DDRNet" }, + { value: "tgs", label: "TGS" }, +] as const; + +function MixedRouteReviewEvidence({ review }: { review: VegetationMixedRouteReview }) { + const [index, setIndex] = useState(0); + const [mode, setMode] = useState("vegetation"); + const [expanded, setExpanded] = useState(false); + const item = review.cases[index]!; + return ( + + setIndex((index - 1 + review.cases.length) % review.cases.length)}> + + + setIndex((index + 1) % review.cases.length)}> + + + + )} + overlay={( +
+ + {item.phase.toUpperCase()} · {index + 1}/{review.cases.length} + + sequence {item.sourceSequence} · +{decimal(item.sessionSeconds, 2)} s + + TGS: {item.tgs.groundCells} ground · {item.tgs.occupiedCells} occupied · {item.tgs.unobservedCells} unobserved + +
+ )} + > +
+ +
+ + ); +} + +function MixedRouteReviewResult({ + rigLabel, + review, +}: { + rigLabel: string; + review: VegetationMixedRouteReview; +}) { + return ( + + )} + evidence={( + + + + )} + result={( + + )} + /> + ); +} + function VegetationRouteEvidence({ result }: { result: VegetationShadowResult }) { const route = result.routeVideo!; const [tgs, setTgs] = useState(null); @@ -99,6 +224,9 @@ export function VegetationShadowResultView({ rigLabel: string; result: VegetationShadowResult; }) { + if (result.routeReview) { + return ; + } const route = result.routeVideo; const selected = result.candidates.find( (candidate) => candidate.candidate === result.selectedCandidate, diff --git a/apps/control-station/src/workspaces/laboratory/annotation/useL34AnnotationCapability.tsx b/apps/control-station/src/workspaces/laboratory/annotation/useL34AnnotationCapability.tsx index cc3ce64..59310f7 100644 --- a/apps/control-station/src/workspaces/laboratory/annotation/useL34AnnotationCapability.tsx +++ b/apps/control-station/src/workspaces/laboratory/annotation/useL34AnnotationCapability.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useState, type ReactNode } from "react"; +import { useCallback, useEffect, useMemo, useState, type ReactNode } from "react"; import type { L34RightYoloxTruthIslandResult } from "../../../core/laboratory/l34RightYoloxTruthIsland"; import type { L34DResult } from "../../../core/laboratory/l34dCumulativePostprocessing"; @@ -28,19 +28,19 @@ export function useL34AnnotationCapability({ }): ReactNode { const [open, setOpen] = useState(false); const openWorkspace = useCallback(() => setOpen(true), []); - const available = selectedWorkId === "l34-right-yolox-truth-island-freeze" - && l34Result - ? { resultId: l34Result.resultId, workflow: "assisted-candidate" as const } - : selectedWorkId === "e46-detector-truth-island" && e46Result - ? { resultId: e46Result.resultId, workflow: "independent-blind" as const } - : selectedWorkId === "e46a-ai-engineering-preannotation" && e46aResult - ? { resultId: e46aResult.resultId, workflow: "engineering-preannotation" as const } - : selectedWorkId === "l34d-cumulative-postprocessing-candidate" - && l34dResult - ? { resultId: l34dResult.resultId, workflow: "prediction-hidden" as const } - : selectedWorkId === "l34e-self-review-diagnostic" && l34eResult - ? { resultId: l34eResult.resultId, workflow: "adjudication" as const } - : null; + const available = useMemo(() => ( + selectedWorkId === "l34-right-yolox-truth-island-freeze" && l34Result + ? { resultId: l34Result.resultId, workflow: "assisted-candidate" as const } + : selectedWorkId === "e46-detector-truth-island" && e46Result + ? { resultId: e46Result.resultId, workflow: "independent-blind" as const } + : selectedWorkId === "e46a-ai-engineering-preannotation" && e46aResult + ? { resultId: e46aResult.resultId, workflow: "engineering-preannotation" as const } + : selectedWorkId === "l34d-cumulative-postprocessing-candidate" && l34dResult + ? { resultId: l34dResult.resultId, workflow: "prediction-hidden" as const } + : selectedWorkId === "l34e-self-review-diagnostic" && l34eResult + ? { resultId: l34eResult.resultId, workflow: "adjudication" as const } + : null + ), [e46Result, e46aResult, l34Result, l34dResult, l34eResult, selectedWorkId]); useEffect(() => { if (!available) { diff --git a/apps/control-station/test/vegetationShadow.test.mjs b/apps/control-station/test/vegetationShadow.test.mjs index d4fd0d2..cb12e72 100644 --- a/apps/control-station/test/vegetationShadow.test.mjs +++ b/apps/control-station/test/vegetationShadow.test.mjs @@ -271,7 +271,8 @@ test("vegetation realtime LAB and archival benchmark use separate admitted instr assert.match(resultSource, /M49TgsFullShadowEvidence/); assert.match(resultSource, /semanticOverride/); assert.match(resultSource, /EoMT CITY \/ DDRNet VEGETATION/); - assert.equal(resultSource.match(/ bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--job", type=Path, required=True) + parser.add_argument("--session", type=Path, required=True) + parser.add_argument("--review-pack", type=Path, required=True) + parser.add_argument("--calibration", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + return parser.parse_args() + + +def _read_review_pack(root: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + resolved = root.resolve(strict=True) + manifest_path = resolved / "manifest.json" + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise MixedRouteLidarPackError("mixed-route review manifest is invalid") from exc + identity = manifest.get("identity") if isinstance(manifest, dict) else None + timeline = manifest.get("timeline") if isinstance(manifest, dict) else None + frames = manifest.get("frames") if isinstance(manifest, dict) else None + if ( + manifest.get("schema_version") != REVIEW_SCHEMA + or not isinstance(identity, dict) + or identity.get("schema_version") != REVIEW_SCHEMA + or identity.get("ground_truth") is not False + or not isinstance(timeline, dict) + or not isinstance(frames, list) + or manifest.get("frame_count") != len(frames) + or not frames + ): + raise MixedRouteLidarPackError("mixed-route review contract changed") + timeline_path = resolved / str(timeline.get("path")) + if ( + not timeline_path.is_file() + or timeline.get("sha256") != _sha256(timeline_path) + or timeline.get("byte_length") != timeline_path.stat().st_size + ): + raise MixedRouteLidarPackError("mixed-route review timeline changed") + rows: list[dict[str, Any]] = [] + previous_seconds = -1.0 + with timeline_path.open(encoding="utf-8") as stream: + for expected, line in enumerate(stream): + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + raise MixedRouteLidarPackError("mixed-route timeline JSON is invalid") from exc + seconds = row.get("session_seconds") if isinstance(row, dict) else None + if ( + not isinstance(row, dict) + or row.get("frame_index") != expected + or row.get("sequence") != expected + 1 + or row.get("source_sequence") != row.get("source_frame_index") + 1 + or not isinstance(seconds, (int, float)) + or isinstance(seconds, bool) + or float(seconds) <= previous_seconds + ): + raise MixedRouteLidarPackError("mixed-route timeline row changed") + rows.append(row) + previous_seconds = float(seconds) + if len(rows) != len(frames): + raise MixedRouteLidarPackError("mixed-route timeline is incomplete") + for frame in frames: + path = resolved / str(frame.get("path")) + if ( + not path.is_file() + or frame.get("byte_length") != path.stat().st_size + or frame.get("sha256") != _sha256(path) + ): + raise MixedRouteLidarPackError("mixed-route source frame changed") + return manifest, rows + + +def _causal_history_clouds( + raw_path: Path, + *, + origin_monotonic_ns: int, + sample_seconds: list[float], +) -> list[np.ndarray]: + grouped: list[list[np.ndarray]] = [[] for _ in sample_seconds] + last = sample_seconds[-1] + for message in iter_replay_messages(raw_path): + monotonic_ns = message.received_monotonic_ns + if not isinstance(monotonic_ns, int) or monotonic_ns < origin_monotonic_ns: + raise MixedRouteLidarPackError("MQTT replay message has no compatible clock") + seconds = (monotonic_ns - origin_monotonic_ns) / 1e9 + if seconds > last: + break + if not message.topic.endswith("/lio_pcl"): + continue + matching = [ + index + for index, sample_time in enumerate(sample_seconds) + if sample_time - CAUSAL_HISTORY_SECONDS <= seconds <= sample_time + ] + if not matching: + continue + frame = decode_lio_pcl(message.payload) + cloud = np.asarray( + [point.scaled_xyz(frame.header.scaler) for point in frame.points], + dtype=np.float32, + ).reshape((-1, 3)) + if cloud.shape[0] == 0 or not np.isfinite(cloud).all(): + raise MixedRouteLidarPackError("causal LiDAR history is empty or non-finite") + for index in matching: + grouped[index].append(cloud) + result: list[np.ndarray] = [] + for clouds in grouped: + if not clouds: + raise MixedRouteLidarPackError("selected frame has no causal LiDAR history") + result.append(np.concatenate(clouds)) + return result + + +def prepare( + *, + job_root: Path, + session_root: Path, + review_pack_root: Path, + calibration_root: Path, + output_root: Path, +) -> Path: + job = validate_camera_compute_job(job_root) + session = session_root.resolve(strict=True) + if not session.is_dir() or session.name != job.session_id: + raise MixedRouteLidarPackError("camera job and observation session differ") + review, timeline = _read_review_pack(review_pack_root) + review_identity = review["identity"] + if ( + review_identity.get("job_id") != job.job_id + or review_identity.get("input_sha256") != job.input_sha256 + or review_identity.get("session_id") != job.session_id + or review_identity.get("source_id") != job.source_id + or review_identity.get("codec_epoch") != job.codec_epoch + ): + raise MixedRouteLidarPackError("review pack and camera job differ") + + calibration, calibration_sha256 = _load_calibration_snapshot( + calibration_root.resolve(strict=True) + ) + projection = Kb4ProjectionProfile.from_factory_calibration(calibration, job.source_id) + capture_root = session / "captures" / "mqtt_live" + origin_path = capture_root / "mqtt.timeline.origin.json" + origin = read_capture_clock_origin(origin_path) + anchors = [ + CameraAnchor( + frame_index=int(row["frame_index"]), + source_frame_index=int(row["source_frame_index"]), + host_session_seconds=( + int(row["host_monotonic_ns"]) - origin.started_monotonic_ns + ) + / 1e9, + video_session_seconds=float(row["session_seconds"]), + ) + for row in timeline + ] + if any( + anchor.host_session_seconds != anchor.video_session_seconds + for anchor in anchors + ): + raise MixedRouteLidarPackError("review timeline does not use host arrival time") + samples = list( + _lidar_samples( + capture_root / "mqtt.raw.k1mqtt", + anchors, + origin_monotonic_ns=origin.started_monotonic_ns, + maximum_lidar_camera_delta_s=MAXIMUM_LIDAR_CAMERA_DELTA_MS / 1000.0, + maximum_pose_point_delta_s=MAXIMUM_POSE_POINT_DELTA_MS / 1000.0, + ) + ) + if len(samples) != len(anchors): + raise MixedRouteLidarPackError("LiDAR sampler did not account for every anchor") + + count = len(anchors) + available = np.zeros((count,), dtype=np.bool_) + offsets = [0] + clouds: list[np.ndarray] = [] + positions = np.full((count, 3), np.nan, dtype=np.float64) + quaternions = np.full((count, 4), np.nan, dtype=np.float64) + lidar_delta = np.full((count,), np.nan, dtype=np.float64) + pose_delta = np.full((count,), np.nan, dtype=np.float64) + sample_seconds: list[float] = [] + for index, (anchor, sample) in enumerate(zip(anchors, samples, strict=True)): + if sample is None: + offsets.append(offsets[-1]) + sample_seconds.append(float("nan")) + continue + cloud = np.asarray( + [ + point.scaled_xyz(sample.point_frame.header.scaler) + for point in sample.point_frame.points + ], + dtype=np.float32, + ).reshape((-1, 3)) + if cloud.shape[0] == 0 or not np.isfinite(cloud).all(): + raise MixedRouteLidarPackError("selected LiDAR sample is empty or non-finite") + available[index] = True + clouds.append(cloud) + offsets.append(offsets[-1] + cloud.shape[0]) + positions[index] = sample.pose_frame.position_xyz + quaternions[index] = sample.pose_frame.orientation_xyzw + lidar_delta[index] = ( + sample.point_session_seconds - anchor.host_session_seconds + ) * 1000.0 + pose_delta[index] = ( + sample.pose_session_seconds - sample.point_session_seconds + ) * 1000.0 + sample_seconds.append(sample.point_session_seconds) + + if not available.all() or not np.isfinite(np.asarray(sample_seconds)).all(): + raise MixedRouteLidarPackError( + "every mixed-route review island must have a temporally admissible LiDAR sample" + ) + history_clouds = _causal_history_clouds( + capture_root / "mqtt.raw.k1mqtt", + origin_monotonic_ns=origin.started_monotonic_ns, + sample_seconds=sample_seconds, + ) + history_offsets = [0] + for cloud in history_clouds: + history_offsets.append(history_offsets[-1] + cloud.shape[0]) + + identity = { + "schema_version": SCHEMA, + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "session_id": job.session_id, + "source_id": job.source_id, + "camera_slot": "camera_1", + "calibration_sha256": calibration_sha256, + "review_pack_id": review["pack_id"], + "review_pack_identity_sha256": review["identity_sha256"], + "selected_source_frame_indices": [ + int(row["source_frame_index"]) for row in timeline + ], + "frame_count": count, + "available_lidar_frames": int(available.sum()), + "point_count": int(offsets[-1]), + "causal_history_seconds": CAUSAL_HISTORY_SECONDS, + "causal_history_point_count": int(history_offsets[-1]), + "temporal_policy": { + "binding": "nearest-host-arrival-best-effort", + "maximum_lidar_camera_delta_ms": MAXIMUM_LIDAR_CAMERA_DELTA_MS, + "maximum_pose_point_delta_ms": MAXIMUM_POSE_POINT_DELTA_MS, + "clock_source": "recorded-host-monotonic-arrival", + }, + "projection": { + "model": "kb4", + "width": projection.width, + "height": projection.height, + "source_coordinates": "k1-map", + "target_camera": job.source_id, + }, + "ground_truth": False, + "authority": { + "navigation_or_safety_accepted": False, + "actuation_allowed": False, + }, + "producer_sha256": _sha256(Path(__file__).resolve(strict=True)), + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + pack_id = f"mixed-route-lidar-pack-{identity_sha256}" + parent = output_root.resolve() + parent.mkdir(mode=0o700, parents=True, exist_ok=True) + final = parent / pack_id + if final.exists(): + return final + staging = Path(tempfile.mkdtemp(prefix=f".{pack_id}.", dir=parent)) + published = False + try: + arrays_path = staging / "lidar-pack.npz" + np.savez_compressed( + arrays_path, + frame_indices=np.arange(count, dtype=np.int64), + source_frame_indices=np.asarray( + [row["source_frame_index"] for row in timeline], dtype=np.int64 + ), + session_seconds=np.asarray( + [anchor.video_session_seconds for anchor in anchors], dtype=np.float64 + ), + host_session_seconds=np.asarray( + [anchor.host_session_seconds for anchor in anchors], dtype=np.float64 + ), + lidar_session_seconds=np.asarray(sample_seconds, dtype=np.float64), + sample_available=available, + cloud_offsets=np.asarray(offsets, dtype=np.int64), + cloud_points_map=( + np.concatenate(clouds) if clouds else np.empty((0, 3), dtype=np.float32) + ), + pose_positions_map=positions, + pose_quaternions_map_from_lidar=quaternions, + lidar_camera_delta_ms=lidar_delta, + pose_point_delta_ms=pose_delta, + causal_history_seconds=np.asarray( + [CAUSAL_HISTORY_SECONDS], dtype=np.float64 + ), + causal_history_offsets=np.asarray(history_offsets, dtype=np.int64), + causal_history_points_map=np.concatenate(history_clouds), + intrinsic_fx_fy_cx_cy=np.asarray( + projection.intrinsic_fx_fy_cx_cy, dtype=np.float64 + ), + distortion_kb4=np.asarray(projection.distortion_kb4, dtype=np.float64), + t_camera_from_lidar=np.asarray(projection.t_camera_from_lidar, dtype=np.float64), + ) + manifest = { + "schema_version": SCHEMA, + "pack_id": pack_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "classification": "private-recorded-sensor-review-input", + "ground_truth": False, + "artifact": { + "path": arrays_path.name, + "media_type": "application/x-npz", + "byte_length": arrays_path.stat().st_size, + "sha256": _sha256(arrays_path), + }, + } + (staging / "manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + encoding="utf-8", + ) + os.replace(staging, final) + published = True + finally: + if not published: + shutil.rmtree(staging, ignore_errors=True) + return final + + +def main() -> int: + args = _arguments() + output = prepare( + job_root=args.job, + session_root=args.session, + review_pack_root=args.review_pack, + calibration_root=args.calibration, + output_root=args.output_root, + ) + manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8")) + print( + json.dumps( + { + "pack_id": manifest["pack_id"], + "output": str(output), + "frames": manifest["identity"]["frame_count"], + "lidar_frames": manifest["identity"]["available_lidar_frames"], + "points": manifest["identity"]["point_count"], + "artifact_sha256": manifest["artifact"]["sha256"], + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/lab_v1_vegetation_goose/run_mixed_route_ddrnet_islands.py b/experiments/perception/worker/lab_v1_vegetation_goose/run_mixed_route_ddrnet_islands.py new file mode 100644 index 0000000..5326e37 --- /dev/null +++ b/experiments/perception/worker/lab_v1_vegetation_goose/run_mixed_route_ddrnet_islands.py @@ -0,0 +1,340 @@ +#!/usr/bin/env python3 +"""Run DDRNet on an immutable mixed-route camera review pack.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import platform +import statistics +import time +from pathlib import Path, PurePosixPath +from typing import Any + +import numpy as np +import torch +from PIL import Image +from run_goose_vegetation_benchmark import ( + CLASS_COUNT, + expand_mask, + infer, + load_mapping, + load_model, + percentile, + preprocess, + read_json, + save_image, + sha256, + stable_digest, + validate_contracts, +) + +SCHEMA = "missioncore.mixed-route-ddrnet-islands/v1" +PACK_SCHEMA = "missioncore.mixed-route-review-pack/v1" +AUTHORITY = { + "ground_truth": False, + "candidate_accepted": False, + "navigation_or_safety_accepted": False, + "camera_semantics_can_clear_rigid_geometry": False, + "actuation_allowed": False, +} + + +class MixedRouteDdrnetError(RuntimeError): + """The route pack or DDRNet evidence changed or is incomplete.""" + + +def arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--pack", type=Path, required=True) + parser.add_argument("--config", type=Path, required=True) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--provider-map", type=Path, required=True) + parser.add_argument("--checkpoint", type=Path, required=True) + parser.add_argument("--dataset-root", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def object_value(value: object, label: str) -> dict[str, Any]: + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise MixedRouteDdrnetError(f"{label} must be an object") + return value + + +def load_pack(root: Path) -> tuple[dict[str, Any], list[dict[str, Any]]]: + pack = root.resolve(strict=True) + if not pack.is_dir() or pack.is_symlink(): + raise MixedRouteDdrnetError("mixed-route review pack is unavailable") + manifest_path = pack / "manifest.json" + manifest = object_value( + json.loads(manifest_path.read_text(encoding="utf-8")), + "mixed-route manifest", + ) + identity = object_value(manifest.get("identity"), "mixed-route identity") + identity_sha256 = manifest.get("identity_sha256") + frames = manifest.get("frames") + frame_count = manifest.get("frame_count") + if ( + manifest.get("schema_version") != PACK_SCHEMA + or identity.get("schema_version") != PACK_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(canonical_json(identity)).hexdigest() != identity_sha256 + or manifest.get("pack_id") != f"mixed-route-review-pack-{identity_sha256}" + or identity.get("ground_truth") is not False + or object_value(identity.get("authority"), "mixed-route authority").get( + "navigation_or_safety_accepted" + ) + is not False + or not isinstance(frame_count, int) + or isinstance(frame_count, bool) + or not 1 <= frame_count <= 64 + or not isinstance(frames, list) + or len(frames) != frame_count + ): + raise MixedRouteDdrnetError("mixed-route review pack identity changed") + timeline_descriptor = object_value(manifest.get("timeline"), "mixed-route timeline") + timeline_path = pack / "timeline.jsonl" + if ( + timeline_descriptor.get("path") != timeline_path.name + or timeline_path.stat().st_size != timeline_descriptor.get("byte_length") + or sha256(timeline_path) != timeline_descriptor.get("sha256") + ): + raise MixedRouteDdrnetError("mixed-route timeline proof changed") + rows: list[dict[str, Any]] = [] + with timeline_path.open(encoding="utf-8") as stream: + for expected, line in enumerate(stream): + row = object_value(json.loads(line), "mixed-route timeline row") + seconds = row.get("session_seconds") + if ( + row.get("frame_index") != expected + or row.get("sequence") != expected + 1 + or not isinstance(row.get("source_sequence"), int) + or row.get("source_frame_index") != row["source_sequence"] - 1 + or not isinstance(seconds, (int, float)) + or isinstance(seconds, bool) + or (rows and float(seconds) <= float(rows[-1]["session_seconds"])) + ): + raise MixedRouteDdrnetError("mixed-route timeline order changed") + rows.append(row) + if len(rows) != frame_count: + raise MixedRouteDdrnetError("mixed-route timeline is incomplete") + for expected, (descriptor_raw, row) in enumerate(zip(frames, rows)): # noqa: B905 + descriptor = object_value(descriptor_raw, "mixed-route frame descriptor") + relative = descriptor.get("path") + if relative != f"frames/frame-{expected + 1:06d}.png": + raise MixedRouteDdrnetError("mixed-route frame path changed") + pure = PurePosixPath(relative) + path = pack.joinpath(*pure.parts) + if ( + path.is_symlink() + or not path.is_file() + or not path.resolve().is_relative_to(pack) + or path.stat().st_size != descriptor.get("byte_length") + or sha256(path) != descriptor.get("sha256") + or not isinstance(descriptor.get("source_segment_sha256"), str) + or row.get("source_sequence") + != identity["selected_sequences"][expected] + ): + raise MixedRouteDdrnetError("mixed-route frame proof changed") + return manifest, rows + + +def overlay(source: Image.Image, semantic: np.ndarray, palette: np.ndarray) -> Image.Image: + if semantic.shape != (600, 800): + raise MixedRouteDdrnetError("expanded semantic mask shape changed") + base = source.convert("RGBA") + colors = Image.fromarray(palette[semantic], mode="RGBA") + return Image.alpha_composite(base, colors) + + +def run() -> int: + args = arguments() + if not torch.cuda.is_available(): + raise MixedRouteDdrnetError("CUDA is required for DDRNet islands") + if args.output.exists(): + raise MixedRouteDdrnetError("DDRNet islands output already exists") + manifest, timeline = load_pack(args.pack) + config = read_json(args.config, "benchmark config") + policy = read_json(args.policy, "mission policy") + provider_map = read_json(args.provider_map, "provider map") + candidate = validate_contracts(config, policy, provider_map, "ddrnet") + checkpoint = args.checkpoint.resolve(strict=True) + if ( + checkpoint.is_symlink() + or checkpoint.stat().st_size != candidate["checkpoint_size_bytes"] + or sha256(checkpoint) != candidate["checkpoint_sha256"] + ): + raise MixedRouteDdrnetError("DDRNet checkpoint identity changed") + dataset_root = args.dataset_root.resolve(strict=True) + mapping_path = dataset_root / config["dataset"]["mapping_relative_path"] + names, palette = load_mapping(mapping_path, config["dataset"]["mapping_sha256"]) + + args.output.mkdir(mode=0o700, parents=True, exist_ok=False) + mask_root = args.output / "semantic-masks" + overlay_root = args.output / "overlay-frames" + mask_root.mkdir(mode=0o700) + overlay_root.mkdir(mode=0o700) + torch.cuda.empty_cache() + model, model_name, architecture_failures = load_model("ddrnet", checkpoint) + first_path = args.pack / manifest["frames"][0]["path"] + with Image.open(first_path) as opened: + warm_source = opened.convert("RGB") + warm_tensor, _ = preprocess(warm_source) + warmup_ms = [infer(model, warm_tensor)[1] for _ in range(3)] + torch.cuda.reset_peak_memory_stats() + + latencies_ms: list[float] = [] + aggregate = np.zeros(CLASS_COUNT, dtype=np.int64) + frame_results: list[dict[str, Any]] = [] + started = time.perf_counter() + for index, (descriptor, timeline_row) in enumerate( + zip(manifest["frames"], timeline) # noqa: B905 - Worker image uses Python 3.9. + ): + source_path = args.pack / descriptor["path"] + with Image.open(source_path) as opened: + source = opened.convert("RGB") + if source.size != (800, 600): + raise MixedRouteDdrnetError("mixed-route source resolution changed") + tensor, crop_box = preprocess(source) + prediction, latency_ms = infer(model, tensor) + expanded = expand_mask(prediction, source.size, crop_box) + latencies_ms.append(latency_ms) + aggregate += np.bincount(expanded.reshape(-1), minlength=CLASS_COUNT) + mask_path = mask_root / f"frame-{index + 1:06d}.png" + overlay_path = overlay_root / f"frame-{index + 1:06d}.png" + mask_sha256 = save_image(mask_path, expanded, "L") + overlay_sha256 = save_image(overlay_path, overlay(source, expanded, palette)) + present = np.flatnonzero(np.bincount(expanded.reshape(-1), minlength=CLASS_COUNT)) + frame_results.append( + { + "frame_index": index, + "source_sequence": timeline_row["source_sequence"], + "source_frame_index": timeline_row["source_frame_index"], + "session_seconds": timeline_row["session_seconds"], + "latency_ms": round(latency_ms, 6), + "present_classes": [ + {"class_id": int(class_id), "label": names[int(class_id)]} + for class_id in present + ], + "mask": { + "path": mask_path.relative_to(args.output).as_posix(), + "byte_length": mask_path.stat().st_size, + "sha256": mask_sha256, + }, + "overlay": { + "path": overlay_path.relative_to(args.output).as_posix(), + "byte_length": overlay_path.stat().st_size, + "sha256": overlay_sha256, + }, + } + ) + wall_seconds = time.perf_counter() - started + if len(frame_results) != manifest["frame_count"]: + raise MixedRouteDdrnetError("DDRNet island accounting changed") + timing = { + "prewarm_inference_count": len(warmup_ms), + "prewarm_latency_ms_first": round(warmup_ms[0], 6), + "prewarm_latency_ms_last": round(warmup_ms[-1], 6), + "inference_wall_seconds": round(wall_seconds, 6), + "latency_ms_mean": round(statistics.fmean(latencies_ms), 6), + "latency_ms_p50": round(percentile(latencies_ms, 0.5), 6), + "latency_ms_p95": round(percentile(latencies_ms, 0.95), 6), + "throughput_fps_from_mean_inference": round( + 1000.0 / statistics.fmean(latencies_ms), 6 + ), + } + if any(not math.isfinite(float(value)) for value in timing.values()): + raise MixedRouteDdrnetError("DDRNet timing is non-finite") + result: dict[str, Any] = { + "schema_version": SCHEMA, + "status": "review-islands-ready-not-accepted", + "worker_id": "worker-006", + "source": { + "pack_id": manifest["pack_id"], + "pack_identity_sha256": manifest["identity_sha256"], + "job_id": manifest["identity"]["job_id"], + "input_sha256": manifest["identity"]["input_sha256"], + "session_id": manifest["identity"]["session_id"], + "source_id": manifest["identity"]["source_id"], + "frame_count": manifest["frame_count"], + "ground_truth_available": False, + }, + "candidate": { + "candidate_key": "ddrnet", + "candidate_id": candidate["candidate_id"], + "loaded_model_name": model_name, + "architecture_probe_failures": architecture_failures, + "checkpoint_size_bytes": checkpoint.stat().st_size, + "checkpoint_sha256": sha256(checkpoint), + }, + "taxonomy": { + "schema_version": "missioncore.lab-v1-vegetation-taxonomy/v1", + "classes": [ + { + "class_id": class_id, + "label": names[class_id], + "color_rgb": palette[class_id, :3].astype(int).tolist(), + "disposition": "undefined" if class_id == 0 else "prediction", + } + for class_id in range(CLASS_COUNT) + ], + }, + "aggregate_prediction_pixels": aggregate.tolist(), + "frames": frame_results, + "timing": timing, + "resource": { + "hostname": platform.node(), + "gpu_name": torch.cuda.get_device_name(0), + "peak_allocated_vram_bytes": int(torch.cuda.max_memory_allocated()), + "peak_reserved_vram_bytes": int(torch.cuda.max_memory_reserved()), + "torch_version": torch.__version__, + "cuda_runtime_version": torch.version.cuda, + "python_version": platform.python_version(), + }, + "provenance": { + "pack_manifest_sha256": sha256(args.pack / "manifest.json"), + "config_sha256": sha256(args.config), + "policy_sha256": sha256(args.policy), + "provider_map_sha256": sha256(args.provider_map), + "runner_sha256": sha256(Path(__file__)), + }, + "limitations": [ + "Selected independently decodable islands are not a complete route timeline.", + "RAVNOVES004TREE has no route truth; class colors are model predictions.", + "DDRNet evidence cannot clear rigid geometry, person or vehicle vetoes.", + ], + "authority": AUTHORITY, + } + result["result_id"] = f"mixed-route-ddrnet-islands-{stable_digest(result)}" + (args.output / "result.json").write_text( + json.dumps(result, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + encoding="utf-8", + ) + print( + json.dumps( + { + "result_id": result["result_id"], + "frames": len(frame_results), + "latency_p95_ms": timing["latency_ms_p95"], + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(run()) diff --git a/experiments/perception/worker/m49_t3_travel/build_mixed_route_tgs_evidence.py b/experiments/perception/worker/m49_t3_travel/build_mixed_route_tgs_evidence.py new file mode 100644 index 0000000..ee3a168 --- /dev/null +++ b/experiments/perception/worker/m49_t3_travel/build_mixed_route_tgs_evidence.py @@ -0,0 +1,277 @@ +#!/usr/bin/env python3 +"""Seal fail-closed TRAVEL/TGS evidence for mixed-route review islands.""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path + +import numpy as np +from build_tgs_fail_closed_evidence import ( + TgsEvidenceError, + _load_float32, + classify_exact_input, + costmap_grid, + rasterize_costmap, + sha256_file, + write_deterministic_npz, +) + +CONFIG_SCHEMA = "missioncore.mixed-route-tgs-review-profile/v1" +INPUT_SCHEMA = "missioncore.mixed-route-tgs-input/v1" +RESULT_SCHEMA = "missioncore.mixed-route-tgs-result/v1" +FRAME_COUNT = 10 + + +def _timing(path: Path) -> dict[str, object]: + rows: list[dict[str, object]] = [] + with path.open(encoding="utf-8", newline="") as stream: + for raw in csv.DictReader(stream, delimiter="\t"): + try: + row = { + "profile_id": str(raw["profile"]), + "slot": int(raw["slot"]), + "wall_seconds": float(raw["wall_seconds"]), + "max_rss_kib": int(raw["max_rss_kib"]), + } + except (KeyError, TypeError, ValueError) as exc: + raise TgsEvidenceError("TGS timing row is invalid") from exc + if ( + row["profile_id"] not in {"current_increment", "causal_rolling_1s"} + or not 0 <= row["slot"] < FRAME_COUNT + or not 0 <= row["wall_seconds"] < 60 + or not 0 < row["max_rss_kib"] < 16 * 1024 * 1024 + ): + raise TgsEvidenceError("TGS timing value is invalid") + rows.append(row) + if len(rows) != FRAME_COUNT * 2: + raise TgsEvidenceError("TGS timing is incomplete") + seconds = np.asarray([row["wall_seconds"] for row in rows], dtype=np.float64) + return { + "runs": rows, + "wall_seconds_mean": round(float(seconds.mean()), 6), + "wall_seconds_p95": round(float(np.percentile(seconds, 95)), 6), + "max_rss_kib": max(int(row["max_rss_kib"]) for row in rows), + } + + +def build(run_root: Path, config_path: Path, output_root: Path) -> dict[str, object]: + if output_root.exists(): + raise TgsEvidenceError("mixed-route TGS evidence already exists") + config = json.loads(config_path.read_text(encoding="utf-8")) + source = config.get("source") if isinstance(config, dict) else None + invariants = config.get("invariants") if isinstance(config, dict) else None + if ( + config.get("schema_version") != CONFIG_SCHEMA + or not isinstance(source, dict) + or not isinstance(invariants, dict) + or invariants.get("aos_allowed") is not False + or invariants.get("missing_support_means_free") is not False + or invariants.get("future_frames_used") is not False + or invariants.get("navigation_or_actuation_allowed") is not False + or config.get("state_codes") + != { + "UNOBSERVED": 0, + "GROUND_SUPPORT": 1, + "NONGROUND_OCCUPIED": 2, + "UNKNOWN_REJECTED": 3, + } + ): + raise TgsEvidenceError("mixed-route TGS profile changed") + input_manifest_path = run_root / "inputs" / "input-manifest.json" + input_manifest = json.loads(input_manifest_path.read_text(encoding="utf-8")) + if ( + input_manifest.get("schema_version") != INPUT_SCHEMA + or input_manifest.get("source_pack_id") != source.get("source_pack_id") + or input_manifest.get("source_pack_sha256") + != source.get("source_pack_sha256") + or input_manifest.get("config_sha256") != sha256_file(config_path) + or input_manifest.get("coordinate_frame") != "map-gravity-local" + or input_manifest.get("future_frames_used") is not False + or input_manifest.get("frame_count") != FRAME_COUNT + or len(input_manifest.get("records", [])) != FRAME_COUNT * 2 + ): + raise TgsEvidenceError("mixed-route TGS input manifest changed") + records = { + (str(row["profile_id"]), int(row["slot"])): row + for row in input_manifest["records"] + } + if len(records) != FRAME_COUNT * 2: + raise TgsEvidenceError("mixed-route TGS input records are not unique") + + cell_size = float(config["costmap"]["cell_size_m"]) + radius = float(config["costmap"]["radius_m"]) + grid = costmap_grid(radius, cell_size) + arrays: dict[str, np.ndarray] = { + "costmap_cell_indices_xy": grid[:, :2].astype(np.int32), + "costmap_cell_centers_xy_m": grid[:, 2:].astype(np.float32), + "source_frame_indices": np.asarray( + [ + records[("current_increment", slot)]["source_frame_index"] + for slot in range(FRAME_COUNT) + ], + dtype=np.int64, + ), + "session_seconds": np.asarray( + [ + records[("current_increment", slot)]["session_seconds"] + for slot in range(FRAME_COUNT) + ], + dtype=np.float64, + ), + } + summaries: list[dict[str, object]] = [] + for profile_id in ("current_increment", "causal_rolling_1s"): + all_points: list[np.ndarray] = [] + all_states: list[np.ndarray] = [] + offsets = [0] + grid_states: list[np.ndarray] = [] + ground_counts: list[np.ndarray] = [] + nonground_counts: list[np.ndarray] = [] + rejected_counts: list[np.ndarray] = [] + z_bounds_rows: list[np.ndarray] = [] + for slot in range(FRAME_COUNT): + record = records[(profile_id, slot)] + native_path = run_root / "inputs" / str(record["relative_path"]) + if ( + not native_path.is_file() + or native_path.stat().st_size != record["bytes"] + or sha256_file(native_path) != record["sha256"] + ): + raise TgsEvidenceError("sealed mixed-route TGS input changed") + output = run_root / "outputs" / profile_id + points, states = classify_exact_input( + _load_float32(native_path, 4), + _load_float32(output / f"{slot}_ground.bin", 4), + _load_float32(output / f"{slot}_nonground.bin", 4), + min_range_m=float(config["tgs"]["min_range_m"]), + max_range_m=float(config["tgs"]["max_range_m"]), + ) + grid_state, ground, nonground, rejected, z_bounds = rasterize_costmap( + points, + states, + grid, + cell_size_m=cell_size, + ) + all_points.append(points.astype(np.float32, copy=False)) + all_states.append(states) + offsets.append(offsets[-1] + points.shape[0]) + grid_states.append(grid_state) + ground_counts.append(ground) + nonground_counts.append(nonground) + rejected_counts.append(rejected) + z_bounds_rows.append(z_bounds) + accounted = ( + np.count_nonzero(states == 1) + + np.count_nonzero(states == 2) + + np.count_nonzero(states == 3) + == points.shape[0] + ) + summaries.append( + { + "profile_id": profile_id, + "slot": slot, + "frame_index": int(record["frame_index"]), + "source_frame_index": int(record["source_frame_index"]), + "source_sequence": int(record["source_sequence"]), + "session_seconds": float(record["session_seconds"]), + "point_count": int(points.shape[0]), + "ground_point_count": int(np.count_nonzero(states == 1)), + "nonground_point_count": int(np.count_nonzero(states == 2)), + "rejected_point_count": int(np.count_nonzero(states == 3)), + "ground_cell_count": int(np.count_nonzero(grid_state == 1)), + "nonground_cell_count": int(np.count_nonzero(grid_state == 2)), + "rejected_cell_count": int(np.count_nonzero(grid_state == 3)), + "unobserved_cell_count": int(np.count_nonzero(grid_state == 0)), + "all_points_accounted": bool(accounted), + } + ) + arrays[f"{profile_id}_points_xyz_m"] = np.concatenate(all_points) + arrays[f"{profile_id}_point_states"] = np.concatenate(all_states) + arrays[f"{profile_id}_point_offsets"] = np.asarray(offsets, dtype=np.int64) + arrays[f"{profile_id}_costmap_states"] = np.stack(grid_states) + arrays[f"{profile_id}_costmap_ground_point_counts"] = np.stack(ground_counts) + arrays[f"{profile_id}_costmap_nonground_point_counts"] = np.stack( + nonground_counts + ) + arrays[f"{profile_id}_costmap_rejected_point_counts"] = np.stack( + rejected_counts + ) + arrays[f"{profile_id}_costmap_z_bounds_m"] = np.stack(z_bounds_rows) + if not all(bool(row["all_points_accounted"]) for row in summaries): + raise TgsEvidenceError("mixed-route TGS lost an eligible point") + + output_root.mkdir(parents=True) + evidence_path = output_root / "evidence.npz" + write_deterministic_npz(evidence_path, arrays) + timing = _timing(run_root / "tgs-timing.tsv") + result = { + "schema_version": RESULT_SCHEMA, + "status": "passed-review-only", + "source": { + "source_id": source["source_id"], + "session_id": source["session_id"], + "review_pack_id": source["review_pack_id"], + "source_pack_id": source["source_pack_id"], + "source_pack_sha256": source["source_pack_sha256"], + }, + "config_sha256": sha256_file(config_path), + "input_manifest_sha256": sha256_file(input_manifest_path), + "evidence": { + "path": "evidence.npz", + "bytes": evidence_path.stat().st_size, + "sha256": sha256_file(evidence_path), + }, + "costmap": { + "coordinate_frame": "map-gravity-local", + "cell_size_m": cell_size, + "radius_m": radius, + "cell_count": int(grid.shape[0]), + }, + "anchors": summaries, + "timing": timing, + "summary": { + "frame_count": FRAME_COUNT, + "anchor_profile_count": len(summaries), + "all_eligible_points_accounted": True, + "aos_used": False, + "primary_profile": "causal_rolling_1s", + }, + "limitations": [ + "Selected review islands are not a complete route timeline.", + ( + "TGS separates local ground support from non-ground evidence; it does not " + "prove ditch or negative-obstacle detection." + ), + "Camera projection is visual evidence only and cannot clear rigid geometry.", + ], + "authority": { + "visual_quality_accepted": False, + "traversability_accepted": False, + "realtime_accepted": False, + "navigation_or_safety_accepted": False, + "actuation_allowed": False, + }, + } + (output_root / "result.json").write_text( + json.dumps(result, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return result + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--run-root", type=Path, required=True) + parser.add_argument("--config", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + result = build(args.run_root, args.config, args.output_root) + print(json.dumps(result["summary"], sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/m49_t3_travel/prepare_mixed_route_tgs_inputs.py b/experiments/perception/worker/m49_t3_travel/prepare_mixed_route_tgs_inputs.py new file mode 100644 index 0000000..a27bbc8 --- /dev/null +++ b/experiments/perception/worker/m49_t3_travel/prepare_mixed_route_tgs_inputs.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +"""Prepare exact mixed-route LiDAR islands for isolated TRAVEL/TGS review.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path + +import numpy as np +from prepare_tgs_fail_closed_inputs import TgsInputError, gravity_local_xyzi + +CONFIG_SCHEMA = "missioncore.mixed-route-tgs-review-profile/v1" +PACK_SCHEMA = "missioncore.mixed-route-lidar-pack/v1" +INPUT_SCHEMA = "missioncore.mixed-route-tgs-input/v1" +FRAME_COUNT = 10 + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _slice(points: np.ndarray, offsets: np.ndarray, index: int) -> np.ndarray: + return points[int(offsets[index]) : int(offsets[index + 1])] + + +def _validate_offsets(offsets: np.ndarray, point_count: int) -> bool: + return bool( + offsets.shape == (FRAME_COUNT + 1,) + and offsets.dtype == np.int64 + and int(offsets[0]) == 0 + and int(offsets[-1]) == point_count + and np.all(np.diff(offsets) > 0) + ) + + +def prepare(source_root: Path, config_path: Path, output_root: Path) -> dict[str, object]: + if output_root.exists(): + raise TgsInputError("mixed-route TGS output already exists") + source = source_root.resolve(strict=True) + config = json.loads(config_path.read_text(encoding="utf-8")) + manifest = json.loads((source / "manifest.json").read_text(encoding="utf-8")) + identity = manifest.get("identity") if isinstance(manifest, dict) else None + artifact = manifest.get("artifact") if isinstance(manifest, dict) else None + source_config = config.get("source") if isinstance(config, dict) else None + invariants = config.get("invariants") if isinstance(config, dict) else None + profiles = config.get("profiles") if isinstance(config, dict) else None + if ( + config.get("schema_version") != CONFIG_SCHEMA + or not isinstance(source_config, dict) + or not isinstance(invariants, dict) + or not isinstance(profiles, dict) + or set(profiles) != {"current_increment", "causal_rolling_1s"} + or source_config.get("input_coordinate_frame") + != "map-gravity-local-translation-only" + or invariants.get("lidar_orientation_applied_to_tgs_input") is not False + or invariants.get("future_frames_used") is not False + or invariants.get("navigation_or_actuation_allowed") is not False + or manifest.get("schema_version") != PACK_SCHEMA + or not isinstance(identity, dict) + or identity.get("schema_version") != PACK_SCHEMA + or identity.get("session_id") != source_config.get("session_id") + or identity.get("review_pack_id") != source_config.get("review_pack_id") + or manifest.get("pack_id") != source_config.get("source_pack_id") + or not isinstance(artifact, dict) + or artifact.get("path") != "lidar-pack.npz" + or artifact.get("sha256") != source_config.get("source_pack_sha256") + or identity.get("frame_count") != FRAME_COUNT + or identity.get("available_lidar_frames") != FRAME_COUNT + or identity.get("causal_history_seconds") + != float(profiles["causal_rolling_1s"]["history_seconds"]) + or identity.get("ground_truth") is not False + ): + raise TgsInputError("mixed-route TGS source contract changed") + pack_path = source / "lidar-pack.npz" + if ( + not pack_path.is_file() + or pack_path.stat().st_size != artifact.get("byte_length") + or sha256_file(pack_path) != artifact.get("sha256") + ): + raise TgsInputError("mixed-route LiDAR pack changed") + + required = { + "frame_indices", + "source_frame_indices", + "session_seconds", + "lidar_session_seconds", + "sample_available", + "cloud_offsets", + "cloud_points_map", + "pose_positions_map", + "lidar_camera_delta_ms", + "pose_point_delta_ms", + "causal_history_seconds", + "causal_history_offsets", + "causal_history_points_map", + } + with np.load(pack_path, allow_pickle=False) as archive: + if not required.issubset(archive.files): + raise TgsInputError("mixed-route LiDAR pack members changed") + arrays = {name: archive[name] for name in required} + current_points = arrays["cloud_points_map"] + history_points = arrays["causal_history_points_map"] + if ( + arrays["frame_indices"].shape != (FRAME_COUNT,) + or arrays["frame_indices"].dtype != np.int64 + or not np.array_equal(arrays["frame_indices"], np.arange(FRAME_COUNT)) + or arrays["source_frame_indices"].shape != (FRAME_COUNT,) + or arrays["source_frame_indices"].dtype != np.int64 + or np.any(np.diff(arrays["source_frame_indices"]) <= 0) + or arrays["session_seconds"].shape != (FRAME_COUNT,) + or arrays["session_seconds"].dtype != np.float64 + or np.any(np.diff(arrays["session_seconds"]) <= 0) + or arrays["lidar_session_seconds"].shape != (FRAME_COUNT,) + or arrays["lidar_session_seconds"].dtype != np.float64 + or arrays["sample_available"].shape != (FRAME_COUNT,) + or arrays["sample_available"].dtype != np.bool_ + or not arrays["sample_available"].all() + or current_points.ndim != 2 + or current_points.shape[1:] != (3,) + or current_points.dtype != np.float32 + or history_points.ndim != 2 + or history_points.shape[1:] != (3,) + or history_points.dtype != np.float32 + or not np.isfinite(current_points).all() + or not np.isfinite(history_points).all() + or not _validate_offsets(arrays["cloud_offsets"], current_points.shape[0]) + or not _validate_offsets( + arrays["causal_history_offsets"], history_points.shape[0] + ) + or arrays["pose_positions_map"].shape != (FRAME_COUNT, 3) + or arrays["pose_positions_map"].dtype != np.float64 + or not np.isfinite(arrays["pose_positions_map"]).all() + or arrays["causal_history_seconds"].shape != (1,) + or float(arrays["causal_history_seconds"][0]) + != float(profiles["causal_rolling_1s"]["history_seconds"]) + or np.any(np.abs(arrays["lidar_camera_delta_ms"]) > 100.0) + or np.any(np.abs(arrays["pose_point_delta_ms"]) > 100.0) + ): + raise TgsInputError("mixed-route LiDAR arrays changed") + + records: list[dict[str, object]] = [] + for profile_id in ("current_increment", "causal_rolling_1s"): + for slot in range(FRAME_COUNT): + if profile_id == "current_increment": + points_map = _slice( + current_points, arrays["cloud_offsets"], slot + ) + else: + points_map = _slice( + history_points, arrays["causal_history_offsets"], slot + ) + radius = float(profiles[profile_id]["local_radius_m"]) + relative_xy = ( + points_map[:, :2].astype(np.float64) + - arrays["pose_positions_map"][slot, :2] + ) + points_map = points_map[np.linalg.norm(relative_xy, axis=1) <= radius] + native = gravity_local_xyzi( + points_map, arrays["pose_positions_map"][slot] + ) + if native.shape[0] == 0: + raise TgsInputError("mixed-route TGS profile produced an empty cloud") + target = ( + output_root + / "profiles" + / profile_id + / "velodyne" + / f"{slot:06d}.bin" + ) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(np.ascontiguousarray(native).tobytes()) + records.append( + { + "profile_id": profile_id, + "slot": slot, + "frame_index": slot, + "source_frame_index": int( + arrays["source_frame_indices"][slot] + ), + "source_sequence": int( + arrays["source_frame_indices"][slot] + ) + + 1, + "session_seconds": float(arrays["session_seconds"][slot]), + "lidar_session_seconds": float( + arrays["lidar_session_seconds"][slot] + ), + "lidar_camera_delta_ms": float( + arrays["lidar_camera_delta_ms"][slot] + ), + "pose_point_delta_ms": float( + arrays["pose_point_delta_ms"][slot] + ), + "point_count": int(native.shape[0]), + "relative_path": target.relative_to(output_root).as_posix(), + "bytes": target.stat().st_size, + "sha256": sha256_file(target), + } + ) + manifest_out = { + "schema_version": INPUT_SCHEMA, + "source_pack_id": manifest["pack_id"], + "source_pack_sha256": artifact["sha256"], + "config_sha256": sha256_file(config_path), + "coordinate_frame": "map-gravity-local", + "transform": "translation-only-preserve-map-gravity-axis", + "intensity_policy": "zero-filled-algorithm-compatibility-only", + "future_frames_used": False, + "frame_count": FRAME_COUNT, + "profile_count": 2, + "records": records, + "authority": { + "navigation_or_safety_accepted": False, + "actuation_allowed": False, + }, + } + manifest_path = output_root / "input-manifest.json" + manifest_path.write_text( + json.dumps(manifest_out, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return manifest_out + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--source-root", type=Path, required=True) + parser.add_argument("--config", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + manifest = prepare(args.source_root, args.config, args.output_root) + print(json.dumps({"ok": True, "records": len(manifest["records"])}, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/prepare_mixed_route_review_pack.py b/scripts/prepare_mixed_route_review_pack.py new file mode 100644 index 0000000..eb10499 --- /dev/null +++ b/scripts/prepare_mixed_route_review_pack.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +"""Publish exact, independently decodable camera islands for mixed-route review.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import tempfile +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from PIL import Image + +from k1link.compute.jobs import validate_camera_compute_job +from k1link.device_plugins.xgrids_k1.mqtt.capture import read_capture_clock_origin + +SCHEMA = "missioncore.mixed-route-review-pack/v1" +MAX_INDEX_LINE_BYTES = 64 * 1024 + + +class MixedRouteReviewPackError(RuntimeError): + """The selected camera evidence cannot be published without ambiguity.""" + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _sequences(value: str) -> tuple[int, ...]: + try: + sequences = tuple(int(item) for item in value.split(",")) + except ValueError as exc: + raise argparse.ArgumentTypeError("sequences must be comma-separated integers") from exc + if not sequences or any(item < 1 for item in sequences): + raise argparse.ArgumentTypeError("sequences must be positive") + if len(set(sequences)) != len(sequences) or tuple(sorted(sequences)) != sequences: + raise argparse.ArgumentTypeError("sequences must be unique and increasing") + return sequences + + +def _arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--job", type=Path, required=True) + parser.add_argument("--session", type=Path, required=True) + parser.add_argument("--sequences", type=_sequences, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--ffmpeg", type=Path, required=True) + return parser.parse_args() + + +def _read_selected_index( + path: Path, + sequences: tuple[int, ...], +) -> list[dict[str, Any]]: + wanted = set(sequences) + selected: dict[int, dict[str, Any]] = {} + with path.open("rb") as stream: + for expected_sequence, line in enumerate(stream, start=1): + if len(line) > MAX_INDEX_LINE_BYTES or not line.endswith(b"\n"): + raise MixedRouteReviewPackError("camera index line is invalid") + if expected_sequence not in wanted: + continue + try: + value = json.loads(line) + except json.JSONDecodeError as exc: + raise MixedRouteReviewPackError("camera index JSON is invalid") from exc + if ( + not isinstance(value, dict) + or value.get("schema_version") + != "missioncore.camera-recording-index/v1" + or value.get("kind") != "media" + or value.get("sequence") != expected_sequence + or value.get("path") != f"segments/{expected_sequence}.m4s" + or not isinstance(value.get("session_monotonic_ns"), int) + or not isinstance(value.get("host_monotonic_ns"), int) + or not isinstance(value.get("host_epoch_ns"), int) + ): + raise MixedRouteReviewPackError("selected camera index row changed") + selected[expected_sequence] = value + if tuple(sorted(selected)) != sequences: + raise MixedRouteReviewPackError("selected camera sequence is incomplete") + return [selected[sequence] for sequence in sequences] + + +def _decode_exact_fragment( + *, + ffmpeg: Path, + init_path: Path, + segment_path: Path, + output_path: Path, +) -> None: + input_value = f"concat:{init_path}|{segment_path}" + completed = subprocess.run( + [ + os.fspath(ffmpeg), + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-y", + "-i", + input_value, + "-frames:v", + "1", + os.fspath(output_path), + ], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + if completed.returncode != 0 or not output_path.is_file(): + detail = completed.stderr.strip().splitlines()[-1:] or ["no decoded frame"] + raise MixedRouteReviewPackError( + f"selected fragment is not independently decodable: {segment_path.name}: {detail[0]}" + ) + with Image.open(output_path) as image: + if image.mode != "RGB" or image.size != (800, 600): + raise MixedRouteReviewPackError("selected camera frame shape changed") + + +def prepare( + *, + job_root: Path, + session_root: Path, + sequences: tuple[int, ...], + output_root: Path, + ffmpeg_path: Path, +) -> Path: + job = validate_camera_compute_job(job_root) + session = session_root.resolve(strict=True) + if not session.is_dir() or session.name != job.session_id: + raise MixedRouteReviewPackError("camera job and observation session differ") + capture_root = session / "captures" / "mqtt_live" + origin_path = capture_root / "mqtt.timeline.origin.json" + origin = read_capture_clock_origin(origin_path) + if sequences[-1] > job.segment_count: + raise MixedRouteReviewPackError("selected sequence escapes the camera epoch") + ffmpeg = ffmpeg_path.resolve(strict=True) + if not ffmpeg.is_file(): + raise MixedRouteReviewPackError("ffmpeg is unavailable") + epoch_root = ( + job.job_root + / "input" + / "camera" + / job.source_id + / f"epoch-{job.codec_epoch}" + ) + selected = _read_selected_index(epoch_root / "index.jsonl", sequences) + identity = { + "schema_version": SCHEMA, + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "session_id": job.session_id, + "source_id": job.source_id, + "codec_epoch": job.codec_epoch, + "clock_origin": { + "artifact_sha256": _sha256(origin_path), + "started_epoch_ns": origin.started_at_epoch_ns, + "started_monotonic_ns": origin.started_monotonic_ns, + }, + "selected_sequences": list(sequences), + "selection_policy": "exact-independently-decodable-fragments/v1", + "ground_truth": False, + "authority": { + "navigation_or_safety_accepted": False, + "actuation_allowed": False, + }, + "producer_sha256": _sha256(Path(__file__).resolve(strict=True)), + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + pack_id = f"mixed-route-review-pack-{identity_sha256}" + parent = output_root.resolve() + parent.mkdir(mode=0o700, parents=True, exist_ok=True) + final = parent / pack_id + if final.exists(): + return final + staging = Path(tempfile.mkdtemp(prefix=f".{pack_id}.", dir=parent)) + published = False + try: + frames_root = staging / "frames" + frames_root.mkdir(mode=0o700) + timeline_rows: list[dict[str, Any]] = [] + artifacts: list[dict[str, Any]] = [] + for frame_index, (sequence, row) in enumerate( + zip(sequences, selected, strict=True) + ): + output_path = frames_root / f"frame-{frame_index + 1:06d}.png" + segment_path = epoch_root / "segments" / f"{sequence}.m4s" + _decode_exact_fragment( + ffmpeg=ffmpeg, + init_path=epoch_root / "init.mp4", + segment_path=segment_path, + output_path=output_path, + ) + host_monotonic_ns = int(row["host_monotonic_ns"]) + if host_monotonic_ns < origin.started_monotonic_ns: + raise MixedRouteReviewPackError("selected frame predates the session clock origin") + session_seconds = ( + host_monotonic_ns - origin.started_monotonic_ns + ) / 1e9 + timeline_rows.append( + { + "frame_index": frame_index, + "sequence": frame_index + 1, + "source_frame_index": sequence - 1, + "source_sequence": sequence, + "session_seconds": session_seconds, + "host_monotonic_ns": row["host_monotonic_ns"], + "host_epoch_ns": row["host_epoch_ns"], + } + ) + artifacts.append( + { + "path": output_path.relative_to(staging).as_posix(), + "byte_length": output_path.stat().st_size, + "sha256": _sha256(output_path), + "source_segment_sha256": row["sha256"], + } + ) + timeline_path = staging / "timeline.jsonl" + timeline_path.write_text( + "".join( + json.dumps( + row, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + for row in timeline_rows + ), + encoding="utf-8", + ) + manifest = { + "schema_version": SCHEMA, + "pack_id": pack_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "frame_count": len(sequences), + "timeline": { + "path": timeline_path.name, + "byte_length": timeline_path.stat().st_size, + "sha256": _sha256(timeline_path), + }, + "frames": artifacts, + } + (staging / "manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + encoding="utf-8", + ) + os.replace(staging, final) + published = True + finally: + if not published: + shutil.rmtree(staging, ignore_errors=True) + return final + + +def main() -> int: + args = _arguments() + result = prepare( + job_root=args.job, + session_root=args.session, + sequences=args.sequences, + output_root=args.output_root, + ffmpeg_path=args.ffmpeg, + ) + print(json.dumps({"pack_id": result.name, "output": os.fspath(result)}, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/laboratory/mixed_route_vegetation_review.py b/src/k1link/laboratory/mixed_route_vegetation_review.py new file mode 100644 index 0000000..ec30939 --- /dev/null +++ b/src/k1link/laboratory/mixed_route_vegetation_review.py @@ -0,0 +1,417 @@ +"""Seal RAVNOVES004TREE mixed-route review into the existing vegetation LAB.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import tempfile +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import numpy as np +from PIL import Image, ImageDraw + +from k1link.laboratory.vegetation_shadow_lab import ( + LAB_SCHEMA, + RESULT_PREFIX, + VegetationShadowLabError, + canonical_json, + sha256_path, +) + +REVIEW_SCHEMA = "missioncore.mixed-route-review-pack/v1" +DDRNET_SCHEMA = "missioncore.mixed-route-ddrnet-islands/v1" +TGS_SCHEMA = "missioncore.mixed-route-tgs-result/v1" +FRAME_COUNT = 10 +PHASES = ( + "rural", + "rural", + "rural", + "rural", + "rural", + "transition", + "urban", + "urban", + "urban", + "urban", +) +TGS_COLORS = { + 0: (5, 7, 9), + 1: (132, 188, 86), + 2: (235, 112, 122), + 3: (150, 154, 163), +} + + +def _read_json(path: Path, label: str) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise VegetationShadowLabError(f"{label} is invalid") from exc + if not isinstance(value, dict): + raise VegetationShadowLabError(f"{label} must be an object") + return value + + +def _artifact( + source: Path, + staging: Path, + relative: str, + artifacts: list[dict[str, object]], + *, + role: str, + media_type: str, +) -> dict[str, object]: + if source.is_symlink() or not source.is_file(): + raise VegetationShadowLabError(f"mixed-route artifact is unavailable: {relative}") + target = staging / relative + target.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + shutil.copyfile(source, target) + descriptor = { + "role": role, + "path": relative, + "byte_length": target.stat().st_size, + "sha256": sha256_path(target), + "media_type": media_type, + } + artifacts.append(descriptor) + return descriptor + + +def _image_proof(descriptor: dict[str, object]) -> dict[str, object]: + return {"path": descriptor["path"], "sha256": descriptor["sha256"]} + + +def _render_tgs_costmaps(tgs_root: Path, destination: Path) -> list[Path]: + result = _read_json(tgs_root / "result.json", "mixed-route TGS result") + evidence = result.get("evidence") + costmap = result.get("costmap") + if ( + result.get("schema_version") != TGS_SCHEMA + or result.get("status") != "passed-review-only" + or not isinstance(evidence, dict) + or not isinstance(costmap, dict) + or result.get("summary", {}).get("frame_count") != FRAME_COUNT + or result.get("authority", {}).get("actuation_allowed") is not False + ): + raise VegetationShadowLabError("mixed-route TGS contract changed") + evidence_path = tgs_root / str(evidence.get("path")) + if ( + not evidence_path.is_file() + or evidence.get("bytes") != evidence_path.stat().st_size + or evidence.get("sha256") != sha256_path(evidence_path) + ): + raise VegetationShadowLabError("mixed-route TGS evidence changed") + with np.load(evidence_path, allow_pickle=False) as archive: + centers = archive["costmap_cell_centers_xy_m"] + states = archive["causal_rolling_1s_costmap_states"] + if centers.shape != (2244, 2) or states.shape != (FRAME_COUNT, 2244): + raise VegetationShadowLabError("mixed-route TGS costmap shape changed") + radius = float(costmap["radius_m"]) + cell_size = float(costmap["cell_size_m"]) + size = 600 + scale = size / (radius * 2.0) + outputs: list[Path] = [] + destination.mkdir(mode=0o700, parents=True, exist_ok=True) + for slot in range(FRAME_COUNT): + image = Image.new("RGB", (size, size), TGS_COLORS[0]) + draw = ImageDraw.Draw(image) + half = cell_size * scale / 2.0 + for center, state in zip(centers, states[slot], strict=True): + x = (float(center[0]) + radius) * scale + y = (radius - float(center[1])) * scale + draw.rectangle((x - half, y - half, x + half, y + half), fill=TGS_COLORS[int(state)]) + rover_w = 0.8 * scale + rover_l = 1.0 * scale + cx = size / 2.0 + cy = size / 2.0 + draw.rectangle( + (cx - rover_w / 2, cy - rover_l / 2, cx + rover_w / 2, cy + rover_l / 2), + outline=(255, 255, 255), + width=3, + ) + path = destination / f"frame-{slot + 1:06d}.png" + image.save(path, format="PNG", optimize=True) + outputs.append(path) + return outputs + + +def seal_mixed_route_vegetation_review( + *, + base_lab_root: Path, + review_pack_root: Path, + eomt_root: Path, + ddrnet_root: Path, + tgs_root: Path, + output_root: Path, +) -> Path: + base_root = base_lab_root.resolve(strict=True) + base = _read_json(base_root / "result.json", "base vegetation LAB") + base_identity = base.get("identity") + if ( + base.get("schema_version") != LAB_SCHEMA + or not isinstance(base_identity, dict) + or hashlib.sha256(canonical_json(base_identity)).hexdigest() + != base.get("identity_sha256") + or base.get("result_id") != base_root.name + or not base_root.name.startswith(RESULT_PREFIX) + or base.get("authority", {}).get("commands_enabled") is not False + ): + raise VegetationShadowLabError("base vegetation LAB proof changed") + + pack_root = review_pack_root.resolve(strict=True) + pack = _read_json(pack_root / "manifest.json", "mixed-route review pack") + timeline_path = pack_root / str(pack.get("timeline", {}).get("path")) + if ( + pack.get("schema_version") != REVIEW_SCHEMA + or pack.get("frame_count") != FRAME_COUNT + or pack.get("identity", {}).get("session_id") != "20260828T130511Z_viewer_live" + or pack.get("identity", {}).get("ground_truth") is not False + or not timeline_path.is_file() + or pack.get("timeline", {}).get("sha256") != sha256_path(timeline_path) + ): + raise VegetationShadowLabError("mixed-route review pack changed") + timeline = [json.loads(line) for line in timeline_path.read_text(encoding="utf-8").splitlines()] + if len(timeline) != FRAME_COUNT: + raise VegetationShadowLabError("mixed-route timeline is incomplete") + + eomt = _read_json(eomt_root / "run-report.partial.json", "mixed-route EoMT result") + ddrnet = _read_json(ddrnet_root / "result.json", "mixed-route DDRNet result") + tgs = _read_json(tgs_root / "result.json", "mixed-route TGS result") + if ( + eomt.get("input", {}).get("frames_admitted") != FRAME_COUNT + or eomt.get("metrics", {}).get("frames_processed") != FRAME_COUNT + or eomt.get("ground_truth") is not False + or ddrnet.get("schema_version") != DDRNET_SCHEMA + or ddrnet.get("source", {}).get("pack_id") != pack["pack_id"] + or len(ddrnet.get("frames", [])) != FRAME_COUNT + or ddrnet.get("authority", {}).get("candidate_accepted") is not False + or tgs.get("schema_version") != TGS_SCHEMA + or tgs.get("source", {}).get("review_pack_id") != pack["pack_id"] + or tgs.get("summary", {}).get("frame_count") != FRAME_COUNT + ): + raise VegetationShadowLabError("mixed-route model identities differ") + + output_root.mkdir(mode=0o700, parents=True, exist_ok=True) + temporary = Path(tempfile.mkdtemp(prefix=".mixed-route-vegetation-", dir=output_root)) + artifacts: list[dict[str, object]] = [] + try: + tgs_images = _render_tgs_costmaps(tgs_root, temporary / ".tgs-render") + cases: list[dict[str, object]] = [] + tgs_anchors = { + int(row["slot"]): row + for row in tgs["anchors"] + if row.get("profile_id") == "causal_rolling_1s" + } + for slot, row in enumerate(timeline): + case_id = f"route-{slot + 1:02d}" + relative_root = f"route-review/{case_id}" + source_descriptor = _artifact( + pack_root / "frames" / f"frame-{slot + 1:06d}.png", + temporary, + f"{relative_root}/source.png", + artifacts, + role="mixed-route-source-frame", + media_type="image/png", + ) + city_descriptor = _artifact( + eomt_root / "overlay-frames" / f"frame-{slot + 1:06d}.png", + temporary, + f"{relative_root}/city.png", + artifacts, + role="mixed-route-eomt-overlay", + media_type="image/png", + ) + vegetation_descriptor = _artifact( + ddrnet_root / "overlay-frames" / f"frame-{slot + 1:06d}.png", + temporary, + f"{relative_root}/vegetation.png", + artifacts, + role="mixed-route-ddrnet-overlay", + media_type="image/png", + ) + tgs_descriptor = _artifact( + tgs_images[slot], + temporary, + f"{relative_root}/tgs.png", + artifacts, + role="mixed-route-tgs-costmap", + media_type="image/png", + ) + anchor = tgs_anchors[slot] + cases.append( + { + "case_id": case_id, + "phase": PHASES[slot], + "source_sequence": int(row["source_sequence"]), + "session_seconds": float(row["session_seconds"]), + "assets": { + "source": _image_proof(source_descriptor), + "city": _image_proof(city_descriptor), + "vegetation": _image_proof(vegetation_descriptor), + "tgs": _image_proof(tgs_descriptor), + }, + "tgs": { + "ground_cells": int(anchor["ground_cell_count"]), + "occupied_cells": int(anchor["nonground_cell_count"]), + "rejected_cells": int(anchor["rejected_cell_count"]), + "unobserved_cells": int(anchor["unobserved_cell_count"]), + }, + } + ) + shutil.rmtree(temporary / ".tgs-render") + + proofs = {} + for key, path in ( + ("base", base_root / "result.json"), + ("eomt", eomt_root / "run-report.partial.json"), + ("ddrnet", ddrnet_root / "result.json"), + ("tgs", tgs_root / "result.json"), + ): + descriptor = _artifact( + path, + temporary, + f"proofs/{key}.json", + artifacts, + role="mixed-route-proof", + media_type="application/json", + ) + proofs[key] = _image_proof(descriptor) + _artifact( + tgs_root / str(tgs["evidence"]["path"]), + temporary, + "proofs/tgs-evidence.npz", + artifacts, + role="mixed-route-tgs-evidence", + media_type="application/x-npz", + ) + + route_review = { + "source_id": "RAVNOVES004TREE", + "session_id": "20260828T130511Z_viewer_live", + "pack_id": pack["pack_id"], + "frame_count": FRAME_COUNT, + "ground_truth": False, + "selection_policy": "same-scene-camera-lidar-aligned-review-islands/v1", + "models": { + "city": { + "name": "EoMT Cityscapes", + "frames": FRAME_COUNT, + "inference_fps": eomt["metrics"]["inference_frames_per_second"], + "end_to_end_p95_ms": eomt["metrics"]["latency_ms"]["end_to_end_ms"]["p95"], + }, + "vegetation": { + "name": ddrnet["candidate"]["loaded_model_name"], + "result_id": ddrnet["result_id"], + "frames": FRAME_COUNT, + "latency_p95_ms": ddrnet["timing"]["latency_ms_p95"], + }, + "tgs": { + "name": "TRAVEL/TGS causal rolling 1 s", + "frames": FRAME_COUNT, + "latency_p95_ms": tgs["timing"]["wall_seconds_p95"] * 1000.0, + "cell_size_m": tgs["costmap"]["cell_size_m"], + "radius_m": tgs["costmap"]["radius_m"], + }, + }, + "cases": cases, + "proofs": proofs, + "limitations": [ + "Ten aligned review islands are not a complete route timeline.", + "RAVNOVES004TREE has no manual truth.", + "DDRNet vegetation subtypes remain visually noisy and are not planner authority.", + "TGS does not prove ditch or negative-obstacle detection.", + "People and vehicles require an independent fail-safe detector and STOP path.", + ], + } + authority = { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + "actuation_accepted": False, + "camera_semantics_can_clear_rigid_geometry": False, + } + identity = { + "lab_id": "lab-v1-vegetation-mission-policy", + "base_result_id": base["result_id"], + "selected_candidate": base_identity["selected_candidate"], + "candidate_metrics": base_identity["candidate_metrics"], + "source": { + "shadow_session": "RAVNOVES004TREE", + "shadow_camera": "sensor.camera.right", + "shadow_frame_count": FRAME_COUNT, + "video_shadow_frame_count": 0, + }, + "route_review": route_review, + "authority": authority, + } + identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest() + result_id = f"{RESULT_PREFIX}{identity_sha256}" + manifest = { + "schema_version": LAB_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "created_at_utc": datetime.now(UTC).isoformat(), + "ground_truth": False, + "status": "visual-shadow-ready-policy-not-authorized", + "identity": identity, + "source": identity["source"], + "route_video": None, + "route_review": route_review, + "method": { + "completeness": "bounded-review-islands", + "execution_class": "ai-inference", + "pipeline_id": "ravnoves004tree-eomt-ddrnet-causal-tgs-review/v1", + }, + "metrics": {"candidates": base["metrics"]["candidates"]}, + "decision": { + "selected_candidate": base_identity["selected_candidate"], + "visual_shadow_ready": True, + "full_video_shadow_ready": False, + "mission_policy_ready_for_configuration": True, + "multilayer_policy_review_ready": True, + "navigation_accepted": False, + "production_accepted": False, + }, + "limitations": route_review["limitations"], + "authority": authority, + "catalogs": {"goose": [], "ravnoves": []}, + "artifacts": artifacts, + } + (temporary / "result.json").write_bytes(canonical_json(manifest) + b"\n") + destination = output_root / result_id + if destination.exists(): + raise VegetationShadowLabError("immutable mixed-route LAB result already exists") + os.replace(temporary, destination) + return destination + except Exception: + shutil.rmtree(temporary, ignore_errors=True) + raise + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-lab-root", type=Path, required=True) + parser.add_argument("--review-pack-root", type=Path, required=True) + parser.add_argument("--eomt-root", type=Path, required=True) + parser.add_argument("--ddrnet-root", type=Path, required=True) + parser.add_argument("--tgs-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + print( + seal_mixed_route_vegetation_review( + base_lab_root=args.base_lab_root, + review_pack_root=args.review_pack_root, + eomt_root=args.eomt_root, + ddrnet_root=args.ddrnet_root, + tgs_root=args.tgs_root, + output_root=args.output_root, + ) + ) + + +if __name__ == "__main__": + main()