"""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]))