mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-21 18:47:13 +02:00
110 lines
4.0 KiB
Python
110 lines
4.0 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 (
|
|
_is_script_plug,
|
|
publish_plug_state,
|
|
run_linked_scripts_after_automatic_switch,
|
|
smart_plug_manager,
|
|
)
|
|
|
|
power_plug_off = None
|
|
async with async_session() as db:
|
|
plugs = await smart_plug_manager._get_plugs_for_printer(printer_id, db)
|
|
for plug in plugs:
|
|
# A script can only be run, not turned off; the "printer turns
|
|
# off" ones run below, after the plug that feeds the printer.
|
|
if not plug.enabled or _is_script_plug(plug):
|
|
continue
|
|
try:
|
|
service = await smart_plug_manager.get_service_for_plug(plug, db)
|
|
if await service.turn_off(plug):
|
|
await publish_plug_state(
|
|
plug.id, plug.name, plug.printer_id, on=False, ha_entity_id=plug.ha_entity_id
|
|
)
|
|
if plug.controls_printer_power:
|
|
power_plug_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)
|
|
if power_plug_off is not None:
|
|
# Once, however many plugs feed the printer.
|
|
await run_linked_scripts_after_automatic_switch(printer_id, power_plug_off, "OFF")
|
|
|
|
|
|
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)
|