mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-04 21:21:54 +02:00
94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
"""Action dispatch for Obico failure detection.
|
|
|
|
Separated from the detection loop so actions can be unit-tested and swapped.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.core.database import async_session
|
|
from backend.app.models.printer import Printer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def execute_action(
|
|
printer_id: int, action: str, task_name: str, score: float, frame: bytes | None = None
|
|
) -> None:
|
|
"""Run the configured action for a detected print failure.
|
|
|
|
action: 'notify' | 'pause' | 'pause_and_off'
|
|
frame: the camera snapshot the ML model flagged, if any — attached to
|
|
the notification so the recipient can see what triggered it.
|
|
"""
|
|
printer_name = await _get_printer_name(printer_id)
|
|
|
|
if action in ("pause", "pause_and_off"):
|
|
_pause_print(printer_id)
|
|
|
|
if action == "pause_and_off":
|
|
await _turn_off_linked_plugs(printer_id)
|
|
|
|
await _notify(printer_id, printer_name, task_name, score, action, frame)
|
|
|
|
|
|
async def _get_printer_name(printer_id: int) -> str:
|
|
async with async_session() as db:
|
|
result = await db.execute(select(Printer).where(Printer.id == printer_id))
|
|
printer = result.scalar_one_or_none()
|
|
return printer.name if printer else f"Printer {printer_id}"
|
|
|
|
|
|
def _pause_print(printer_id: int) -> None:
|
|
from backend.app.services.printer_manager import printer_manager
|
|
|
|
client = printer_manager.get_client(printer_id)
|
|
if not client:
|
|
logger.warning("Obico pause: no MQTT client for printer %s", printer_id)
|
|
return
|
|
if not client.pause_print():
|
|
logger.warning("Obico pause: pause_print() returned False for printer %s", printer_id)
|
|
|
|
|
|
async def _turn_off_linked_plugs(printer_id: int) -> None:
|
|
from backend.app.services.smart_plug_manager import smart_plug_manager
|
|
|
|
async with async_session() as db:
|
|
plugs = await smart_plug_manager._get_plugs_for_printer(printer_id, db)
|
|
for plug in plugs:
|
|
if not plug.enabled:
|
|
continue
|
|
try:
|
|
service = await smart_plug_manager.get_service_for_plug(plug, db)
|
|
await service.turn_off(plug)
|
|
logger.info("Obico action: turned off plug %s for printer %s", plug.name, printer_id)
|
|
except Exception as e:
|
|
logger.error("Obico action: failed to turn off plug %s: %s", plug.name, e)
|
|
|
|
|
|
async def _notify(
|
|
printer_id: int, printer_name: str, task_name: str, score: float, action: str, frame: bytes | None = None
|
|
) -> None:
|
|
"""Fire the AI Failure Detection notification (#1794).
|
|
|
|
Routed to its own event in 0.2.5b1; previously rode the multiplexed
|
|
on_printer_error toggle, which made it indistinguishable from HMS
|
|
hardware errors in the UI.
|
|
"""
|
|
from backend.app.services.notification_service import notification_service
|
|
|
|
async with async_session() as db:
|
|
try:
|
|
await notification_service.on_ai_failure_detection(
|
|
printer_id=printer_id,
|
|
printer_name=printer_name,
|
|
task_name=task_name,
|
|
confidence=score,
|
|
action=action,
|
|
db=db,
|
|
image_data=frame,
|
|
)
|
|
except Exception as e:
|
|
logger.error("Obico notify failed for printer %s: %s", printer_id, e)
|