diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7baf44509..a4ef35abc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,12 @@ All notable changes to Bambuddy will be documented in this file.
## [0.1.6] - 2026-01-24
### New Features
+- **External Network Camera Support** - Add external cameras (MJPEG, RTSP, HTTP snapshot) to replace built-in printer cameras (Issue #143):
+ - Configure per-printer external camera URL and type in Settings → Camera
+ - Live streaming uses external camera when enabled
+ - Finish photo capture uses external camera
+ - Layer-based timelapse: captures frame on each layer change, stitches to MP4 on print completion
+ - Test connection button to verify camera accessibility
- **Recalculate Costs Button** - New button on Dashboard to recalculate all archive costs using current filament prices (Issue #120)
- **Create Folder from ZIP** - New option in File Manager upload to automatically create a folder named after the ZIP file (Issue #121)
- **Multi-File Selection in Printer Files** - Printer card file browser now supports multiple file selection (Issue #144):
diff --git a/README.md b/README.md
index 193aab0bc..642c1c79a 100644
--- a/README.md
+++ b/README.md
@@ -57,6 +57,7 @@
### 📊 Monitoring & Control
- Real-time printer status via WebSocket
- Live camera streaming (MJPEG) & snapshots with multi-viewer support
+- External network camera support (MJPEG, RTSP, HTTP snapshot) with layer-based timelapse
- Fan status monitoring (part cooling, auxiliary, chamber)
- Printer control (stop, pause, resume, chamber light)
- Resizable printer cards (S/M/L/XL)
diff --git a/backend/app/api/routes/camera.py b/backend/app/api/routes/camera.py
index 70de2c24f..67aa413f4 100644
--- a/backend/app/api/routes/camera.py
+++ b/backend/app/api/routes/camera.py
@@ -39,6 +39,9 @@ _last_frame_times: dict[int, float] = {}
# Track stream start times for each printer
_stream_start_times: dict[int, float] = {}
+# Track active external camera streams by printer ID
+_active_external_streams: set[int] = set()
+
def get_buffered_frame(printer_id: int) -> bytes | None:
"""Get the last buffered frame for a printer from an active stream.
@@ -350,7 +353,8 @@ async def camera_stream(
This endpoint returns a multipart MJPEG stream that can be used directly
in an tag or video player.
- Uses the appropriate protocol based on printer model:
+ Uses external camera if configured, otherwise uses built-in camera:
+ - External: MJPEG, RTSP, or HTTP snapshot
- A1/P1: Chamber image protocol (port 6000)
- X1/H2/P2: RTSP via ffmpeg (port 322)
@@ -362,6 +366,50 @@ async def camera_stream(
printer = await get_printer_or_404(printer_id, db)
+ # Check for external camera first
+ if printer.external_camera_enabled and printer.external_camera_url:
+ import time
+
+ from backend.app.services.external_camera import generate_mjpeg_stream
+
+ # Limit external camera FPS to reduce browser load
+ fps = min(max(fps, 1), 15)
+ logger.info(f"Using external camera ({printer.external_camera_type}) for printer {printer_id} at {fps} fps")
+
+ # Track stream start
+ _stream_start_times[printer_id] = time.time()
+ _active_external_streams.add(printer_id)
+
+ async def external_stream_wrapper():
+ """Wrap external stream to track start/stop and update frame times."""
+ frame_interval = 1.0 / fps
+ last_yield_time = 0.0
+ try:
+ async for frame in generate_mjpeg_stream(
+ printer.external_camera_url, printer.external_camera_type, fps
+ ):
+ # Rate limit to prevent overwhelming browser
+ current_time = time.time()
+ elapsed = current_time - last_yield_time
+ if elapsed < frame_interval:
+ await asyncio.sleep(frame_interval - elapsed)
+ last_yield_time = time.time()
+ _last_frame_times[printer_id] = last_yield_time
+ yield frame
+ finally:
+ _active_external_streams.discard(printer_id)
+ logger.info(f"External camera stream ended for printer {printer_id}")
+
+ return StreamingResponse(
+ external_stream_wrapper(),
+ media_type="multipart/x-mixed-replace; boundary=frame",
+ headers={
+ "Cache-Control": "no-cache, no-store, must-revalidate",
+ "Pragma": "no-cache",
+ "Expires": "0",
+ },
+ )
+
# Validate FPS - A1/P1 models max out at ~5 FPS
if is_chamber_image_model(printer.model):
fps = min(max(fps, 1), 5)
@@ -554,13 +602,18 @@ async def camera_status(printer_id: int):
# Check if there's an active stream for this printer
has_active_stream = False
+ # Check external camera streams
+ if printer_id in _active_external_streams:
+ has_active_stream = True
+
# Check ffmpeg/RTSP streams
- for stream_id in _active_streams:
- if stream_id.startswith(f"{printer_id}-"):
- process = _active_streams[stream_id]
- if process.returncode is None:
- has_active_stream = True
- break
+ if not has_active_stream:
+ for stream_id in _active_streams:
+ if stream_id.startswith(f"{printer_id}-"):
+ process = _active_streams[stream_id]
+ if process.returncode is None:
+ has_active_stream = True
+ break
# Check chamber image streams
if not has_active_stream:
@@ -597,3 +650,28 @@ async def camera_status(printer_id: int):
and (seconds_since_frame is None or seconds_since_frame > 10)
),
}
+
+
+@router.post("/{printer_id}/camera/external/test")
+async def test_external_camera(
+ printer_id: int,
+ url: str,
+ camera_type: str,
+ db: AsyncSession = Depends(get_db),
+):
+ """Test external camera connection.
+
+ Args:
+ printer_id: Printer ID (for authorization)
+ url: Camera URL to test
+ camera_type: Camera type ("mjpeg", "rtsp", "snapshot")
+
+ Returns:
+ Dict with {success: bool, error?: str, resolution?: str}
+ """
+ # Verify printer exists (for authorization)
+ await get_printer_or_404(printer_id, db)
+
+ from backend.app.services.external_camera import test_connection
+
+ return await test_connection(url, camera_type)
diff --git a/backend/app/api/routes/settings.py b/backend/app/api/routes/settings.py
index 82991e022..7512f0fca 100644
--- a/backend/app/api/routes/settings.py
+++ b/backend/app/api/routes/settings.py
@@ -404,6 +404,9 @@ async def export_backup(
"auto_archive": printer.auto_archive,
"print_hours_offset": printer.print_hours_offset,
"runtime_seconds": printer.runtime_seconds,
+ "external_camera_url": printer.external_camera_url,
+ "external_camera_type": printer.external_camera_type,
+ "external_camera_enabled": printer.external_camera_enabled,
}
if include_access_codes:
printer_data["access_code"] = printer.access_code
@@ -984,6 +987,9 @@ async def import_backup(
auto_archive=printer_data.get("auto_archive", True),
print_hours_offset=printer_data.get("print_hours_offset", 0.0),
runtime_seconds=printer_data.get("runtime_seconds", 0),
+ external_camera_url=printer_data.get("external_camera_url"),
+ external_camera_type=printer_data.get("external_camera_type"),
+ external_camera_enabled=printer_data.get("external_camera_enabled", False),
)
db.add(printer)
restored["printers"] += 1
diff --git a/backend/app/core/database.py b/backend/app/core/database.py
index 3308d7dbd..8a3e5c926 100644
--- a/backend/app/core/database.py
+++ b/backend/app/core/database.py
@@ -675,6 +675,20 @@ async def run_migrations(conn):
except Exception:
pass
+ # Migration: Add external camera columns to printers
+ try:
+ await conn.execute(text("ALTER TABLE printers ADD COLUMN external_camera_url VARCHAR(500)"))
+ except Exception:
+ pass
+ try:
+ await conn.execute(text("ALTER TABLE printers ADD COLUMN external_camera_type VARCHAR(20)"))
+ except Exception:
+ pass
+ try:
+ await conn.execute(text("ALTER TABLE printers ADD COLUMN external_camera_enabled BOOLEAN DEFAULT 0"))
+ except Exception:
+ pass
+
async def seed_notification_templates():
"""Seed default notification templates if they don't exist."""
diff --git a/backend/app/main.py b/backend/app/main.py
index 599fb3838..7d1cee863 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -798,6 +798,18 @@ async def on_print_start(printer_id: int, data: dict):
logger.info(f"Created fallback archive {fallback_archive.id} for {print_name} (no 3MF available)")
+ # Start timelapse session if external camera is enabled
+ if printer.external_camera_enabled and printer.external_camera_url:
+ from backend.app.services.layer_timelapse import start_session
+
+ start_session(
+ printer_id,
+ fallback_archive.id,
+ printer.external_camera_url,
+ printer.external_camera_type or "mjpeg",
+ )
+ logger.info(f"Started layer timelapse for printer {printer_id}, archive {fallback_archive.id}")
+
# Track as active print
_active_prints[(printer_id, fallback_archive.filename)] = fallback_archive.id
if filename:
@@ -872,6 +884,18 @@ async def on_print_start(printer_id: int, data: dict):
logger.info(f"Created archive {archive.id} for {downloaded_filename}")
+ # Start timelapse session if external camera is enabled
+ if printer.external_camera_enabled and printer.external_camera_url:
+ from backend.app.services.layer_timelapse import start_session
+
+ start_session(
+ printer_id,
+ archive.id,
+ printer.external_camera_url,
+ printer.external_camera_type or "mjpeg",
+ )
+ logger.info(f"Started layer timelapse for printer {printer_id}, archive {archive.id}")
+
# Record starting energy from smart plug if available
try:
plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
@@ -1365,35 +1389,52 @@ async def on_print_complete(printer_id: int, data: dict):
archive_dir = app_settings.base_dir / Path(archive.file_path).parent
photo_filename = None
- # Check if camera stream is active - use buffered frame to avoid freeze
- # Check both RTSP streams (_active_streams) and chamber image streams (_active_chamber_streams)
- active_for_printer = [k for k in _active_streams if k.startswith(f"{printer_id}-")]
- active_chamber_for_printer = [
- k for k in _active_chamber_streams if k.startswith(f"{printer_id}-")
- ]
- buffered_frame = get_buffered_frame(printer_id)
+ # Check for external camera first
+ if printer.external_camera_enabled and printer.external_camera_url:
+ logger.info("[PHOTO-BG] Using external camera")
+ from backend.app.services.external_camera import capture_frame
- if (active_for_printer or active_chamber_for_printer) and buffered_frame:
- # Use frame from active stream
- logger.info("[PHOTO-BG] Using buffered frame from active stream")
- photos_dir = archive_dir / "photos"
- photos_dir.mkdir(parents=True, exist_ok=True)
- timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
- photo_filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg"
- photo_path = photos_dir / photo_filename
- await asyncio.to_thread(photo_path.write_bytes, buffered_frame)
- logger.info(f"[PHOTO-BG] Saved buffered frame: {photo_filename}")
- else:
- # No active stream - capture new frame
- from backend.app.services.camera import capture_finish_photo
-
- photo_filename = await capture_finish_photo(
- printer_id=printer_id,
- ip_address=printer.ip_address,
- access_code=printer.access_code,
- model=printer.model,
- archive_dir=archive_dir,
+ frame_data = await capture_frame(
+ printer.external_camera_url, printer.external_camera_type or "mjpeg"
)
+ if frame_data:
+ photos_dir = archive_dir / "photos"
+ photos_dir.mkdir(parents=True, exist_ok=True)
+ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+ photo_filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg"
+ photo_path = photos_dir / photo_filename
+ await asyncio.to_thread(photo_path.write_bytes, frame_data)
+ logger.info(f"[PHOTO-BG] Saved external camera frame: {photo_filename}")
+ else:
+ # Check if camera stream is active - use buffered frame to avoid freeze
+ # Check both RTSP streams (_active_streams) and chamber image streams (_active_chamber_streams)
+ active_for_printer = [k for k in _active_streams if k.startswith(f"{printer_id}-")]
+ active_chamber_for_printer = [
+ k for k in _active_chamber_streams if k.startswith(f"{printer_id}-")
+ ]
+ buffered_frame = get_buffered_frame(printer_id)
+
+ if (active_for_printer or active_chamber_for_printer) and buffered_frame:
+ # Use frame from active stream
+ logger.info("[PHOTO-BG] Using buffered frame from active stream")
+ photos_dir = archive_dir / "photos"
+ photos_dir.mkdir(parents=True, exist_ok=True)
+ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+ photo_filename = f"finish_{timestamp}_{uuid.uuid4().hex[:8]}.jpg"
+ photo_path = photos_dir / photo_filename
+ await asyncio.to_thread(photo_path.write_bytes, buffered_frame)
+ logger.info(f"[PHOTO-BG] Saved buffered frame: {photo_filename}")
+ else:
+ # No active stream - capture new frame
+ from backend.app.services.camera import capture_finish_photo
+
+ photo_filename = await capture_finish_photo(
+ printer_id=printer_id,
+ ip_address=printer.ip_address,
+ access_code=printer.access_code,
+ model=printer.model,
+ archive_dir=archive_dir,
+ )
if photo_filename:
photos = archive.photos or []
@@ -1529,6 +1570,43 @@ async def on_print_complete(printer_id: int, data: dict):
await _background_notifications(None)
asyncio.create_task(_photo_then_notify())
+
+ # Stitch external camera layer timelapse if session was active
+ print_status = data.get("status", "completed")
+
+ async def _background_layer_timelapse():
+ """Stitch layer timelapse and attach to archive."""
+ from backend.app.services.layer_timelapse import cancel_session, on_print_complete as tl_complete
+
+ try:
+ if print_status == "completed":
+ logger.info(f"[LAYER-TL] Stitching layer timelapse for printer {printer_id}")
+ timelapse_path = await tl_complete(printer_id)
+ if timelapse_path and archive_id:
+ logger.info(f"[LAYER-TL] Attaching timelapse {timelapse_path} to archive {archive_id}")
+ async with async_session() as db:
+ service = ArchiveService(db)
+ timelapse_data = await asyncio.to_thread(timelapse_path.read_bytes)
+ await service.attach_timelapse(archive_id, timelapse_data, "layer_timelapse.mp4")
+ # Clean up the temp file
+ await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
+ logger.info("[LAYER-TL] Layer timelapse attached successfully")
+ elif timelapse_path:
+ # Timelapse created but no archive - just clean up
+ await asyncio.to_thread(timelapse_path.unlink, missing_ok=True)
+ else:
+ # Print failed or cancelled - cancel timelapse session
+ cancel_session(printer_id)
+ logger.info(f"[LAYER-TL] Cancelled layer timelapse for printer {printer_id} (status: {print_status})")
+ except Exception as e:
+ logger.warning(f"[LAYER-TL] Failed: {e}")
+ # Try to cancel session on error
+ try:
+ cancel_session(printer_id)
+ except Exception:
+ pass
+
+ asyncio.create_task(_background_layer_timelapse())
log_timing("All background tasks scheduled")
# Auto-scan for timelapse if recording was active during the print
@@ -1928,6 +2006,15 @@ async def lifespan(app: FastAPI):
printer_manager.set_print_complete_callback(on_print_complete)
printer_manager.set_ams_change_callback(on_ams_change)
+ # Layer change callback for external camera timelapse
+ async def on_layer_change(printer_id: int, layer_num: int):
+ """Capture timelapse frame on layer change."""
+ from backend.app.services.layer_timelapse import on_layer_change as tl_layer_change
+
+ await tl_layer_change(printer_id, layer_num)
+
+ printer_manager.set_layer_change_callback(on_layer_change)
+
# Initialize MQTT relay from settings
async with async_session() as db:
from backend.app.api.routes.settings import get_setting
diff --git a/backend/app/models/printer.py b/backend/app/models/printer.py
index 1bbc3a354..f1105d3c7 100644
--- a/backend/app/models/printer.py
+++ b/backend/app/models/printer.py
@@ -24,6 +24,10 @@ class Printer(Base):
last_runtime_update: Mapped[datetime | None] = mapped_column(
DateTime, nullable=True
) # Last time runtime was updated
+ # External camera configuration
+ external_camera_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
+ external_camera_type: Mapped[str | None] = mapped_column(String(20), nullable=True) # mjpeg, rtsp, snapshot
+ external_camera_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
diff --git a/backend/app/schemas/printer.py b/backend/app/schemas/printer.py
index aade85181..5daf6a508 100644
--- a/backend/app/schemas/printer.py
+++ b/backend/app/schemas/printer.py
@@ -11,6 +11,9 @@ class PrinterBase(BaseModel):
model: str | None = None
location: str | None = None # Group/location name
auto_archive: bool = True
+ external_camera_url: str | None = None
+ external_camera_type: str | None = None # "mjpeg", "rtsp", "snapshot"
+ external_camera_enabled: bool = False
class PrinterCreate(PrinterBase):
@@ -26,6 +29,9 @@ class PrinterUpdate(BaseModel):
is_active: bool | None = None
auto_archive: bool | None = None
print_hours_offset: float | None = None
+ external_camera_url: str | None = None
+ external_camera_type: str | None = None
+ external_camera_enabled: bool | None = None
class PrinterResponse(PrinterBase):
@@ -33,6 +39,9 @@ class PrinterResponse(PrinterBase):
is_active: bool
nozzle_count: int = 1 # 1 or 2, auto-detected from MQTT
print_hours_offset: float = 0.0
+ external_camera_url: str | None = None
+ external_camera_type: str | None = None
+ external_camera_enabled: bool = False
created_at: datetime
updated_at: datetime
diff --git a/backend/app/services/bambu_mqtt.py b/backend/app/services/bambu_mqtt.py
index 713db02ef..49728df7c 100644
--- a/backend/app/services/bambu_mqtt.py
+++ b/backend/app/services/bambu_mqtt.py
@@ -253,6 +253,7 @@ class BambuMQTTClient:
on_print_start: Callable[[dict], None] | None = None,
on_print_complete: Callable[[dict], None] | None = None,
on_ams_change: Callable[[list], None] | None = None,
+ on_layer_change: Callable[[int], None] | None = None,
):
self.ip_address = ip_address
self.serial_number = serial_number
@@ -261,6 +262,7 @@ class BambuMQTTClient:
self.on_print_start = on_print_start
self.on_print_complete = on_print_complete
self.on_ams_change = on_ams_change
+ self.on_layer_change = on_layer_change
self.state = PrinterState()
self._client: mqtt.Client | None = None
@@ -1030,7 +1032,12 @@ class BambuMQTTClient:
)
self.state.mc_print_sub_stage = new_sub_stage
if "layer_num" in data:
- self.state.layer_num = int(data["layer_num"])
+ new_layer = int(data["layer_num"])
+ old_layer = self.state.layer_num
+ self.state.layer_num = new_layer
+ # Trigger layer change callback if layer increased
+ if new_layer > old_layer and self.on_layer_change:
+ self.on_layer_change(new_layer)
if "total_layer_num" in data:
self.state.total_layers = int(data["total_layer_num"])
@@ -1736,6 +1743,8 @@ class BambuMQTTClient:
if is_new_print or is_file_change:
# Clear any old HMS errors when a new print starts
self.state.hms_errors = []
+ # Reset layer tracking for new print (needed for layer-based timelapse)
+ self.state.layer_num = 0
# Reset completion tracking for new print
self._was_running = True
self._completion_triggered = False
diff --git a/backend/app/services/external_camera.py b/backend/app/services/external_camera.py
new file mode 100644
index 000000000..2d0656491
--- /dev/null
+++ b/backend/app/services/external_camera.py
@@ -0,0 +1,409 @@
+"""External network camera service.
+
+Supports MJPEG streams, RTSP streams (via ffmpeg), and HTTP snapshot URLs.
+"""
+
+import asyncio
+import logging
+import shutil
+from collections.abc import AsyncGenerator
+from pathlib import Path
+
+import aiohttp
+
+logger = logging.getLogger(__name__)
+
+
+def get_ffmpeg_path() -> str | None:
+ """Get the path to ffmpeg executable."""
+ # Try shutil.which first
+ path = shutil.which("ffmpeg")
+ if path:
+ return path
+ # Check common locations (systemd services may have limited PATH)
+ for common_path in ["/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg", "/opt/homebrew/bin/ffmpeg"]:
+ if Path(common_path).exists():
+ return common_path
+ return None
+
+
+async def capture_frame(url: str, camera_type: str, timeout: int = 15) -> bytes | None:
+ """Capture single frame from external camera.
+
+ Args:
+ url: Camera URL (MJPEG stream, RTSP URL, or HTTP snapshot URL)
+ camera_type: "mjpeg", "rtsp", or "snapshot"
+ timeout: Connection timeout in seconds
+
+ Returns:
+ JPEG bytes or None on failure
+ """
+ logger.debug(f"capture_frame called: type={camera_type}, url={url[:50] if url else 'None'}...")
+ if camera_type == "mjpeg":
+ return await _capture_mjpeg_frame(url, timeout)
+ elif camera_type == "rtsp":
+ return await _capture_rtsp_frame(url, timeout)
+ elif camera_type == "snapshot":
+ return await _capture_snapshot(url, timeout)
+ else:
+ logger.warning(f"Unknown camera type: {camera_type}")
+ return None
+
+
+async def _capture_mjpeg_frame(url: str, timeout: int) -> bytes | None:
+ """Extract single frame from MJPEG stream."""
+ try:
+ async with (
+ aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session,
+ session.get(url) as response,
+ ):
+ if response.status != 200:
+ logger.error(f"MJPEG stream returned status {response.status}")
+ return None
+
+ # Read chunks until we find a complete JPEG frame
+ buffer = b""
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ async for chunk in response.content.iter_chunked(8192):
+ buffer += chunk
+
+ # Look for complete JPEG frame
+ start_idx = buffer.find(jpeg_start)
+ if start_idx == -1:
+ continue
+
+ end_idx = buffer.find(jpeg_end, start_idx + 2)
+ if end_idx != -1:
+ # Found complete frame
+ frame = buffer[start_idx : end_idx + 2]
+ return frame
+
+ # Keep searching, but limit buffer size
+ if len(buffer) > 5 * 1024 * 1024: # 5MB limit
+ logger.warning("MJPEG buffer exceeded 5MB without finding frame")
+ return None
+
+ except TimeoutError:
+ logger.warning(f"MJPEG frame capture timed out after {timeout}s")
+ return None
+ except Exception as e:
+ logger.error(f"MJPEG frame capture failed: {e}")
+ return None
+
+ return None
+
+
+async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None:
+ """Capture frame from RTSP using ffmpeg."""
+ ffmpeg = get_ffmpeg_path()
+ if not ffmpeg:
+ logger.error("ffmpeg not found - required for RTSP capture")
+ return None
+
+ # Use ffmpeg to grab a single frame from RTSP stream
+ # ffmpeg handles both rtsp:// and rtsps:// URLs automatically
+ cmd = [
+ ffmpeg,
+ "-rtsp_transport",
+ "tcp",
+ "-i",
+ url,
+ "-frames:v",
+ "1",
+ "-f",
+ "image2pipe",
+ "-vcodec",
+ "mjpeg",
+ "-q:v",
+ "2",
+ "-",
+ ]
+
+ try:
+ print(f"[EXT-CAM] Running ffmpeg command: {' '.join(cmd[:6])}...")
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ )
+
+ stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+ print(
+ f"[EXT-CAM] ffmpeg returned: code={process.returncode}, stdout={len(stdout)} bytes, stderr={len(stderr)} bytes"
+ )
+
+ if process.returncode != 0:
+ logger.error(f"ffmpeg RTSP capture failed: {stderr.decode()[:200]}")
+ print(f"[EXT-CAM] ffmpeg error: {stderr.decode()[:300]}")
+ return None
+
+ if not stdout or len(stdout) < 100:
+ logger.error("ffmpeg returned empty or too small frame")
+ return None
+
+ return stdout
+
+ except TimeoutError:
+ logger.warning(f"RTSP frame capture timed out after {timeout}s")
+ if process:
+ process.kill()
+ return None
+ except Exception as e:
+ logger.error(f"RTSP frame capture failed: {e}")
+ return None
+
+
+async def _capture_snapshot(url: str, timeout: int) -> bytes | None:
+ """Fetch snapshot from HTTP URL."""
+ try:
+ async with (
+ aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session,
+ session.get(url) as response,
+ ):
+ if response.status != 200:
+ logger.error(f"Snapshot URL returned status {response.status}")
+ return None
+
+ data = await response.read()
+
+ # Validate it looks like JPEG
+ if not data.startswith(b"\xff\xd8"):
+ logger.warning("Snapshot does not appear to be JPEG")
+ # Still return it - might be valid with different header
+
+ return data
+
+ except TimeoutError:
+ logger.warning(f"Snapshot capture timed out after {timeout}s")
+ return None
+ except Exception as e:
+ logger.error(f"Snapshot capture failed: {e}")
+ return None
+
+
+async def test_connection(url: str, camera_type: str) -> dict:
+ """Test camera connection.
+
+ Returns:
+ Dict with {success: bool, error?: str, resolution?: str}
+ """
+ print(f"[EXT-CAM] Testing camera connection: type={camera_type}, url={url[:50]}...")
+ logger.info(f"Testing camera connection: type={camera_type}, url={url[:50]}...")
+ try:
+ frame = await capture_frame(url, camera_type, timeout=10)
+ print(f"[EXT-CAM] Capture result: {len(frame) if frame else 0} bytes")
+ logger.info(f"Capture result: {len(frame) if frame else 0} bytes")
+
+ if frame:
+ # Try to get resolution from JPEG header
+ resolution = None
+ try:
+ # Simple JPEG dimension extraction
+ # SOF0 marker is FF C0, followed by length, precision, height, width
+ sof_markers = [b"\xff\xc0", b"\xff\xc1", b"\xff\xc2"]
+ for marker in sof_markers:
+ idx = frame.find(marker)
+ if idx != -1 and idx + 9 <= len(frame):
+ height = (frame[idx + 5] << 8) | frame[idx + 6]
+ width = (frame[idx + 7] << 8) | frame[idx + 8]
+ resolution = f"{width}x{height}"
+ break
+ except Exception:
+ pass
+
+ return {"success": True, "resolution": resolution}
+ else:
+ return {"success": False, "error": "Failed to capture frame from camera"}
+
+ except Exception as e:
+ return {"success": False, "error": str(e)}
+
+
+async def generate_mjpeg_stream(url: str, camera_type: str, fps: int = 10) -> AsyncGenerator[bytes, None]:
+ """Generator yielding MJPEG frames for streaming.
+
+ Args:
+ url: Camera URL
+ camera_type: "mjpeg", "rtsp", or "snapshot"
+ fps: Target frames per second
+
+ Yields:
+ MJPEG frame data with HTTP multipart boundaries
+ """
+ frame_interval = 1.0 / max(fps, 1)
+ last_frame_time = 0.0
+
+ if camera_type == "mjpeg":
+ # Proxy MJPEG stream directly
+ async for frame in _stream_mjpeg(url):
+ 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)
+
+ elif camera_type == "rtsp":
+ # Use ffmpeg to convert RTSP to MJPEG
+ async for frame in _stream_rtsp(url, fps):
+ yield _format_mjpeg_frame(frame)
+
+ elif camera_type == "snapshot":
+ # Poll snapshot URL at interval
+ while True:
+ try:
+ frame = await _capture_snapshot(url, timeout=10)
+ if frame:
+ yield _format_mjpeg_frame(frame)
+ await asyncio.sleep(frame_interval)
+ except asyncio.CancelledError:
+ break
+ except Exception as e:
+ logger.warning(f"Snapshot poll failed: {e}")
+ await asyncio.sleep(frame_interval)
+
+
+def _format_mjpeg_frame(frame: bytes) -> bytes:
+ """Format frame for MJPEG HTTP response."""
+ return (
+ b"--frame\r\n"
+ b"Content-Type: image/jpeg\r\n"
+ b"Content-Length: " + str(len(frame)).encode() + b"\r\n"
+ b"\r\n" + frame + b"\r\n"
+ )
+
+
+async def _stream_mjpeg(url: str) -> AsyncGenerator[bytes, None]:
+ """Stream frames from MJPEG URL."""
+ try:
+ timeout = aiohttp.ClientTimeout(total=None, sock_read=30)
+ async with aiohttp.ClientSession(timeout=timeout) as session, session.get(url) as response:
+ if response.status != 200:
+ logger.error(f"MJPEG stream returned status {response.status}")
+ return
+
+ buffer = b""
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ async for chunk in response.content.iter_chunked(8192):
+ buffer += chunk
+
+ # Extract complete frames from buffer
+ while True:
+ start_idx = buffer.find(jpeg_start)
+ if start_idx == -1:
+ buffer = buffer[-2:] if len(buffer) > 2 else buffer
+ break
+
+ if start_idx > 0:
+ buffer = buffer[start_idx:]
+
+ end_idx = buffer.find(jpeg_end, 2)
+ if end_idx == -1:
+ break
+
+ frame = buffer[: end_idx + 2]
+ buffer = buffer[end_idx + 2 :]
+ yield frame
+
+ except asyncio.CancelledError:
+ logger.info("MJPEG stream cancelled")
+ except Exception as e:
+ logger.error(f"MJPEG stream error: {e}")
+
+
+async def _stream_rtsp(url: str, fps: int) -> AsyncGenerator[bytes, None]:
+ """Stream frames from RTSP URL via ffmpeg."""
+ ffmpeg = get_ffmpeg_path()
+ if not ffmpeg:
+ logger.error("ffmpeg not found - required for RTSP streaming")
+ return
+
+ # ffmpeg handles both rtsp:// and rtsps:// URLs automatically
+ cmd = [
+ ffmpeg,
+ "-rtsp_transport",
+ "tcp",
+ "-rtsp_flags",
+ "prefer_tcp",
+ "-timeout",
+ "30000000",
+ "-buffer_size",
+ "1024000",
+ "-max_delay",
+ "500000",
+ "-i",
+ url,
+ "-f",
+ "mjpeg",
+ "-q:v",
+ "5",
+ "-r",
+ str(fps),
+ "-an",
+ "-",
+ ]
+
+ process = None
+ try:
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ )
+
+ # Give ffmpeg a moment to start and check for immediate failures
+ await asyncio.sleep(0.5)
+ if process.returncode is not None:
+ stderr = await process.stderr.read()
+ logger.error(f"ffmpeg RTSP stream failed immediately: {stderr.decode()[:300]}")
+ return
+
+ buffer = b""
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ while True:
+ try:
+ chunk = await asyncio.wait_for(process.stdout.read(8192), timeout=30.0)
+
+ if not chunk:
+ break
+
+ buffer += chunk
+
+ # Extract complete frames
+ while True:
+ start_idx = buffer.find(jpeg_start)
+ if start_idx == -1:
+ buffer = buffer[-2:] if len(buffer) > 2 else buffer
+ break
+
+ if start_idx > 0:
+ buffer = buffer[start_idx:]
+
+ end_idx = buffer.find(jpeg_end, 2)
+ if end_idx == -1:
+ break
+
+ frame = buffer[: end_idx + 2]
+ buffer = buffer[end_idx + 2 :]
+ yield frame
+
+ except TimeoutError:
+ logger.warning("RTSP stream read timeout")
+ break
+
+ except asyncio.CancelledError:
+ logger.info("RTSP stream cancelled")
+ except Exception as e:
+ logger.error(f"RTSP stream error: {e}")
+ finally:
+ if process and process.returncode is None:
+ process.terminate()
+ try:
+ await asyncio.wait_for(process.wait(), timeout=2.0)
+ except TimeoutError:
+ process.kill()
+ await process.wait()
diff --git a/backend/app/services/layer_timelapse.py b/backend/app/services/layer_timelapse.py
new file mode 100644
index 000000000..6ddec9b0d
--- /dev/null
+++ b/backend/app/services/layer_timelapse.py
@@ -0,0 +1,274 @@
+"""Layer-based timelapse for external cameras.
+
+Captures a frame on each layer change and stitches them into a video on print completion.
+"""
+
+import asyncio
+import logging
+import shutil
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+
+from backend.app.core.config import settings
+from backend.app.services.external_camera import capture_frame
+
+logger = logging.getLogger(__name__)
+
+# Active timelapse sessions: {printer_id: TimelapseSession}
+_active_sessions: dict[int, "TimelapseSession"] = {}
+
+
+def get_ffmpeg_path() -> str | None:
+ """Get the path to ffmpeg executable."""
+ # Try shutil.which first
+ path = shutil.which("ffmpeg")
+ if path:
+ return path
+ # Check common locations (systemd services may have limited PATH)
+ for common_path in ["/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg", "/opt/homebrew/bin/ffmpeg"]:
+ if Path(common_path).exists():
+ return common_path
+ return None
+
+
+@dataclass
+class TimelapseSession:
+ """Active timelapse recording session."""
+
+ printer_id: int
+ archive_id: int | None
+ camera_url: str
+ camera_type: str
+ last_layer: int = -1
+ frame_count: int = 0
+ session_id: str = field(default_factory=lambda: datetime.now().strftime("%Y%m%d_%H%M%S"))
+ frames_dir: Path = field(init=False)
+
+ def __post_init__(self):
+ self.frames_dir = settings.base_dir / "timelapse_frames" / str(self.printer_id) / self.session_id
+ self.frames_dir.mkdir(parents=True, exist_ok=True)
+ logger.info(f"Created timelapse session {self.session_id} for printer {self.printer_id}")
+
+ async def capture_layer(self, layer_num: int) -> bool:
+ """Capture frame if layer changed.
+
+ Args:
+ layer_num: Current layer number from printer
+
+ Returns:
+ True if frame was captured, False otherwise
+ """
+ # Only capture if layer increased
+ if layer_num <= self.last_layer:
+ return False
+
+ self.last_layer = layer_num
+
+ try:
+ frame_data = await capture_frame(self.camera_url, self.camera_type)
+ if frame_data:
+ frame_path = self.frames_dir / f"layer_{layer_num:05d}.jpg"
+ await asyncio.to_thread(frame_path.write_bytes, frame_data)
+ self.frame_count += 1
+ logger.debug(f"Captured layer {layer_num} for printer {self.printer_id} (frame {self.frame_count})")
+ return True
+ else:
+ logger.warning(f"Failed to capture frame for layer {layer_num}")
+ return False
+ except Exception as e:
+ logger.error(f"Error capturing timelapse frame: {e}")
+ return False
+
+ async def stitch(self, output_path: Path, fps: int = 30) -> bool:
+ """Create MP4 from captured frames using ffmpeg.
+
+ Args:
+ output_path: Path for output video file
+ fps: Frames per second for output video
+
+ Returns:
+ True if stitching succeeded, False otherwise
+ """
+ if self.frame_count == 0:
+ logger.warning("No frames to stitch")
+ return False
+
+ ffmpeg = get_ffmpeg_path()
+ if not ffmpeg:
+ logger.error("ffmpeg not found - required for timelapse stitching")
+ return False
+
+ # Find all frame files and create a sequential list
+ # This handles gaps in layer numbers (e.g., if some captures failed)
+ frame_files = sorted(self.frames_dir.glob("layer_*.jpg"))
+ if not frame_files:
+ logger.warning("No frame files found in timelapse directory")
+ return False
+
+ # Create a concat file listing all frames
+ concat_file = self.frames_dir / "frames.txt"
+ try:
+ with open(concat_file, "w") as f:
+ for frame in frame_files:
+ # Each frame shown for 1/fps duration
+ f.write(f"file '{frame.name}'\n")
+ f.write(f"duration {1.0 / fps}\n")
+ # Add last frame again (required by concat demuxer)
+ if frame_files:
+ f.write(f"file '{frame_files[-1].name}'\n")
+ except Exception as e:
+ logger.error(f"Failed to create concat file: {e}")
+ return False
+
+ # Use ffmpeg concat demuxer for variable-gap frame sequences
+ cmd = [
+ ffmpeg,
+ "-y", # Overwrite output
+ "-f",
+ "concat",
+ "-safe",
+ "0",
+ "-i",
+ str(concat_file),
+ "-c:v",
+ "libx264",
+ "-pix_fmt",
+ "yuv420p",
+ "-preset",
+ "medium",
+ "-crf",
+ "23",
+ str(output_path),
+ ]
+
+ try:
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ cwd=str(self.frames_dir), # Run in frames dir so relative paths work
+ )
+
+ stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300)
+
+ if process.returncode != 0:
+ logger.error(f"ffmpeg timelapse stitch failed: {stderr.decode()[:500]}")
+ return False
+
+ logger.info(f"Created timelapse video: {output_path} ({self.frame_count} frames)")
+ return True
+
+ except TimeoutError:
+ logger.error("Timelapse stitching timed out")
+ if process:
+ process.kill()
+ return False
+ except Exception as e:
+ logger.error(f"Timelapse stitch failed: {e}")
+ return False
+
+ def cleanup(self):
+ """Remove temporary frames directory."""
+ try:
+ if self.frames_dir.exists():
+ shutil.rmtree(self.frames_dir, ignore_errors=True)
+ logger.info(f"Cleaned up timelapse frames for session {self.session_id}")
+ except Exception as e:
+ logger.warning(f"Failed to cleanup timelapse frames: {e}")
+
+
+def start_session(printer_id: int, archive_id: int | None, url: str, cam_type: str) -> TimelapseSession:
+ """Start new timelapse session for a printer.
+
+ Args:
+ printer_id: The printer ID
+ archive_id: Associated print archive ID (optional)
+ url: External camera URL
+ cam_type: Camera type ("mjpeg", "rtsp", "snapshot")
+
+ Returns:
+ The new TimelapseSession
+ """
+ # Cancel any existing session
+ cancel_session(printer_id)
+
+ session = TimelapseSession(
+ printer_id=printer_id,
+ archive_id=archive_id,
+ camera_url=url,
+ camera_type=cam_type,
+ )
+ _active_sessions[printer_id] = session
+ logger.info(f"Started timelapse session for printer {printer_id}")
+ return session
+
+
+def get_session(printer_id: int) -> TimelapseSession | None:
+ """Get active timelapse session for a printer."""
+ return _active_sessions.get(printer_id)
+
+
+async def on_layer_change(printer_id: int, layer_num: int):
+ """Called on layer change - captures frame if session active.
+
+ Args:
+ printer_id: The printer ID
+ layer_num: Current layer number
+ """
+ session = get_session(printer_id)
+ if session:
+ await session.capture_layer(layer_num)
+
+
+async def on_print_complete(printer_id: int) -> Path | None:
+ """Stitch timelapse and return path. Cleans up session.
+
+ Args:
+ printer_id: The printer ID
+
+ Returns:
+ Path to stitched video, or None if no session or stitching failed
+ """
+ session = _active_sessions.pop(printer_id, None)
+ if not session:
+ return None
+
+ if session.frame_count == 0:
+ logger.info(f"No timelapse frames captured for printer {printer_id}")
+ session.cleanup()
+ return None
+
+ # Create output path in parent of frames dir
+ output_path = session.frames_dir.parent / f"timelapse_{session.session_id}.mp4"
+
+ try:
+ success = await session.stitch(output_path)
+ if success:
+ # Cleanup frames after successful stitch
+ session.cleanup()
+ return output_path
+ else:
+ session.cleanup()
+ return None
+ except Exception as e:
+ logger.error(f"Timelapse completion failed: {e}")
+ session.cleanup()
+ return None
+
+
+def cancel_session(printer_id: int):
+ """Cancel and cleanup timelapse session (on print fail/cancel).
+
+ Args:
+ printer_id: The printer ID
+ """
+ session = _active_sessions.pop(printer_id, None)
+ if session:
+ session.cleanup()
+ logger.info(f"Cancelled timelapse session for printer {printer_id}")
+
+
+def get_active_sessions() -> dict[int, TimelapseSession]:
+ """Get all active timelapse sessions."""
+ return _active_sessions.copy()
diff --git a/backend/app/services/printer_manager.py b/backend/app/services/printer_manager.py
index 4a956649e..ace0e8106 100644
--- a/backend/app/services/printer_manager.py
+++ b/backend/app/services/printer_manager.py
@@ -66,6 +66,7 @@ class PrinterManager:
self._on_print_complete: Callable[[int, dict], None] | None = None
self._on_status_change: Callable[[int, PrinterState], None] | None = None
self._on_ams_change: Callable[[int, list], None] | None = None
+ self._on_layer_change: Callable[[int, int], None] | None = None
self._loop: asyncio.AbstractEventLoop | None = None
def get_printer(self, printer_id: int) -> PrinterInfo | None:
@@ -92,6 +93,10 @@ class PrinterManager:
"""Set callback for AMS data change events."""
self._on_ams_change = callback
+ def set_layer_change_callback(self, callback: Callable[[int, int], None]):
+ """Set callback for layer change events. Receives (printer_id, layer_num)."""
+ self._on_layer_change = callback
+
def _schedule_async(self, coro):
"""Schedule an async coroutine from a sync context.
@@ -135,6 +140,10 @@ class PrinterManager:
if self._on_ams_change:
self._schedule_async(self._on_ams_change(printer_id, ams_data))
+ def on_layer_change(layer_num: int):
+ if self._on_layer_change:
+ self._schedule_async(self._on_layer_change(printer_id, layer_num))
+
client = BambuMQTTClient(
ip_address=printer.ip_address,
serial_number=printer.serial_number,
@@ -143,6 +152,7 @@ class PrinterManager:
on_print_start=on_print_start,
on_print_complete=on_print_complete,
on_ams_change=on_ams_change,
+ on_layer_change=on_layer_change,
)
client.connect()
diff --git a/backend/tests/unit/services/test_external_camera.py b/backend/tests/unit/services/test_external_camera.py
new file mode 100644
index 000000000..b04118605
--- /dev/null
+++ b/backend/tests/unit/services/test_external_camera.py
@@ -0,0 +1,217 @@
+"""
+Tests for the external camera service.
+
+These tests cover pure functions and frame parsing logic.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+class TestFormatMjpegFrame:
+ """Tests for MJPEG frame formatting."""
+
+ def test_format_mjpeg_frame_basic(self):
+ """Verify MJPEG frame is formatted correctly with boundary and headers."""
+ from backend.app.services.external_camera import _format_mjpeg_frame
+
+ # Minimal JPEG data (just SOI and EOI markers)
+ jpeg_data = b"\xff\xd8\xff\xd9"
+
+ result = _format_mjpeg_frame(jpeg_data)
+
+ # Check boundary
+ assert result.startswith(b"--frame\r\n")
+ # Check content type
+ assert b"Content-Type: image/jpeg\r\n" in result
+ # Check content length
+ assert b"Content-Length: 4\r\n" in result
+ # Check frame data is included
+ assert jpeg_data in result
+ # Check ends with CRLF
+ assert result.endswith(b"\r\n")
+
+ def test_format_mjpeg_frame_larger_data(self):
+ """Verify content length is correct for larger frames."""
+ from backend.app.services.external_camera import _format_mjpeg_frame
+
+ # Simulate a larger JPEG (1000 bytes)
+ jpeg_data = b"\xff\xd8" + b"\x00" * 996 + b"\xff\xd9"
+
+ result = _format_mjpeg_frame(jpeg_data)
+
+ assert b"Content-Length: 1000\r\n" in result
+
+
+class TestGetFfmpegPath:
+ """Tests for ffmpeg path detection."""
+
+ def test_get_ffmpeg_path_from_shutil_which(self):
+ """Verify ffmpeg found via shutil.which is returned."""
+ from backend.app.services.external_camera import get_ffmpeg_path
+
+ with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
+ result = get_ffmpeg_path()
+ assert result == "/usr/bin/ffmpeg"
+
+ def test_get_ffmpeg_path_fallback_to_common_paths(self):
+ """Verify common paths are checked when shutil.which fails."""
+ from backend.app.services.external_camera import get_ffmpeg_path
+
+ with patch("shutil.which", return_value=None), patch("pathlib.Path.exists") as mock_exists:
+ # First common path exists
+ mock_exists.return_value = True
+ result = get_ffmpeg_path()
+ assert result in ["/usr/bin/ffmpeg", "/usr/local/bin/ffmpeg", "/opt/homebrew/bin/ffmpeg"]
+
+ def test_get_ffmpeg_path_returns_none_when_not_found(self):
+ """Verify None is returned when ffmpeg not found anywhere."""
+ from backend.app.services.external_camera import get_ffmpeg_path
+
+ with patch("shutil.which", return_value=None), patch("pathlib.Path.exists", return_value=False):
+ result = get_ffmpeg_path()
+ assert result is None
+
+
+class TestJpegFrameExtraction:
+ """Tests for JPEG frame extraction from buffer."""
+
+ def test_extract_single_frame_from_buffer(self):
+ """Test extracting a complete JPEG frame from buffer."""
+ # JPEG markers
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ # Create a buffer with one complete frame
+ frame_content = b"\x00" * 100
+ buffer = jpeg_start + frame_content + jpeg_end
+
+ # Find frame boundaries
+ start_idx = buffer.find(jpeg_start)
+ end_idx = buffer.find(jpeg_end, start_idx + 2)
+
+ assert start_idx == 0
+ assert end_idx == 102
+
+ # Extract frame
+ frame = buffer[start_idx : end_idx + 2]
+ assert frame == buffer
+ assert len(frame) == 104
+
+ def test_extract_frame_with_leading_garbage(self):
+ """Test extracting frame when buffer has leading garbage data."""
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ # Buffer with garbage before the JPEG
+ garbage = b"\x00\x01\x02\x03"
+ frame_content = b"\xff" * 50
+ buffer = garbage + jpeg_start + frame_content + jpeg_end
+
+ start_idx = buffer.find(jpeg_start)
+ assert start_idx == 4 # After garbage
+
+ end_idx = buffer.find(jpeg_end, start_idx + 2)
+ frame = buffer[start_idx : end_idx + 2]
+
+ assert frame.startswith(jpeg_start)
+ assert frame.endswith(jpeg_end)
+ assert len(frame) == 54 # 2 + 50 + 2
+
+ def test_incomplete_frame_detection(self):
+ """Test detection of incomplete frame (no end marker)."""
+ jpeg_start = b"\xff\xd8"
+
+ # Incomplete buffer - no end marker
+ buffer = jpeg_start + b"\x00" * 100
+
+ start_idx = buffer.find(jpeg_start)
+ end_idx = buffer.find(b"\xff\xd9", start_idx + 2)
+
+ assert start_idx == 0
+ assert end_idx == -1 # Not found
+
+ def test_multiple_frames_in_buffer(self):
+ """Test extracting first frame when buffer contains multiple frames."""
+ jpeg_start = b"\xff\xd8"
+ jpeg_end = b"\xff\xd9"
+
+ # Two complete frames
+ frame1 = jpeg_start + b"\x01" * 10 + jpeg_end
+ frame2 = jpeg_start + b"\x02" * 20 + jpeg_end
+ buffer = frame1 + frame2
+
+ # Extract first frame
+ start_idx = buffer.find(jpeg_start)
+ end_idx = buffer.find(jpeg_end, start_idx + 2)
+ first_frame = buffer[start_idx : end_idx + 2]
+
+ assert first_frame == frame1
+ assert len(first_frame) == 14
+
+ # Remaining buffer should contain second frame
+ remaining = buffer[end_idx + 2 :]
+ assert remaining == frame2
+
+
+class TestCameraTypeValidation:
+ """Tests for camera type handling."""
+
+ @pytest.mark.asyncio
+ async def test_capture_frame_unknown_type_returns_none(self):
+ """Verify unknown camera type returns None."""
+ from backend.app.services.external_camera import capture_frame
+
+ result = await capture_frame("http://example.com", "unknown_type")
+ assert result is None
+
+ @pytest.mark.asyncio
+ async def test_capture_frame_valid_types(self):
+ """Verify valid camera types are accepted (they may fail but shouldn't error on type)."""
+ from backend.app.services.external_camera import capture_frame
+
+ # These will fail to connect but shouldn't raise type errors
+ for camera_type in ["mjpeg", "rtsp", "snapshot"]:
+ # Use a non-routable IP to fail fast
+ result = await capture_frame("http://192.0.2.1/test", camera_type, timeout=1)
+ # Should return None (failed connection) not raise exception
+ assert result is None
+
+
+class TestRtspUrlHandling:
+ """Tests for RTSP/RTSPS URL handling."""
+
+ def test_rtsps_url_detection(self):
+ """Verify rtsps:// and rtsp:// URL schemes are distinct."""
+ url_rtsps = "rtsps://user:pass@192.168.1.1:554/stream"
+ url_rtsp = "rtsp://user:pass@192.168.1.1:554/stream"
+
+ assert url_rtsps.startswith("rtsps://")
+ assert not url_rtsp.startswith("rtsps://")
+ assert url_rtsp.startswith("rtsp://")
+
+ def test_ffmpeg_handles_both_rtsp_and_rtsps(self):
+ """Verify ffmpeg command structure handles both URL schemes identically.
+
+ ffmpeg automatically handles TLS for rtsps:// URLs, so no special
+ flags are needed - both URL schemes use the same command structure.
+ """
+ # Both URL types should use the same basic ffmpeg options
+ base_cmd = [
+ "ffmpeg",
+ "-rtsp_transport",
+ "tcp",
+ "-i",
+ ]
+
+ rtsp_url = "rtsp://user:pass@192.168.1.1:554/stream"
+ rtsps_url = "rtsps://user:pass@192.168.1.1:554/stream"
+
+ # Command structure is identical for both
+ cmd_rtsp = base_cmd + [rtsp_url]
+ cmd_rtsps = base_cmd + [rtsps_url]
+
+ # Only the URL differs
+ assert cmd_rtsp[:-1] == cmd_rtsps[:-1]
+ assert cmd_rtsp[-1] != cmd_rtsps[-1]
diff --git a/backend/tests/unit/services/test_layer_timelapse.py b/backend/tests/unit/services/test_layer_timelapse.py
new file mode 100644
index 000000000..893713346
--- /dev/null
+++ b/backend/tests/unit/services/test_layer_timelapse.py
@@ -0,0 +1,320 @@
+"""
+Tests for the layer timelapse service.
+
+These tests cover session management and pure logic functions.
+"""
+
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+class TestTimelapseSessionManagement:
+ """Tests for timelapse session lifecycle."""
+
+ def test_start_session_creates_new_session(self):
+ """Verify start_session creates and registers a new session."""
+ from backend.app.services.layer_timelapse import (
+ _active_sessions,
+ cancel_session,
+ get_session,
+ start_session,
+ )
+
+ # Clear any existing sessions
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test_bambuddy")
+
+ session = start_session(
+ printer_id=1,
+ archive_id=100,
+ url="http://camera.local/mjpeg",
+ cam_type="mjpeg",
+ )
+
+ assert session is not None
+ assert session.printer_id == 1
+ assert session.archive_id == 100
+ assert session.camera_url == "http://camera.local/mjpeg"
+ assert session.camera_type == "mjpeg"
+ assert session.last_layer == -1
+ assert session.frame_count == 0
+
+ # Session should be retrievable
+ retrieved = get_session(1)
+ assert retrieved is session
+
+ # Cleanup
+ cancel_session(1)
+
+ def test_start_session_cancels_existing(self):
+ """Verify starting a new session cancels any existing session."""
+ from backend.app.services.layer_timelapse import (
+ _active_sessions,
+ cancel_session,
+ get_session,
+ start_session,
+ )
+
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test_bambuddy")
+
+ # Start first session
+ session1 = start_session(1, 100, "http://cam1/", "mjpeg")
+
+ # Mock cleanup to track if it was called
+ session1.cleanup = MagicMock()
+
+ # Start second session for same printer
+ session2 = start_session(1, 101, "http://cam2/", "rtsp")
+
+ # First session should be replaced
+ current = get_session(1)
+ assert current is session2
+ assert current.archive_id == 101 # Verify it's the new session
+ assert current.camera_url == "http://cam2/"
+
+ # First session's cleanup should have been called
+ session1.cleanup.assert_called_once()
+
+ # Cleanup
+ cancel_session(1)
+
+ def test_get_session_returns_none_for_unknown(self):
+ """Verify get_session returns None for unknown printer."""
+ from backend.app.services.layer_timelapse import _active_sessions, get_session
+
+ _active_sessions.clear()
+
+ result = get_session(999)
+ assert result is None
+
+ def test_cancel_session_removes_and_cleans_up(self):
+ """Verify cancel_session removes session and cleans up."""
+ from backend.app.services.layer_timelapse import (
+ _active_sessions,
+ cancel_session,
+ get_session,
+ start_session,
+ )
+
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test_bambuddy")
+
+ session = start_session(1, 100, "http://cam/", "mjpeg")
+
+ # Mock cleanup to avoid filesystem operations
+ session.cleanup = MagicMock()
+
+ cancel_session(1)
+
+ # Session should be removed
+ assert get_session(1) is None
+ # Cleanup should have been called
+ session.cleanup.assert_called_once()
+
+ def test_cancel_nonexistent_session_is_safe(self):
+ """Verify canceling a non-existent session doesn't error."""
+ from backend.app.services.layer_timelapse import _active_sessions, cancel_session
+
+ _active_sessions.clear()
+
+ # Should not raise
+ cancel_session(999)
+
+
+class TestTimelapseSession:
+ """Tests for TimelapseSession class."""
+
+ def test_session_id_format(self):
+ """Verify session ID follows expected datetime format."""
+ from backend.app.services.layer_timelapse import TimelapseSession, _active_sessions
+
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test_bambuddy")
+
+ session = TimelapseSession(
+ printer_id=1,
+ archive_id=100,
+ camera_url="http://test/",
+ camera_type="mjpeg",
+ )
+
+ # Session ID should be timestamp format YYYYMMDD_HHMMSS
+ assert len(session.session_id) == 15
+ assert session.session_id[8] == "_"
+
+ # Should be parseable as datetime
+ try:
+ datetime.strptime(session.session_id, "%Y%m%d_%H%M%S")
+ except ValueError:
+ pytest.fail("Session ID is not valid datetime format")
+
+ def test_frames_dir_path_structure(self):
+ """Verify frames directory path is structured correctly."""
+ from backend.app.services.layer_timelapse import TimelapseSession
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/data/bambuddy")
+
+ with patch.object(Path, "mkdir"): # Avoid creating real directories
+ session = TimelapseSession(
+ printer_id=42,
+ archive_id=100,
+ camera_url="http://test/",
+ camera_type="mjpeg",
+ )
+
+ expected_path = Path("/data/bambuddy/timelapse_frames/42") / session.session_id
+ assert session.frames_dir == expected_path
+
+
+class TestLayerChangeLogic:
+ """Tests for layer change capture logic."""
+
+ @pytest.mark.asyncio
+ async def test_capture_layer_only_on_increase(self):
+ """Verify frames are only captured when layer increases."""
+ from backend.app.services.layer_timelapse import TimelapseSession
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test")
+
+ with patch.object(Path, "mkdir"):
+ session = TimelapseSession(1, 100, "http://test/", "mjpeg")
+
+ # Mock capture_frame to return data
+ with patch(
+ "backend.app.services.layer_timelapse.capture_frame", new_callable=AsyncMock
+ ) as mock_capture:
+ mock_capture.return_value = b"\xff\xd8test\xff\xd9"
+
+ with patch.object(Path, "write_bytes"):
+ # First layer should capture
+ result = await session.capture_layer(1)
+ assert result is True
+ assert session.last_layer == 1
+ assert session.frame_count == 1
+
+ # Same layer should NOT capture
+ result = await session.capture_layer(1)
+ assert result is False
+ assert session.frame_count == 1
+
+ # Lower layer should NOT capture
+ result = await session.capture_layer(0)
+ assert result is False
+ assert session.frame_count == 1
+
+ # Higher layer should capture
+ result = await session.capture_layer(5)
+ assert result is True
+ assert session.last_layer == 5
+ assert session.frame_count == 2
+
+ @pytest.mark.asyncio
+ async def test_capture_layer_handles_failed_capture(self):
+ """Verify failed capture returns False but updates layer."""
+ from backend.app.services.layer_timelapse import TimelapseSession
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test")
+
+ with patch.object(Path, "mkdir"):
+ session = TimelapseSession(1, 100, "http://test/", "mjpeg")
+
+ # Mock capture_frame to return None (failure)
+ with patch(
+ "backend.app.services.layer_timelapse.capture_frame", new_callable=AsyncMock
+ ) as mock_capture:
+ mock_capture.return_value = None
+
+ result = await session.capture_layer(1)
+
+ assert result is False
+ assert session.last_layer == 1 # Layer is still updated
+ assert session.frame_count == 0 # But frame count not incremented
+
+
+class TestOnLayerChange:
+ """Tests for the on_layer_change callback."""
+
+ @pytest.mark.asyncio
+ async def test_on_layer_change_captures_when_session_exists(self):
+ """Verify on_layer_change triggers capture when session exists."""
+ from backend.app.services.layer_timelapse import (
+ _active_sessions,
+ cancel_session,
+ on_layer_change,
+ start_session,
+ )
+
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test")
+
+ with patch.object(Path, "mkdir"):
+ session = start_session(1, 100, "http://test/", "mjpeg")
+
+ with patch.object(session, "capture_layer", new_callable=AsyncMock) as mock_capture:
+ mock_capture.return_value = True
+
+ await on_layer_change(1, 5)
+
+ mock_capture.assert_called_once_with(5)
+
+ cancel_session(1)
+
+ @pytest.mark.asyncio
+ async def test_on_layer_change_does_nothing_without_session(self):
+ """Verify on_layer_change is safe when no session exists."""
+ from backend.app.services.layer_timelapse import _active_sessions, on_layer_change
+
+ _active_sessions.clear()
+
+ # Should not raise
+ await on_layer_change(999, 10)
+
+
+class TestGetActiveSessions:
+ """Tests for get_active_sessions."""
+
+ def test_get_active_sessions_returns_copy(self):
+ """Verify get_active_sessions returns a copy, not the original dict."""
+ from backend.app.services.layer_timelapse import (
+ _active_sessions,
+ cancel_session,
+ get_active_sessions,
+ start_session,
+ )
+
+ _active_sessions.clear()
+
+ with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
+ mock_settings.base_dir = Path("/tmp/test")
+
+ with patch.object(Path, "mkdir"):
+ start_session(1, 100, "http://test/", "mjpeg")
+
+ sessions = get_active_sessions()
+
+ # Should be a copy
+ assert sessions is not _active_sessions
+ assert 1 in sessions
+
+ # Modifying copy shouldn't affect original
+ sessions.clear()
+ assert 1 in _active_sessions
+
+ cancel_session(1)
diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts
index f2e8361da..89f3336d2 100644
--- a/frontend/src/api/client.ts
+++ b/frontend/src/api/client.ts
@@ -65,6 +65,9 @@ export interface Printer {
nozzle_count: number; // 1 or 2, auto-detected from MQTT
is_active: boolean;
auto_archive: boolean;
+ external_camera_url: string | null;
+ external_camera_type: string | null; // "mjpeg", "rtsp", "snapshot"
+ external_camera_enabled: boolean;
created_at: string;
updated_at: string;
}
@@ -209,6 +212,9 @@ export interface PrinterCreate {
model?: string;
location?: string;
auto_archive?: boolean;
+ external_camera_url?: string | null;
+ external_camera_type?: string | null;
+ external_camera_enabled?: boolean;
}
// Archive types
@@ -1548,6 +1554,11 @@ export const api = {
request<{ connected: boolean }>(`/printers/${id}/disconnect`, {
method: 'POST',
}),
+ testExternalCamera: (printerId: number, url: string, cameraType: string) =>
+ request<{ success: boolean; error?: string; resolution?: string }>(
+ `/printers/${printerId}/camera/external/test?url=${encodeURIComponent(url)}&camera_type=${encodeURIComponent(cameraType)}`,
+ { method: 'POST' }
+ ),
// Print Control
stopPrint: (printerId: number) =>
diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx
index eebd58bdc..d13ab0854 100644
--- a/frontend/src/pages/SettingsPage.tsx
+++ b/frontend/src/pages/SettingsPage.tsx
@@ -95,6 +95,10 @@ export function SettingsPage() {
const [haTestResult, setHaTestResult] = useState<{ success: boolean; message: string | null; error: string | null } | null>(null);
const [haTestLoading, setHaTestLoading] = useState(false);
+ // External camera test state
+ const [extCameraTestResults, setExtCameraTestResults] = useState>({});
+ const [extCameraTestLoading, setExtCameraTestLoading] = useState>({});
+
const handleDefaultViewChange = (path: string) => {
setDefaultViewState(path);
setDefaultView(path);
@@ -366,6 +370,18 @@ export function SettingsPage() {
},
});
+ const updatePrinterMutation = useMutation({
+ mutationFn: ({ id, data }: { id: number; data: Partial<{ external_camera_url: string | null; external_camera_type: string | null; external_camera_enabled: boolean }> }) =>
+ api.updatePrinter(id, data),
+ onSuccess: () => {
+ queryClient.invalidateQueries({ queryKey: ['printers'] });
+ showToast('Camera settings saved', 'success');
+ },
+ onError: (error: Error) => {
+ showToast(`Failed to update printer: ${error.message}`, 'error');
+ },
+ });
+
// Debounced auto-save when localSettings change
useEffect(() => {
// Skip if initial load or no settings
@@ -489,6 +505,38 @@ export function SettingsPage() {
setLocalSettings(prev => prev ? { ...prev, [key]: value } : null);
}, []);
+ const handleTestExternalCamera = async (printerId: number, url: string, cameraType: string) => {
+ if (!url) {
+ showToast('Please enter a camera URL', 'error');
+ return;
+ }
+ setExtCameraTestLoading(prev => ({ ...prev, [printerId]: true }));
+ setExtCameraTestResults(prev => ({ ...prev, [printerId]: null }));
+ try {
+ const result = await api.testExternalCamera(printerId, url, cameraType);
+ setExtCameraTestResults(prev => ({ ...prev, [printerId]: result }));
+ if (result.success) {
+ showToast(`Camera connected${result.resolution ? ` (${result.resolution})` : ''}`, 'success');
+ } else {
+ showToast(result.error || 'Connection failed', 'error');
+ }
+ } catch (error) {
+ const message = error instanceof Error ? error.message : 'Test failed';
+ setExtCameraTestResults(prev => ({ ...prev, [printerId]: { success: false, error: message } }));
+ showToast(message, 'error');
+ } finally {
+ setExtCameraTestLoading(prev => ({ ...prev, [printerId]: false }));
+ }
+ };
+
+ const handleUpdatePrinterCamera = (printerId: number, updates: { url?: string; type?: string; enabled?: boolean }) => {
+ const data: Partial<{ external_camera_url: string | null; external_camera_type: string | null; external_camera_enabled: boolean }> = {};
+ if (updates.url !== undefined) data.external_camera_url = updates.url || null;
+ if (updates.type !== undefined) data.external_camera_type = updates.type || null;
+ if (updates.enabled !== undefined) data.external_camera_enabled = updates.enabled;
+ updatePrinterMutation.mutate({ id: printerId, data });
+ };
+
if (isLoading || !localSettings) {
return (
@@ -952,6 +1000,87 @@ export function SettingsPage() {
: 'Camera opens in a separate browser window'}
+
+ {/* External Cameras Section */}
+
+
External Cameras
+
+ Configure network cameras to replace the built-in printer camera. Supports MJPEG streams, RTSP, and HTTP snapshots. When enabled, the external camera is used for live view and finish photos.
+