mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
06e5114a03
The printer card's AI badge collapsed every class that was not Warning or Failure into green Safe, and the service reported `safe` whenever it had no verdict. The state entry is created when a monitored print is first seen -- before the first snapshot, let alone the first inference -- so a rejected ML API token, an unreachable ML API, a failed capture and an unset External URL all rendered as a healthy watched print: green Safe at score 0.000. For a safety feature that is the worst failure mode available: it asserts the print is being watched exactly when it is not. The reporter read that badge and concluded the loop had never started. It had been calling the ML API every ten seconds and being turned away with a 401 -- invisible because Obico's auth layer rejects a bad token before its request log sees it, and because successful checks log nothing there either. Add two honest states. Not checking (amber) when the last poll produced no result, carrying the reason; Starting while a monitored print waits for its first result. Score and frame count are withheld while not checking, since 0.000 beside "Not checking" reads as a measurement rather than its absence. The reason is per printer, so a card names its own problem rather than whichever printer failed most recently, and stays behind settings:read because it can quote configured URLs -- the badge state does not, because whether a print is watched is not configuration. An unrecognised class now falls back to Starting, not Safe. Test Connection saves the form before probing, so a green result describes the configuration the loop actually runs with rather than what is typed in the boxes.
106 lines
4.2 KiB
Python
106 lines
4.2 KiB
Python
"""API routes for Obico AI failure detection."""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, HTTPException, Response
|
|
from pydantic import BaseModel
|
|
|
|
from backend.app.core.auth import RequirePermissionIfAuthEnabled
|
|
from backend.app.core.permissions import Permission
|
|
from backend.app.models.user import User
|
|
from backend.app.services.obico_detection import obico_detection_service, pop_frame
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/obico", tags=["obico"])
|
|
|
|
|
|
class TestConnectionRequest(BaseModel):
|
|
url: str
|
|
# Omitted entirely = test with the saved token; "" = test with no token.
|
|
token: str | None = None
|
|
|
|
|
|
@router.get("/status")
|
|
async def get_status(
|
|
_: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
|
|
):
|
|
"""Scheduler status, per-printer classification, and recent detection history."""
|
|
settings = await obico_detection_service._load_settings()
|
|
status = obico_detection_service.get_status(settings["sensitivity"])
|
|
return {
|
|
**status,
|
|
"enabled": settings["enabled"],
|
|
"ml_url": settings["ml_url"],
|
|
"sensitivity": settings["sensitivity"],
|
|
"action": settings["action"],
|
|
"poll_interval": settings["poll_interval"],
|
|
"external_url_configured": bool(settings["external_url"]),
|
|
}
|
|
|
|
|
|
@router.get("/printer-status")
|
|
async def get_printer_status(
|
|
user: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
|
|
):
|
|
"""Per-printer live classification for the printer cards (#1546).
|
|
|
|
Deliberately excludes configuration (ML URL, action, history) so users
|
|
with printers:read but no settings:read can still render the badge.
|
|
"""
|
|
settings = await obico_detection_service._load_settings()
|
|
enabled_printers = settings["enabled_printers"]
|
|
# Error strings can embed configured URLs (ML API base, external URL), so
|
|
# they stay behind settings:read like the rest of the configuration.
|
|
can_see_error = user is None or user.has_permission(Permission.SETTINGS_READ.value)
|
|
per_printer = obico_detection_service.get_per_printer()
|
|
if not can_see_error:
|
|
# The "error" *class* is not configuration — a printers:read user still
|
|
# needs to know their print is not being watched. Only the reason, which
|
|
# can name a URL, is withheld.
|
|
per_printer = {pid: {**entry, "error": None} for pid, entry in per_printer.items()}
|
|
return {
|
|
"enabled": settings["enabled"],
|
|
# None = all printers are monitored
|
|
"monitored_printers": sorted(enabled_printers) if enabled_printers is not None else None,
|
|
"per_printer": per_printer,
|
|
"last_error": obico_detection_service._last_error if can_see_error else None,
|
|
}
|
|
|
|
|
|
@router.post("/test-connection")
|
|
async def test_connection(
|
|
req: TestConnectionRequest,
|
|
_: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
|
|
):
|
|
"""Ping the Obico ML API health endpoint and check the token. Returns ok + raw body."""
|
|
if not req.url:
|
|
return {"ok": False, "status_code": None, "body": None, "error": "URL is empty", "auth_ok": None}
|
|
token = req.token
|
|
if token is None:
|
|
# Field omitted entirely — test what the service actually uses.
|
|
settings = await obico_detection_service._load_settings()
|
|
token = settings.get("ml_token") or ""
|
|
return await obico_detection_service.test_connection(req.url, token)
|
|
|
|
|
|
@router.get("/cached-frame/{nonce}")
|
|
async def cached_frame(nonce: str):
|
|
"""Serve a pre-captured JPEG to the Obico ML API.
|
|
|
|
The detection loop captures a snapshot locally (where we control the timeout),
|
|
stashes the bytes under a one-shot random nonce, then hands this URL to Obico's
|
|
ML API. Obico's hardcoded 5s read timeout never races our snapshot pipeline.
|
|
|
|
Unauthenticated: the unguessable 32-byte nonce is single-use and expires in
|
|
seconds, so exposing this path doesn't widen the camera access surface.
|
|
"""
|
|
data = await pop_frame(nonce)
|
|
if data is None:
|
|
raise HTTPException(status_code=404, detail="Frame not found or expired")
|
|
return Response(
|
|
content=data,
|
|
media_type="image/jpeg",
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|