mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-20 17:50:12 +02:00
Add build plate empty detection feature
Automatically detect if objects are on the build plate before printing and pause the print immediately if detected. Features: - Per-printer toggle to enable/disable plate detection - Multi-reference calibration: store up to 5 reference images per printer for different plate types (textured, smooth, high-temp, etc.) - Automatic print pause when objects detected at print start - Push notification and WebSocket alert when print is paused - ROI (Region of Interest) calibration UI with sliders to adjust detection area - Reference management: view thumbnails, add labels, delete references - Works with both built-in and external cameras - Uses buffered camera frames when stream is active (no blocking) - Split button UI: main button opens modal, chevron toggles on/off - Green visual indicator when plate detection is enabled - Included in backup/restore
This commit is contained in:
@@ -2,6 +2,22 @@
|
||||
|
||||
All notable changes to Bambuddy will be documented in this file.
|
||||
|
||||
## [0.1.7] - 2026-01-26
|
||||
|
||||
### New Features
|
||||
- **Build Plate Empty Detection** - Automatically detect if objects are on the build plate before printing:
|
||||
- Per-printer toggle to enable/disable plate detection
|
||||
- Multi-reference calibration: Store up to 5 reference images of empty plates (different plate types)
|
||||
- Automatic print pause when objects detected on plate at print start
|
||||
- Push notification and WebSocket alert when print is paused due to plate detection
|
||||
- ROI (Region of Interest) calibration UI with sliders to focus detection on build plate area
|
||||
- Reference management: View thumbnails, add labels, delete references
|
||||
- Works with both built-in and external cameras
|
||||
- Uses buffered camera frames when stream is active (no blocking)
|
||||
- Split button UI: Main button opens calibration modal, chevron toggles detection on/off
|
||||
- Green visual indicator when plate detection is enabled
|
||||
- Included in backup/restore
|
||||
|
||||
## [0.1.6] - 2026-01-24
|
||||
|
||||
### New Features
|
||||
|
||||
@@ -58,6 +58,7 @@
|
||||
- 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
|
||||
- **Build plate empty detection** - Auto-pause print if objects detected on plate (multi-reference calibration, ROI adjustment)
|
||||
- Fan status monitoring (part cooling, auxiliary, chamber)
|
||||
- Printer control (stop, pause, resume, chamber light)
|
||||
- Resizable printer cards (S/M/L/XL)
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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())
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"references": {
|
||||
"0": {
|
||||
"label": "CyroGrip",
|
||||
"timestamp": "2026-01-26T12:06:09.612288"
|
||||
},
|
||||
"1": {
|
||||
"label": "Carbon",
|
||||
"timestamp": "2026-01-26T12:07:01.878776"
|
||||
},
|
||||
"2": {
|
||||
"label": "Textured PEI",
|
||||
"timestamp": "2026-01-26T12:07:31.460596"
|
||||
},
|
||||
"3": {
|
||||
"label": "Wham Bam",
|
||||
"timestamp": "2026-01-26T12:07:53.889845"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"references": {
|
||||
"0": {
|
||||
"label": "Textured PEI",
|
||||
"timestamp": "2026-01-26T12:11:30.814062"
|
||||
},
|
||||
"1": {
|
||||
"label": "",
|
||||
"timestamp": "2026-01-26T12:34:31.393095"
|
||||
}
|
||||
}
|
||||
}
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 123 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 118 KiB |
@@ -122,4 +122,65 @@ describe('Layout', () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('plate detection alert modal', () => {
|
||||
it('shows modal when plate-not-empty event is dispatched', async () => {
|
||||
render(<Layout />);
|
||||
|
||||
// Dispatch the plate-not-empty event
|
||||
window.dispatchEvent(
|
||||
new CustomEvent('plate-not-empty', {
|
||||
detail: {
|
||||
printer_id: 1,
|
||||
printer_name: 'Test Printer',
|
||||
message: 'Objects detected on build plate',
|
||||
},
|
||||
})
|
||||
);
|
||||
|
||||
await waitFor(() => {
|
||||
// Modal should appear with "Print Paused!" text
|
||||
expect(document.body.textContent).toContain('Print Paused!');
|
||||
expect(document.body.textContent).toContain('Test Printer');
|
||||
});
|
||||
});
|
||||
|
||||
it('closes modal when I Understand button is clicked', async () => {
|
||||
render(<Layout />);
|
||||
|
||||
// Dispatch the plate-not-empty event
|
||||
window.dispatchEvent(
|
||||
new CustomEvent('plate-not-empty', {
|
||||
detail: {
|
||||
printer_id: 1,
|
||||
printer_name: 'Test Printer',
|
||||
message: 'Objects detected on build plate',
|
||||
},
|
||||
})
|
||||
);
|
||||
|
||||
await waitFor(() => {
|
||||
expect(document.body.textContent).toContain('Print Paused!');
|
||||
});
|
||||
|
||||
// Click the "I Understand" button
|
||||
const button = document.querySelector('button');
|
||||
if (button && button.textContent?.includes('I Understand')) {
|
||||
button.click();
|
||||
}
|
||||
|
||||
// Find and click the "I Understand" button by searching all buttons
|
||||
const buttons = document.querySelectorAll('button');
|
||||
buttons.forEach((btn) => {
|
||||
if (btn.textContent?.includes('I Understand')) {
|
||||
btn.click();
|
||||
}
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
// Modal should be closed
|
||||
expect(document.body.textContent).not.toContain('Print Paused!');
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -68,6 +68,8 @@ export interface Printer {
|
||||
external_camera_url: string | null;
|
||||
external_camera_type: string | null; // "mjpeg", "rtsp", "snapshot"
|
||||
external_camera_enabled: boolean;
|
||||
plate_detection_enabled: boolean; // Check plate before print
|
||||
plate_detection_roi?: PlateDetectionROI; // ROI for plate detection
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
}
|
||||
@@ -215,6 +217,51 @@ export interface PrinterCreate {
|
||||
external_camera_url?: string | null;
|
||||
external_camera_type?: string | null;
|
||||
external_camera_enabled?: boolean;
|
||||
plate_detection_enabled?: boolean;
|
||||
plate_detection_roi?: PlateDetectionROI;
|
||||
}
|
||||
|
||||
// Plate Detection
|
||||
export interface PlateDetectionROI {
|
||||
x: number; // X start % (0.0-1.0)
|
||||
y: number; // Y start % (0.0-1.0)
|
||||
w: number; // Width % (0.0-1.0)
|
||||
h: number; // Height % (0.0-1.0)
|
||||
}
|
||||
|
||||
export interface PlateDetectionResult {
|
||||
is_empty: boolean;
|
||||
confidence: number;
|
||||
difference_percent: number;
|
||||
message: string;
|
||||
has_debug_image: boolean;
|
||||
debug_image_url?: string;
|
||||
needs_calibration: boolean;
|
||||
light_warning?: boolean;
|
||||
reference_count?: number;
|
||||
max_references?: number;
|
||||
roi?: PlateDetectionROI;
|
||||
}
|
||||
|
||||
export interface PlateDetectionStatus {
|
||||
available: boolean;
|
||||
calibrated: boolean;
|
||||
reference_count: number;
|
||||
max_references: number;
|
||||
message: string;
|
||||
}
|
||||
|
||||
export interface CalibrationResult {
|
||||
success: boolean;
|
||||
message: string;
|
||||
}
|
||||
|
||||
export interface PlateReference {
|
||||
index: number;
|
||||
label: string;
|
||||
timestamp: string;
|
||||
has_image: boolean;
|
||||
thumbnail_url: string;
|
||||
}
|
||||
|
||||
// Archive types
|
||||
@@ -2596,6 +2643,59 @@ export const api = {
|
||||
testCameraConnection: (printerId: number) =>
|
||||
request<{ success: boolean; message?: string; error?: string }>(`/printers/${printerId}/camera/test`),
|
||||
|
||||
// Plate Detection - Multi-reference calibration (stores up to 5 references per printer)
|
||||
checkPlateEmpty: (printerId: number, options?: { useExternal?: boolean; includeDebugImage?: boolean }) => {
|
||||
const params = new URLSearchParams();
|
||||
params.set('use_external', String(options?.useExternal ?? false));
|
||||
params.set('include_debug_image', String(options?.includeDebugImage ?? false));
|
||||
return request<PlateDetectionResult>(
|
||||
`/printers/${printerId}/camera/check-plate?${params.toString()}`
|
||||
);
|
||||
},
|
||||
getPlateDetectionStatus: (printerId: number) => {
|
||||
return request<PlateDetectionStatus & { chamber_light?: boolean }>(
|
||||
`/printers/${printerId}/camera/plate-detection/status`
|
||||
);
|
||||
},
|
||||
calibratePlateDetection: (printerId: number, options?: { label?: string; useExternal?: boolean }) => {
|
||||
const params = new URLSearchParams();
|
||||
if (options?.label) params.set('label', options.label);
|
||||
params.set('use_external', String(options?.useExternal ?? false));
|
||||
return request<CalibrationResult & { index: number }>(
|
||||
`/printers/${printerId}/camera/plate-detection/calibrate?${params.toString()}`,
|
||||
{ method: 'POST' }
|
||||
);
|
||||
},
|
||||
deletePlateCalibration: (printerId: number) => {
|
||||
return request<CalibrationResult>(
|
||||
`/printers/${printerId}/camera/plate-detection/calibrate`,
|
||||
{ method: 'DELETE' }
|
||||
);
|
||||
},
|
||||
getPlateReferences: (printerId: number) => {
|
||||
return request<{
|
||||
references: PlateReference[];
|
||||
max_references: number;
|
||||
}>(`/printers/${printerId}/camera/plate-detection/references`);
|
||||
},
|
||||
getPlateReferenceThumbnailUrl: (printerId: number, index: number) => {
|
||||
return `${API_BASE}/printers/${printerId}/camera/plate-detection/references/${index}/thumbnail`;
|
||||
},
|
||||
updatePlateReferenceLabel: (printerId: number, index: number, label: string) => {
|
||||
const params = new URLSearchParams();
|
||||
params.set('label', label);
|
||||
return request<{ success: boolean; index: number; label: string }>(
|
||||
`/printers/${printerId}/camera/plate-detection/references/${index}?${params.toString()}`,
|
||||
{ method: 'PUT' }
|
||||
);
|
||||
},
|
||||
deletePlateReference: (printerId: number, index: number) => {
|
||||
return request<{ success: boolean; message: string }>(
|
||||
`/printers/${printerId}/camera/plate-detection/references/${index}`,
|
||||
{ method: 'DELETE' }
|
||||
);
|
||||
},
|
||||
|
||||
// External Links
|
||||
getExternalLinks: () => request<ExternalLink[]>('/external-links/'),
|
||||
getExternalLink: (id: number) => request<ExternalLink>(`/external-links/${id}`),
|
||||
|
||||
@@ -84,6 +84,11 @@ export function Layout() {
|
||||
const [dismissedUpdateVersion, setDismissedUpdateVersion] = useState<string | null>(() =>
|
||||
sessionStorage.getItem('dismissedUpdateVersion')
|
||||
);
|
||||
const [plateDetectionAlert, setPlateDetectionAlert] = useState<{
|
||||
printer_id: number;
|
||||
printer_name: string;
|
||||
message: string;
|
||||
} | null>(null);
|
||||
|
||||
// Check for updates
|
||||
const { data: versionInfo } = useQuery({
|
||||
@@ -300,6 +305,20 @@ export function Layout() {
|
||||
}
|
||||
}, [location.pathname, isMobile]);
|
||||
|
||||
// Listen for plate detection warnings (objects on plate, print paused)
|
||||
useEffect(() => {
|
||||
const handlePlateNotEmpty = (event: Event) => {
|
||||
const detail = (event as CustomEvent).detail;
|
||||
setPlateDetectionAlert({
|
||||
printer_id: detail.printer_id,
|
||||
printer_name: detail.printer_name,
|
||||
message: detail.message,
|
||||
});
|
||||
};
|
||||
window.addEventListener('plate-not-empty', handlePlateNotEmpty);
|
||||
return () => window.removeEventListener('plate-not-empty', handlePlateNotEmpty);
|
||||
}, []);
|
||||
|
||||
// Global keyboard shortcuts for navigation
|
||||
const handleKeyDown = useCallback((e: KeyboardEvent) => {
|
||||
const target = e.target as HTMLElement;
|
||||
@@ -750,6 +769,37 @@ export function Layout() {
|
||||
}).filter(Boolean) as { type: 'nav' | 'external'; label: string; labelKey?: string }[]}
|
||||
/>
|
||||
)}
|
||||
|
||||
{/* Plate Detection Alert Modal */}
|
||||
{plateDetectionAlert && (
|
||||
<div className="fixed inset-0 bg-black/70 flex items-center justify-center z-[100] p-4">
|
||||
<div className="bg-bambu-dark-secondary border-2 border-yellow-500 rounded-xl shadow-2xl max-w-md w-full animate-in fade-in zoom-in duration-200">
|
||||
<div className="p-6 text-center">
|
||||
<div className="w-16 h-16 mx-auto mb-4 rounded-full bg-yellow-500/20 flex items-center justify-center">
|
||||
<svg className="w-10 h-10 text-yellow-500" fill="none" viewBox="0 0 24 24" stroke="currentColor" strokeWidth={2}>
|
||||
<path strokeLinecap="round" strokeLinejoin="round" d="M12 9v2m0 4h.01m-6.938 4h13.856c1.54 0 2.502-1.667 1.732-3L13.732 4c-.77-1.333-2.694-1.333-3.464 0L3.34 16c-.77 1.333.192 3 1.732 3z" />
|
||||
</svg>
|
||||
</div>
|
||||
<h2 className="text-xl font-bold text-yellow-400 mb-2">
|
||||
Print Paused!
|
||||
</h2>
|
||||
<p className="text-lg text-white mb-2">
|
||||
{plateDetectionAlert.printer_name}
|
||||
</p>
|
||||
<p className="text-bambu-gray mb-6">
|
||||
Objects detected on build plate. The print has been automatically paused.
|
||||
Please clear the plate and resume the print.
|
||||
</p>
|
||||
<button
|
||||
onClick={() => setPlateDetectionAlert(null)}
|
||||
className="w-full py-3 px-6 bg-yellow-500 hover:bg-yellow-600 text-black font-semibold rounded-lg transition-colors"
|
||||
>
|
||||
I Understand
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -214,6 +214,18 @@ export function useWebSocket() {
|
||||
case 'pong':
|
||||
// Keepalive response, ignore
|
||||
break;
|
||||
|
||||
case 'plate_not_empty':
|
||||
// Plate detection found objects - print was paused
|
||||
// Dispatch event for toast notification
|
||||
window.dispatchEvent(new CustomEvent('plate-not-empty', {
|
||||
detail: {
|
||||
printer_id: message.printer_id,
|
||||
printer_name: (message as unknown as { printer_name?: string }).printer_name,
|
||||
message: (message as unknown as { message?: string }).message,
|
||||
}
|
||||
}));
|
||||
break;
|
||||
}
|
||||
}, [queryClient, debouncedInvalidate, throttledPrinterStatusUpdate]);
|
||||
|
||||
|
||||
@@ -36,6 +36,9 @@ import {
|
||||
Wind,
|
||||
AirVent,
|
||||
Download,
|
||||
ScanSearch,
|
||||
CheckCircle,
|
||||
XCircle,
|
||||
} from 'lucide-react';
|
||||
|
||||
// Custom Skip Objects icon - arrow jumping over boxes
|
||||
@@ -965,6 +968,22 @@ function PrinterCard({
|
||||
trayInfoIdx?: string;
|
||||
} | null>(null);
|
||||
const [showFirmwareModal, setShowFirmwareModal] = useState(false);
|
||||
const [plateCheckResult, setPlateCheckResult] = useState<{
|
||||
is_empty: boolean;
|
||||
confidence: number;
|
||||
difference_percent: number;
|
||||
message: string;
|
||||
debug_image_url?: string;
|
||||
needs_calibration: boolean;
|
||||
light_warning?: boolean;
|
||||
reference_count?: number;
|
||||
max_references?: number;
|
||||
roi?: { x: number; y: number; w: number; h: number };
|
||||
} | null>(null);
|
||||
const [isCheckingPlate, setIsCheckingPlate] = useState(false);
|
||||
const [isCalibrating, setIsCalibrating] = useState(false);
|
||||
const [editingRoi, setEditingRoi] = useState<{ x: number; y: number; w: number; h: number } | null>(null);
|
||||
const [isSavingRoi, setIsSavingRoi] = useState(false);
|
||||
|
||||
const { data: status } = useQuery({
|
||||
queryKey: ['printerStatus', printer.id],
|
||||
@@ -1187,6 +1206,16 @@ function PrinterCard({
|
||||
},
|
||||
});
|
||||
|
||||
// Plate detection setting mutation
|
||||
const plateDetectionMutation = useMutation({
|
||||
mutationFn: (enabled: boolean) => api.updatePrinter(printer.id, { plate_detection_enabled: enabled }),
|
||||
onSuccess: () => {
|
||||
queryClient.invalidateQueries({ queryKey: ['printers'] });
|
||||
showToast(plateDetectionMutation.variables ? 'Plate check enabled' : 'Plate check disabled');
|
||||
},
|
||||
onError: (error: Error) => showToast(error.message || 'Failed to update setting', 'error'),
|
||||
});
|
||||
|
||||
// Query for printable objects (for skip functionality)
|
||||
// Fetch when printing with 2+ objects OR when modal is open
|
||||
const isPrintingWithObjects = (status?.state === 'RUNNING' || status?.state === 'PAUSE' || status?.state === 'PAUSED') && (status?.printable_objects_count ?? 0) >= 2;
|
||||
@@ -1250,6 +1279,118 @@ function PrinterCard({
|
||||
},
|
||||
});
|
||||
|
||||
// Plate references state
|
||||
const [plateReferences, setPlateReferences] = useState<{
|
||||
references: Array<{ index: number; label: string; timestamp: string; has_image: boolean; thumbnail_url: string }>;
|
||||
max_references: number;
|
||||
} | null>(null);
|
||||
const [editingRefLabel, setEditingRefLabel] = useState<{ index: number; label: string } | null>(null);
|
||||
|
||||
// Fetch plate references
|
||||
const fetchPlateReferences = async () => {
|
||||
try {
|
||||
const data = await api.getPlateReferences(printer.id);
|
||||
setPlateReferences(data);
|
||||
} catch {
|
||||
// Ignore errors - references will show as empty
|
||||
}
|
||||
};
|
||||
|
||||
// Toggle plate detection enabled/disabled
|
||||
const handleTogglePlateDetection = () => {
|
||||
plateDetectionMutation.mutate(!printer.plate_detection_enabled);
|
||||
};
|
||||
|
||||
// Open plate detection management modal (for calibration/references)
|
||||
const handleOpenPlateManagement = async () => {
|
||||
setIsCheckingPlate(true);
|
||||
setPlateCheckResult(null);
|
||||
try {
|
||||
const result = await api.checkPlateEmpty(printer.id, { includeDebugImage: true });
|
||||
setPlateCheckResult(result);
|
||||
fetchPlateReferences();
|
||||
} catch (error) {
|
||||
showToast(error instanceof Error ? error.message : 'Failed to check plate', 'error');
|
||||
} finally {
|
||||
setIsCheckingPlate(false);
|
||||
}
|
||||
};
|
||||
|
||||
// Calibrate plate detection handler
|
||||
const handleCalibratePlate = async (label?: string) => {
|
||||
setIsCalibrating(true);
|
||||
try {
|
||||
const result = await api.calibratePlateDetection(printer.id, { label });
|
||||
if (result.success) {
|
||||
showToast(result.message || 'Calibration saved!', 'success');
|
||||
// Refresh references and re-check
|
||||
fetchPlateReferences();
|
||||
const checkResult = await api.checkPlateEmpty(printer.id, { includeDebugImage: true });
|
||||
setPlateCheckResult(checkResult);
|
||||
} else {
|
||||
showToast(result.message || 'Calibration failed', 'error');
|
||||
}
|
||||
} catch (error) {
|
||||
showToast(error instanceof Error ? error.message : 'Calibration failed', 'error');
|
||||
} finally {
|
||||
setIsCalibrating(false);
|
||||
}
|
||||
};
|
||||
|
||||
// Update reference label
|
||||
const handleUpdateRefLabel = async (index: number, label: string) => {
|
||||
try {
|
||||
await api.updatePlateReferenceLabel(printer.id, index, label);
|
||||
setEditingRefLabel(null);
|
||||
fetchPlateReferences();
|
||||
} catch (error) {
|
||||
showToast(error instanceof Error ? error.message : 'Failed to update label', 'error');
|
||||
}
|
||||
};
|
||||
|
||||
// Delete reference
|
||||
const handleDeleteRef = async (index: number) => {
|
||||
try {
|
||||
await api.deletePlateReference(printer.id, index);
|
||||
showToast('Reference deleted', 'success');
|
||||
fetchPlateReferences();
|
||||
// Re-check to update counts
|
||||
const checkResult = await api.checkPlateEmpty(printer.id, { includeDebugImage: true });
|
||||
setPlateCheckResult(checkResult);
|
||||
} catch (error) {
|
||||
showToast(error instanceof Error ? error.message : 'Failed to delete reference', 'error');
|
||||
}
|
||||
};
|
||||
|
||||
// Save ROI settings
|
||||
const handleSaveRoi = async () => {
|
||||
if (!editingRoi) return;
|
||||
setIsSavingRoi(true);
|
||||
try {
|
||||
await api.updatePrinter(printer.id, { plate_detection_roi: editingRoi });
|
||||
showToast('Detection area saved', 'success');
|
||||
setEditingRoi(null);
|
||||
// Re-check to see new ROI in action
|
||||
const checkResult = await api.checkPlateEmpty(printer.id, { includeDebugImage: true });
|
||||
setPlateCheckResult(checkResult);
|
||||
} catch (error) {
|
||||
showToast(error instanceof Error ? error.message : 'Failed to save detection area', 'error');
|
||||
} finally {
|
||||
setIsSavingRoi(false);
|
||||
}
|
||||
};
|
||||
|
||||
// Close plate check modal on Escape key
|
||||
useEffect(() => {
|
||||
const handleEscape = (e: KeyboardEvent) => {
|
||||
if (e.key === 'Escape' && plateCheckResult) {
|
||||
setPlateCheckResult(null);
|
||||
}
|
||||
};
|
||||
window.addEventListener('keydown', handleEscape);
|
||||
return () => window.removeEventListener('keydown', handleEscape);
|
||||
}, [plateCheckResult]);
|
||||
|
||||
// Watch ams_status_main to detect when RFID read completes
|
||||
// ams_status_main: 0=idle, 2=rfid_identifying
|
||||
const deferredClearRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||
@@ -2560,6 +2701,37 @@ function PrinterCard({
|
||||
>
|
||||
<Video className="w-4 h-4" />
|
||||
</Button>
|
||||
{/* Split button: main part opens modal, chevron toggles */}
|
||||
<div className={`inline-flex rounded-md ${printer.plate_detection_enabled ? 'ring-1 ring-green-500' : ''}`}>
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="sm"
|
||||
onClick={handleOpenPlateManagement}
|
||||
disabled={!status?.connected || isCheckingPlate}
|
||||
title="Manage plate detection calibration"
|
||||
className={`!rounded-r-none !border-r-0 ${printer.plate_detection_enabled ? "!border-green-500 !text-green-400 hover:!bg-green-500/20" : ""}`}
|
||||
>
|
||||
{isCheckingPlate ? (
|
||||
<Loader2 className="w-4 h-4 animate-spin" />
|
||||
) : (
|
||||
<ScanSearch className="w-4 h-4" />
|
||||
)}
|
||||
</Button>
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="sm"
|
||||
onClick={handleTogglePlateDetection}
|
||||
disabled={!status?.connected || plateDetectionMutation.isPending}
|
||||
title={printer.plate_detection_enabled ? "Plate check enabled - Click to disable" : "Plate check disabled - Click to enable"}
|
||||
className={`!rounded-l-none !px-1.5 ${printer.plate_detection_enabled ? "!border-green-500 !text-green-400 hover:!bg-green-500/20" : ""}`}
|
||||
>
|
||||
{plateDetectionMutation.isPending ? (
|
||||
<Loader2 className="w-3 h-3 animate-spin" />
|
||||
) : (
|
||||
<ChevronDown className="w-3 h-3" />
|
||||
)}
|
||||
</Button>
|
||||
</div>
|
||||
<Button
|
||||
variant="secondary"
|
||||
size="sm"
|
||||
@@ -2592,6 +2764,287 @@ function PrinterCard({
|
||||
/>
|
||||
)}
|
||||
|
||||
{/* Plate Check Result Modal */}
|
||||
{plateCheckResult && (
|
||||
<div className="fixed inset-0 bg-black/50 flex items-center justify-center z-50 p-4" onClick={() => setPlateCheckResult(null)}>
|
||||
<div className="bg-bambu-dark-secondary border border-bambu-dark-tertiary rounded-xl shadow-2xl max-w-lg w-full" onClick={e => e.stopPropagation()}>
|
||||
<div className="flex items-center justify-between p-4 border-b border-bambu-dark-tertiary">
|
||||
<div className="flex items-center gap-2">
|
||||
{plateCheckResult.needs_calibration ? (
|
||||
<ScanSearch className="w-5 h-5 text-blue-500" />
|
||||
) : plateCheckResult.is_empty ? (
|
||||
<CheckCircle className="w-5 h-5 text-green-500" />
|
||||
) : (
|
||||
<XCircle className="w-5 h-5 text-yellow-500" />
|
||||
)}
|
||||
<h2 className="text-lg font-semibold text-white">
|
||||
Build Plate Check
|
||||
</h2>
|
||||
{plateCheckResult.reference_count !== undefined && plateCheckResult.max_references && (
|
||||
<span className="text-xs text-bambu-gray bg-bambu-dark-tertiary px-2 py-1 rounded">
|
||||
{plateCheckResult.reference_count}/{plateCheckResult.max_references} refs
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
<button
|
||||
onClick={() => setPlateCheckResult(null)}
|
||||
className="p-1 text-bambu-gray hover:text-white rounded transition-colors"
|
||||
>
|
||||
<X className="w-5 h-5" />
|
||||
</button>
|
||||
</div>
|
||||
<div className="p-4 space-y-4">
|
||||
{/* Light Warning */}
|
||||
{plateCheckResult.light_warning && (
|
||||
<div className="p-3 rounded-lg bg-yellow-500/20 border border-yellow-500/50">
|
||||
<p className="font-medium text-yellow-400 flex items-center gap-2">
|
||||
<AlertTriangle className="w-4 h-4" />
|
||||
Chamber light is OFF
|
||||
</p>
|
||||
<p className="text-sm text-bambu-gray mt-1">
|
||||
For reliable detection, please turn ON the chamber light before checking or calibrating.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{plateCheckResult.needs_calibration ? (
|
||||
<>
|
||||
<div className="p-3 rounded-lg bg-blue-500/20 border border-blue-500/50">
|
||||
<p className="font-medium text-blue-400">
|
||||
Calibration Required
|
||||
</p>
|
||||
<p className="text-sm text-bambu-gray mt-1">
|
||||
Please ensure the build plate is <strong>completely empty</strong> and chamber light is <strong>ON</strong>, then click Calibrate.
|
||||
</p>
|
||||
</div>
|
||||
<div className="text-sm text-bambu-gray space-y-2">
|
||||
<p>Calibration captures a reference image of the empty plate. Future checks will compare against this reference to detect objects.</p>
|
||||
<p><strong>Tip:</strong> You can store up to 5 calibrations for different plates. The system automatically uses the best match when checking.</p>
|
||||
</div>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<div className={`p-3 rounded-lg ${plateCheckResult.is_empty ? 'bg-green-500/20 border border-green-500/50' : 'bg-yellow-500/20 border border-yellow-500/50'}`}>
|
||||
<p className={`font-medium ${plateCheckResult.is_empty ? 'text-green-400' : 'text-yellow-400'}`}>
|
||||
{plateCheckResult.is_empty ? 'Plate appears empty' : 'Objects detected on plate'}
|
||||
</p>
|
||||
<p className="text-sm text-bambu-gray mt-1">
|
||||
Confidence: {Math.round(plateCheckResult.confidence * 100)}% | Difference: {plateCheckResult.difference_percent.toFixed(1)}%
|
||||
</p>
|
||||
</div>
|
||||
{plateCheckResult.debug_image_url && (
|
||||
<div>
|
||||
<p className="text-sm text-bambu-gray mb-2">Analysis preview:</p>
|
||||
<img
|
||||
src={plateCheckResult.debug_image_url}
|
||||
alt="Plate detection analysis"
|
||||
className="w-full rounded-lg border border-bambu-dark-tertiary"
|
||||
/>
|
||||
<p className="text-xs text-bambu-gray mt-2">
|
||||
Green box = detection area, Red overlay = differences from calibration
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
<p className="text-xs text-bambu-gray">
|
||||
{plateCheckResult.message}
|
||||
</p>
|
||||
</>
|
||||
)}
|
||||
|
||||
{/* Saved References Grid */}
|
||||
{plateReferences && plateReferences.references.length > 0 && (
|
||||
<div className="mt-4 pt-4 border-t border-bambu-dark-tertiary">
|
||||
<p className="text-sm font-medium text-white mb-2">
|
||||
Saved References ({plateReferences.references.length}/{plateReferences.max_references})
|
||||
</p>
|
||||
<div className="grid grid-cols-5 gap-2">
|
||||
{plateReferences.references.map((ref) => (
|
||||
<div key={ref.index} className="relative group">
|
||||
<img
|
||||
src={api.getPlateReferenceThumbnailUrl(printer.id, ref.index)}
|
||||
alt={ref.label || `Reference ${ref.index + 1}`}
|
||||
className="w-full aspect-video object-cover rounded border border-bambu-dark-tertiary"
|
||||
/>
|
||||
{/* Delete button */}
|
||||
<button
|
||||
onClick={() => handleDeleteRef(ref.index)}
|
||||
className="absolute top-1 right-1 p-0.5 bg-red-500/80 rounded opacity-0 group-hover:opacity-100 transition-opacity"
|
||||
title="Delete reference"
|
||||
>
|
||||
<X className="w-3 h-3 text-white" />
|
||||
</button>
|
||||
{/* Label */}
|
||||
{editingRefLabel?.index === ref.index ? (
|
||||
<input
|
||||
type="text"
|
||||
value={editingRefLabel.label}
|
||||
onChange={(e) => setEditingRefLabel({ ...editingRefLabel, label: e.target.value })}
|
||||
onBlur={() => handleUpdateRefLabel(ref.index, editingRefLabel.label)}
|
||||
onKeyDown={(e) => {
|
||||
if (e.key === 'Enter') handleUpdateRefLabel(ref.index, editingRefLabel.label);
|
||||
if (e.key === 'Escape') setEditingRefLabel(null);
|
||||
}}
|
||||
className="w-full mt-1 px-1 py-0.5 text-xs bg-bambu-dark-tertiary border border-bambu-green rounded text-white"
|
||||
autoFocus
|
||||
placeholder="Label..."
|
||||
/>
|
||||
) : (
|
||||
<p
|
||||
className="text-xs text-bambu-gray mt-1 truncate cursor-pointer hover:text-white"
|
||||
onClick={() => setEditingRefLabel({ index: ref.index, label: ref.label })}
|
||||
title={ref.label ? `${ref.label} - Click to edit` : 'Click to add label'}
|
||||
>
|
||||
{ref.label || <span className="italic opacity-50">No label</span>}
|
||||
</p>
|
||||
)}
|
||||
{/* Timestamp */}
|
||||
<p className="text-[10px] text-bambu-gray/60">
|
||||
{ref.timestamp ? new Date(ref.timestamp).toLocaleDateString() : ''}
|
||||
</p>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* ROI Editor */}
|
||||
{!plateCheckResult.needs_calibration && (
|
||||
<div className="mt-4 pt-4 border-t border-bambu-dark-tertiary">
|
||||
<div className="flex items-center justify-between mb-2">
|
||||
<p className="text-sm font-medium text-white">Detection Area (ROI)</p>
|
||||
{!editingRoi ? (
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
onClick={() => setEditingRoi(plateCheckResult.roi || { x: 0.15, y: 0.35, w: 0.70, h: 0.55 })}
|
||||
>
|
||||
<Pencil className="w-3 h-3 mr-1" />
|
||||
Edit
|
||||
</Button>
|
||||
) : (
|
||||
<div className="flex gap-1">
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
onClick={() => setEditingRoi(null)}
|
||||
disabled={isSavingRoi}
|
||||
>
|
||||
Cancel
|
||||
</Button>
|
||||
<Button
|
||||
size="sm"
|
||||
onClick={handleSaveRoi}
|
||||
disabled={isSavingRoi}
|
||||
>
|
||||
{isSavingRoi ? <Loader2 className="w-3 h-3 animate-spin" /> : 'Save'}
|
||||
</Button>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
{editingRoi ? (
|
||||
<div className="space-y-3 bg-bambu-dark-tertiary/50 p-3 rounded-lg">
|
||||
<div className="grid grid-cols-2 gap-3">
|
||||
<div>
|
||||
<label className="text-xs text-bambu-gray">X Start</label>
|
||||
<input
|
||||
type="range"
|
||||
min="0"
|
||||
max="0.9"
|
||||
step="0.01"
|
||||
value={editingRoi.x}
|
||||
onChange={(e) => setEditingRoi({ ...editingRoi, x: parseFloat(e.target.value) })}
|
||||
className="w-full h-1.5 bg-bambu-dark-tertiary rounded-lg cursor-pointer accent-green-500"
|
||||
/>
|
||||
<span className="text-xs text-bambu-gray">{Math.round(editingRoi.x * 100)}%</span>
|
||||
</div>
|
||||
<div>
|
||||
<label className="text-xs text-bambu-gray">Y Start</label>
|
||||
<input
|
||||
type="range"
|
||||
min="0"
|
||||
max="0.9"
|
||||
step="0.01"
|
||||
value={editingRoi.y}
|
||||
onChange={(e) => setEditingRoi({ ...editingRoi, y: parseFloat(e.target.value) })}
|
||||
className="w-full h-1.5 bg-bambu-dark-tertiary rounded-lg cursor-pointer accent-green-500"
|
||||
/>
|
||||
<span className="text-xs text-bambu-gray">{Math.round(editingRoi.y * 100)}%</span>
|
||||
</div>
|
||||
<div>
|
||||
<label className="text-xs text-bambu-gray">Width</label>
|
||||
<input
|
||||
type="range"
|
||||
min="0.1"
|
||||
max="1"
|
||||
step="0.01"
|
||||
value={editingRoi.w}
|
||||
onChange={(e) => setEditingRoi({ ...editingRoi, w: parseFloat(e.target.value) })}
|
||||
className="w-full h-1.5 bg-bambu-dark-tertiary rounded-lg cursor-pointer accent-green-500"
|
||||
/>
|
||||
<span className="text-xs text-bambu-gray">{Math.round(editingRoi.w * 100)}%</span>
|
||||
</div>
|
||||
<div>
|
||||
<label className="text-xs text-bambu-gray">Height</label>
|
||||
<input
|
||||
type="range"
|
||||
min="0.1"
|
||||
max="1"
|
||||
step="0.01"
|
||||
value={editingRoi.h}
|
||||
onChange={(e) => setEditingRoi({ ...editingRoi, h: parseFloat(e.target.value) })}
|
||||
className="w-full h-1.5 bg-bambu-dark-tertiary rounded-lg cursor-pointer accent-green-500"
|
||||
/>
|
||||
<span className="text-xs text-bambu-gray">{Math.round(editingRoi.h * 100)}%</span>
|
||||
</div>
|
||||
</div>
|
||||
<p className="text-xs text-bambu-gray">
|
||||
Adjust the detection area to focus on the build plate. The green box in the preview shows the current area.
|
||||
</p>
|
||||
</div>
|
||||
) : (
|
||||
<p className="text-xs text-bambu-gray">
|
||||
Current: X={Math.round((plateCheckResult.roi?.x || 0.15) * 100)}%, Y={Math.round((plateCheckResult.roi?.y || 0.35) * 100)}%,
|
||||
W={Math.round((plateCheckResult.roi?.w || 0.70) * 100)}%, H={Math.round((plateCheckResult.roi?.h || 0.55) * 100)}%
|
||||
</p>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
<div className="flex justify-end gap-2 p-4 border-t border-bambu-dark-tertiary">
|
||||
{plateCheckResult.needs_calibration ? (
|
||||
<>
|
||||
<Button variant="ghost" onClick={() => setPlateCheckResult(null)}>
|
||||
Cancel
|
||||
</Button>
|
||||
<Button
|
||||
onClick={() => handleCalibratePlate()}
|
||||
disabled={isCalibrating}
|
||||
>
|
||||
{isCalibrating ? (
|
||||
<>
|
||||
<Loader2 className="w-4 h-4 mr-2 animate-spin" />
|
||||
Calibrating...
|
||||
</>
|
||||
) : (
|
||||
'Calibrate Empty Plate'
|
||||
)}
|
||||
</Button>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<Button variant="ghost" onClick={() => handleCalibratePlate()} disabled={isCalibrating}>
|
||||
{isCalibrating ? 'Adding...' : `Add Reference (${plateReferences?.references.length || 0}/${plateReferences?.max_references || 5})`}
|
||||
</Button>
|
||||
<Button onClick={() => setPlateCheckResult(null)}>
|
||||
Close
|
||||
</Button>
|
||||
</>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Power On Confirmation */}
|
||||
{showPowerOnConfirm && smartPlug && (
|
||||
<ConfirmModal
|
||||
|
||||
@@ -41,6 +41,10 @@ psutil>=6.0.0
|
||||
python-jose[cryptography]>=3.3.0
|
||||
passlib[bcrypt]>=1.7.4
|
||||
|
||||
# Plate Detection (optional - enables build plate empty detection)
|
||||
opencv-python-headless>=4.8.0
|
||||
numpy>=1.24.0
|
||||
|
||||
# Development
|
||||
pytest>=8.0.0
|
||||
pytest-asyncio>=0.23.0
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+2
-2
@@ -23,8 +23,8 @@
|
||||
|
||||
<!-- Splash screens for iOS -->
|
||||
<link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
|
||||
<script type="module" crossorigin src="/assets/index-BO_iYk_v.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-BKSrBx0A.css">
|
||||
<script type="module" crossorigin src="/assets/index-CsmyN56S.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="/assets/index-BI0S_RVt.css">
|
||||
</head>
|
||||
<body>
|
||||
<div id="root"></div>
|
||||
|
||||
Reference in New Issue
Block a user