public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-17 19:36:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #158 from maziggy/feature/empty_plate_detection

Browse Source 
Add build plate empty detection feature
This commit is contained in:
MartinNYHC
2026-01-26 14:38:19 +01:00
committed by GitHub
parent 6b0e3e7964 888891fdc6
commit 4ff9378a6b
26 changed files with 2536 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/camera.py
+357
View File
@@ -675,3 +675,360 @@ async def test_external_camera(
from backend.app.services.external_camera import test_connection
return await test_connection(url, camera_type)
@router.get("/{printer_id}/camera/check-plate")
async def check_plate_empty(
printer_id: int,
plate_type: str | None = None,
use_external: bool = False,
include_debug_image: bool = False,
db: AsyncSession = Depends(get_db),
):
"""Check if the build plate is empty using camera vision.
Uses calibration-based difference detection - compares current frame
to a reference image of the empty plate.
IMPORTANT: Chamber light must be ON for reliable detection.
Args:
printer_id: Printer ID
plate_type: Type of build plate (e.g., "High Temp Plate") for calibration lookup
use_external: If True, prefer external camera over built-in
include_debug_image: If True, return URL to annotated debug image
Returns:
Dict with detection results:
- is_empty: bool - Whether plate appears empty
- confidence: float - Confidence level (0.0 to 1.0)
- difference_percent: float - How different from calibration reference
- message: str - Human-readable result message
- needs_calibration: bool - True if calibration is required
- light_warning: bool - True if chamber light is off
"""
from backend.app.services.plate_detection import (
check_plate_empty as do_check,
is_plate_detection_available,
)
from backend.app.services.printer_manager import printer_manager
if not is_plate_detection_available():
raise HTTPException(
status_code=503,
detail="Plate detection not available. Install opencv-python-headless to enable.",
)
printer = await get_printer_or_404(printer_id, db)
# Check chamber light status
light_warning = False
state = printer_manager.get_status(printer_id)
if state and not state.chamber_light:
light_warning = True
from backend.app.services.plate_detection import PlateDetector
# Build ROI tuple from printer settings if available
roi = None
if all(
[
printer.plate_detection_roi_x is not None,
printer.plate_detection_roi_y is not None,
printer.plate_detection_roi_w is not None,
printer.plate_detection_roi_h is not None,
]
):
roi = (
printer.plate_detection_roi_x,
printer.plate_detection_roi_y,
printer.plate_detection_roi_w,
printer.plate_detection_roi_h,
)
result = await do_check(
printer_id=printer.id,
ip_address=printer.ip_address,
access_code=printer.access_code,
model=printer.model,
plate_type=plate_type,
include_debug_image=include_debug_image,
external_camera_url=printer.external_camera_url if printer.external_camera_enabled else None,
external_camera_type=printer.external_camera_type if printer.external_camera_enabled else None,
use_external=use_external,
roi=roi,
)
# Get reference count for the response
detector = PlateDetector()
ref_count = detector.get_calibration_count(printer.id)
response = result.to_dict()
response["light_warning"] = light_warning
response["reference_count"] = ref_count
response["max_references"] = detector.MAX_REFERENCES
# Include current ROI in response
if roi:
response["roi"] = {"x": roi[0], "y": roi[1], "w": roi[2], "h": roi[3]}
else:
# Return default ROI
response["roi"] = {"x": 0.15, "y": 0.35, "w": 0.70, "h": 0.55}
# If debug image requested and available, encode as base64 data URL
if include_debug_image and result.debug_image:
import base64
b64_image = base64.b64encode(result.debug_image).decode("utf-8")
response["debug_image_url"] = f"data:image/jpeg;base64,{b64_image}"
return response
@router.post("/{printer_id}/camera/plate-detection/calibrate")
async def calibrate_plate_detection(
printer_id: int,
label: str | None = None,
use_external: bool = False,
db: AsyncSession = Depends(get_db),
):
"""Calibrate plate detection by capturing a reference image of the empty plate.
The plate MUST be empty when calling this endpoint. The captured image
will be used as the reference for future detection comparisons.
Supports up to 5 reference images per printer. When adding a 6th, the oldest
is automatically removed.
IMPORTANT: Chamber light should be ON for calibration.
Args:
printer_id: Printer ID
label: Optional label for this reference (e.g., "High Temp Plate", "Wham Bam")
use_external: If True, prefer external camera over built-in
Returns:
Dict with:
- success: bool - Whether calibration succeeded
- message: str - Status message
- index: int - The reference slot used (0-4)
"""
from backend.app.services.plate_detection import (
calibrate_plate,
is_plate_detection_available,
)
from backend.app.services.printer_manager import printer_manager
if not is_plate_detection_available():
raise HTTPException(
status_code=503,
detail="Plate detection not available. Install opencv-python-headless to enable.",
)
printer = await get_printer_or_404(printer_id, db)
# Check chamber light - warn but don't block
state = printer_manager.get_status(printer_id)
light_warning = state and not state.chamber_light
success, message, index = await calibrate_plate(
printer_id=printer.id,
ip_address=printer.ip_address,
access_code=printer.access_code,
model=printer.model,
label=label,
external_camera_url=printer.external_camera_url if printer.external_camera_enabled else None,
external_camera_type=printer.external_camera_type if printer.external_camera_enabled else None,
use_external=use_external,
)
if light_warning and success:
message += " (Warning: Chamber light was off)"
return {"success": success, "message": message, "index": index}
@router.delete("/{printer_id}/camera/plate-detection/calibrate")
async def delete_plate_calibration(
printer_id: int,
plate_type: str | None = None,
db: AsyncSession = Depends(get_db),
):
"""Delete the plate detection calibration for a printer and plate type.
Args:
printer_id: Printer ID
plate_type: Type of build plate (if None, deletes legacy non-plate-specific calibration)
Returns:
Dict with:
- success: bool - Whether deletion succeeded
- message: str - Status message
"""
from backend.app.services.plate_detection import (
delete_calibration,
is_plate_detection_available,
)
if not is_plate_detection_available():
raise HTTPException(
status_code=503,
detail="Plate detection not available. Install opencv-python-headless to enable.",
)
# Verify printer exists
await get_printer_or_404(printer_id, db)
deleted = delete_calibration(printer_id, plate_type)
plate_msg = f" for '{plate_type}'" if plate_type else ""
return {
"success": deleted,
"message": f"Calibration deleted{plate_msg}" if deleted else f"No calibration found{plate_msg}",
}
@router.get("/{printer_id}/camera/plate-detection/status")
async def get_plate_detection_status(
printer_id: int,
plate_type: str | None = None,
db: AsyncSession = Depends(get_db),
):
"""Check plate detection status for a printer and plate type.
Returns:
Dict with:
- available: bool - Whether OpenCV is installed
- calibrated: bool - Whether printer has calibration for this plate type
- plate_type: str - The plate type queried
- chamber_light: bool - Whether chamber light is on
- message: str - Status message
"""
from backend.app.services.plate_detection import (
get_calibration_status,
is_plate_detection_available,
)
from backend.app.services.printer_manager import printer_manager
if not is_plate_detection_available():
return {
"available": False,
"calibrated": False,
"plate_type": plate_type,
"chamber_light": False,
"message": "OpenCV not installed",
}
# Verify printer exists
await get_printer_or_404(printer_id, db)
# Get chamber light status
state = printer_manager.get_status(printer_id)
chamber_light = state.chamber_light if state else False
status = get_calibration_status(printer_id, plate_type)
status["chamber_light"] = chamber_light
return status
@router.get("/{printer_id}/camera/plate-detection/references")
async def get_plate_references(
printer_id: int,
db: AsyncSession = Depends(get_db),
):
"""Get all calibration references for a printer with metadata.
Returns list of references with index, label, timestamp, and thumbnail URL.
"""
from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available
if not is_plate_detection_available():
raise HTTPException(503, "Plate detection not available")
await get_printer_or_404(printer_id, db)
detector = PlateDetector()
references = detector.get_references(printer_id)
# Add thumbnail URLs
for ref in references:
ref["thumbnail_url"] = (
f"/api/v1/printers/{printer_id}/camera/plate-detection/references/{ref['index']}/thumbnail"
)
return {
"references": references,
"max_references": detector.MAX_REFERENCES,
}
@router.get("/{printer_id}/camera/plate-detection/references/{index}/thumbnail")
async def get_reference_thumbnail(
printer_id: int,
index: int,
db: AsyncSession = Depends(get_db),
):
"""Get thumbnail image for a calibration reference."""
from fastapi.responses import Response
from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available
if not is_plate_detection_available():
raise HTTPException(503, "Plate detection not available")
await get_printer_or_404(printer_id, db)
detector = PlateDetector()
thumbnail = detector.get_reference_thumbnail(printer_id, index)
if thumbnail is None:
raise HTTPException(404, "Reference not found")
return Response(content=thumbnail, media_type="image/jpeg")
@router.put("/{printer_id}/camera/plate-detection/references/{index}")
async def update_reference_label(
printer_id: int,
index: int,
label: str,
db: AsyncSession = Depends(get_db),
):
"""Update the label for a calibration reference."""
from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available
if not is_plate_detection_available():
raise HTTPException(503, "Plate detection not available")
await get_printer_or_404(printer_id, db)
detector = PlateDetector()
success = detector.update_reference_label(printer_id, index, label)
if not success:
raise HTTPException(404, "Reference not found")
return {"success": True, "index": index, "label": label}
@router.delete("/{printer_id}/camera/plate-detection/references/{index}")
async def delete_reference(
printer_id: int,
index: int,
db: AsyncSession = Depends(get_db),
):
"""Delete a specific calibration reference."""
from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available
if not is_plate_detection_available():
raise HTTPException(503, "Plate detection not available")
await get_printer_or_404(printer_id, db)
detector = PlateDetector()
success = detector.delete_reference(printer_id, index)
if not success:
raise HTTPException(404, "Reference not found")
return {"success": True, "message": "Reference deleted"}
backend/app/api/routes/printers.py
+16
View File
@@ -92,6 +92,22 @@ async def update_printer(
raise HTTPException(404, "Printer not found")
update_data = printer_data.model_dump(exclude_unset=True)
# Handle nested ROI object - flatten to individual columns
if "plate_detection_roi" in update_data:
roi = update_data.pop("plate_detection_roi")
if roi:
update_data["plate_detection_roi_x"] = roi.get("x")
update_data["plate_detection_roi_y"] = roi.get("y")
update_data["plate_detection_roi_w"] = roi.get("w")
update_data["plate_detection_roi_h"] = roi.get("h")
else:
# Clear ROI if set to null
update_data["plate_detection_roi_x"] = None
update_data["plate_detection_roi_y"] = None
update_data["plate_detection_roi_w"] = None
update_data["plate_detection_roi_h"] = None
for field, value in update_data.items():
setattr(printer, field, value)
backend/app/api/routes/settings.py
+10
View File
@@ -416,6 +416,11 @@ async def export_backup(
"external_camera_url": printer.external_camera_url,
"external_camera_type": printer.external_camera_type,
"external_camera_enabled": printer.external_camera_enabled,
"plate_detection_enabled": printer.plate_detection_enabled,
"plate_detection_roi_x": printer.plate_detection_roi_x,
"plate_detection_roi_y": printer.plate_detection_roi_y,
"plate_detection_roi_w": printer.plate_detection_roi_w,
"plate_detection_roi_h": printer.plate_detection_roi_h,
}
if include_access_codes:
printer_data["access_code"] = printer.access_code
@@ -1000,6 +1005,11 @@ async def import_backup(
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),
plate_detection_enabled=printer_data.get("plate_detection_enabled", False),
plate_detection_roi_x=printer_data.get("plate_detection_roi_x"),
plate_detection_roi_y=printer_data.get("plate_detection_roi_y"),
plate_detection_roi_w=printer_data.get("plate_detection_roi_w"),
plate_detection_roi_h=printer_data.get("plate_detection_roi_h"),
)
db.add(printer)
restored["printers"] += 1
backend/app/core/database.py
+24
View File
@@ -729,6 +729,30 @@ async def run_migrations(conn):
except Exception:
pass
# Migration: Add plate_detection_enabled column to printers
try:
await conn.execute(text("ALTER TABLE printers ADD COLUMN plate_detection_enabled BOOLEAN DEFAULT 0"))
except Exception:
pass
# Migration: Add plate detection ROI columns to printers
try:
await conn.execute(text("ALTER TABLE printers ADD COLUMN plate_detection_roi_x REAL"))
except Exception:
pass
try:
await conn.execute(text("ALTER TABLE printers ADD COLUMN plate_detection_roi_y REAL"))
except Exception:
pass
try:
await conn.execute(text("ALTER TABLE printers ADD COLUMN plate_detection_roi_w REAL"))
except Exception:
pass
try:
await conn.execute(text("ALTER TABLE printers ADD COLUMN plate_detection_roi_h REAL"))
except Exception:
pass
async def seed_notification_templates():
"""Seed default notification templates if they don't exist."""
backend/app/main.py
+74
View File
@@ -505,6 +505,80 @@ async def on_print_start(printer_id: int, data: dict):
result = await db.execute(select(Printer).where(Printer.id == printer_id))
printer = result.scalar_one_or_none()
# Plate detection check - pause if objects detected on build plate
if printer and printer.plate_detection_enabled:
try:
from backend.app.services.plate_detection import check_plate_empty
# Build ROI tuple from printer settings if available
roi = None
if all(
[
printer.plate_detection_roi_x is not None,
printer.plate_detection_roi_y is not None,
printer.plate_detection_roi_w is not None,
printer.plate_detection_roi_h is not None,
]
):
roi = (
printer.plate_detection_roi_x,
printer.plate_detection_roi_y,
printer.plate_detection_roi_w,
printer.plate_detection_roi_h,
)
logger.info(f"[PLATE CHECK] Running plate detection for printer {printer_id}")
plate_result = await check_plate_empty(
printer_id=printer_id,
ip_address=printer.ip_address,
access_code=printer.access_code,
model=printer.model,
include_debug_image=False,
external_camera_url=printer.external_camera_url,
external_camera_type=printer.external_camera_type,
use_external=printer.external_camera_enabled,
roi=roi,
)
if not plate_result.needs_calibration and not plate_result.is_empty:
# Objects detected - pause the print!
logger.warning(
f"[PLATE CHECK] Objects detected on plate for printer {printer_id}! "
f"Confidence: {plate_result.confidence:.0%}, Diff: {plate_result.difference_percent:.1f}%"
)
client = printer_manager.get_client(printer_id)
if client:
client.pause_print()
logger.info(f"[PLATE CHECK] Print paused for printer {printer_id}")
# Send notification about plate not empty
await ws_manager.broadcast(
{
"type": "plate_not_empty",
"printer_id": printer_id,
"printer_name": printer.name,
"message": f"Objects detected on build plate! Print paused. (Diff: {plate_result.difference_percent:.1f}%)",
}
)
# Also send push notification
try:
await notification_service.send_notification(
printer_id=printer_id,
printer_name=printer.name,
event_type="plate_not_empty",
title=f"⚠️ {printer.name}: Plate Not Empty!",
body="Objects detected on build plate. Print has been paused. Please clear the plate and resume.",
db=db,
)
except Exception as notif_err:
logger.warning(f"[PLATE CHECK] Failed to send notification: {notif_err}")
else:
logger.info(f"[PLATE CHECK] Plate is empty for printer {printer_id}, proceeding with print")
except Exception as plate_err:
# Don't block print on plate detection errors
logger.warning(f"[PLATE CHECK] Plate detection failed for printer {printer_id}: {plate_err}")
if not printer or not printer.auto_archive:
# Send notification without archive data (auto-archive disabled)
logger.info(
backend/app/models/printer.py
+7
View File
@@ -28,6 +28,13 @@ class Printer(Base):
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)
# Plate detection - check if build plate is empty before starting print
plate_detection_enabled: Mapped[bool] = mapped_column(Boolean, default=False)
# ROI for plate detection (percentages: 0.0-1.0)
plate_detection_roi_x: Mapped[float | None] = mapped_column(Float, nullable=True) # X start %
plate_detection_roi_y: Mapped[float | None] = mapped_column(Float, nullable=True) # Y start %
plate_detection_roi_w: Mapped[float | None] = mapped_column(Float, nullable=True) # Width %
plate_detection_roi_h: Mapped[float | None] = mapped_column(Float, nullable=True) # Height %
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())
backend/app/schemas/printer.py
+52
View File
@@ -20,6 +20,15 @@ class PrinterCreate(PrinterBase):
pass
class PlateDetectionROI(BaseModel):
"""Region of interest for plate detection (percentages 0.0-1.0)."""
x: float = Field(..., ge=0.0, le=1.0) # X start %
y: float = Field(..., ge=0.0, le=1.0) # Y start %
w: float = Field(..., ge=0.0, le=1.0) # Width %
h: float = Field(..., ge=0.0, le=1.0) # Height %
class PrinterUpdate(BaseModel):
name: str | None = None
ip_address: str | None = None
@@ -32,6 +41,8 @@ class PrinterUpdate(BaseModel):
external_camera_url: str | None = None
external_camera_type: str | None = None
external_camera_enabled: bool | None = None
plate_detection_enabled: bool | None = None
plate_detection_roi: PlateDetectionROI | None = None
class PrinterResponse(PrinterBase):
@@ -42,12 +53,53 @@ class PrinterResponse(PrinterBase):
external_camera_url: str | None = None
external_camera_type: str | None = None
external_camera_enabled: bool = False
plate_detection_enabled: bool = False
plate_detection_roi: PlateDetectionROI | None = None
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
@classmethod
def from_orm_with_roi(cls, printer) -> "PrinterResponse":
"""Create response from ORM model, converting ROI fields to nested object."""
data = {
"id": printer.id,
"name": printer.name,
"serial_number": printer.serial_number,
"ip_address": printer.ip_address,
"access_code": printer.access_code,
"model": printer.model,
"location": printer.location,
"auto_archive": printer.auto_archive,
"external_camera_url": printer.external_camera_url,
"external_camera_type": printer.external_camera_type,
"external_camera_enabled": printer.external_camera_enabled,
"is_active": printer.is_active,
"nozzle_count": printer.nozzle_count,
"print_hours_offset": printer.print_hours_offset,
"plate_detection_enabled": printer.plate_detection_enabled,
"created_at": printer.created_at,
"updated_at": printer.updated_at,
}
# Build ROI object if any ROI field is set
if any(
[
printer.plate_detection_roi_x is not None,
printer.plate_detection_roi_y is not None,
printer.plate_detection_roi_w is not None,
printer.plate_detection_roi_h is not None,
]
):
data["plate_detection_roi"] = PlateDetectionROI(
x=printer.plate_detection_roi_x or 0.15,
y=printer.plate_detection_roi_y or 0.35,
w=printer.plate_detection_roi_w or 0.70,
h=printer.plate_detection_roi_h or 0.55,
)
return cls(**data)
class HMSErrorResponse(BaseModel):
code: str
backend/app/services/plate_detection.py
+774
View File
@@ -0,0 +1,774 @@
"""Build plate empty detection using OpenCV.
Analyzes camera frames to detect if there are objects on the build plate.
Uses calibration-based difference detection - compares current frame to
a reference image of the empty plate.
"""
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
# Optional OpenCV import - feature disabled if not available
try:
import cv2
import numpy as np
OPENCV_AVAILABLE = True
except ImportError:
OPENCV_AVAILABLE = False
logger.info("OpenCV not available - plate detection feature disabled")
# Directory to store calibration reference images
CALIBRATION_DIR = Path(__file__).parent.parent.parent.parent / "data" / "plate_calibration"
class PlateDetectionResult:
"""Result of plate detection analysis."""
def __init__(
self,
is_empty: bool,
confidence: float,
difference_percent: float,
message: str,
debug_image: bytes | None = None,
needs_calibration: bool = False,
):
self.is_empty = is_empty
self.confidence = confidence # 0.0 to 1.0
self.difference_percent = difference_percent # How different from reference
self.message = message
self.debug_image = debug_image # Optional annotated image for debugging
self.needs_calibration = needs_calibration # True if no reference image exists
def to_dict(self) -> dict:
return {
"is_empty": bool(self.is_empty),
"confidence": float(round(self.confidence, 2)),
"difference_percent": float(round(self.difference_percent, 2)),
"message": self.message,
"has_debug_image": self.debug_image is not None,
"needs_calibration": bool(self.needs_calibration),
}
class PlateDetector:
"""Detects if the build plate is empty using calibration-based difference detection."""
# Default region of interest (ROI) as percentage of image dimensions
# These define where the build plate typically appears in the camera view
# Format: (x_start%, y_start%, width%, height%)
DEFAULT_ROI = (0.15, 0.35, 0.70, 0.55) # Center-lower portion of frame
# Detection thresholds for difference detection
# Using mean pixel difference (0-100% scale)
# Small objects may only cause 1-2% mean difference
DEFAULT_DIFFERENCE_THRESHOLD = 1.0
DEFAULT_BLUR_SIZE = 21 # Gaussian blur kernel size (must be odd) - unused with edge detection
def __init__(
self,
roi: tuple[float, float, float, float] | None = None,
difference_threshold: float = DEFAULT_DIFFERENCE_THRESHOLD,
blur_size: int = DEFAULT_BLUR_SIZE,
):
"""Initialize the plate detector.
Args:
roi: Region of interest as (x%, y%, w%, h%) - percentages of image size
difference_threshold: Percentage of pixels that must differ to trigger "not empty"
blur_size: Gaussian blur kernel size for noise reduction
"""
if not OPENCV_AVAILABLE:
raise RuntimeError("OpenCV is not installed. Install with: pip install opencv-python-headless")
self.roi = roi or self.DEFAULT_ROI
self.difference_threshold = difference_threshold
self.blur_size = blur_size if blur_size % 2 == 1 else blur_size + 1 # Must be odd
# Maximum number of reference images to store per printer
MAX_REFERENCES = 5
def _get_metadata_path(self, printer_id: int) -> Path:
"""Get the path to the metadata JSON file for a printer."""
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
return CALIBRATION_DIR / f"printer_{printer_id}_metadata.json"
def _load_metadata(self, printer_id: int) -> dict:
"""Load metadata for a printer's references."""
import json
meta_path = self._get_metadata_path(printer_id)
if meta_path.exists():
try:
with open(meta_path) as f:
return json.load(f)
except Exception:
pass
return {"references": {}}
def _save_metadata(self, printer_id: int, metadata: dict) -> None:
"""Save metadata for a printer's references."""
import json
meta_path = self._get_metadata_path(printer_id)
with open(meta_path, "w") as f:
json.dump(metadata, f, indent=2)
def _get_reference_paths(self, printer_id: int) -> list[Path]:
"""Get all existing reference image paths for a printer."""
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
paths = []
for i in range(self.MAX_REFERENCES):
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i}.jpg"
if path.exists():
paths.append(path)
return paths
def _get_next_reference_slot(self, printer_id: int) -> Path:
"""Get the path for the next reference image slot (cycles through slots)."""
CALIBRATION_DIR.mkdir(parents=True, exist_ok=True)
# Find first empty slot, or use oldest (slot 0) and shift others
for i in range(self.MAX_REFERENCES):
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i}.jpg"
if not path.exists():
return path
# All slots full - return slot 0 (will be overwritten, but we rotate first)
return CALIBRATION_DIR / f"printer_{printer_id}_ref_0.jpg"
def _rotate_references(self, printer_id: int) -> None:
"""Rotate references: delete oldest (0), shift others down."""
# Delete slot 0
slot0 = CALIBRATION_DIR / f"printer_{printer_id}_ref_0.jpg"
if slot0.exists():
slot0.unlink()
# Shift others down
for i in range(1, self.MAX_REFERENCES):
old_path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i}.jpg"
new_path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i - 1}.jpg"
if old_path.exists():
old_path.rename(new_path)
# Also rotate metadata
metadata = self._load_metadata(printer_id)
refs = metadata.get("references", {})
new_refs = {}
for i in range(1, self.MAX_REFERENCES):
if str(i) in refs:
new_refs[str(i - 1)] = refs[str(i)]
metadata["references"] = new_refs
self._save_metadata(printer_id, metadata)
def get_references(self, printer_id: int) -> list[dict]:
"""Get all references with metadata for a printer.
Returns list of dicts with: index, label, timestamp, has_image
"""
metadata = self._load_metadata(printer_id)
refs = metadata.get("references", {})
result = []
for i in range(self.MAX_REFERENCES):
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i}.jpg"
if path.exists():
ref_meta = refs.get(str(i), {})
result.append(
{
"index": i,
"label": ref_meta.get("label", ""),
"timestamp": ref_meta.get("timestamp", ""),
"has_image": True,
}
)
return result
def update_reference_label(self, printer_id: int, index: int, label: str) -> bool:
"""Update the label for a reference."""
if index < 0 or index >= self.MAX_REFERENCES:
return False
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{index}.jpg"
if not path.exists():
return False
metadata = self._load_metadata(printer_id)
if "references" not in metadata:
metadata["references"] = {}
if str(index) not in metadata["references"]:
metadata["references"][str(index)] = {}
metadata["references"][str(index)]["label"] = label
self._save_metadata(printer_id, metadata)
return True
def delete_reference(self, printer_id: int, index: int) -> bool:
"""Delete a specific reference by index."""
if index < 0 or index >= self.MAX_REFERENCES:
return False
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{index}.jpg"
if not path.exists():
return False
# Delete image
path.unlink()
# Remove from metadata
metadata = self._load_metadata(printer_id)
refs = metadata.get("references", {})
if str(index) in refs:
del refs[str(index)]
metadata["references"] = refs
self._save_metadata(printer_id, metadata)
# Shift remaining references down to fill the gap
for i in range(index + 1, self.MAX_REFERENCES):
old_img = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i}.jpg"
new_img = CALIBRATION_DIR / f"printer_{printer_id}_ref_{i - 1}.jpg"
if old_img.exists():
old_img.rename(new_img)
# Also shift metadata
if str(i) in refs:
refs[str(i - 1)] = refs[str(i)]
del refs[str(i)]
metadata["references"] = refs
self._save_metadata(printer_id, metadata)
return True
def get_reference_thumbnail(self, printer_id: int, index: int, max_size: int = 150) -> bytes | None:
"""Get a thumbnail of a reference image.
Returns JPEG bytes or None if not found.
"""
path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{index}.jpg"
if not path.exists():
return None
try:
img = cv2.imread(str(path))
if img is None:
return None
# Calculate thumbnail size maintaining aspect ratio
h, w = img.shape[:2]
if w > h:
new_w = max_size
new_h = int(h * max_size / w)
else:
new_h = max_size
new_w = int(w * max_size / h)
thumb = cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_AREA)
_, buffer = cv2.imencode(".jpg", thumb, [cv2.IMWRITE_JPEG_QUALITY, 80])
return buffer.tobytes()
except Exception as e:
logger.error(f"Error creating thumbnail: {e}")
return None
def _extract_roi(self, frame: np.ndarray) -> tuple[np.ndarray, int, int, int, int]:
"""Extract the region of interest from a frame.
Returns:
Tuple of (roi_frame, x_start, y_start, roi_width, roi_height)
"""
height, width = frame.shape[:2]
x_start = int(width * self.roi[0])
y_start = int(height * self.roi[1])
roi_width = int(width * self.roi[2])
roi_height = int(height * self.roi[3])
roi_frame = frame[y_start : y_start + roi_height, x_start : x_start + roi_width]
return roi_frame, x_start, y_start, roi_width, roi_height
def _preprocess_for_comparison(self, frame: np.ndarray) -> np.ndarray:
"""Preprocess a frame for comparison.
Uses heavy blur to create "blob" representation - smooths out texture
and noise while preserving large objects. Then normalizes brightness
to reduce lighting sensitivity.
"""
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
# Very heavy blur to smooth texture, keep only large shapes
blurred = cv2.GaussianBlur(gray, (51, 51), 0)
# Normalize to 0-255 range to reduce brightness sensitivity
normalized = cv2.normalize(blurred, None, 0, 255, cv2.NORM_MINMAX)
return normalized
def calibrate(self, image_data: bytes, printer_id: int, label: str | None = None) -> tuple[bool, str, int]:
"""Calibrate by saving a reference image of the empty plate.
Stores up to MAX_REFERENCES (5) images per printer. When all slots are full,
the oldest reference is removed and others are shifted.
Args:
image_data: JPEG image data as bytes
printer_id: Printer database ID
label: Optional label for this reference (e.g., "High Temp Plate")
Returns:
Tuple of (success, message, index) where index is the slot used
"""
from datetime import datetime
try:
# Decode image
nparr = np.frombuffer(image_data, np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if frame is None:
return False, "Failed to decode image", -1
# Get existing references count
existing_refs = self._get_reference_paths(printer_id)
num_existing = len(existing_refs)
# If all slots are full, rotate (remove oldest)
if num_existing >= self.MAX_REFERENCES:
self._rotate_references(printer_id)
num_existing = self.MAX_REFERENCES - 1
# Save to next available slot
slot_index = num_existing
reference_path = CALIBRATION_DIR / f"printer_{printer_id}_ref_{slot_index}.jpg"
cv2.imwrite(str(reference_path), frame, [cv2.IMWRITE_JPEG_QUALITY, 95])
# Save metadata
metadata = self._load_metadata(printer_id)
if "references" not in metadata:
metadata["references"] = {}
metadata["references"][str(slot_index)] = {
"label": label or "",
"timestamp": datetime.now().isoformat(),
}
self._save_metadata(printer_id, metadata)
logger.info(
f"Saved plate calibration reference {slot_index + 1}/{self.MAX_REFERENCES} for printer {printer_id}"
)
return True, f"Calibration saved ({slot_index + 1}/{self.MAX_REFERENCES} references)", slot_index
except Exception as e:
logger.exception("Error during plate calibration")
return False, f"Calibration error: {e!s}", -1
def get_calibration_count(self, printer_id: int) -> int:
"""Get the number of calibration references for a printer."""
return len(self._get_reference_paths(printer_id))
def has_calibration(self, printer_id: int, plate_type: str | None = None) -> bool:
"""Check if a printer has any calibration reference images."""
return len(self._get_reference_paths(printer_id)) > 0
def delete_calibration(self, printer_id: int, plate_type: str | None = None) -> bool:
"""Delete all calibration reference images for a printer."""
paths = self._get_reference_paths(printer_id)
if not paths:
return False
for path in paths:
path.unlink()
logger.info(f"Deleted {len(paths)} plate calibration reference(s) for printer {printer_id}")
return True
def analyze_frame(
self, image_data: bytes, printer_id: int, plate_type: str | None = None, include_debug_image: bool = False
) -> PlateDetectionResult:
"""Analyze a camera frame to detect if the plate is empty.
Compares the current frame to all calibration reference images and uses
the best match (lowest difference) for the final result.
Args:
image_data: JPEG image data as bytes
printer_id: Printer database ID (for reference lookup)
plate_type: Unused - kept for API compatibility
include_debug_image: If True, include annotated image in result
Returns:
PlateDetectionResult with analysis results
"""
try:
# Check for calibration
reference_paths = self._get_reference_paths(printer_id)
if not reference_paths:
return PlateDetectionResult(
is_empty=True, # Default to empty when not calibrated
confidence=0.0,
difference_percent=0.0,
message="No calibration - please calibrate with empty plate first",
needs_calibration=True,
)
# Decode current image
nparr = np.frombuffer(image_data, np.uint8)
current_frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if current_frame is None:
return PlateDetectionResult(
is_empty=True,
confidence=0.0,
difference_percent=0.0,
message="Failed to decode current image",
)
# Extract ROI from current frame
current_roi, x_start, y_start, roi_width, roi_height = self._extract_roi(current_frame)
current_processed = self._preprocess_for_comparison(current_roi)
# Compare against all references, find best match (lowest difference)
best_difference_percent = float("inf")
best_ref_idx = -1
best_diff = None
for idx, ref_path in enumerate(reference_paths):
# Load reference image
reference_frame = cv2.imread(str(ref_path), cv2.IMREAD_COLOR)
if reference_frame is None:
continue
# Ensure same dimensions
if current_frame.shape != reference_frame.shape:
reference_frame = cv2.resize(reference_frame, (current_frame.shape[1], current_frame.shape[0]))
# Extract ROI and preprocess
reference_roi, _, _, _, _ = self._extract_roi(reference_frame)
reference_processed = self._preprocess_for_comparison(reference_roi)
# Calculate absolute difference
diff = cv2.absdiff(current_processed, reference_processed)
# Calculate mean difference as percentage
mean_diff = np.mean(diff)
difference_percent = (mean_diff / 255.0) * 100
if difference_percent < best_difference_percent:
best_difference_percent = difference_percent
best_ref_idx = idx
best_diff = diff
if best_ref_idx == -1:
return PlateDetectionResult(
is_empty=True,
confidence=0.0,
difference_percent=0.0,
message="Failed to load any reference images - please recalibrate",
needs_calibration=True,
)
difference_percent = best_difference_percent
# Determine if plate is empty (use best match)
is_empty = difference_percent < self.difference_threshold
# Calculate confidence
if is_empty:
# Higher confidence when very little difference
confidence = 1.0 - min(1.0, difference_percent / self.difference_threshold)
else:
# Higher confidence when clearly different
confidence = min(1.0, difference_percent / (self.difference_threshold * 2))
# Generate message
num_refs = len(reference_paths)
if is_empty:
message = (
f"Plate appears empty (difference: {difference_percent:.1f}%, ref {best_ref_idx + 1}/{num_refs})"
)
else:
message = f"Objects detected on plate (difference: {difference_percent:.1f}%, best ref {best_ref_idx + 1}/{num_refs})"
# Generate debug image if requested
debug_image = None
if include_debug_image and best_diff is not None:
debug_frame = current_frame.copy()
# Draw ROI rectangle
cv2.rectangle(
debug_frame,
(x_start, y_start),
(x_start + roi_width, y_start + roi_height),
(0, 255, 0),
2,
)
# Create colored difference overlay
# Red = areas that are different from reference
# Amplify diff for visibility (multiply by 3, cap at 255)
diff_amplified = np.minimum(best_diff * 3, 255).astype(np.uint8)
diff_colored = cv2.cvtColor(diff_amplified, cv2.COLOR_GRAY2BGR)
diff_colored[:, :, 0] = 0 # Remove blue
diff_colored[:, :, 1] = 0 # Remove green
# Red channel has the diff
# Overlay difference on ROI
roi_overlay = debug_frame[y_start : y_start + roi_height, x_start : x_start + roi_width]
cv2.addWeighted(diff_colored, 0.5, roi_overlay, 0.5, 0, roi_overlay)
# Add status text
status_text = "EMPTY" if is_empty else "OBJECTS DETECTED"
color = (0, 255, 0) if is_empty else (0, 0, 255)
cv2.putText(debug_frame, status_text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, color, 2)
cv2.putText(
debug_frame,
f"Diff: {difference_percent:.1f}% (ref {best_ref_idx + 1}/{num_refs})",
(10, 60),
cv2.FONT_HERSHEY_SIMPLEX,
0.7,
color,
2,
)
cv2.putText(
debug_frame,
f"Confidence: {confidence:.0%}",
(10, 90),
cv2.FONT_HERSHEY_SIMPLEX,
0.7,
color,
2,
)
# Encode debug image as JPEG
_, buffer = cv2.imencode(".jpg", debug_frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
debug_image = buffer.tobytes()
return PlateDetectionResult(
is_empty=is_empty,
confidence=confidence,
difference_percent=difference_percent,
message=message,
debug_image=debug_image,
)
except Exception as e:
logger.exception("Error analyzing frame for plate detection")
return PlateDetectionResult(
is_empty=True, # Default to empty on error (don't block prints)
confidence=0.0,
difference_percent=0.0,
message=f"Analysis error: {e!s}",
)
async def capture_camera_image(
printer_id: int,
ip_address: str,
access_code: str,
model: str,
external_camera_url: str | None = None,
external_camera_type: str | None = None,
use_external: bool = False,
) -> tuple[bytes | None, str]:
"""Capture an image from the printer camera.
If there's an active camera stream, uses the buffered frame instead of
creating a new connection (which would fail while stream is active).
Returns:
Tuple of (image_data, camera_source) or (None, error_message)
"""
image_data: bytes | None = None
camera_source = "unknown"
# Try external camera first if requested and available
if use_external and external_camera_url and external_camera_type:
try:
from backend.app.services.external_camera import capture_frame
image_data = await capture_frame(external_camera_url, external_camera_type)
if image_data:
camera_source = "external"
logger.debug(f"Captured frame from external camera for printer {printer_id}")
except Exception as e:
logger.warning(f"Failed to capture from external camera: {e}")
# Fall back to built-in camera
if image_data is None:
# First, check if there's an active stream with a buffered frame
# This avoids blocking when camera viewer is open
try:
from backend.app.api.routes.camera import get_buffered_frame
buffered = get_buffered_frame(printer_id)
if buffered:
image_data = buffered
camera_source = "built-in (buffered)"
logger.debug(f"Using buffered frame from active stream for printer {printer_id}")
except Exception as e:
logger.debug(f"Could not get buffered frame: {e}")
# If no buffered frame, try to capture a new one
if image_data is None:
import tempfile
from backend.app.services.camera import capture_camera_frame
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
tmp_path = Path(tmp.name)
try:
success = await capture_camera_frame(ip_address, access_code, model, tmp_path, timeout=10)
if success:
with open(tmp_path, "rb") as f:
image_data = f.read()
camera_source = "built-in"
logger.debug(f"Captured frame from built-in camera for printer {printer_id}")
finally:
try:
tmp_path.unlink()
except Exception:
pass
return image_data, camera_source
async def check_plate_empty(
printer_id: int,
ip_address: str,
access_code: str,
model: str,
plate_type: str | None = None,
include_debug_image: bool = False,
external_camera_url: str | None = None,
external_camera_type: str | None = None,
use_external: bool = False,
roi: tuple[float, float, float, float] | None = None,
) -> PlateDetectionResult:
"""Check if the build plate is empty for a printer.
Args:
printer_id: Printer database ID
ip_address: Printer IP address
access_code: Printer access code
model: Printer model string
plate_type: Type of build plate for calibration lookup
include_debug_image: If True, include annotated image in result
external_camera_url: URL of external camera (if configured)
external_camera_type: Type of external camera (mjpeg, rtsp, snapshot)
use_external: If True, prefer external camera over built-in
roi: Region of interest as (x%, y%, w%, h%) - percentages of image size
Returns:
PlateDetectionResult with analysis results
"""
if not OPENCV_AVAILABLE:
return PlateDetectionResult(
is_empty=True,
confidence=0.0,
difference_percent=0.0,
message="OpenCV not available - plate detection disabled",
)
image_data, camera_source = await capture_camera_image(
printer_id, ip_address, access_code, model, external_camera_url, external_camera_type, use_external
)
if image_data is None:
return PlateDetectionResult(
is_empty=True, # Default to empty on error
confidence=0.0,
difference_percent=0.0,
message="Failed to capture camera frame from any source",
)
# Analyze the captured frame
detector = PlateDetector(roi=roi)
result = detector.analyze_frame(image_data, printer_id, plate_type, include_debug_image)
# Add camera source to message
result.message = f"[{camera_source}] {result.message}"
return result
async def calibrate_plate(
printer_id: int,
ip_address: str,
access_code: str,
model: str,
label: str | None = None,
external_camera_url: str | None = None,
external_camera_type: str | None = None,
use_external: bool = False,
) -> tuple[bool, str, int]:
"""Calibrate plate detection by capturing a reference image of the empty plate.
Args:
printer_id: Printer database ID
ip_address: Printer IP address
access_code: Printer access code
model: Printer model string
label: Optional label for this reference (e.g., "High Temp Plate")
external_camera_url: URL of external camera (if configured)
external_camera_type: Type of external camera (mjpeg, rtsp, snapshot)
use_external: If True, prefer external camera over built-in
Returns:
Tuple of (success, message, index)
"""
if not OPENCV_AVAILABLE:
return False, "OpenCV not available - plate detection disabled", -1
image_data, camera_source = await capture_camera_image(
printer_id, ip_address, access_code, model, external_camera_url, external_camera_type, use_external
)
if image_data is None:
return False, "Failed to capture camera frame for calibration", -1
detector = PlateDetector()
success, message, index = detector.calibrate(image_data, printer_id, label)
if success:
message = f"[{camera_source}] {message}"
return success, message, index
def get_calibration_status(printer_id: int, plate_type: str | None = None) -> dict:
"""Get calibration status for a printer.
Returns:
Dict with calibration info including reference count
"""
if not OPENCV_AVAILABLE:
return {
"available": False,
"calibrated": False,
"reference_count": 0,
"max_references": 5,
"message": "OpenCV not available",
}
detector = PlateDetector()
calibrated = detector.has_calibration(printer_id)
ref_count = detector.get_calibration_count(printer_id)
if calibrated:
message = f"Calibrated with {ref_count}/{detector.MAX_REFERENCES} reference(s)"
else:
message = "Not calibrated - please calibrate with empty plate"
return {
"available": True,
"calibrated": calibrated,
"reference_count": ref_count,
"max_references": detector.MAX_REFERENCES,
"message": message,
}
def delete_calibration(printer_id: int, plate_type: str | None = None) -> bool:
"""Delete calibration for a printer and plate type."""
if not OPENCV_AVAILABLE:
return False
detector = PlateDetector()
return detector.delete_calibration(printer_id, plate_type)
def is_plate_detection_available() -> bool:
"""Check if plate detection feature is available (OpenCV installed)."""
return OPENCV_AVAILABLE
backend/tests/integration/test_camera_api.py
+185
View File
@@ -223,3 +223,188 @@ class TestCameraAPI:
)
# Response will be a streaming response with error
assert response.status_code == 200
# ========================================================================
# Plate Detection Endpoints
# ========================================================================
@pytest.mark.asyncio
@pytest.mark.integration
async def test_plate_detection_status_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when checking plate detection status for non-existent printer."""
response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/status")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_plate_detection_status_opencv_not_available(self, async_client: AsyncClient, printer_factory):
"""Verify plate detection status returns unavailable when OpenCV not installed."""
printer = await printer_factory()
with patch("backend.app.services.plate_detection.OPENCV_AVAILABLE", False):
response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/status")
assert response.status_code == 200
result = response.json()
assert result["available"] is False
assert result["calibrated"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_plate_detection_status_success(self, async_client: AsyncClient, printer_factory):
"""Verify plate detection status returns correctly when OpenCV available."""
printer = await printer_factory()
# OpenCV is available in test environment, just check the response structure
response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/status")
assert response.status_code == 200
result = response.json()
assert "available" in result
assert "calibrated" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_check_plate_empty_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when checking plate for non-existent printer."""
response = await async_client.get("/api/v1/printers/99999/camera/check-plate")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_check_plate_empty_success_structure(self, async_client: AsyncClient, printer_factory):
"""Verify check plate returns proper structure when OpenCV available."""
printer = await printer_factory()
# Mock PlateDetectionResult to avoid camera timeout
mock_result = MagicMock()
mock_result.is_empty = True
mock_result.confidence = 0.95
mock_result.difference_percent = 0.5
mock_result.message = "Plate appears empty"
mock_result.needs_calibration = False
mock_result.to_dict.return_value = {
"is_empty": True,
"confidence": 0.95,
"difference_percent": 0.5,
"message": "Plate appears empty",
"has_debug_image": False,
"needs_calibration": False,
}
with patch("backend.app.services.plate_detection.check_plate_empty", new_callable=AsyncMock) as mock_check:
mock_check.return_value = mock_result
response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/check-plate")
assert response.status_code == 200
result = response.json()
assert "is_empty" in result
assert "confidence" in result
assert "message" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_calibrate_plate_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when calibrating plate for non-existent printer."""
response = await async_client.post("/api/v1/printers/99999/camera/plate-detection/calibrate")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_calibrate_plate_success_structure(self, async_client: AsyncClient, printer_factory):
"""Verify calibrate endpoint responds with proper structure."""
printer = await printer_factory()
# Mock calibrate_plate at the source module to avoid camera timeout
with patch("backend.app.services.plate_detection.calibrate_plate", new_callable=AsyncMock) as mock_calibrate:
mock_calibrate.return_value = (True, "Calibration saved (1/5 references)", 0)
response = await async_client.post(f"/api/v1/printers/{printer.id}/camera/plate-detection/calibrate")
assert response.status_code == 200
result = response.json()
assert result["success"] is True
assert "index" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_calibration_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when deleting calibration for non-existent printer."""
response = await async_client.delete("/api/v1/printers/99999/camera/plate-detection/calibrate")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_calibration_success(self, async_client: AsyncClient, printer_factory):
"""Verify delete calibration returns proper structure."""
printer = await printer_factory()
response = await async_client.delete(f"/api/v1/printers/{printer.id}/camera/plate-detection/calibrate")
assert response.status_code == 200
result = response.json()
assert "success" in result
assert "message" in result
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_references_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when getting references for non-existent printer."""
response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_references_opencv_not_available(self, async_client: AsyncClient, printer_factory):
"""Verify get references returns unavailable when OpenCV not installed."""
printer = await printer_factory()
with patch("backend.app.services.plate_detection.OPENCV_AVAILABLE", False):
response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/references")
assert response.status_code == 503
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_references_success(self, async_client: AsyncClient, printer_factory):
"""Verify get references returns proper structure."""
printer = await printer_factory()
# OpenCV is available in test environment, just check the response structure
response = await async_client.get(f"/api/v1/printers/{printer.id}/camera/plate-detection/references")
assert response.status_code == 200
result = response.json()
assert "references" in result
assert "max_references" in result
assert isinstance(result["references"], list)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_update_reference_label_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when updating reference label for non-existent printer."""
response = await async_client.put(
"/api/v1/printers/99999/camera/plate-detection/references/0", params={"label": "New Label"}
)
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_delete_reference_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when deleting reference for non-existent printer."""
response = await async_client.delete("/api/v1/printers/99999/camera/plate-detection/references/0")
assert response.status_code == 404
@pytest.mark.asyncio
@pytest.mark.integration
async def test_get_reference_thumbnail_printer_not_found(self, async_client: AsyncClient):
"""Verify 404 when getting reference thumbnail for non-existent printer."""
response = await async_client.get("/api/v1/printers/99999/camera/plate-detection/references/0/thumbnail")
assert response.status_code == 404
backend/tests/unit/services/test_plate_detection.py
+185
View File
@@ -0,0 +1,185 @@
"""Unit tests for plate detection service."""
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Mock cv2 and numpy before importing the module
cv2_mock = MagicMock()
np_mock = MagicMock()
class TestPlateDetectionResult:
"""Tests for PlateDetectionResult class."""
def test_result_to_dict(self):
"""Verify PlateDetectionResult.to_dict() returns correct structure."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
from backend.app.services.plate_detection import PlateDetectionResult
result = PlateDetectionResult(
is_empty=True,
confidence=0.95,
difference_percent=0.5,
message="Test message",
debug_image=None,
needs_calibration=False,
)
d = result.to_dict()
assert d["is_empty"] is True
assert d["confidence"] == 0.95
assert d["difference_percent"] == 0.5
assert d["message"] == "Test message"
assert d["has_debug_image"] is False
assert d["needs_calibration"] is False
def test_result_with_debug_image(self):
"""Verify has_debug_image is True when debug_image is provided."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
from backend.app.services.plate_detection import PlateDetectionResult
result = PlateDetectionResult(
is_empty=False,
confidence=0.8,
difference_percent=5.0,
message="Objects detected",
debug_image=b"fake_image_data",
needs_calibration=False,
)
d = result.to_dict()
assert d["has_debug_image"] is True
def test_result_needs_calibration(self):
"""Verify needs_calibration flag is preserved."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
from backend.app.services.plate_detection import PlateDetectionResult
result = PlateDetectionResult(
is_empty=True,
confidence=0.0,
difference_percent=0.0,
message="No calibration",
needs_calibration=True,
)
d = result.to_dict()
assert d["needs_calibration"] is True
class TestPlateDetector:
"""Tests for PlateDetector class."""
def test_detector_initialization(self):
"""Verify PlateDetector initializes with default values."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
# Re-import to get fresh module
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
# Mock OPENCV_AVAILABLE
pd_module.OPENCV_AVAILABLE = True
detector = pd_module.PlateDetector()
assert detector.roi == (0.15, 0.35, 0.70, 0.55)
assert detector.difference_threshold == 1.0
def test_detector_custom_roi(self):
"""Verify PlateDetector accepts custom ROI."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = True
custom_roi = (0.1, 0.2, 0.8, 0.6)
detector = pd_module.PlateDetector(roi=custom_roi)
assert detector.roi == custom_roi
def test_detector_raises_without_opencv(self):
"""Verify PlateDetector raises when OpenCV not available."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = False
with pytest.raises(RuntimeError, match="OpenCV is not installed"):
pd_module.PlateDetector()
class TestCalibrationStatus:
"""Tests for calibration status functions."""
def test_get_calibration_status_no_opencv(self):
"""Verify calibration status when OpenCV not available."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = False
status = pd_module.get_calibration_status(1)
assert status["available"] is False
assert status["calibrated"] is False
assert status["reference_count"] == 0
assert "OpenCV not available" in status["message"]
def test_is_plate_detection_available_true(self):
"""Verify is_plate_detection_available returns True when OpenCV available."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = True
assert pd_module.is_plate_detection_available() is True
def test_is_plate_detection_available_false(self):
"""Verify is_plate_detection_available returns False when OpenCV not available."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = False
assert pd_module.is_plate_detection_available() is False
class TestDeleteCalibration:
"""Tests for delete_calibration function."""
def test_delete_calibration_no_opencv(self):
"""Verify delete_calibration returns False when OpenCV not available."""
with patch.dict("sys.modules", {"cv2": cv2_mock, "numpy": np_mock}):
import importlib
import backend.app.services.plate_detection as pd_module
importlib.reload(pd_module)
pd_module.OPENCV_AVAILABLE = False
result = pd_module.delete_calibration(1)
assert result is False