#!/usr/bin/env python3 """Camera capture supervisor. Wraps one physical video device, stream-copying (never re-encoding) into hourly, crash-safe segments. MJPEG modes record raw .mjpeg segments (bare concatenated JPEG frames -- inherently crash-safe, truncation only ever costs the final frame, and trivially parseable); H.264 modes record Matroska (.mkv), since a raw H.264 elementary stream would lose framing/timestamps. MJPEG modes capture via `v4l2-ctl ... --stream-to=- | ffmpeg -f mjpeg -i -` rather than letting ffmpeg open the V4L2 device itself. The reason is memory: ffmpeg's v4l2 input allocates a fixed 32 capture buffers, and at 3200x1200 the driver reports 7.68MB per buffer -- ~246MB, which the OOM killer refuses on a 512MB Zero 2 W (measured: ffmpeg died with total-vm 243308kB). v4l2-ctl's --stream-mmap=N caps that at camera.capture_buffers (default 3, ~23MB), and ffmpeg -- reading an already-framed MJPEG byte stream on stdin -- never makes the big allocation. H.264 modes keep the single-ffmpeg path; they've not been exercised on real hardware and their per-buffer sizes are far smaller. Actively confirms that capture is genuinely progressing -- not just that the processes launched -- before reporting state="running". This is what makes the LED's RED->GREEN transition (see led_status.py) trustworthy. Usage: capture_supervisor.py is "primary" for a single combined side-by-side stream, or "left"/"right" if the camera enumerates as two independent UVC nodes. It matches the systemd template instance (bubblecam-camera@) and selects which config-file device path / output filename prefix to use. """ from __future__ import annotations import signal import subprocess import sys import time from pathlib import Path from typing import List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from bubblecam import config, events, state_lib # noqa: E402 # Consecutive no-progress checks before escalating from "stalled" to a hard # error (which exits non-zero and lets systemd's Restart=on-failure retry). # At growth_check_delay_seconds=5 (default), 3 strikes is ~15s of silence. STALL_LIMIT = 3 # How long the writer (ffmpeg) gets to finalize the current segment after the # source closes, before being signalled. Generous on purpose: this runs at # service stop, and systemd's own TimeoutStopSec (90s default) is the real # outer bound. Finishing the segment properly matters more than stopping fast. WRITER_EOF_GRACE_SECONDS = 30 # v4l2 demuxer -input_format string per confirmed camera.mode. NOTE: the # exact string the kernel/ffmpeg expect for an onboard-H.264 UVC source # should be double-checked against `v4l2-ctl --list-formats-ext` output on # real hardware -- "h264" is the common case but isn't guaranteed universal. INPUT_FORMAT_BY_MODE = { "sbs_mjpeg": "mjpeg", "sbs_h264": "h264", "dual_node_mjpeg": "mjpeg", "dual_node_h264": "h264", } # (ffmpeg segment_format, file extension) per mode -- see module docstring. SEGMENT_FORMAT_BY_MODE = { "sbs_mjpeg": ("mjpeg", "mjpeg"), "sbs_h264": ("matroska", "mkv"), "dual_node_mjpeg": ("mjpeg", "mjpeg"), "dual_node_h264": ("matroska", "mkv"), } def segment_extension(cfg: dict) -> str: return SEGMENT_FORMAT_BY_MODE[cfg["camera"]["mode"]][1] def uses_v4l2ctl_source(cfg: dict) -> bool: """MJPEG modes use the low-memory v4l2-ctl source (see module docstring).""" return INPUT_FORMAT_BY_MODE[cfg["camera"]["mode"]] == "mjpeg" def device_for_instance(cfg: dict, instance: str) -> str: camera_cfg = cfg["camera"] if instance == "primary": return camera_cfg["device_primary"] if instance == "left": return camera_cfg["device_left"] if instance == "right": return camera_cfg["device_right"] raise ValueError(f"unknown camera instance {instance!r}") def latest_segment(video_dir: Path, instance: str, extension: str = "mjpeg") -> Optional[Path]: matches = sorted(video_dir.glob(f"cam_{instance}_*.{extension}"), key=lambda p: p.stat().st_mtime) return matches[-1] if matches else None def build_v4l2_ctl_command(cfg: dict, device: str) -> List[str]: """Low-memory MJPEG source: raw JPEG frames to stdout, N mmap buffers.""" camera_cfg = cfg["camera"] width, height = camera_cfg["resolution"].split("x") return [ "v4l2-ctl", "-d", device, f"--set-fmt-video=width={width},height={height},pixelformat=MJPG", f"--set-parm={camera_cfg['framerate']}", f"--stream-mmap={camera_cfg['capture_buffers']}", "--stream-to=-", ] def build_ffmpeg_command( cfg: dict, device: str, output_pattern: str, from_stdin: bool = False ) -> List[str]: camera_cfg = cfg["camera"] segment_format = SEGMENT_FORMAT_BY_MODE[camera_cfg["mode"]][0] if from_stdin: # Raw MJPEG carries no timestamps, so -framerate is what the segment # muxer uses to decide segment length. Get it wrong and hourly # segments come out the wrong wall-clock duration. source = [ "-f", "mjpeg", "-framerate", str(camera_cfg["framerate"]), "-i", "-", ] # No -nostdin: stdin IS the video source here. prefix = ["ffmpeg", "-hide_banner", "-loglevel", "warning"] else: source = [ "-f", "v4l2", "-input_format", INPUT_FORMAT_BY_MODE[camera_cfg["mode"]], "-video_size", camera_cfg["resolution"], "-framerate", str(camera_cfg["framerate"]), "-i", device, ] prefix = ["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning"] return prefix + source + [ "-c", "copy", "-f", "segment", "-segment_time", str(camera_cfg["segment_seconds"]), "-segment_format", segment_format, "-strftime", "1", "-reset_timestamps", "1", output_pattern, ] def main() -> int: if len(sys.argv) != 2: print(f"usage: {sys.argv[0]} ", file=sys.stderr) return 2 instance = sys.argv[1] cfg = config.load() state_name = f"camera_{instance}" video_dir = Path(cfg["paths"]["video_dir"]) video_dir.mkdir(parents=True, exist_ok=True) device = device_for_instance(cfg, instance) growth_check_delay = cfg["camera"]["growth_check_delay_seconds"] state_lib.write_state(state_name, "starting", device=device) extension = segment_extension(cfg) output_pattern = str(video_dir / f"cam_{instance}_%Y%m%dT%H%M%S.{extension}") source_proc: Optional[subprocess.Popen] = None if uses_v4l2ctl_source(cfg): source_proc = subprocess.Popen( build_v4l2_ctl_command(cfg, device), stdout=subprocess.PIPE, # v4l2-ctl prints a running "<<< N fps" progress line roughly # once a second; that would swamp a size-capped journal over a # multi-day mission. Its exit code still surfaces failures. stderr=subprocess.DEVNULL, ) writer_proc = subprocess.Popen( build_ffmpeg_command(cfg, device, output_pattern, from_stdin=True), stdin=source_proc.stdout, ) # Close our copy so the writer sees EOF when the source exits -- # that's what lets a stopped source finalize the segment cleanly. source_proc.stdout.close() else: writer_proc = subprocess.Popen(build_ffmpeg_command(cfg, device, output_pattern)) def _stop_pipeline() -> None: """Stop source first, then let the writer finish on its own. Closing the source's stdout gives ffmpeg EOF, which makes it flush and write its trailer -- the whole reason for stopping in this order. It must be given time to actually do that: signalling it straight away produced "Error writing trailer: Immediate exit requested" and an unfinalized segment. SIGTERM is only the fallback if EOF alone doesn't end it, and SIGKILL only if SIGTERM doesn't. """ if source_proc is not None and source_proc.poll() is None: source_proc.terminate() try: source_proc.wait(timeout=5) except subprocess.TimeoutExpired: source_proc.kill() if writer_proc.poll() is None: try: writer_proc.wait(timeout=WRITER_EOF_GRACE_SECONDS) except subprocess.TimeoutExpired: writer_proc.send_signal(signal.SIGTERM) try: writer_proc.wait(timeout=10) except subprocess.TimeoutExpired: writer_proc.kill() def _forward_sigterm(signum, frame): _stop_pipeline() # Explicit fresh "stopped" so readers (e-paper, LED) flip to NOT # RECORDING immediately, rather than showing the last "running" # state for the ~30s it takes to go stale. state_lib.write_state(state_name, "stopped", device=device) events.log("camera", "recording stopped (segment finalized)") raise SystemExit(0) signal.signal(signal.SIGTERM, _forward_sigterm) def _pipeline_failure() -> Optional[str]: if writer_proc.poll() is not None: return f"ffmpeg exited with code {writer_proc.returncode}" if source_proc is not None and source_proc.poll() is not None: return f"v4l2-ctl exited with code {source_proc.returncode}" return None confirmed_running = False consecutive_stalls = 0 try: while True: failure = _pipeline_failure() if failure: state_lib.write_state(state_name, "error", device=device, detail=failure) events.log("camera", f"ERROR: {failure}") return 1 segment_before = latest_segment(video_dir, instance, extension) size_before = segment_before.stat().st_size if segment_before else None time.sleep(growth_check_delay) failure = _pipeline_failure() if failure: state_lib.write_state(state_name, "error", device=device, detail=failure) return 1 segment_after = latest_segment(video_dir, instance, extension) if segment_after is None: # Processes alive but nothing written yet. consecutive_stalls += 1 progressed = False else: appeared = segment_before is None rolled_over = (not appeared) and segment_after.name != segment_before.name grew = (not appeared) and (not rolled_over) and segment_after.stat().st_size > size_before progressed = appeared or rolled_over or grew if progressed: consecutive_stalls = 0 if not confirmed_running: events.log("camera", f"recording confirmed -- writing {segment_after.name}") elif rolled_over: events.log("camera", f"segment rollover -- {segment_after.name}") confirmed_running = True state_lib.write_state( state_name, "running", device=device, segment=segment_after.name, ) else: consecutive_stalls += 1 if consecutive_stalls >= STALL_LIMIT: state_lib.write_state( state_name, "error", device=device, detail=f"no capture progress for {STALL_LIMIT} consecutive checks", ) events.log("camera", f"ERROR: no capture progress for {STALL_LIMIT} checks") return 1 if not progressed and not confirmed_running: state_lib.write_state(state_name, "starting", device=device) finally: _stop_pipeline() if __name__ == "__main__": sys.exit(main())