bambuddy/backend/app/services/obico_actions.py

"""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 | None,
    frame: bytes | None = None,
    *,
    print_quality: int | 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, print_quality=print_quality)


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("AI failure detection: no MQTT client for printer %s", printer_id)
        return
    if not client.pause_print():
        logger.warning("AI failure detection: 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("AI failure detection: turned off plug %s for printer %s", plug.name, printer_id)
            except Exception as e:
                logger.error("AI failure detection: 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 | None,
    action: str,
    frame: bytes | None = None,
    *,
    print_quality: int | 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,
                print_quality=print_quality,
                image_data=frame,
            )
        except Exception as e:
            logger.error("AI failure detection notify failed for printer %s: %s", printer_id, e)