diff --git a/bridge_logging.py b/bridge_logging.py new file mode 100644 index 0000000..24ff7d6 --- /dev/null +++ b/bridge_logging.py @@ -0,0 +1,60 @@ +""" +bridge_logging.py - browser log stream plumbing for the bridge. + +Holds the ring buffer + SSE queues that feed the Web UI's live log view, the +logging.Handler that populates them, and the verbose-HTTP-log toggle. +Extracted from kobrax_moonraker_bridge.py; the buffer/queues are re-imported +there so the log-stream/download endpoints keep working against the same +shared objects. + +──────────────────────────────────────────────────────────────────────────── +Copyright (C) 2026 viewit (KX-Bridge contributors) + +Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. +""" + +import collections as _collections +import logging + + +def _set_verbose_http_log(enabled: bool): + logging.getLogger("aiohttp.access").setLevel(logging.INFO if enabled else logging.WARNING) + + +# Ring buffer for the browser log stream + the open SSE consumer queues. +# These are shared mutable objects: kobrax_moonraker_bridge re-imports the +# same buffer/list so its log-stream/download handlers append/read the very +# same instances this handler writes to. +_log_buffer: "_collections.deque[dict]" = _collections.deque(maxlen=500) +_log_sse_queues: list = [] + + +class _BrowserLogHandler(logging.Handler): + """Sends log records to the ring buffer and all open SSE queues.""" + _fmt = logging.Formatter(datefmt="%H:%M:%S") + + def emit(self, record: logging.LogRecord): + msg = record.getMessage() + # Pass exceptions with traceback through to the browser (otherwise the + # user only sees "Error: X" without context). + if record.exc_info: + try: + msg += "\n" + self._fmt.formatException(record.exc_info) + except Exception: + pass + entry = { + "ts": self._fmt.formatTime(record, "%H:%M:%S"), + "lvl": record.levelname, + "name": record.name, + "msg": msg, + } + _log_buffer.append(entry) + for q in list(_log_sse_queues): + try: + q.put_nowait(entry) + except Exception: + pass + + +_browser_handler = _BrowserLogHandler() +logging.getLogger().addHandler(_browser_handler) diff --git a/camera.py b/camera.py new file mode 100644 index 0000000..f5a21f5 --- /dev/null +++ b/camera.py @@ -0,0 +1,409 @@ +""" +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) + diff --git a/gcode_meta.py b/gcode_meta.py new file mode 100644 index 0000000..eced7d5 --- /dev/null +++ b/gcode_meta.py @@ -0,0 +1,182 @@ +""" +gcode_meta.py - GCode file metadata extraction helpers (estimated print time, +layer heights, embedded thumbnail, per-slot filament info). + +Extracted from kobrax_moonraker_bridge.py; re-exported from there so existing +call sites keep working. Used by the file-upload and print-start paths. + +──────────────────────────────────────────────────────────────────────────── +Copyright (C) 2026 viewit (KX-Bridge contributors) + +Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. +""" + +import re +import base64 +import logging + +log = logging.getLogger("bridge") + + +def _parse_gcode_estimated_time(data: bytes) -> int: + """Reads the estimated print time from GCode (OrcaSlicer + PrusaSlicer). + Returns seconds, 0 when not found. + PrusaSlicer writes the time into the header (first 16KB), + OrcaSlicer writes it at the end of the file (last 16KB).""" + import re + # Search the beginning + end of the file (OrcaSlicer writes the time at the end) + search_text = (data[:16384] + data[-65536:]).decode("utf-8", errors="ignore") + # OrcaSlicer: ; total estimated time: 9m 20s + # PrusaSlicer: ; estimated printing time (normal mode) = 1h 9m 20s + m = (re.search(r";\s*total estimated time:\s*(.*)", search_text) or + re.search(r";\s*estimated printing time \(normal mode\)\s*=\s*(.*)", search_text)) + if not m: + return 0 + parts = re.findall(r"(\d+)\s*([hms])", m.group(1)) + secs = 0 + for val, unit in parts: + if unit == "h": secs += int(val) * 3600 + elif unit == "m": secs += int(val) * 60 + elif unit == "s": secs += int(val) + if secs: + log.info(f"Slicer estimate: {secs}s ({m.group(1).strip()})") + return secs + + +def _parse_gcode_layer_heights(data: bytes) -> tuple[float, float]: + """Reads (layer_height, initial_layer_height) from the OrcaSlicer/PrusaSlicer + GCode header. Both are stored as a config block at the end of the GCode. + + Beispiel-Zeilen: + ; layer_height = 0.2 + ; initial_layer_print_height = 0.2 + + Returns (0.0, 0.0) when not found - the caller decides what to do + (typisch: keinen Z-Wert anzeigen).""" + import re + head = data[:16384].decode("utf-8", errors="ignore") + tail = data[-65536:].decode("utf-8", errors="ignore") + search = head + "\n" + tail + def _grab(pat): + m = re.search(pat, search) + if not m: + return 0.0 + try: + return float(m.group(1)) + except Exception: + return 0.0 + layer_h = _grab(r";\s*layer_height\s*=\s*([0-9.]+)") + first_h = (_grab(r";\s*initial_layer_print_height\s*=\s*([0-9.]+)") or + _grab(r";\s*first_layer_height\s*=\s*([0-9.]+)") or + layer_h) + return layer_h, first_h + + +def _extract_thumbnail(data: bytes) -> str: + """Extracts the base64 PNG thumbnail from GCode (OrcaSlicer format).""" + try: + marker = b"; thumbnail begin" + end_marker = b"; thumbnail end" + start = data.find(marker) + if start == -1: + return "" + start = data.find(b"\n", start) + 1 + end = data.find(end_marker, start) + if end == -1: + return "" + lines = data[start:end].split(b"\n") + b64 = b"".join( + line[2:].strip() if line.startswith(b"; ") else line.strip() + for line in lines + ) + return b64.decode("ascii") + except Exception: + return "" + + +def _extract_filament_info(data: bytes) -> list[dict]: + """Reads filament colors/materials incl. tool order from Orca/Prusa GCode. + + Gibt Liste von {slot_index, color_hex, material} in Tool-/Paint-Reihenfolge + (T0, T1, ...). + Searches both the start and the end of the file since Orca can insert + large thumbnail blocks, pushing the metadata into the tail. + """ + try: + head = data[:131072] + tail = data[-131072:] if len(data) > 131072 else b"" + header = (head + b"\n" + tail).decode("utf-8", errors="ignore") + colors, materials = [], [] + paint_count_hint = 0 + tool_filament_order = [] + for line in header.splitlines(): + if re.match(r"^\s*;\s*filament_colour\s*=", line): + val = line.split("=", 1)[-1].strip() + colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()] + elif re.match(r"^\s*;\s*filament_multi_colour\s*=", line) and not colors: + val = line.split("=", 1)[-1].strip() + colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()] + elif re.match(r"^\s*;\s*filament_type\s*=", line): + val = line.split("=", 1)[-1].strip() + parts = [m.strip() for m in re.split(r"[;,]", val) if m.strip()] + materials = parts + paint_count_hint = max(paint_count_hint, len(parts)) + elif re.match(r"^\s*;\s*filament_density\s*:", line): + val = line.split(":", 1)[-1].strip() + parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()] + paint_count_hint = max(paint_count_hint, len(parts)) + elif re.match(r"^\s*;\s*filament_diameter\s*:", line): + val = line.split(":", 1)[-1].strip() + parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()] + paint_count_hint = max(paint_count_hint, len(parts)) + elif re.match(r"^\s*;\s*filament\s*:", line): + raw = line.split(":", 1)[-1] + parsed = [] + for p in [x.strip() for x in raw.split(",") if x.strip()]: + try: + parsed.append(int(p)) + except Exception: + pass + if parsed: + tool_filament_order = parsed + total_paints = max(len(colors), len(materials), paint_count_hint) + if tool_filament_order: + total_paints = max(total_paints, max(tool_filament_order)) + if total_paints <= 0: + return [] + + # Keep full paint list visible; mark paints referenced by Orca tool order as used. + if len(colors) < total_paints: + colors.extend(["FFFFFF"] * (total_paints - len(colors))) + if len(materials) < total_paints: + materials.extend(["PLA"] * (total_paints - len(materials))) + # Prefer actual tool-change commands from the GCode body. + # This avoids forwarding paints that are present in metadata but never used. + used_paints_zero_based = set() + try: + for m in re.finditer(br"(?m)^[ \t]*T([0-9]+)\b", data): + used_paints_zero_based.add(int(m.group(1))) + except Exception: + used_paints_zero_based = set() + + # Fallback for slicers that only provide paint usage in header metadata. + used_paints_from_header = set() + for n in tool_filament_order: + try: + # Orca/Prusa filament: list is typically 1-based. + used_paints_from_header.add(max(0, int(n) - 1)) + except Exception: + pass + + result = [] + for i in range(total_paints): + hex_color = colors[i] if i < len(colors) else "FFFFFF" + result.append({ + "slot_index": i, + "color_hex": "#" + hex_color.upper() if hex_color else "#FFFFFF", + "material": materials[i] if i < len(materials) else "PLA", + "is_used": (i in used_paints_zero_based) if used_paints_zero_based else ((i in used_paints_from_header) if used_paints_from_header else True), + }) + return result + except Exception: + return [] diff --git a/gcode_store.py b/gcode_store.py new file mode 100644 index 0000000..ad6d7f1 --- /dev/null +++ b/gcode_store.py @@ -0,0 +1,226 @@ +""" +gcode_store.py - persistent per-bridge SQLite store for uploaded GCode files +and print-job history. + +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 json +import time +import uuid +import sqlite3 +import threading + + +class GCodeStore: + """Persistenter GCode-Store pro Bridge-Instanz (SQLite).""" + + def __init__(self, data_dir: str): + os.makedirs(data_dir, exist_ok=True) + self._gcode_dir = os.path.join(data_dir, "gcodes") + os.makedirs(self._gcode_dir, exist_ok=True) + db_path = os.path.join(data_dir, "kx-bridge.db") + self._conn = sqlite3.connect(db_path, check_same_thread=False) + self._conn.row_factory = sqlite3.Row + self._lock = threading.Lock() + self._init_schema() + + def _init_schema(self): + with self._lock: + self._conn.executescript(""" + CREATE TABLE IF NOT EXISTS gcode_files ( + id TEXT PRIMARY KEY, + filename TEXT NOT NULL, + path TEXT NOT NULL, + size_bytes INTEGER NOT NULL, + uploaded_at TEXT NOT NULL, + thumbnail_b64 TEXT, + est_print_time_sec INTEGER, + filament_used_mm REAL, + layer_count INTEGER, + gcode_filaments TEXT, + objects_skip_parts TEXT, + svg_image TEXT, + web_unverified INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS print_jobs ( + id TEXT PRIMARY KEY, + gcode_file_id TEXT NOT NULL, + printer_id TEXT NOT NULL, + started_at TEXT NOT NULL, + ended_at TEXT, + status TEXT NOT NULL, + duration_sec INTEGER, + filament_assignments TEXT, + abort_reason TEXT + ); + """) + # Migration: add gcode_filaments column for older databases + try: + self._conn.execute("ALTER TABLE gcode_files ADD COLUMN gcode_filaments TEXT") + self._conn.commit() + except Exception: + pass + # Migration: columns objects_skip_parts + svg_image (part-skip feature, v0.9.10) + # Plus layer_height / first_layer_height (Obico Z height, v0.9.18) + for col, typ in ( + ("objects_skip_parts", "TEXT"), + ("svg_image", "TEXT"), + ("layer_height", "REAL"), + ("first_layer_height", "REAL"), + ): + try: + self._conn.execute(f"ALTER TABLE gcode_files ADD COLUMN {col} {typ}") + self._conn.commit() + except Exception: + pass + # Migration: flag for web uploads (warning before print) + try: + self._conn.execute("ALTER TABLE gcode_files ADD COLUMN web_unverified INTEGER NOT NULL DEFAULT 0") + self._conn.commit() + except Exception: + pass + + def save_file(self, file_id: str, filename: str, data: bytes, + est_time_sec: int = 0, thumbnail_b64: str = "", + gcode_filaments: list | None = None, + web_unverified: bool = False, + layer_height: float = 0.0, + first_layer_height: float = 0.0) -> str: + """Saves a GCode file to disk and DB. Returns the path.""" + safe_name = os.path.basename(filename) + path = os.path.join(self._gcode_dir, safe_name) + with open(path, "wb") as f: + f.write(data) + now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + with self._lock: + filaments_json = json.dumps(gcode_filaments) if gcode_filaments else None + self._conn.execute( + """INSERT OR REPLACE INTO gcode_files + (id, filename, path, size_bytes, uploaded_at, thumbnail_b64, est_print_time_sec, gcode_filaments, web_unverified, layer_height, first_layer_height) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + (file_id, filename, path, len(data), now, thumbnail_b64 or None, est_time_sec or None, filaments_json, 1 if web_unverified else 0, layer_height or None, first_layer_height or None) + ) + self._conn.commit() + return path + + def list_files(self) -> list: + with self._lock: + rows = self._conn.execute( + "SELECT * FROM gcode_files ORDER BY uploaded_at DESC" + ).fetchall() + return [dict(r) for r in rows] + + def get_file(self, file_id: str) -> dict | None: + with self._lock: + row = self._conn.execute( + "SELECT * FROM gcode_files WHERE id=?", (file_id,) + ).fetchone() + return dict(row) if row else None + + def get_file_by_name(self, filename: str) -> dict | None: + with self._lock: + row = self._conn.execute( + "SELECT * FROM gcode_files WHERE filename=? ORDER BY uploaded_at DESC LIMIT 1", + (filename,) + ).fetchone() + return dict(row) if row else None + + def update_file_objects(self, filename: str, objects: list, svg: str = "") -> None: + """Saves the object list + optional SVG for a file (matched via filename).""" + if not filename: + return + with self._lock: + self._conn.execute( + "UPDATE gcode_files SET objects_skip_parts=?, svg_image=? " + "WHERE filename=?", + (json.dumps(objects), svg or "", filename), + ) + self._conn.commit() + + def update_file_filaments(self, file_id: str, gcode_filaments: list | None) -> None: + """Updates parsed GCode filaments for an existing DB entry.""" + with self._lock: + self._conn.execute( + "UPDATE gcode_files SET gcode_filaments=? WHERE id=?", + (json.dumps(gcode_filaments) if gcode_filaments else None, file_id), + ) + self._conn.commit() + + def clear_web_unverified(self, file_id: str) -> bool: + with self._lock: + cur = self._conn.execute( + "UPDATE gcode_files SET web_unverified=0 WHERE id=?", + (file_id,), + ) + self._conn.commit() + return cur.rowcount > 0 + + def delete_file(self, file_id: str) -> bool: + row = self.get_file(file_id) + if not row: + return False + try: + os.remove(row["path"]) + except OSError: + pass + with self._lock: + self._conn.execute("DELETE FROM gcode_files WHERE id=?", (file_id,)) + self._conn.commit() + return True + + def start_job(self, gcode_file_id: str, printer_id: str, + filament_assignments: list | None = None) -> str: + job_id = str(uuid.uuid4()) + now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + assignments_json = json.dumps(filament_assignments) if filament_assignments else None + with self._lock: + self._conn.execute( + """INSERT INTO print_jobs + (id, gcode_file_id, printer_id, started_at, status, filament_assignments) + VALUES (?,?,?,?,'printing',?)""", + (job_id, gcode_file_id, printer_id, now, assignments_json) + ) + self._conn.commit() + return job_id + + def finish_job(self, job_id: str, status: str = "completed", + abort_reason: str = "") -> None: + now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + with self._lock: + row = self._conn.execute( + "SELECT started_at FROM print_jobs WHERE id=?", (job_id,) + ).fetchone() + duration = None + if row: + try: + import calendar + start = time.strptime(row["started_at"], "%Y-%m-%dT%H:%M:%SZ") + duration = int(time.time() - calendar.timegm(start)) + except Exception: + pass + self._conn.execute( + """UPDATE print_jobs SET ended_at=?, status=?, duration_sec=?, abort_reason=? + WHERE id=?""", + (now, status, duration, abort_reason or None, job_id) + ) + self._conn.commit() + + def list_jobs(self, limit: int = 50, offset: int = 0) -> list: + with self._lock: + rows = self._conn.execute( + """SELECT j.*, f.filename, f.thumbnail_b64 + FROM print_jobs j + LEFT JOIN gcode_files f ON j.gcode_file_id = f.id + ORDER BY j.started_at DESC LIMIT ? OFFSET ?""", + (limit, offset) + ).fetchall() + return [dict(r) for r in rows] + diff --git a/kobrax_moonraker_bridge.py b/kobrax_moonraker_bridge.py index e4e8245..5f7f371 100644 --- a/kobrax_moonraker_bridge.py +++ b/kobrax_moonraker_bridge.py @@ -55,20 +55,19 @@ sys.path.insert(0, _BASE) # sys._MEIPASS entpackt; im Script-/Docker-Modus liegen sie neben dieser Datei. _WEB_BASE = getattr(sys, "_MEIPASS", _BASE) from kobrax_client import KobraXClient +# Extracted modules, re-exported here so existing imports (tests, callers) +# keep working against kobrax_moonraker_bridge unchanged. +from spoolman_client import SpoolmanClient +from gcode_store import GCodeStore +from gcode_meta import ( + _parse_gcode_estimated_time, + _parse_gcode_layer_heights, + _extract_thumbnail, + _extract_filament_info, +) +from camera import CameraCache, _find_ffmpeg -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" - try: from aiohttp import web import aiohttp @@ -140,9 +139,6 @@ log = logging.getLogger("bridge") logging.getLogger("aiohttp.access").setLevel(logging.WARNING) -def _set_verbose_http_log(enabled: bool): - logging.getLogger("aiohttp.access").setLevel(logging.INFO if enabled else logging.WARNING) - # Web UI: subdirectory under web/themes//index.html _UI_THEME_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$") # Allowed static theme files under /kx/ui/ @@ -157,39 +153,16 @@ _KX_UI_LIB_TYPES: dict[str, str] = { } _KX_UI_TRANSLATION_RE = re.compile(r"^translations/([a-z]{2}(?:-[a-z]{2})?)\.json$") -# Ring buffer for the browser log stream (last 200 entries) -import collections as _collections -_log_buffer: "_collections.deque[dict]" = _collections.deque(maxlen=500) -_log_sse_queues: "list[asyncio.Queue]" = [] - -class _BrowserLogHandler(logging.Handler): - """Sends log records to the ring buffer and all open SSE queues.""" - _fmt = logging.Formatter(datefmt="%H:%M:%S") - - def emit(self, record: logging.LogRecord): - msg = record.getMessage() - # Pass exceptions with traceback through to the browser (otherwise the - # user only sees "Error: X" without context). - if record.exc_info: - try: - msg += "\n" + self._fmt.formatException(record.exc_info) - except Exception: - pass - entry = { - "ts": self._fmt.formatTime(record, "%H:%M:%S"), - "lvl": record.levelname, - "name": record.name, - "msg": msg, - } - _log_buffer.append(entry) - for q in list(_log_sse_queues): - try: - q.put_nowait(entry) - except Exception: - pass - -_browser_handler = _BrowserLogHandler() -logging.getLogger().addHandler(_browser_handler) +# Browser log stream (ring buffer + SSE queues + handler) lives in +# bridge_logging; import the shared buffer/queues so the log-stream and +# log-download endpoints below operate on the same objects the handler writes. +from bridge_logging import ( + _set_verbose_http_log, + _log_buffer, + _log_sse_queues, + _BrowserLogHandler, + _browser_handler, +) KOBRA_TO_KLIPPER_STATE = { "free": "standby", @@ -215,781 +188,6 @@ MOONRAKER_VERSION = "v0.9.3-1" KLIPPER_VERSION = "v0.12.0-1" -def _parse_gcode_estimated_time(data: bytes) -> int: - """Reads the estimated print time from GCode (OrcaSlicer + PrusaSlicer). - Returns seconds, 0 when not found. - PrusaSlicer writes the time into the header (first 16KB), - OrcaSlicer writes it at the end of the file (last 16KB).""" - import re - # Search the beginning + end of the file (OrcaSlicer writes the time at the end) - search_text = (data[:16384] + data[-65536:]).decode("utf-8", errors="ignore") - # OrcaSlicer: ; total estimated time: 9m 20s - # PrusaSlicer: ; estimated printing time (normal mode) = 1h 9m 20s - m = (re.search(r";\s*total estimated time:\s*(.*)", search_text) or - re.search(r";\s*estimated printing time \(normal mode\)\s*=\s*(.*)", search_text)) - if not m: - return 0 - parts = re.findall(r"(\d+)\s*([hms])", m.group(1)) - secs = 0 - for val, unit in parts: - if unit == "h": secs += int(val) * 3600 - elif unit == "m": secs += int(val) * 60 - elif unit == "s": secs += int(val) - if secs: - log.info(f"Slicer estimate: {secs}s ({m.group(1).strip()})") - return secs - - -def _parse_gcode_layer_heights(data: bytes) -> tuple[float, float]: - """Reads (layer_height, initial_layer_height) from the OrcaSlicer/PrusaSlicer - GCode header. Both are stored as a config block at the end of the GCode. - - Beispiel-Zeilen: - ; layer_height = 0.2 - ; initial_layer_print_height = 0.2 - - Returns (0.0, 0.0) when not found - the caller decides what to do - (typisch: keinen Z-Wert anzeigen).""" - import re - head = data[:16384].decode("utf-8", errors="ignore") - tail = data[-65536:].decode("utf-8", errors="ignore") - search = head + "\n" + tail - def _grab(pat): - m = re.search(pat, search) - if not m: - return 0.0 - try: - return float(m.group(1)) - except Exception: - return 0.0 - layer_h = _grab(r";\s*layer_height\s*=\s*([0-9.]+)") - first_h = (_grab(r";\s*initial_layer_print_height\s*=\s*([0-9.]+)") or - _grab(r";\s*first_layer_height\s*=\s*([0-9.]+)") or - layer_h) - return layer_h, first_h - - -def _extract_thumbnail(data: bytes) -> str: - """Extracts the base64 PNG thumbnail from GCode (OrcaSlicer format).""" - try: - marker = b"; thumbnail begin" - end_marker = b"; thumbnail end" - start = data.find(marker) - if start == -1: - return "" - start = data.find(b"\n", start) + 1 - end = data.find(end_marker, start) - if end == -1: - return "" - lines = data[start:end].split(b"\n") - b64 = b"".join( - line[2:].strip() if line.startswith(b"; ") else line.strip() - for line in lines - ) - return b64.decode("ascii") - except Exception: - return "" - - -def _extract_filament_info(data: bytes) -> list[dict]: - """Reads filament colors/materials incl. tool order from Orca/Prusa GCode. - - Gibt Liste von {slot_index, color_hex, material} in Tool-/Paint-Reihenfolge - (T0, T1, ...). - Searches both the start and the end of the file since Orca can insert - large thumbnail blocks, pushing the metadata into the tail. - """ - try: - head = data[:131072] - tail = data[-131072:] if len(data) > 131072 else b"" - header = (head + b"\n" + tail).decode("utf-8", errors="ignore") - colors, materials = [], [] - paint_count_hint = 0 - tool_filament_order = [] - for line in header.splitlines(): - if re.match(r"^\s*;\s*filament_colour\s*=", line): - val = line.split("=", 1)[-1].strip() - colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()] - elif re.match(r"^\s*;\s*filament_multi_colour\s*=", line) and not colors: - val = line.split("=", 1)[-1].strip() - colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()] - elif re.match(r"^\s*;\s*filament_type\s*=", line): - val = line.split("=", 1)[-1].strip() - parts = [m.strip() for m in re.split(r"[;,]", val) if m.strip()] - materials = parts - paint_count_hint = max(paint_count_hint, len(parts)) - elif re.match(r"^\s*;\s*filament_density\s*:", line): - val = line.split(":", 1)[-1].strip() - parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()] - paint_count_hint = max(paint_count_hint, len(parts)) - elif re.match(r"^\s*;\s*filament_diameter\s*:", line): - val = line.split(":", 1)[-1].strip() - parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()] - paint_count_hint = max(paint_count_hint, len(parts)) - elif re.match(r"^\s*;\s*filament\s*:", line): - raw = line.split(":", 1)[-1] - parsed = [] - for p in [x.strip() for x in raw.split(",") if x.strip()]: - try: - parsed.append(int(p)) - except Exception: - pass - if parsed: - tool_filament_order = parsed - total_paints = max(len(colors), len(materials), paint_count_hint) - if tool_filament_order: - total_paints = max(total_paints, max(tool_filament_order)) - if total_paints <= 0: - return [] - - # Keep full paint list visible; mark paints referenced by Orca tool order as used. - if len(colors) < total_paints: - colors.extend(["FFFFFF"] * (total_paints - len(colors))) - if len(materials) < total_paints: - materials.extend(["PLA"] * (total_paints - len(materials))) - # Prefer actual tool-change commands from the GCode body. - # This avoids forwarding paints that are present in metadata but never used. - used_paints_zero_based = set() - try: - for m in re.finditer(br"(?m)^[ \t]*T([0-9]+)\b", data): - used_paints_zero_based.add(int(m.group(1))) - except Exception: - used_paints_zero_based = set() - - # Fallback for slicers that only provide paint usage in header metadata. - used_paints_from_header = set() - for n in tool_filament_order: - try: - # Orca/Prusa filament: list is typically 1-based. - used_paints_from_header.add(max(0, int(n) - 1)) - except Exception: - pass - - result = [] - for i in range(total_paints): - hex_color = colors[i] if i < len(colors) else "FFFFFF" - result.append({ - "slot_index": i, - "color_hex": "#" + hex_color.upper() if hex_color else "#FFFFFF", - "material": materials[i] if i < len(materials) else "PLA", - "is_used": (i in used_paints_zero_based) if used_paints_zero_based else ((i in used_paints_from_header) if used_paints_from_header else True), - }) - return result - except Exception: - return [] - - -class GCodeStore: - """Persistenter GCode-Store pro Bridge-Instanz (SQLite).""" - - def __init__(self, data_dir: str): - os.makedirs(data_dir, exist_ok=True) - self._gcode_dir = os.path.join(data_dir, "gcodes") - os.makedirs(self._gcode_dir, exist_ok=True) - db_path = os.path.join(data_dir, "kx-bridge.db") - self._conn = sqlite3.connect(db_path, check_same_thread=False) - self._conn.row_factory = sqlite3.Row - self._lock = threading.Lock() - self._init_schema() - - def _init_schema(self): - with self._lock: - self._conn.executescript(""" - CREATE TABLE IF NOT EXISTS gcode_files ( - id TEXT PRIMARY KEY, - filename TEXT NOT NULL, - path TEXT NOT NULL, - size_bytes INTEGER NOT NULL, - uploaded_at TEXT NOT NULL, - thumbnail_b64 TEXT, - est_print_time_sec INTEGER, - filament_used_mm REAL, - layer_count INTEGER, - gcode_filaments TEXT, - objects_skip_parts TEXT, - svg_image TEXT, - web_unverified INTEGER NOT NULL DEFAULT 0 - ); - CREATE TABLE IF NOT EXISTS print_jobs ( - id TEXT PRIMARY KEY, - gcode_file_id TEXT NOT NULL, - printer_id TEXT NOT NULL, - started_at TEXT NOT NULL, - ended_at TEXT, - status TEXT NOT NULL, - duration_sec INTEGER, - filament_assignments TEXT, - abort_reason TEXT - ); - """) - # Migration: add gcode_filaments column for older databases - try: - self._conn.execute("ALTER TABLE gcode_files ADD COLUMN gcode_filaments TEXT") - self._conn.commit() - except Exception: - pass - # Migration: columns objects_skip_parts + svg_image (part-skip feature, v0.9.10) - # Plus layer_height / first_layer_height (Obico Z height, v0.9.18) - for col, typ in ( - ("objects_skip_parts", "TEXT"), - ("svg_image", "TEXT"), - ("layer_height", "REAL"), - ("first_layer_height", "REAL"), - ): - try: - self._conn.execute(f"ALTER TABLE gcode_files ADD COLUMN {col} {typ}") - self._conn.commit() - except Exception: - pass - # Migration: flag for web uploads (warning before print) - try: - self._conn.execute("ALTER TABLE gcode_files ADD COLUMN web_unverified INTEGER NOT NULL DEFAULT 0") - self._conn.commit() - except Exception: - pass - - def save_file(self, file_id: str, filename: str, data: bytes, - est_time_sec: int = 0, thumbnail_b64: str = "", - gcode_filaments: list | None = None, - web_unverified: bool = False, - layer_height: float = 0.0, - first_layer_height: float = 0.0) -> str: - """Saves a GCode file to disk and DB. Returns the path.""" - safe_name = os.path.basename(filename) - path = os.path.join(self._gcode_dir, safe_name) - with open(path, "wb") as f: - f.write(data) - now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - with self._lock: - filaments_json = json.dumps(gcode_filaments) if gcode_filaments else None - self._conn.execute( - """INSERT OR REPLACE INTO gcode_files - (id, filename, path, size_bytes, uploaded_at, thumbnail_b64, est_print_time_sec, gcode_filaments, web_unverified, layer_height, first_layer_height) - VALUES (?,?,?,?,?,?,?,?,?,?,?)""", - (file_id, filename, path, len(data), now, thumbnail_b64 or None, est_time_sec or None, filaments_json, 1 if web_unverified else 0, layer_height or None, first_layer_height or None) - ) - self._conn.commit() - return path - - def list_files(self) -> list: - with self._lock: - rows = self._conn.execute( - "SELECT * FROM gcode_files ORDER BY uploaded_at DESC" - ).fetchall() - return [dict(r) for r in rows] - - def get_file(self, file_id: str) -> dict | None: - with self._lock: - row = self._conn.execute( - "SELECT * FROM gcode_files WHERE id=?", (file_id,) - ).fetchone() - return dict(row) if row else None - - def get_file_by_name(self, filename: str) -> dict | None: - with self._lock: - row = self._conn.execute( - "SELECT * FROM gcode_files WHERE filename=? ORDER BY uploaded_at DESC LIMIT 1", - (filename,) - ).fetchone() - return dict(row) if row else None - - def update_file_objects(self, filename: str, objects: list, svg: str = "") -> None: - """Saves the object list + optional SVG for a file (matched via filename).""" - if not filename: - return - with self._lock: - self._conn.execute( - "UPDATE gcode_files SET objects_skip_parts=?, svg_image=? " - "WHERE filename=?", - (json.dumps(objects), svg or "", filename), - ) - self._conn.commit() - - def update_file_filaments(self, file_id: str, gcode_filaments: list | None) -> None: - """Updates parsed GCode filaments for an existing DB entry.""" - with self._lock: - self._conn.execute( - "UPDATE gcode_files SET gcode_filaments=? WHERE id=?", - (json.dumps(gcode_filaments) if gcode_filaments else None, file_id), - ) - self._conn.commit() - - def clear_web_unverified(self, file_id: str) -> bool: - with self._lock: - cur = self._conn.execute( - "UPDATE gcode_files SET web_unverified=0 WHERE id=?", - (file_id,), - ) - self._conn.commit() - return cur.rowcount > 0 - - def delete_file(self, file_id: str) -> bool: - row = self.get_file(file_id) - if not row: - return False - try: - os.remove(row["path"]) - except OSError: - pass - with self._lock: - self._conn.execute("DELETE FROM gcode_files WHERE id=?", (file_id,)) - self._conn.commit() - return True - - def start_job(self, gcode_file_id: str, printer_id: str, - filament_assignments: list | None = None) -> str: - job_id = str(uuid.uuid4()) - now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - assignments_json = json.dumps(filament_assignments) if filament_assignments else None - with self._lock: - self._conn.execute( - """INSERT INTO print_jobs - (id, gcode_file_id, printer_id, started_at, status, filament_assignments) - VALUES (?,?,?,?,'printing',?)""", - (job_id, gcode_file_id, printer_id, now, assignments_json) - ) - self._conn.commit() - return job_id - - def finish_job(self, job_id: str, status: str = "completed", - abort_reason: str = "") -> None: - now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - with self._lock: - row = self._conn.execute( - "SELECT started_at FROM print_jobs WHERE id=?", (job_id,) - ).fetchone() - duration = None - if row: - try: - import calendar - start = time.strptime(row["started_at"], "%Y-%m-%dT%H:%M:%SZ") - duration = int(time.time() - calendar.timegm(start)) - except Exception: - pass - self._conn.execute( - """UPDATE print_jobs SET ended_at=?, status=?, duration_sec=?, abort_reason=? - WHERE id=?""", - (now, status, duration, abort_reason or None, job_id) - ) - self._conn.commit() - - def list_jobs(self, limit: int = 50, offset: int = 0) -> list: - with self._lock: - rows = self._conn.execute( - """SELECT j.*, f.filename, f.thumbnail_b64 - FROM print_jobs j - LEFT JOIN gcode_files f ON j.gcode_file_id = f.id - ORDER BY j.started_at DESC LIMIT ? OFFSET ?""", - (limit, offset) - ).fetchall() - return [dict(r) for r in rows] - - -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) - - -class SpoolmanClient: - """Thin synchronous HTTP client for Spoolman filament tracking. - - Designed to be called from daemon threads (poll loop, _on_print callbacks). - Uses requests (already in requirements) so no event-loop dependency. - """ - - def __init__(self, server_url: str, sync_rate: int = 0): - self.server_url = server_url.rstrip("/") - self.sync_rate = sync_rate - - def _req(self, method: str, path: str, **kwargs): - import requests - r = requests.request(method, f"{self.server_url}{path}", timeout=5, **kwargs) - r.raise_for_status() - return r.json() - - def health_check(self) -> bool: - try: - self._req("GET", "/api/v1/health") - return True - except Exception: - return False - - def list_spools(self) -> list: - return self._req("GET", "/api/v1/spool") - - def use_filament(self, spool_id: int, use_length_mm: float) -> None: - """Report consumed filament length in mm. Spoolman converts to weight - using the spool's filament profile density.""" - self._req("PUT", f"/api/v1/spool/{spool_id}/use", - json={"use_length": round(use_length_mm, 2)}) - - class KobraXBridge: def __init__(self, client: KobraXClient, args=None, store=None, printer_id: str = "1", all_bridges=None): self.client = client diff --git a/spoolman_client.py b/spoolman_client.py new file mode 100644 index 0000000..b4e1fb0 --- /dev/null +++ b/spoolman_client.py @@ -0,0 +1,45 @@ +""" +spoolman_client.py - thin synchronous HTTP client for Spoolman filament tracking. + +Extracted from kobrax_moonraker_bridge.py as part of splitting that module up; +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. +""" + + +class SpoolmanClient: + """Thin synchronous HTTP client for Spoolman filament tracking. + + Designed to be called from daemon threads (poll loop, _on_print callbacks). + Uses requests (already in requirements) so no event-loop dependency. + """ + + def __init__(self, server_url: str, sync_rate: int = 0): + self.server_url = server_url.rstrip("/") + self.sync_rate = sync_rate + + def _req(self, method: str, path: str, **kwargs): + import requests + r = requests.request(method, f"{self.server_url}{path}", timeout=5, **kwargs) + r.raise_for_status() + return r.json() + + def health_check(self) -> bool: + try: + self._req("GET", "/api/v1/health") + return True + except Exception: + return False + + def list_spools(self) -> list: + return self._req("GET", "/api/v1/spool") + + def use_filament(self, spool_id: int, use_length_mm: float) -> None: + """Report consumed filament length in mm. Spoolman converts to weight + using the spool's filament profile density.""" + self._req("PUT", f"/api/v1/spool/{spool_id}/use", + json={"use_length": round(use_length_mm, 2)})