""" camera.py - central camera demuxer (CameraCache) plus the ffmpeg locator. Keeps one ffmpeg process per output type (jpeg/h264/mjpeg) open, reading the printer's FLV stream and fanning it out to dashboard/OrcaSlicer/Obico consumers. Extracted from kobrax_moonraker_bridge.py; re-exported from there so existing imports keep working. ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. """ import os import sys import time import asyncio import logging log = logging.getLogger("bridge") # Same base-path logic as the main module: next to sys.executable in a # PyInstaller binary, otherwise next to this file. _BASE = os.path.dirname(sys.executable) if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__)) try: import imageio_ffmpeg def _find_ffmpeg() -> str: return imageio_ffmpeg.get_ffmpeg_exe() except ImportError: def _find_ffmpeg() -> str: exe_name = "ffmpeg.exe" if sys.platform == "win32" else "ffmpeg" local = os.path.join(_BASE, exe_name) if os.path.isfile(local): return local return "ffmpeg" class CameraCache: """Zentraler Kamera-Demuxer. Keeps ONE ffmpeg process per output type open that reads the FLV stream from the printer and produces: - MJPEG @ 2fps -> last frame in RAM for /api/camera/snapshot - MPEG-TS (-c:v copy) -> fanout to all /api/camera/h264 subscribers - MJPEG @ 15fps/640px -> fanout to all /api/camera/stream subscribers (the live-view used by the dashboard AND by every Moonraker-compatible client, since server.webcams.list advertises this same stream_url) Damit: * Only ONE FLV connection to the printer per output type (solves the single-client limit / 429) - previously /api/camera/stream opened a brand-new, uncached ffmpeg + printer connection per HTTP client, which competed with the cached jpeg/h264 connections for the printer's very limited number of concurrent camera clients and caused intermittent "stream unavailable" failures. * Snapshots are instant (memory read, no ffmpeg spawn per request) * Multiple parallel H.264/MJPEG consumers possible (plugin + web UI + ...) Lazy start on the first consumer, auto-restart on ffmpeg crash. """ JPEG_SOI = b"\xff\xd8" JPEG_EOI = b"\xff\xd9" TS_CHUNK = 65536 def __init__(self): self._url: str = "" self.latest_jpeg: bytes = b"" self.latest_jpeg_ts: float = 0.0 self.h264_subscribers: "set[asyncio.Queue[bytes]]" = set() self.mjpeg_subscribers: "set[asyncio.Queue[bytes]]" = set() self._proc_jpeg: "asyncio.subprocess.Process | None" = None self._proc_h264: "asyncio.subprocess.Process | None" = None self._proc_mjpeg: "asyncio.subprocess.Process | None" = None self._task_jpeg: "asyncio.Task | None" = None self._task_h264: "asyncio.Task | None" = None self._task_mjpeg: "asyncio.Task | None" = None self._lock = asyncio.Lock() self._fail_count_jpeg: int = 0 self._fail_count_h264: int = 0 self._fail_count_mjpeg: int = 0 def set_url(self, url: str): # A changed URL means the printer rotated its stream token (typically # after a reboot). Running ffmpeg processes still hold the stale URL # and will never pick it up on their own - they only re-read self._url # at the top of their outer loop, which they never reach while blocked # in a stdout read on the old, now-silent connection. Tear them down; # the next ensure_running() respawns them against the new URL. changed = bool(url and self._url and url != self._url) self._url = url if changed: self.reset() def reset(self): """Reset backoff counters and forcefully tear down any running ffmpeg loops - including cancelling their background tasks. Only killing the ffmpeg subprocess is not enough: the owning task might currently be sitting in `await asyncio.sleep(delay)` from a previous exponential backoff (up to 300s) after an earlier failure. Resetting the fail-count doesn't wake it up early, so a user clicking "reset" could see nothing happen for minutes. Cancelling the task guarantees an immediate, clean restart on the next ensure_running() call. """ self._fail_count_jpeg = 0 self._fail_count_h264 = 0 self._fail_count_mjpeg = 0 for task in (self._task_jpeg, self._task_h264, self._task_mjpeg): if task is not None and not task.done(): task.cancel() for proc in (self._proc_jpeg, self._proc_h264, self._proc_mjpeg): if proc is not None: try: proc.kill() except Exception: pass self._task_jpeg = self._task_h264 = self._task_mjpeg = None self._proc_jpeg = self._proc_h264 = self._proc_mjpeg = None async def ensure_running(self): # NOTE: we check the *task* state, not self._proc_* - the process # handle is only assigned later, inside the task body, once ffmpeg # has actually been spawned. Checking self._proc_* here left a race # window: two callers arriving before the newly-created task got a # chance to run would both see "no process yet" and each spawn a # duplicate ffmpeg + duplicate printer connection, silently # orphaning the older one (whichever task's coroutine runs last # overwrites the shared self._proc_* reference, so nobody keeps a # handle to kill the earlier orphaned process). Task creation is # synchronous, so checking self._task_* here is race-free. if self._task_jpeg is None or self._task_jpeg.done(): self._task_jpeg = asyncio.create_task(self._run_jpeg_loop()) if self._task_h264 is None or self._task_h264.done(): self._task_h264 = asyncio.create_task(self._run_h264_loop()) if self._task_mjpeg is None or self._task_mjpeg.done(): self._task_mjpeg = asyncio.create_task(self._run_mjpeg_loop()) def _input_args(self, url: str) -> list[str]: args = ["-fflags", "nobuffer", "-flags", "low_delay", # Bail out if the source goes silent. A printer reboot or # network loss leaves the TCP connection ESTABLISHED with no # data and no FIN, so a passive stdout read blocks forever # without this (Issue #99). Value is microseconds. "-timeout", "10000000"] if url.lower().startswith("rtsp://"): args += ["-probesize", "32", "-analyzeduration", "0", "-rtsp_transport", "tcp"] else: # The printer's FLV source occasionally emits non-monotonic container # timestamps (PTS jumps of days) while the video data itself stays # valid. Without this flag ffmpeg's realtime pacing breaks on such a # jump and the stream stalls after ~15-30 min (Issue #90). args += ["-use_wallclock_as_timestamps", "1", "-probesize", "500000", "-analyzeduration", "500000"] return args async def _run_jpeg_loop(self): """Keeps an ffmpeg process alive that writes MJPEG@2fps into the cache.""" while True: url = self._url if not url: await asyncio.sleep(2.0) continue try: proc = await asyncio.create_subprocess_exec( _find_ffmpeg(), "-loglevel", "warning", *self._input_args(url), "-i", url, "-vf", "fps=2", "-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "3", "-flush_packets", "1", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) self._proc_jpeg = proc except Exception as e: log.warning(f"CameraCache: ffmpeg-jpeg start failed: {e}") await asyncio.sleep(3.0) continue buf = b"" rc = None try: while True: chunk = await proc.stdout.read(self.TS_CHUNK) if not chunk: break buf += chunk # extract complete JPEG frames while True: start = buf.find(self.JPEG_SOI) if start == -1: buf = b"" break end = buf.find(self.JPEG_EOI, start + 2) if end == -1: buf = buf[start:] break self.latest_jpeg = buf[start:end + 2] self.latest_jpeg_ts = time.time() buf = buf[end + 2:] except Exception as e: log.debug(f"CameraCache: jpeg-loop unterbrochen: {e}") finally: # NOTE: cleanup operates on the local `proc` reference, not on # self._proc_jpeg - see _run_mjpeg_loop's identical comment. # If this task got cancelled (e.g. by reset()), a new task may # already have started and assigned its own process to # self._proc_jpeg by the time we reach here; killing that # shared attribute instead of our own local proc would kill # the WRONG (newer) process and leak this one as an orphan. try: proc.kill() except Exception: pass try: await proc.wait() except Exception: pass rc = proc.returncode if rc: try: err = await proc.stderr.read(500) if err: log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}") except Exception: pass if self._proc_jpeg is proc: self._proc_jpeg = None if rc: self._fail_count_jpeg += 1 delay = min(2.0 * (2 ** self._fail_count_jpeg), 300.0) log.warning(f"CameraCache: ffmpeg-jpeg exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_jpeg})") await asyncio.sleep(delay) else: self._fail_count_jpeg = 0 await asyncio.sleep(2.0) async def _run_h264_loop(self): """Keeps an ffmpeg process alive that fans out MPEG-TS to all subscribers.""" while True: url = self._url if not url: await asyncio.sleep(2.0) continue try: proc = await asyncio.create_subprocess_exec( _find_ffmpeg(), "-loglevel", "warning", *self._input_args(url), "-i", url, "-c:v", "copy", "-an", "-f", "mpegts", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) self._proc_h264 = proc except Exception as e: log.warning(f"CameraCache: ffmpeg-h264 start failed: {e}") await asyncio.sleep(3.0) continue rc = None try: while True: chunk = await proc.stdout.read(self.TS_CHUNK) if not chunk: break # Fanout: non-blocking per subscriber; slow clients # get their oldest chunk dropped (queue full -> drop). for q in list(self.h264_subscribers): if q.full(): try: q.get_nowait() except Exception: pass try: q.put_nowait(chunk) except Exception: pass except Exception as e: log.debug(f"CameraCache: h264-loop unterbrochen: {e}") finally: # NOTE: cleanup operates on the local `proc` reference, not on # self._proc_h264 - see _run_mjpeg_loop's identical comment. # If this task got cancelled (e.g. by reset()), a new task may # already have started and assigned its own process to # self._proc_h264 by the time we reach here; killing that # shared attribute instead of our own local proc would kill # the WRONG (newer) process and leak this one as an orphan. try: proc.kill() except Exception: pass try: await proc.wait() except Exception: pass rc = proc.returncode if rc: try: err = await proc.stderr.read(500) if err: log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}") except Exception: pass if self._proc_h264 is proc: self._proc_h264 = None if rc: self._fail_count_h264 += 1 delay = min(2.0 * (2 ** self._fail_count_h264), 300.0) log.warning(f"CameraCache: ffmpeg-h264 exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_h264})") await asyncio.sleep(delay) else: self._fail_count_h264 = 0 await asyncio.sleep(2.0) async def _run_mjpeg_loop(self): """Keeps an ffmpeg process alive that fans out MJPEG@15fps/640px (complete JPEG frames) to all /api/camera/stream subscribers.""" while True: url = self._url if not url: await asyncio.sleep(2.0) continue try: proc = await asyncio.create_subprocess_exec( _find_ffmpeg(), "-loglevel", "warning", *self._input_args(url), "-i", url, "-vf", "fps=15,scale=640:-1", "-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "3", "-flush_packets", "1", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) self._proc_mjpeg = proc except Exception as e: log.warning(f"CameraCache: ffmpeg-mjpeg start failed: {e}") await asyncio.sleep(3.0) continue buf = b"" rc = None try: while True: chunk = await proc.stdout.read(self.TS_CHUNK) if not chunk: break buf += chunk # extract complete JPEG frames and fan them out whole # (so every subscriber gets clean multipart boundaries, # not arbitrary byte chunks like the h264/mpegts fanout) while True: start = buf.find(self.JPEG_SOI) if start == -1: buf = b"" break end = buf.find(self.JPEG_EOI, start + 2) if end == -1: buf = buf[start:] break frame = buf[start:end + 2] buf = buf[end + 2:] for q in list(self.mjpeg_subscribers): if q.full(): try: q.get_nowait() except Exception: pass try: q.put_nowait(frame) except Exception: pass except Exception as e: log.debug(f"CameraCache: mjpeg-loop unterbrochen: {e}") finally: # NOTE: cleanup operates on the local `proc` reference, not # on self._proc_mjpeg. If this task got cancelled (e.g. by # reset()) a new task may already have started and assigned # its own process to self._proc_mjpeg by the time we reach # here - killing that shared attribute instead of our own # local proc would kill the WRONG (newer) process. try: proc.kill() except Exception: pass try: await proc.wait() except Exception: pass rc = proc.returncode if rc: try: err = await proc.stderr.read(500) if err: log.warning(f"CameraCache: ffmpeg-mjpeg stderr: {err.decode(errors='replace').strip()}") except Exception: pass if self._proc_mjpeg is proc: self._proc_mjpeg = None if rc: self._fail_count_mjpeg += 1 delay = min(2.0 * (2 ** self._fail_count_mjpeg), 300.0) log.warning(f"CameraCache: ffmpeg-mjpeg exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_mjpeg})") await asyncio.sleep(delay) else: self._fail_count_mjpeg = 0 await asyncio.sleep(2.0)