mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-19 15:54:27 +02:00
feat(obico): AI print-failure detection via self-hosted Obico ML API (#172)
Adds a Failure Detection tab under Settings that wires Bambuddy to a
self-hosted Obico ml_api container — no cloud, no account, no WebSocket.
While a print is running, the detection service periodically hands the
printer's camera snapshot URL to the ML API and smooths scores over
time (30-frame warmup + EWM, alpha=2/13, short/long rolling means) so
one noisy frame can't trigger an action. When the smoothed score
crosses HIGH, the configured action fires exactly once per print:
notify, pause, or pause-and-cut-power (via linked smart plugs).
- Backend: new obico_detection + obico_smoothing + obico_actions
services, /obico/status and /obico/test-connection routes
(SETTINGS_READ / SETTINGS_UPDATE), six obico_* AppSettings fields
with validators for sensitivity/action/enabled_printers.
- Frontend: FailureDetectionSettings component (enable, ML URL + test,
sensitivity, action, poll interval, per-printer monitor list, live
status + detection history), new sidebar tab with service-active
bullet, toast on save.
- Tests: 17 detection unit tests + 15 smoothing unit tests + 4
frontend component tests.
- Docs: README bullet, CHANGELOG entry, wiki page under Analytics,
website features.html entry.
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
"""API routes for Obico AI failure detection."""
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter
|
||||
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
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/obico", tags=["obico"])
|
||||
|
||||
|
||||
class TestConnectionRequest(BaseModel):
|
||||
url: str
|
||||
|
||||
|
||||
@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()
|
||||
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.post("/test-connection")
|
||||
async def test_connection(
|
||||
req: TestConnectionRequest,
|
||||
_: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
|
||||
):
|
||||
"""Ping the Obico ML API `/hc/` health endpoint. Returns ok + raw body."""
|
||||
if not req.url:
|
||||
return {"ok": False, "status_code": None, "body": None, "error": "URL is empty"}
|
||||
return await obico_detection_service.test_connection(req.url)
|
||||
Reference in New Issue
Block a user