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.
180 lines
7.5 KiB
Python
180 lines
7.5 KiB
Python
"""Retained fMP4 through the real durable gateway and a bounded decoder reader.
|
|
|
|
The producer reads local files at original receipt cadence. It cannot connect to
|
|
RTSP or send device commands. This qualifies downstream camera contention, not
|
|
network transport, device authority, or browser MSE.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from k1link.device_plugins.xgrids_k1.camera import (
|
|
XgridsK1CameraGateway,
|
|
_CameraProducer,
|
|
_read_fmp4_stdout,
|
|
)
|
|
from k1link.missions.causal_replay import digest
|
|
from k1link.web.camera_archive import CameraArchiveWriter
|
|
|
|
|
|
def camera_inputs(epoch):
|
|
rows = [json.loads(line) for line in (epoch / "index.jsonl").read_text().splitlines()]
|
|
assert rows
|
|
inputs = {str(epoch / "index.jsonl"): digest(epoch / "index.jsonl")}
|
|
if rows[0]["kind"] != "init":
|
|
# The canonical August baseline indexes media only. Its init timestamp
|
|
# is unavailable; deliver init at the first media receipt without
|
|
# inventing a measured initialization latency. All media deltas survive.
|
|
summary = json.loads((epoch / "summary.json").read_text())
|
|
inputs[str(epoch / "summary.json")] = digest(epoch / "summary.json")
|
|
rows.insert(0, dict(kind="init", path="init.mp4", sha256=summary["init_sha256"],
|
|
host_monotonic_ns=rows[0]["host_monotonic_ns"]))
|
|
previous = rows[0]["host_monotonic_ns"]
|
|
for row in rows:
|
|
path = (epoch / row["path"]).resolve()
|
|
assert path.is_relative_to(epoch.resolve()), "Camera path escaped retained epoch."
|
|
assert row["host_monotonic_ns"] >= previous
|
|
previous = row["host_monotonic_ns"]
|
|
assert digest(path) == row["sha256"]
|
|
inputs[str(path)] = row["sha256"]
|
|
assert (previous - rows[0]["host_monotonic_ns"]) / 1e9 <= 65
|
|
return rows, inputs
|
|
|
|
|
|
class ArchiveCamera:
|
|
def __init__(self, epoch, root, ffmpeg):
|
|
self.epoch, self.root, self.ffmpeg = epoch, root, ffmpeg
|
|
self.rows, self.inputs = camera_inputs(epoch)
|
|
self.gateway = None
|
|
self.producer = None
|
|
self.decoder = None
|
|
self.threads = []
|
|
self.files = []
|
|
self.errors = []
|
|
self.commits = []
|
|
self.received = []
|
|
|
|
def start(self):
|
|
self.root.mkdir(parents=True, exist_ok=False)
|
|
self.started = time.monotonic_ns()
|
|
# Explicit local emitter only; never use the gateway's RTSP spawn path.
|
|
process = subprocess.Popen(
|
|
[sys.executable, str(Path(__file__).resolve()), str(self.epoch)],
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
|
)
|
|
self.gateway = XgridsK1CameraGateway(
|
|
self.root, "archive-qualification", committed_segment_observer=self.observe,
|
|
)
|
|
self.producer = _CameraProducer(
|
|
1, "sensor.camera.right", process,
|
|
CameraArchiveWriter(self.root, "sensor.camera.right", 1),
|
|
)
|
|
# Private instance uses the same parser/archive/queue implementation.
|
|
# No singleton, active acquisition, authority ledger, or network endpoint.
|
|
self.gateway._producer = self.producer
|
|
self.gateway._generation = 1
|
|
self.gateway._source_id = self.producer.source_id
|
|
self.gateway._recording_root = self.root
|
|
self.gateway._phase = "connecting"
|
|
self.lease = self.gateway.open_delivery(1, require_recording=True)
|
|
self.gateway.expect_source_end_for_device_stop() # Expected fixture EOF only.
|
|
frames = (self.root / "decoded.framemd5").open("wb")
|
|
errors = (self.root / "decode.log").open("wb")
|
|
self.files.extend([frames, errors])
|
|
self.decoder = subprocess.Popen(
|
|
[str(self.ffmpeg), "-hide_banner", "-loglevel", "warning", "-threads", "1",
|
|
"-i", "pipe:0", "-map", "0:v:0", "-an", "-f", "framemd5", "pipe:1"],
|
|
stdin=subprocess.PIPE, stdout=frames, stderr=errors,
|
|
)
|
|
self.threads = [
|
|
threading.Thread(target=self.decode, name="archive-camera-preview"),
|
|
threading.Thread(
|
|
target=_read_fmp4_stdout, args=(self.gateway, self.producer),
|
|
name="archive-camera-parser",
|
|
),
|
|
]
|
|
for thread in self.threads:
|
|
thread.start()
|
|
|
|
def observe(self, segment):
|
|
self.commits.append(dict(
|
|
kind=segment.kind, sequence=segment.sequence,
|
|
at_s=(time.monotonic_ns() - self.started) / 1e9,
|
|
sha256=hashlib.sha256(segment.payload).hexdigest(),
|
|
))
|
|
|
|
def decode(self):
|
|
try:
|
|
while (segment := self.lease.segments.get()) is not None:
|
|
kind, payload = segment
|
|
self.decoder.stdin.write(payload)
|
|
self.decoder.stdin.flush()
|
|
self.received.append(dict(kind=kind, sha256=hashlib.sha256(payload).hexdigest()))
|
|
except Exception as exc:
|
|
self.errors.append(f"{type(exc).__name__}: {exc}")
|
|
finally:
|
|
self.decoder.stdin.close()
|
|
|
|
def close(self):
|
|
if self.gateway is None:
|
|
return
|
|
for thread in self.threads:
|
|
thread.join(3)
|
|
self.gateway.close()
|
|
for thread in self.threads:
|
|
thread.join(3)
|
|
for process in (self.producer.process, self.decoder):
|
|
try:
|
|
process.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=3)
|
|
self.errors.append("Child exceeded cleanup deadline.")
|
|
for stream in self.files:
|
|
stream.close()
|
|
if self.producer.process.stdout:
|
|
self.producer.process.stdout.close()
|
|
|
|
def report(self):
|
|
archive = self.root / "media/sensor.camera.right/epoch-1"
|
|
summary = json.loads((archive / "summary.json").read_text())
|
|
expected = [row["sha256"] for row in self.rows]
|
|
frames = [line for line in (self.root / "decoded.framemd5").read_text().splitlines()
|
|
if line and not line.startswith("#")]
|
|
decoder_log = (self.root / "decode.log").read_text()
|
|
checks = dict(
|
|
archive_complete=summary["status"] == "complete",
|
|
committed_all_exact_bytes=[item["sha256"] for item in self.commits] == expected,
|
|
delivered_all_exact_bytes=[item["sha256"] for item in self.received] == expected,
|
|
preview_not_retired=self.lease.failure_code is None,
|
|
clean_decode=self.decoder.returncode == 0 and not decoder_log.strip() and bool(frames),
|
|
workers_stopped=all(not t.is_alive() for t in self.threads),
|
|
no_errors=not self.errors,
|
|
input_unchanged=all(digest(Path(p)) == sha for p, sha in self.inputs.items()),
|
|
)
|
|
return dict(checks=checks, errors=self.errors, commits=self.commits,
|
|
media_segments=len(self.rows) - 1, decoded_frames=len(frames),
|
|
decoder_returncode=self.decoder.returncode, browser_mse_tested=False,
|
|
rtsp_tested=False, acquisition_authority_tested=False)
|
|
|
|
|
|
def emit(epoch):
|
|
rows, _ = camera_inputs(epoch)
|
|
started, origin = time.monotonic_ns(), rows[0]["host_monotonic_ns"]
|
|
for row in rows:
|
|
delay = started + row["host_monotonic_ns"] - origin - time.monotonic_ns()
|
|
if delay > 0:
|
|
time.sleep(delay / 1e9)
|
|
sys.stdout.buffer.write((epoch / row["path"]).read_bytes())
|
|
sys.stdout.buffer.flush()
|
|
time.sleep(0.5) # Allow the disposable reader to drain before expected EOF.
|
|
|
|
|
|
if __name__ == "__main__":
|
|
emit(Path(sys.argv[1]))
|