fix(camera): reuse the live view's frame for external-camera captures (#2707)

On a printer with an external camera, watching the live view while a print
ran meant the layer timelapse recorded almost nothing and the finish photo
went out with no image. The reporter measured 0 of 87 layer captures on one
print and 0 of 105 on another, both watched throughout. A USB camera allows
one V4L2 handle, so a capture during a live view fails outright.

The built-in camera has had this rule since #1348 and #1271: reuse the
viewer's buffered frame rather than opening a second connection. It was
never extended to the external paths, and it could not have been -- the
buffer it depends on was only ever populated by the built-in paths.
generate_mjpeg_stream yields multipart-wrapped chunks, so the route layer
could not recover the JPEG, and a guarded caller would have found an empty
buffer and skipped every time.

So the stream now publishes each raw frame through a new on_frame callback
(parallel to on_process from #2675), and the six one-shot consumers reuse
it: layer timelapse, the finish-photo moment and its background fallback,
the notification snapshot, Obico polling, and the plate check. A viewer
attached with nothing buffered yet skips that one attempt rather than
competing -- kicking the viewer off is worse than missing a frame.

on_frame exceptions are logged and swallowed, like iter_subscriber's
on_unsubscribe: buffering is a side effect and must never be able to take
the live stream down with it. The external stream's teardown now releases
the buffered frame too, ownership-checked so a concurrent viewer of the
same printer keeps its own.

Two side effects on paths not touched here, both improvements: the snapshot
endpoint and the finish-photo fallback chain consult get_buffered_frame and
can now serve an external camera's live frame. plate_detection's docstring
already claimed this behaviour while implementing it only for the built-in
fallback; that drift is resolved.
This commit is contained in:
maziggy
2026-07-30 13:05:15 +02:00
parent 40e7b60e8c
commit d9da60dd8d
8 changed files with 457 additions and 28 deletions
+41
View File
@@ -276,6 +276,29 @@ def _new_fanout_stream_id(printer_id: int) -> str:
return f"{printer_id}-fanout-{uuid.uuid4().hex[:8]}"
def live_frame_for_capture(printer_id: int) -> tuple[bool, bytes | None]:
"""Should a one-shot capture stand down for the live view, and to what frame?
Returns ``(defer, frame)``. ``defer`` True means DO NOT open a capture of
your own: use ``frame`` when it isn't None, and otherwise skip this attempt
rather than competing.
Both camera kinds allow exactly one reader — Bambu firmware permits one
connection, and a USB camera permits one V4L2 handle — so a capture that
races the live view doesn't degrade, it fails outright. #2707 measured 0 of
87 and 0 of 105 layer-timelapse captures on prints watched throughout, and
finish photos going out with no image attached.
Skipping when the buffer is momentarily empty (stream starting, mid-
reconnect) rather than falling through to a capture is the #1348 rule:
opening a competing handle kicks the viewer off, which is a worse outcome
than missing one frame.
"""
if not is_stream_active(printer_id):
return False, None
return True, _last_frames.get(printer_id)
def _release_printer_frame_state(printer_id: int | None) -> None:
"""Drop a printer's buffered frame and timings — unless a stream still owns them.
@@ -926,6 +949,18 @@ async def camera_stream(
_spawned_ffmpeg_pids[proc.pid] = time.time()
_stream_last_frame_times[stream_id] = time.time()
def _publish_external_frame(frame: bytes) -> None:
"""Make the live frame reusable by one-shot consumers (#2707).
Only the built-in camera paths populated _last_frames, so every
external-camera consumer — layer timelapse, finish photo, Obico,
plate check — found an empty buffer and opened its own handle on a
device that allows exactly one reader, which simply failed while a
viewer was attached. Raw frame, not the multipart-wrapped chunk the
generator yields, because that is what those consumers expect.
"""
_last_frames[printer_id] = frame
async def external_stream_wrapper():
"""Wrap external stream to track start/stop and update frame times."""
try:
@@ -934,6 +969,7 @@ async def camera_stream(
printer.external_camera_type,
fps,
on_process=_register_external_process,
on_frame=_publish_external_frame,
stop_event=stop_event,
):
# generate_mjpeg_stream already handles rate limiting;
@@ -954,6 +990,11 @@ async def camera_stream(
_disconnect_events.pop(stream_id, None)
_stream_last_frame_times.pop(stream_id, None)
_active_external_streams.discard(printer_id)
# Now that this path publishes a buffered frame, it has to
# retract it too — ownership-checked, so a concurrent viewer of
# the same printer keeps its own. Also clears the per-printer
# timings this path used to leave behind.
_release_printer_frame_state(printer_id)
logger.info("External camera stream ended for printer %s", printer_id)
return StreamingResponse(
+39 -15
View File
@@ -2172,13 +2172,21 @@ async def _capture_snapshot_for_notification(printer_id: int, printer, logger) -
# Try external camera first
if printer.external_camera_enabled and printer.external_camera_url:
logger.info("[SNAPSHOT] Capturing from external camera for printer %s", printer_id)
from backend.app.api.routes.camera import live_frame_for_capture
from backend.app.services.external_camera import capture_frame
frame_data = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
# An external camera allows one reader, so capturing while a viewer
# is attached fails (#2707). A None here falls through to the paths
# below exactly as a failed capture did.
defer, buffered = live_frame_for_capture(printer_id)
if defer:
frame_data = buffered
else:
frame_data = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
if frame_data and len(frame_data) <= 2_500_000:
logger.info("[SNAPSHOT] External camera frame: %s bytes", len(frame_data))
return _apply_camera_rotation(frame_data, printer, logger)
@@ -4337,13 +4345,21 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
)
if frame_bytes is None and printer.external_camera_enabled and printer.external_camera_url:
from backend.app.api.routes.camera import live_frame_for_capture
from backend.app.services.external_camera import capture_frame
frame_bytes = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
# #2707: this used to collide with the live view and fail, which is
# how finish-photo notifications went out with no image attached.
# Leaving frame_bytes None keeps the rest of the fallback chain.
defer, buffered = live_frame_for_capture(printer_id)
if defer:
frame_bytes = buffered
else:
frame_bytes = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
if frame_bytes:
logger.info(
"[FINISH-PHOTO-MOMENT] captured external-camera frame (%d bytes)",
@@ -5307,13 +5323,21 @@ async def on_print_complete(printer_id: int, data: dict):
if not photo_filename:
if printer.external_camera_enabled and printer.external_camera_url:
logger.info("[PHOTO-BG] Using external camera")
from backend.app.api.routes.camera import live_frame_for_capture
from backend.app.services.external_camera import capture_frame
frame_data = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
# #2707: the second half of the finish-photo failure — the
# pre-capture and this fallback both collided with the live
# view. None here continues down the fallback chain.
defer, buffered = live_frame_for_capture(printer_id)
if defer:
frame_data = buffered
else:
frame_data = await capture_frame(
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
if frame_data:
photos_dir = archive_dir / "photos"
photos_dir.mkdir(parents=True, exist_ok=True)
+24 -4
View File
@@ -607,6 +607,7 @@ async def generate_mjpeg_stream(
fps: int = 10,
*,
on_process: Callable[[asyncio.subprocess.Process], None] | None = None,
on_frame: Callable[[bytes], None] | None = None,
stop_event: asyncio.Event | None = None,
) -> AsyncGenerator[bytes, None]:
"""Generator yielding MJPEG frames for streaming.
@@ -622,6 +623,16 @@ async def generate_mjpeg_stream(
open (#2675). Without it the process is reachable only from this
generator's own ``finally``, which an abrupt client disconnect can
skip (same cancellation-timing class as #776).
on_frame: Called with each RAW frame, before it is wrapped for the wire,
so the route layer can publish it as the printer's buffered frame
(#2707). It has to be a callback: what this generator yields is
multipart-wrapped, so a consumer of the stream cannot recover the
JPEG, and until now nothing populated the buffer for external
cameras at all — leaving every one-shot consumer (layer timelapse,
finish photo, Obico, plate check) with nothing to reuse and no
option but to open a competing handle on a single-reader device.
Exceptions are logged and swallowed: buffering must never be able
to break the live stream.
stop_event: When set, the reconnect loops stop retrying — so an explicit
stop (which kills the current ffmpeg) doesn't immediately respawn a
new process and reacquire the device.
@@ -632,6 +643,15 @@ async def generate_mjpeg_stream(
frame_interval = 1.0 / max(fps, 1)
last_frame_time = 0.0
def _publish(frame: bytes) -> bytes:
"""Hand the raw frame to on_frame, then format it for the wire."""
if on_frame is not None:
try:
on_frame(frame)
except Exception:
logger.exception("on_frame callback raised")
return _format_mjpeg_frame(frame)
if camera_type == "mjpeg":
# Proxy MJPEG stream directly, with reconnect on timeout
max_retries = 3
@@ -642,7 +662,7 @@ async def generate_mjpeg_stream(
current_time = asyncio.get_event_loop().time()
if current_time - last_frame_time >= frame_interval:
last_frame_time = current_time
yield _format_mjpeg_frame(frame)
yield _publish(frame)
if not frame_yielded or attempt == max_retries or (stop_event is not None and stop_event.is_set()):
break
logger.warning(
@@ -659,7 +679,7 @@ async def generate_mjpeg_stream(
frame_yielded = False
async for frame in _stream_rtsp(url, fps, on_process=on_process):
frame_yielded = True
yield _format_mjpeg_frame(frame)
yield _publish(frame)
if not frame_yielded or attempt == max_retries or (stop_event is not None and stop_event.is_set()):
break
logger.warning(
@@ -672,7 +692,7 @@ async def generate_mjpeg_stream(
elif camera_type == "usb":
# Use ffmpeg to stream from USB camera
async for frame in _stream_usb(url, fps, on_process=on_process):
yield _format_mjpeg_frame(frame)
yield _publish(frame)
elif camera_type == "snapshot":
# Poll snapshot URL at interval
@@ -680,7 +700,7 @@ async def generate_mjpeg_stream(
try:
frame = await _capture_snapshot(url, timeout=10)
if frame:
yield _format_mjpeg_frame(frame)
yield _publish(frame)
await asyncio.sleep(frame_interval)
except asyncio.CancelledError:
break
+20 -1
View File
@@ -67,7 +67,26 @@ class TimelapseSession:
self.last_layer = layer_num
try:
frame_data = await capture_frame(self.camera_url, self.camera_type, snapshot_url=self.snapshot_url)
# Reuse the live view's frame instead of opening a second handle on
# a single-reader device (#2707). Unguarded, a print watched from
# start to finish recorded zero successful layer captures, and the
# stitched video came out empty or badly truncated.
from backend.app.api.routes.camera import live_frame_for_capture
defer, buffered = live_frame_for_capture(self.printer_id)
if defer:
if not buffered:
# Viewer attached but nothing buffered yet: skip this layer
# rather than compete and kick them off (#1348).
logger.debug(
"Skipping layer %s for printer %s: viewer attached, no buffered frame yet",
layer_num,
self.printer_id,
)
return False
frame_data = buffered
else:
frame_data = await capture_frame(self.camera_url, self.camera_type, snapshot_url=self.snapshot_url)
if frame_data:
frame_path = self.frames_dir / f"layer_{layer_num:05d}.jpg"
await asyncio.to_thread(frame_path.write_bytes, frame_data)
+15
View File
@@ -193,6 +193,21 @@ class ObicoDetectionService:
return None
if printer.external_camera_enabled and printer.external_camera_url:
# Same rule as the built-in branch below, which this used to skip:
# an external camera is single-reader too, so polling while a viewer
# is attached just fails (#2707).
from backend.app.api.routes.camera import live_frame_for_capture
defer, buffered = live_frame_for_capture(printer_id)
if defer:
if buffered:
return buffered
logger.info(
"Obico: viewer attached for printer %s but buffer empty; "
"skipping this poll to avoid competing camera handle (#2707)",
printer_id,
)
return None
return await capture_external_frame(
printer.external_camera_url,
printer.external_camera_type,
+20 -8
View File
@@ -604,16 +604,28 @@ async def capture_camera_image(
# Try external camera first if requested and available
if use_external and external_camera_url and external_camera_type:
try:
from backend.app.api.routes.camera import live_frame_for_capture
from backend.app.services.external_camera import capture_frame
image_data = await capture_frame(
external_camera_url,
external_camera_type,
snapshot_url=external_camera_snapshot_url,
)
if image_data:
camera_source = "external"
logger.debug("Captured frame from external camera for printer %s", printer_id)
# What this function's docstring already promised, but only the
# built-in fallback below delivered: an external camera is
# single-reader too, so capturing while a viewer watches fails
# (#2707).
defer, buffered = live_frame_for_capture(printer_id)
if defer:
if buffered:
image_data = buffered
camera_source = "external (buffered)"
logger.debug("Using buffered external frame for printer %s", printer_id)
else:
image_data = await capture_frame(
external_camera_url,
external_camera_type,
snapshot_url=external_camera_snapshot_url,
)
if image_data:
camera_source = "external"
logger.debug("Captured frame from external camera for printer %s", printer_id)
except Exception as e:
logger.warning("Failed to capture from external camera: %s", e)
@@ -0,0 +1,297 @@
"""External-camera captures must reuse the live view's frame (#2707).
A USB camera allows exactly one V4L2 handle, so a one-shot capture taken while
somebody is watching the live view doesn't degrade — it fails. The reporter
measured 0 of 87 and 0 of 105 layer-timelapse captures on prints watched from
start to finish, and finish-photo notifications going out with no image.
The guards for the built-in camera (#1348, #1271) were never extended to the
external paths, and the deeper reason they couldn't be: ``_last_frames`` was
only ever populated by the built-in paths. ``generate_mjpeg_stream`` yields
multipart-wrapped chunks, so the route layer had no way to recover the JPEG —
hence the ``on_frame`` callback, and hence a guard alone would have found an
empty buffer and skipped every time.
These tests cover the plumbing (raw frames reach the callback) and each consumer
that used to compete: layer timelapse, Obico polling, and plate detection.
"""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.api.routes import camera
from backend.app.services import external_camera, layer_timelapse
from backend.app.services.obico_detection import ObicoDetectionService
pytestmark = pytest.mark.asyncio
LIVE_FRAME = b"\xff\xd8live-viewer-frame\xff\xd9"
FRESH_FRAME = b"\xff\xd8fresh-capture\xff\xd9"
PRINTER_ID = 9310
@pytest.fixture(autouse=True)
def _clean_registries():
def _purge():
for sid in [k for k in camera._active_streams if k.startswith(f"{PRINTER_ID}-")]:
camera._active_streams.pop(sid, None)
camera._last_frames.pop(PRINTER_ID, None)
camera._last_frame_times.pop(PRINTER_ID, None)
camera._stream_start_times.pop(PRINTER_ID, None)
_purge()
yield
_purge()
def _attach_viewer(frame: bytes | None = LIVE_FRAME) -> None:
"""Register a live external stream, as the stream route does."""
camera._active_streams[f"{PRINTER_ID}-ext-deadbeef"] = object()
if frame is not None:
camera._last_frames[PRINTER_ID] = frame
# ---------------------------------------------------------------------------
# live_frame_for_capture — the shared decision
# ---------------------------------------------------------------------------
async def test_no_viewer_means_capture_normally():
defer, frame = camera.live_frame_for_capture(PRINTER_ID)
assert defer is False
assert frame is None
async def test_viewer_with_a_buffered_frame_is_reused():
_attach_viewer()
defer, frame = camera.live_frame_for_capture(PRINTER_ID)
assert defer is True
assert frame == LIVE_FRAME
async def test_viewer_with_an_empty_buffer_means_skip_not_capture():
"""#1348: competing for the device is worse than missing one frame."""
_attach_viewer(frame=None)
defer, frame = camera.live_frame_for_capture(PRINTER_ID)
assert defer is True
assert frame is None
# ---------------------------------------------------------------------------
# on_frame plumbing — without this the buffer is always empty
# ---------------------------------------------------------------------------
async def test_on_frame_receives_the_raw_jpeg_not_the_multipart_chunk():
"""The consumers want a JPEG; the stream yields multipart. Hence a callback."""
captured: list[bytes] = []
async def _fake_usb(_url, _fps, on_process=None):
yield FRESH_FRAME
with patch.object(external_camera, "_stream_usb", _fake_usb):
chunks = [
chunk
async for chunk in external_camera.generate_mjpeg_stream(
"/dev/video0", "usb", fps=15, on_frame=captured.append
)
]
assert captured == [FRESH_FRAME], "callback did not get the raw frame"
assert b"--frame" in chunks[0], "wire format should still be multipart"
assert b"--frame" not in captured[0]
async def test_a_raising_on_frame_callback_cannot_break_the_stream():
"""Buffering is a side effect; it must never take the live view down."""
async def _fake_usb(_url, _fps, on_process=None):
yield FRESH_FRAME
yield FRESH_FRAME
def _boom(_frame: bytes) -> None:
raise RuntimeError("buffering blew up")
with patch.object(external_camera, "_stream_usb", _fake_usb):
chunks = [
chunk async for chunk in external_camera.generate_mjpeg_stream("/dev/video0", "usb", fps=15, on_frame=_boom)
]
assert len(chunks) == 2, "stream stopped because the callback raised"
# ---------------------------------------------------------------------------
# Layer timelapse — the 0-of-87 case
# ---------------------------------------------------------------------------
def _session(tmp_path) -> layer_timelapse.TimelapseSession:
with patch.object(layer_timelapse.settings, "base_dir", tmp_path):
return layer_timelapse.TimelapseSession(
printer_id=PRINTER_ID,
archive_id=None,
camera_url="/dev/video0",
camera_type="usb",
)
async def test_timelapse_uses_the_live_frame_instead_of_competing(tmp_path):
session = _session(tmp_path)
_attach_viewer()
with patch.object(layer_timelapse, "capture_frame", new=AsyncMock(return_value=FRESH_FRAME)) as mock_capture:
captured = await session.capture_layer(1)
assert captured is True, "layer capture failed with a viewer attached"
# Would have opened a competing handle on a single-reader device.
mock_capture.assert_not_called()
written = sorted(session.frames_dir.glob("layer_*.jpg"))
assert len(written) == 1
assert written[0].read_bytes() == LIVE_FRAME
async def test_timelapse_skips_a_layer_rather_than_competing_on_an_empty_buffer(tmp_path):
session = _session(tmp_path)
_attach_viewer(frame=None)
with patch.object(layer_timelapse, "capture_frame", new=AsyncMock(return_value=FRESH_FRAME)) as mock_capture:
captured = await session.capture_layer(1)
assert captured is False
mock_capture.assert_not_called()
assert sorted(session.frames_dir.glob("layer_*.jpg")) == []
async def test_timelapse_captures_normally_with_no_viewer(tmp_path):
"""The unwatched path must be untouched — this is the common case."""
session = _session(tmp_path)
with patch.object(layer_timelapse, "capture_frame", new=AsyncMock(return_value=FRESH_FRAME)) as mock_capture:
captured = await session.capture_layer(1)
assert captured is True
mock_capture.assert_awaited_once()
written = sorted(session.frames_dir.glob("layer_*.jpg"))
assert written[0].read_bytes() == FRESH_FRAME
# ---------------------------------------------------------------------------
# Obico polling — external branch, mirroring the built-in one
# ---------------------------------------------------------------------------
def _external_printer() -> MagicMock:
return MagicMock(
external_camera_enabled=True,
external_camera_url="/dev/video0",
external_camera_type="usb",
external_camera_snapshot_url=None,
ip_address="192.168.1.10",
access_code="12345678",
model="A1",
)
def _db_returning(printer) -> MagicMock:
session = MagicMock()
session.get = AsyncMock(return_value=printer)
ctx = MagicMock()
ctx.__aenter__ = AsyncMock(return_value=session)
ctx.__aexit__ = AsyncMock(return_value=None)
return ctx
async def test_obico_reuses_the_live_external_frame():
_attach_viewer()
svc = ObicoDetectionService()
with (
patch(
"backend.app.services.obico_detection.async_session",
return_value=_db_returning(_external_printer()),
),
patch(
"backend.app.services.external_camera.capture_frame",
new=AsyncMock(return_value=FRESH_FRAME),
) as mock_capture,
):
result = await svc._capture_frame(printer_id=PRINTER_ID)
assert result == LIVE_FRAME
mock_capture.assert_not_called()
async def test_obico_skips_the_poll_when_the_external_buffer_is_empty():
_attach_viewer(frame=None)
svc = ObicoDetectionService()
with (
patch(
"backend.app.services.obico_detection.async_session",
return_value=_db_returning(_external_printer()),
),
patch(
"backend.app.services.external_camera.capture_frame",
new=AsyncMock(return_value=FRESH_FRAME),
) as mock_capture,
):
result = await svc._capture_frame(printer_id=PRINTER_ID)
assert result is None
mock_capture.assert_not_called()
async def test_obico_still_captures_when_nobody_is_watching():
svc = ObicoDetectionService()
with (
patch(
"backend.app.services.obico_detection.async_session",
return_value=_db_returning(_external_printer()),
),
patch(
"backend.app.services.external_camera.capture_frame",
new=AsyncMock(return_value=FRESH_FRAME),
) as mock_capture,
):
result = await svc._capture_frame(printer_id=PRINTER_ID)
assert result == FRESH_FRAME
mock_capture.assert_awaited_once()
# ---------------------------------------------------------------------------
# Plate detection — its docstring already promised this
# ---------------------------------------------------------------------------
async def test_plate_detection_reuses_the_live_external_frame():
from backend.app.services import plate_detection
_attach_viewer()
with patch(
"backend.app.services.external_camera.capture_frame",
new=AsyncMock(return_value=FRESH_FRAME),
) as mock_capture:
image, source = await plate_detection.capture_camera_image(
printer_id=PRINTER_ID,
ip_address="192.168.1.10",
access_code="12345678",
model="A1",
external_camera_url="/dev/video0",
external_camera_type="usb",
use_external=True,
)
assert image == LIVE_FRAME
assert source == "external (buffered)"
mock_capture.assert_not_called()