refactor(bridge): extract self-contained classes/helpers into modules (stage 1)
First stage of splitting the 6368-line kobrax_moonraker_bridge.py monolith.
The low-coupling, self-contained pieces move into their own modules;
kobrax_moonraker_bridge.py re-exports them so every caller (12 test files,
the PyInstaller spec) keeps working unchanged - not a single test or the
spec needed editing.
Extracted:
- spoolman_client.py <- SpoolmanClient (zero coupling)
- gcode_store.py <- GCodeStore (stdlib only)
- gcode_meta.py <- _parse_gcode_*/_extract_* metadata helpers
- camera.py <- CameraCache + _find_ffmpeg
- bridge_logging.py <- _BrowserLogHandler, the log ring buffer + SSE
queues, _set_verbose_http_log (the shared mutable
buffer/queues are re-imported so the log endpoints
still operate on the same objects the handler writes)
Facade shrinks from 6368 to 5566 lines. All 184 tests green after each
extraction. Verified the re-exported names resolve and the shared log
objects are identical by reference across modules. No behavior change.
This commit is contained in:
60
bridge_logging.py
Normal file
60
bridge_logging.py
Normal file
@@ -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)
|
||||
409
camera.py
Normal file
409
camera.py
Normal file
@@ -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)
|
||||
|
||||
182
gcode_meta.py
Normal file
182
gcode_meta.py
Normal file
@@ -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 []
|
||||
226
gcode_store.py
Normal file
226
gcode_store.py
Normal file
@@ -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]
|
||||
|
||||
@@ -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/<name>/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/<name>
|
||||
@@ -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
|
||||
|
||||
45
spoolman_client.py
Normal file
45
spoolman_client.py
Normal file
@@ -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)})
|
||||
Reference in New Issue
Block a user