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

