bambuddy/backend/app/services/printer_manager.py

import asyncio
import logging
import math
import re
import traceback
from collections.abc import Callable

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.models.printer import Printer
from backend.app.services.bambu_mqtt import (
    STAGE_NAMES,
    BambuMQTTClient,
    DryingCycleEnd,
    MQTTLogEntry,
    PrinterState,
    get_stage_name,
)
from backend.app.utils.ams_humidity import ams_humidity_percent
from backend.app.utils.kprofile_lookup import build_slot_k_resolver

logger = logging.getLogger(__name__)

# Models that have a real chamber temperature sensor
# Based on Home Assistant Bambu Lab integration
# P1P/P1S and A1/A1Mini do NOT have chamber temp sensors
# Includes both display names and internal codes from MQTT/SSDP
CHAMBER_TEMP_SUPPORTED_MODELS = frozenset(
    [
        # Display names
        "X1",
        "X1C",
        "X1E",  # X1 series
        "X2D",  # X2 series
        "P2S",  # P2 series
        "H2C",
        "H2D",
        "H2DPRO",
        "H2S",  # H2 series
        # Internal codes (from MQTT/SSDP)
        "BL-P001",  # X1/X1C
        "C13",  # X1E
        "N6",  # X2D
        "O1D",  # H2D
        "O1C",  # H2C
        "O1C2",  # H2C (dual nozzle variant)
        "O1S",  # H2S
        "O1E",  # H2D Pro
        "O2D",  # H2D Pro (alternate code)
        "N7",  # P2S
    ]
)

# Models that may incorrectly report stg_cur=0 when idle (firmware bug)
# Based on Home Assistant Bambu Lab integration observations
# See: https://github.com/greghesp/ha-bambulab/blob/main/custom_components/bambu_lab/pybambu/models.py
A1_MODELS = frozenset(
    [
        # Display names
        "A1",
        "A1-MINI",
        "A1MINI",
        # Internal codes (from MQTT/SSDP)
        "N1",  # A1 Mini
        "N2S",  # A1
    ]
)

# Models affected by the stg_cur=0 idle bug (firmware reports stg_cur=0 when idle,
# which maps to "Printing" in STAGE_NAMES and overrides the correct IDLE state)
STG_CUR_IDLE_BUG_MODELS = A1_MODELS | frozenset(
    [
        # Display names
        "P1P",
        "P1S",
        # Internal codes (from MQTT/SSDP)
        "C11",  # P1P
        "C12",  # P1S
    ]
)


def _normalize_model(model: str) -> str:
    """Upper case without spaces, the form the model sets below hold (#3335).

    A printer is stored as "H2D Pro" or "A1 Mini"; upper-casing alone kept the
    space, and "H2D PRO" matched no set. Dashes stay: the sets name the X1C's
    code as "BL-P001".
    """
    return model.strip().upper().replace(" ", "")


def supports_chamber_temp(model: str | None) -> bool:
    """Check if a printer model has a real chamber temperature sensor.

    P1P, P1S, A1, and A1Mini do NOT have chamber temp sensors.
    The 'chamber_temper' value they report is meaningless.
    """
    if not model:
        return False
    return _normalize_model(model) in CHAMBER_TEMP_SUPPORTED_MODELS


# Models with an ACTIVE chamber heater (M141 has an effect).
# Many printers in CHAMBER_TEMP_SUPPORTED_MODELS only have a passive sensor —
# X1C and P2S report chamber temperature but cannot actively heat it. Only
# the models below ship a heater that responds to M141; the X1E heats to 60 °C.
CHAMBER_HEATER_MODELS = frozenset(
    [
        # Display names
        "H2C",
        "H2D",
        "H2DPRO",
        "H2S",
        "X1E",
        "X2D",
        # Internal codes (from MQTT/SSDP)
        "C13",  # X1E
        "O1C",  # H2C
        "O1C2",  # H2C dual-nozzle variant
        "O1D",  # H2D
        "O1E",  # H2D Pro
        "O2D",  # H2D Pro alternate code
        "O1S",  # H2S
        "N6",  # X2D
    ]
)


def supports_chamber_heater(model: str | None) -> bool:
    """Check if a printer model has an active chamber heater (responds to M141).

    The chamber temperature SENSOR is more widely deployed than the chamber
    HEATER — X1C/P2S report chamber temp but ignore M141. Only H2C, H2D,
    H2D Pro, H2S, X1E, X2D actually heat. Sensor-only models silently swallow the
    command at the firmware level, so we 400 at the route to surface that.
    """
    if not model:
        return False
    return _normalize_model(model) in CHAMBER_HEATER_MODELS


# Models with a cooling / heating airduct flap. Same set as the frontend
# PrintersPage airduct-toggle whitelist (P2S, X2D, H2D, H2D Pro, H2C, H2S).
# X1E has a chamber heater but NO airduct flap — the warm-air recirculation
# happens via the fixed front-door inlet, so no `set_airduct` command is
# needed (and the firmware ignores it). P2S has an airduct but no heater —
# the flap manages chamber airflow even without an active heater. The
# intersection (chamber heater AND airduct) is what the preheat stage cares
# about: when M141 fires we also need to assert heating mode, otherwise the
# default cooling mode actively fights the chamber heater.
CHAMBER_AIRDUCT_MODELS = frozenset(
    [
        # Display names
        "P2S",
        "X2D",
        "H2C",
        "H2D",
        "H2DPRO",
        "H2S",
        # Internal codes (from MQTT/SSDP)
        "N7",  # P2S
        "N6",  # X2D
        "O1C",  # H2C
        "O1C2",  # H2C dual-nozzle variant
        "O1D",  # H2D
        "O1E",  # H2D Pro
        "O2D",  # H2D Pro alternate code
        "O1S",  # H2S
    ]
)


def supports_airduct(model: str | None) -> bool:
    """Check if a printer model has a cooling / heating airduct mode toggle.

    Mirrors the frontend PrintersPage `['P2S', 'X2D', 'H2D', 'H2C', 'H2S']`
    + H2D Pro whitelist. Distinct from `supports_chamber_heater` — P2S has
    the airduct toggle but no active heater, and X1E has the heater but no
    airduct. The preheat stage cares about the intersection (heater AND
    airduct) so it can flip the flap to heating before energising M141.
    """
    if not model:
        return False
    return _normalize_model(model) in CHAMBER_AIRDUCT_MODELS


def has_stg_cur_idle_bug(model: str | None) -> bool:
    """Check if a printer model may incorrectly report stg_cur=0 when idle.

    Some firmware versions report stg_cur=0 (which maps to "Printing")
    even when the printer is idle. Originally observed on A1/A1 Mini via the
    Home Assistant Bambu Lab integration, also confirmed on P1S.
    """
    if not model:
        return False
    return _normalize_model(model) in STG_CUR_IDLE_BUG_MODELS


# Minimum firmware versions for AMS drying support (confirmed via capture testing)
# Keys are exact model names (upper-cased, no spaces). Do NOT use substring matching — it would
# incorrectly gate X1E (matched by "X1") and H2D Pro (matched by "H2D").
_DRYING_MIN_FIRMWARE: dict[str, str] = {
    "H2D": "01.02.30.00",
    "H2S": "01.02.00.00",
    "H2C": "01.02.00.00",
    "O1C": "01.02.00.00",  # H2C SSDP model code
    "O1C2": "01.02.00.00",  # H2C dual-nozzle SSDP model code
    "X1": "01.09.00.00",
    "X1C": "01.09.00.00",
    "P2S": "01.02.00.00",
    "N7": "01.02.00.00",  # P2S internal model code
}
# Models that definitely don't support AMS drying (no AMS 2 Pro / AMS-HT compatibility)
_DRYING_UNSUPPORTED_MODELS = frozenset({"A1", "A1MINI", "A1-MINI", "O1S", "N1", "N2S"})

# Models whose AMS can dry, but only from the printer's own touchscreen. Bambu's P1
# manual is explicit: "P1S connected AMS drying functions may only be controlled from
# the P1S screen." The firmware still answers `ams_filament_drying` with
# result: success and then does nothing — the reporter of #2533 sent it three times
# on an idle P1S with an AMS 2 Pro and the unit never left dry_status 0. Bambuddy
# originally listed P1P/P1S here as fw-gated (01.08+, #292); that version is when P1
# firmware gained AMS 2 Pro *support*, not remote drying, and it was never verified
# against a live P1. Nothing we can send will start a cycle, so we don't offer to.
_DRYING_SCREEN_ONLY_MODELS = frozenset({"P1P", "P1S"})


def drying_screen_only(model: str | None) -> bool:
    """True when the model's AMS dries only via the printer's own screen (#2533).

    Distinct from "unsupported": these printers *can* dry, and Bambuddy still shows
    a cycle started on the printer. They just can't be commanded to start or stop
    one remotely, so the UI explains that instead of silently dropping the control.
    """
    if not model:
        return False
    return _normalize_model(model) in _DRYING_SCREEN_ONLY_MODELS


# Temperature keys the UI actually draws. `state.temperatures` is also working
# memory: it carries private bookkeeping (`_nozzle_target_set_time`) and derived
# flags (`nozzle_heating`) that no consumer outside this module should see. The
# full-status path hands out the whole dict to logged-in callers; the streaming
# overlay gets only this list, because an overlay token is a narrower grant than
# a login and should not pick up fields by accident as the dict grows.
DISPLAY_TEMPERATURE_KEYS = (
    "nozzle",
    "nozzle_target",
    "nozzle_2",
    "nozzle_2_target",
    "bed",
    "bed_target",
    "chamber",
    "chamber_target",
)


def display_temperatures(temperatures: dict | None, model: str | None) -> dict[str, float]:
    """Filter `state.temperatures` down to the readings a viewer is shown.

    Drops chamber readings on models without a real chamber sensor — P1P, P1S,
    A1 and A1 mini all report a meaningless `chamber_temper` — matching what
    ``printer_state_to_dict`` already does for the full status payload.
    """
    if not temperatures:
        return {}
    allow_chamber = supports_chamber_temp(model)
    out: dict[str, float] = {}
    for key in DISPLAY_TEMPERATURE_KEYS:
        if key.startswith("chamber") and not allow_chamber:
            continue
        value = temperatures.get(key)
        if value is None:
            continue
        try:
            out[key] = float(value)
        except (TypeError, ValueError):
            continue
    return out


def uniform_tray_filament_hint(loaded_types: list[str]) -> str | None:
    """Guess an active cycle's filament from the loaded trays.

    Bambu never echoes back which filament or temperature a drying cycle is
    running, so the badge normally reads the target we cached when we sent the
    command. This is the fallback for when we have no record — drying started in
    a previous backend lifetime, or from the printer's own screen.

    It answers only when every loaded tray holds the same filament type. On a
    mixed unit the first tray is evidence of nothing: an AMS holding two PETG
    and two PLA spools, drying PLA at the 45°C the user picked, was labelled
    "PETG @ 65°C" purely because slot 1 happened to be PETG (#2759).

    Deliberately no temperature. The RFID-recommended ``drying_temp`` used to be
    returned alongside a uniform filament, which narrowed #2759 to units whose
    spools disagree but left the uniform case stating a temperature just as
    invented: a unit loaded entirely with PLA, drying at the 45°C the user
    picked, read "PLA @ 55°C" the moment the cached target went missing. The
    filament type is real evidence — every spool in the unit agrees on it, and
    the dryer heats all of them — but the temperature is a free choice in the
    popover, so a recommendation is never evidence of what is running. The badge
    shows the filament and the countdown, and names a temperature only when we
    actually sent it.

    Args:
        loaded_types: ``tray_type`` for each tray, in slot order. Empty slots
            (falsy) are ignored.

    Returns:
        The shared filament type, or None if the loaded trays disagree or the
        unit is empty.
    """
    types = {str(tray_type) for tray_type in loaded_types if tray_type}
    if len(types) != 1:
        return None
    return next(iter(types))


def supports_drying(model: str | None, firmware: str | None) -> bool:
    """Check if a printer model accepts remote AMS drying commands.

    Known models with confirmed min firmware get version-gated.
    Known unsupported models, and models that only dry from their own screen,
    are blocked.
    All other models (H2D Pro, X1E, future models) are allowed —
    the command fails gracefully with result: "fail" if unsupported.
    """
    if not model:
        return False
    model_upper = _normalize_model(model)
    if model_upper in _DRYING_UNSUPPORTED_MODELS or model_upper in _DRYING_SCREEN_ONLY_MODELS:
        return False
    if model_upper in _DRYING_MIN_FIRMWARE:
        return bool(firmware and firmware >= _DRYING_MIN_FIRMWARE[model_upper])
    # For all other models: allow
    return True


# Minimum firmware versions for AMS "Print While Drying" — drying that runs CONCURRENTLY
# with an active print. Strictly stricter than _DRYING_MIN_FIRMWARE (idle drying). Verified
# against Bambu wiki release notes — the canonical phrasing on every supported model is
# "printing while filament is drying" / "Print While Drying". Models absent from the wiki
# release notes (A1, A1 Mini, P1*, X1 non-C, X1E) are intentionally excluded — the firmware
# will reject the command in those cases anyway via dry_sf_reason=[0] (TaskOccupied).
_DRY_WHILE_PRINTING_MIN_FIRMWARE: dict[str, str] = {
    "H2D": "01.03.00.00",
    "H2DPRO": "01.02.00.00",
    "O1E": "01.02.00.00",  # H2D Pro SSDP code
    "O2D": "01.02.00.00",  # H2D Pro alternate code
    "H2C": "01.02.00.00",
    "O1C": "01.02.00.00",  # H2C SSDP code
    "O1C2": "01.02.00.00",  # H2C dual-nozzle SSDP code
    "H2S": "01.02.00.00",
    "X2D": "01.01.00.00",
    "N6": "01.01.00.00",  # X2D internal code
    "X1C": "01.11.02.00",
    "BL-P001": "01.11.02.00",  # X1C internal code
    "P2S": "01.02.00.00",
    "N7": "01.02.00.00",  # P2S internal code
    "A2L": "01.01.00.00",
    "N9": "01.01.00.00",  # A2L internal code
}


def supports_drying_while_printing(model: str | None, firmware: str | None) -> bool:
    """Check if a printer model+firmware supports running AMS drying CONCURRENTLY
    with an active print.

    Distinct from supports_drying() — that gates idle drying. This gate is strict:
    only models explicitly confirmed by Bambu wiki release notes are allowed.
    On unsupported models the firmware returns dry_sf_reason=[0] (TaskOccupied)
    while a print is running, so being conservative here costs nothing — the
    firmware is the ultimate arbiter, this gate just hides UI affordances.
    """
    if not model:
        return False
    model_upper = _normalize_model(model)
    if model_upper not in _DRY_WHILE_PRINTING_MIN_FIRMWARE:
        return False
    return bool(firmware and firmware >= _DRY_WHILE_PRINTING_MIN_FIRMWARE[model_upper])


class PrinterInfo:
    """Basic printer info for callbacks."""

    def __init__(self, name: str, serial_number: str):
        self.name = name
        self.serial_number = serial_number


class PrinterManager:
    """Manager for multiple printer connections."""

    def __init__(self):
        self._clients: dict[int, BambuMQTTClient] = {}
        self._models: dict[int, str | None] = {}  # Cache printer models for feature detection
        self._printer_info: dict[int, PrinterInfo] = {}  # Cache printer name/serial for callbacks
        # Last AMS / external-spool reading of a printer whose client has been
        # dropped, so the queue can still tell which machine holds which colour
        # (#2876). Deliberately outside the client's own state: it answers
        # "what did this printer last have loaded", not "what is it reporting
        # now", and the two must not be confused by anything that displays or
        # merges live status.
        self._last_trays: dict[int, dict] = {}
        self._on_print_start: Callable[[int, dict], None] | None = None
        self._on_print_complete: Callable[[int, dict], None] | None = None
        self._on_print_running_observed: Callable[[int, dict], None] | None = None
        self._on_finish_photo_moment: Callable[[int, dict], None] | None = None
        self._on_status_change: Callable[[int, PrinterState], None] | None = None
        self._on_ams_change: Callable[[int, list], None] | None = None
        self._on_fts_inlet_change: Callable[[int, int, str], None] | None = None
        self._on_layer_change: Callable[[int, int], None] | None = None
        self._on_print_progress: Callable[[int, int], None] | None = None
        self._on_bed_temp_update: Callable[[int, float], None] | None = None
        self._on_drying_complete: Callable[[int, int], None] | None = None
        self._on_drying_cycle_end: Callable[[int, DryingCycleEnd], None] | None = None
        self._on_assignment_verified: Callable[[int, int, int, bool, dict], None] | None = None
        self._on_tray_change: Callable[[int, int, int], None] | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        # Track who started the current print (Issue #206)
        self._current_print_user: dict[int, dict] = {}  # {printer_id: {"user_id": int, "username": str}}
        # Track printers awaiting plate-clear acknowledgment after a finished/failed print.
        # Persisted to DB (printers.awaiting_plate_clear) so the gate survives restarts/power
        # cycles — see issue #961. Loaded into this set at startup via load_awaiting_plate_clear_from_db().
        self._awaiting_plate_clear: set[int] = set()

    def get_printer(self, printer_id: int) -> PrinterInfo | None:
        """Get printer info by ID."""
        return self._printer_info.get(printer_id)

    def set_current_print_user(self, printer_id: int, user_id: int, username: str, queue_item_id: int | None = None):
        """Track who started the current print (Issue #206), and for which queue item."""
        self._current_print_user[printer_id] = {
            "user_id": user_id,
            "username": username,
            "queue_item_id": queue_item_id,
        }

    def get_current_print_user(self, printer_id: int) -> dict | None:
        """Get the user who started the current print (Issue #206)."""
        return self._current_print_user.get(printer_id)

    def clear_current_print_user(self, printer_id: int):
        """Clear the current print user when print completes (Issue #206)."""
        self._current_print_user.pop(printer_id, None)

    def clear_current_print_user_for_item(self, printer_id: int, queue_item_id: int) -> None:
        """Clear the print user if a queue item's dispatch set it and that print never started.

        A later dispatch's user, set for another item, stays.
        """
        user = self._current_print_user.get(printer_id)
        if user is not None and user.get("queue_item_id") == queue_item_id:
            del self._current_print_user[printer_id]

    def forget_user(self, user_id: int) -> None:
        """A deleted user is credited with none of the prints still running."""
        for printer_id, user in list(self._current_print_user.items()):
            if user.get("user_id") == user_id:
                del self._current_print_user[printer_id]

    def forget_printer(self, printer_id: int) -> None:
        """A deleted printer's print owner, plate-clear gate and last trays go
        with it, not to the next printer given its id (SQLite reuses it)."""
        self._current_print_user.pop(printer_id, None)
        self._awaiting_plate_clear.discard(printer_id)
        self._last_trays.pop(printer_id, None)

    def is_awaiting_plate_clear(self, printer_id: int) -> bool:
        """Return True when the printer finished/failed a print and is waiting for the
        user to acknowledge the plate is cleared before the queue may dispatch the next job.
        """
        return printer_id in self._awaiting_plate_clear

    def set_awaiting_plate_clear(self, printer_id: int, awaiting: bool):
        """Set/clear the awaiting-plate-clear gate and persist it to DB.

        Persisted so the gate survives Bambuddy/printer restarts (#961): after Auto Off
        cycles the printer, the printer boots into IDLE with no memory of the previous
        finish, and without persistence the queue would bypass the confirmation prompt.

        Also broadcasts an updated ``printer_status`` over the WebSocket (#1128).
        ``awaiting_plate_clear`` is a Bambuddy-side flag — toggling it does not
        produce an MQTT push from the printer, so without an explicit broadcast
        any UI subscriber that's NOT the originating tab would stay stale until
        the next coincidental status refresh. The plate-clear button on the
        printer card disappeared "immediately" only because of an optimistic
        React Query cache update on the click path; clearing the flag through
        any other route (an admin script, a second tab, an automation that
        hits ``POST /printers/{id}/clear-plate`` directly) silently broke the
        UI without it. Centralised here so every current AND future caller is
        covered without each one having to remember to broadcast.
        """
        # Callers re-assert the current value routinely (the queue clears the gate
        # on every dispatch, whether or not it was up), so the outward-facing
        # emissions below are edge-triggered — an MQTT subscriber or a phone
        # notification must not see a "plate cleared" for a plate that was never
        # dirty. Persistence and the WebSocket broadcast stay unconditional: they
        # are idempotent and predate this (#961/#1128).
        changed = awaiting != (printer_id in self._awaiting_plate_clear)
        if awaiting:
            self._awaiting_plate_clear.add(printer_id)
        else:
            self._awaiting_plate_clear.discard(printer_id)
        # Only create the coroutine when there is a loop to run it on — otherwise Python
        # emits "coroutine was never awaited" warnings (e.g. in sync unit tests).
        if self._loop and self._loop.is_running():
            self._schedule_async(self._persist_awaiting_plate_clear(printer_id, awaiting))
            self._schedule_async(self._broadcast_status_change(printer_id))
            if changed:
                self._schedule_async(self._emit_plate_clear_change(printer_id, awaiting))

    async def _emit_plate_clear_change(self, printer_id: int, awaiting: bool) -> None:
        """Relay a plate-clear gate transition to MQTT and notifications (#2525).

        The flag is Bambuddy-side, so nothing about it reaches an external
        automation on its own — the printer's own MQTT push knows only
        RUNNING/PAUSE/FAILED/FINISH/IDLE. Emitted from here rather than from the
        three call sites so every current and future caller is covered, the same
        reasoning as the WebSocket broadcast above.

        Imports are local: ``mqtt_relay`` and ``notification_service`` both sit
        above this module in the dependency order.
        """
        printer = self.get_printer(printer_id)
        if not printer:
            # No cached info means no client is registered — the printer was
            # disconnected outright rather than merely powered off. The gate is
            # still releasable from the API in that state (#2864), and a retained
            # MQTT topic left saying "awaiting" would outlive the truth, so fall
            # back to the row rather than dropping the emission.
            printer = await self._printer_info_from_db(printer_id)
            if not printer:
                return

        try:
            from backend.app.services.mqtt_relay import mqtt_relay

            await mqtt_relay.on_plate_clear_state(printer_id, printer.name, printer.serial_number, awaiting)
        except Exception as e:
            logger.warning("Failed to publish plate-clear state for printer %d: %s", printer_id, e)

        # Only the rising edge is worth a notification — "the bed is now free"
        # is not an action item, and the queue clears the gate by itself.
        if not awaiting:
            return

        try:
            from backend.app.core.database import async_session
            from backend.app.services.notification_service import notification_service

            async with async_session() as db:
                await notification_service.on_plate_clear_required(printer_id, printer.name, db)
        except Exception as e:
            logger.warning("Failed to send plate-clear notification for printer %d: %s", printer_id, e)

    async def _printer_info_from_db(self, printer_id: int) -> PrinterInfo | None:
        """Name and serial for a printer with no registered client."""
        from backend.app.core.database import async_session

        try:
            async with async_session() as db:
                row = (
                    await db.execute(select(Printer.name, Printer.serial_number).where(Printer.id == printer_id))
                ).first()
        except Exception as e:
            logger.warning("Failed to load printer %d info from DB: %s", printer_id, e)
            return None
        return PrinterInfo(row[0], row[1]) if row else None

    async def _broadcast_status_change(self, printer_id: int) -> None:
        """Emit a ``printer_status`` WebSocket update for this printer (#1128).

        Used for state changes that don't come from MQTT — currently just the
        ``awaiting_plate_clear`` flag, but any future Bambuddy-side flag added
        to ``printer_state_to_dict`` should plumb through here too. The
        existing MQTT-driven broadcast in ``main.on_printer_status_change``
        deduplicates on a status_key that intentionally excludes Bambuddy
        flags (so e.g. queue-state changes don't get echoed as printer
        events), which is precisely why those flags need their own emit.

        Lazy-imports ``ws_manager`` to keep ``printer_manager`` clean of
        application-layer infra at module-import time — the broadcast is the
        only thing here that needs it.
        """
        state = self.get_status(printer_id)
        if not state:
            # Printer disconnected or unknown — nothing to broadcast. The
            # next reconnect will produce a fresh status push anyway, so the
            # UI eventually catches up without us forcing a stale snapshot
            # on subscribers now.
            return
        try:
            from backend.app.core.websocket import ws_manager

            await ws_manager.send_printer_status(
                printer_id,
                printer_state_to_dict(
                    state,
                    printer_id,
                    self.get_model(printer_id),
                    self.get_drying_targets(printer_id),
                ),
            )
        except Exception as e:
            logger.warning(
                "Failed to broadcast printer_status after Bambuddy-side state change for printer %d: %s",
                printer_id,
                e,
            )

    async def _persist_awaiting_plate_clear(self, printer_id: int, awaiting: bool):
        from backend.app.core.database import run_with_retry

        async def _do(db):
            printer = await db.get(Printer, printer_id)
            if printer is not None:
                printer.awaiting_plate_clear = awaiting
                await db.commit()

        try:
            await run_with_retry(_do, label=f"persist awaiting_plate_clear printer={printer_id}")
        except Exception as e:
            logger.warning("Failed to persist awaiting_plate_clear for printer %d: %s", printer_id, e)

    async def load_awaiting_plate_clear_from_db(self):
        """Rehydrate the awaiting-plate-clear set from the printers table on startup."""
        from backend.app.core.database import async_session

        try:
            async with async_session() as db:
                result = await db.execute(select(Printer.id).where(Printer.awaiting_plate_clear.is_(True)))
                ids = {row[0] for row in result.all()}
                self._awaiting_plate_clear = ids
                if ids:
                    logger.info("Loaded %d printer(s) awaiting plate-clear acknowledgment: %s", len(ids), sorted(ids))
        except Exception as e:
            logger.warning("Failed to load awaiting_plate_clear from DB: %s", e)

    def set_event_loop(self, loop: asyncio.AbstractEventLoop):
        """Set the event loop for async callbacks."""
        self._loop = loop

    def set_print_start_callback(self, callback: Callable[[int, dict], None]):
        """Set callback for print start events."""
        self._on_print_start = callback

    def set_print_complete_callback(self, callback: Callable[[int, dict], None]):
        """Set callback for print completion events."""
        self._on_print_complete = callback

    def set_print_running_observed_callback(self, callback: Callable[[int, dict], None]):
        """Set callback for restart-recovery RUNNING-state observations (#1485
        follow-up). Fires the first time we see ``state == RUNNING`` for a
        printer that started its print before Bambuddy came up — the #1304
        guard suppresses ``on_print_start`` for these, so anything that
        normally hangs off it (e.g. timelapse baseline capture) needs this
        hook to recover."""
        self._on_print_running_observed = callback

    def set_finish_photo_moment_callback(self, callback: Callable[[int, dict], None]):
        """Set callback for the #1721 finish-photo moment.

        Fires on the stage-22 (\"Filament unloading\") edge at end-of-print
        — the framing window where the toolhead is parked but the bed
        hasn't dropped yet. Falls back to firing at the FINISH-state
        transition for prints that skip stage 22 (cancel, external-spool-
        only, HMS halt, firmware variants). Payload includes the
        ``trigger`` key (``\"stage_22\"`` or ``\"finish_state\"``) and
        ``timelapse_was_active`` so the photo path can choose between
        live-camera capture and timelapse last-frame extraction."""
        self._on_finish_photo_moment = callback

    def set_status_change_callback(self, callback: Callable[[int, PrinterState], None]):
        """Set callback for status change events."""
        self._on_status_change = callback

    def set_ams_change_callback(self, callback: Callable[[int, list], None]):
        """Set callback for AMS data change events."""
        self._on_ams_change = callback

    def set_fts_inlet_change_callback(self, callback: Callable[[int, int, str], None]):
        """Set callback for Filament Track Switch inlet moves.

        Receives ``(printer_id, ams_id, inlet)``. Fired only when an AMS moves
        between inlets, not on the first sighting of a binding.
        """
        self._on_fts_inlet_change = callback

    def set_layer_change_callback(self, callback: Callable[[int, int], None]):
        """Set callback for layer change events. Receives (printer_id, layer_num)."""
        self._on_layer_change = callback

    def set_print_progress_callback(self, callback: Callable[[int, int], None]):
        """Set callback for print-progress advances (#2547).

        Receives (printer_id, percent) each time `mc_percent` increases during a
        running print — including the final layer, where layer-change events
        have already stopped.
        """
        self._on_print_progress = callback

    def set_bed_temp_update_callback(self, callback: Callable[[int, float], None]):
        """Set callback for bed temperature updates. Receives (printer_id, bed_temp)."""
        self._on_bed_temp_update = callback

    def set_drying_complete_callback(self, callback: Callable[[int, int], None]):
        """Set callback for AMS drying completion events (#1349).

        Receives ``(printer_id, ams_id)``. Fires once per falling edge of
        ``dry_time`` (>0 → 0) for each AMS unit.
        """
        self._on_drying_complete = callback

    def set_drying_cycle_end_callback(self, callback: Callable[[int, DryingCycleEnd], None]):
        """Set callback for the end of an AMS drying cycle (#2863).

        Receives ``(printer_id, DryingCycleEnd)`` on the same edge as the
        drying-complete callback, with what was seen of the cycle.
        """
        self._on_drying_cycle_end = callback

    def set_assignment_verified_callback(self, callback: Callable[[int, int, int, bool, dict], None]):
        """Set callback for spool-assignment read-back verification (#2582).

        Receives ``(printer_id, ams_id, tray_id, verified, detail)``. Fires once
        per assignment either when the tray telemetry confirms the pushed
        filament id or when the verification window elapses without it.
        """
        self._on_assignment_verified = callback

    def set_tray_change_callback(self, callback: Callable[[int, int, int], None]):
        """Set callback for mid-print tray changes.

        Receives ``(printer_id, global_tray_id, layer_num)`` for every entry
        appended to the printer's tray-change log, so it can be persisted for
        the completion-time weight split.
        """
        self._on_tray_change = callback

    def _schedule_async(self, coro):
        """Schedule an async coroutine from a sync context.

        Captures exceptions from the coroutine and logs them to prevent
        silent failures in callbacks.
        """
        if self._loop and self._loop.is_running():
            future = asyncio.run_coroutine_threadsafe(coro, self._loop)

            def handle_exception(f):
                # Stopping the loop cancels callbacks still pending. That is
                # shutdown, not a failure (#3243), and concurrent.futures'
                # CancelledError is an Exception, so it would be logged below.
                if f.cancelled():
                    return
                try:
                    # This will re-raise any exception from the coroutine
                    f.result()
                except Exception as e:
                    import logging

                    logging.getLogger(__name__).error(f"Exception in scheduled callback: {e}", exc_info=True)

            future.add_done_callback(handle_exception)

    def last_known_trays(self, printer_id: int) -> dict:
        """What this printer last had loaded, for a printer with no live client.

        Only the tray keys, and only as history: a caller that wants to know
        what a printer is reporting *now* must use :meth:`get_status`. This
        exists because dropping a client drops its status with it, and the
        queue reads the loaded filament to decide which offline printer is
        worth switching on (#2876) — ``_power_on_and_wait`` replaces the client
        on every attempt, so without this each attempt erased the reading the
        next one needs.
        """
        return self._last_trays.get(printer_id, {})

    def _remember_trays(self, printer_id: int) -> None:
        """Keep the tray reading of a client that is about to be dropped."""
        client = self._clients.get(printer_id)
        if not client:
            return
        raw = client.state.raw_data or {}
        remembered = {key: raw[key] for key in ("ams", "vt_tray") if raw.get(key)}
        if remembered:
            self._last_trays[printer_id] = remembered

    async def connect_printer(self, printer: Printer) -> bool:
        """Connect to a printer."""
        if printer.id in self._clients:
            self.disconnect_printer(printer.id)

        printer_id = printer.id

        def on_state_change(state: PrinterState):
            if self._on_status_change:
                self._schedule_async(self._on_status_change(printer_id, state))

        def on_print_start(data: dict):
            if self._on_print_start:
                self._schedule_async(self._on_print_start(printer_id, data))

        def on_print_complete(data: dict):
            if self._on_print_complete:
                self._schedule_async(self._on_print_complete(printer_id, data))

        def on_print_running_observed(data: dict):
            if self._on_print_running_observed:
                self._schedule_async(self._on_print_running_observed(printer_id, data))

        def on_finish_photo_moment(data: dict):
            if self._on_finish_photo_moment:
                self._schedule_async(self._on_finish_photo_moment(printer_id, data))

        def on_ams_change(ams_data: list):
            if self._on_ams_change:
                self._schedule_async(self._on_ams_change(printer_id, ams_data))

        def on_fts_inlet_change(ams_id: int, inlet: str):
            if self._on_fts_inlet_change:
                self._schedule_async(self._on_fts_inlet_change(printer_id, ams_id, inlet))

        def on_layer_change(layer_num: int):
            if self._on_layer_change:
                self._schedule_async(self._on_layer_change(printer_id, layer_num))

        def on_print_progress(percent: int):
            if self._on_print_progress:
                self._schedule_async(self._on_print_progress(printer_id, percent))

        def on_bed_temp_update(bed_temp: float):
            if self._on_bed_temp_update:
                self._schedule_async(self._on_bed_temp_update(printer_id, bed_temp))

        def on_drying_complete(ams_id: int):
            if self._on_drying_complete:
                self._schedule_async(self._on_drying_complete(printer_id, ams_id))

        def on_drying_cycle_end(cycle: DryingCycleEnd):
            if self._on_drying_cycle_end:
                self._schedule_async(self._on_drying_cycle_end(printer_id, cycle))

        def on_assignment_verified(ams_id: int, tray_id: int, verified: bool, detail: dict):
            if self._on_assignment_verified:
                self._schedule_async(self._on_assignment_verified(printer_id, ams_id, tray_id, verified, detail))

        def on_tray_change(tray_global: int, layer_num: int):
            if self._on_tray_change:
                self._schedule_async(self._on_tray_change(printer_id, tray_global, layer_num))

        client = BambuMQTTClient(
            ip_address=printer.ip_address,
            serial_number=printer.serial_number,
            access_code=printer.access_code,
            model=printer.model,
            on_state_change=on_state_change,
            on_print_start=on_print_start,
            on_print_complete=on_print_complete,
            on_ams_change=on_ams_change,
            on_fts_inlet_change=on_fts_inlet_change,
            on_layer_change=on_layer_change,
            on_print_progress=on_print_progress,
            on_bed_temp_update=on_bed_temp_update,
            on_drying_complete=on_drying_complete,
            on_drying_cycle_end=on_drying_cycle_end,
            on_print_running_observed=on_print_running_observed,
            on_finish_photo_moment=on_finish_photo_moment,
            on_assignment_verified=on_assignment_verified,
            on_tray_change=on_tray_change,
        )

        client.connect()
        self._clients[printer_id] = client
        self._models[printer_id] = printer.model  # Cache model for feature detection
        self._printer_info[printer_id] = PrinterInfo(printer.name, printer.serial_number)

        # Wait a moment for connection
        await asyncio.sleep(1)
        return client.state.connected

    def disconnect_printer(self, printer_id: int, timeout: float = 0):
        """Disconnect from a printer."""
        if printer_id in self._clients:
            self._remember_trays(printer_id)
            self._clients[printer_id].disconnect(timeout=timeout)
            del self._clients[printer_id]
        self._models.pop(printer_id, None)  # Clean up model cache
        self._printer_info.pop(printer_id, None)  # Clean up printer info cache

    def disconnect_all(self, timeout: float = 0):
        """Disconnect from all printers."""
        for printer_id in list(self._clients.keys()):
            self.disconnect_printer(printer_id, timeout=timeout)

    def get_status(self, printer_id: int) -> PrinterState | None:
        """Get the current status of a printer (checks for stale connections)."""
        if printer_id in self._clients:
            client = self._clients[printer_id]
            # Check staleness and update connected state if needed
            client.check_staleness()
            return client.state
        return None

    # Gcode states in which a job is loaded / in progress and cutting power
    # would ruin the print. PAUSE is included on purpose — a paused print is
    # still loaded on the bed. Used by the smart-plug auto-off guard (#1890) so
    # a re-print started from the touchscreen isn't killed mid-print.
    ACTIVE_PRINT_STATES = ("RUNNING", "PAUSE", "PREPARE", "SLICING")

    def is_print_active(self, printer_id: int) -> bool:
        """True when the printer currently has a print loaded / in progress.

        Returns False when disconnected or in any idle/terminal state
        (IDLE / FINISH / FAILED / unknown), so callers fail *open* only for
        the safe "nothing is printing" case. #1890.
        """
        state = self.get_status(printer_id)
        if not state or not state.connected:
            return False
        return state.state in self.ACTIVE_PRINT_STATES

    def get_model(self, printer_id: int) -> str | None:
        """Get the cached model for a printer."""
        return self._models.get(printer_id)

    def get_drying_targets(self, printer_id: int) -> dict[int, dict] | None:
        """Get cached active drying target params keyed by AMS id.

        Returned dict shape: ``{ams_id: {"filament": str, "temp": int}}``.
        Returns ``None`` when the printer is not connected. The cache is
        seeded by ``send_drying_command(mode=1)`` and cleared when drying
        stops or on the ``dry_time`` falling edge (handled inside
        ``BambuMQTTClient``).
        """
        client = self._clients.get(printer_id)
        return client._drying_targets if client else None

    def get_all_statuses(self) -> dict[int, PrinterState]:
        """Get status of all connected printers (checks for stale connections)."""
        result = {}
        for printer_id, client in self._clients.items():
            # Check staleness and update connected state if needed
            client.check_staleness()
            result[printer_id] = client.state
        return result

    def is_connected(self, printer_id: int) -> bool:
        """Check if a printer is connected (checks for stale connections)."""
        if printer_id in self._clients:
            client = self._clients[printer_id]
            # Check staleness and update connected state if needed
            return client.check_staleness()
        return False

    def get_client(self, printer_id: int) -> BambuMQTTClient | None:
        """Get the MQTT client for a printer."""
        return self._clients.get(printer_id)

    def mark_printer_offline(self, printer_id: int):
        """Mark a printer as offline and trigger status callback.

        This is used when we know the printer power was cut (e.g., smart plug turned off)
        to immediately update the UI without waiting for MQTT timeout.

        The mark is a presumption, not a fact: the plug may not actually feed
        the printer. ``BambuMQTTClient.mark_power_off`` records the state it
        overwrites so the client can undo it as soon as the printer sends
        another report (#2629).
        """
        import logging

        logger = logging.getLogger(__name__)

        if printer_id in self._clients:
            client = self._clients[printer_id]
            if client.mark_power_off():
                logger.info("Marking printer %s as offline (smart plug power off)", printer_id)
                # Trigger the status change callback to broadcast via WebSocket
                if self._on_status_change:
                    self._schedule_async(self._on_status_change(printer_id, client.state))

    def start_print(
        self,
        printer_id: int,
        filename: str,
        plate_id: int = 1,
        ams_mapping: list[int] | None = None,
        bed_levelling: str = "auto",
        flow_cali: str = "auto",
        vibration_cali: bool = True,
        layer_inspect: bool = False,
        timelapse: bool = False,
        use_ams: bool = True,
        nozzle_offset_cali: str = "auto",
        nozzle_mapping: str | None = None,
        nozzle_slot_extruders: str | None = None,
        plate_uses_only_external: bool = False,
    ) -> bool:
        """Start a print on a connected printer.

        ``nozzle_mapping`` is an opaque JSON string captured from BambuStudio's
        project_file MQTT command (H2C rack-swap slicer pick preservation,
        #1780). It rides through to the MQTT client untouched; the dispatch
        builder there parses + injects it only on dual-nozzle models.

        ``nozzle_slot_extruders`` is the fallback for a job that never passed
        through BambuStudio (#2800): per-slot extruder indices the MQTT layer
        resolves into physical rack positions, and only on rack models.
        """
        caller = traceback.extract_stack(limit=3)[0]
        logger.info(
            "PRINT COMMAND: printer=%s, file=%s, caller=%s:%s:%s",
            printer_id,
            filename,
            caller.filename.split("/")[-1],
            caller.lineno,
            caller.name,
        )
        if printer_id in self._clients:
            return self._clients[printer_id].start_print(
                filename,
                plate_id,
                ams_mapping=ams_mapping,
                timelapse=timelapse,
                bed_levelling=bed_levelling,
                flow_cali=flow_cali,
                vibration_cali=vibration_cali,
                layer_inspect=layer_inspect,
                use_ams=use_ams,
                nozzle_offset_cali=nozzle_offset_cali,
                nozzle_mapping=nozzle_mapping,
                nozzle_slot_extruders=nozzle_slot_extruders,
                plate_uses_only_external=plate_uses_only_external,
            )
        return False

    def stop_print(self, printer_id: int) -> bool:
        """Stop the current print on a connected printer."""
        if printer_id in self._clients:
            return self._clients[printer_id].stop_print()
        return False

    def enable_logging(self, printer_id: int, enabled: bool = True) -> bool:
        """Enable or disable MQTT logging for a printer."""
        if printer_id in self._clients:
            self._clients[printer_id].enable_logging(enabled)
            return True
        return False

    def get_logs(self, printer_id: int) -> list[MQTTLogEntry]:
        """Get MQTT logs for a printer."""
        if printer_id in self._clients:
            return self._clients[printer_id].get_logs()
        return []

    def clear_logs(self, printer_id: int) -> bool:
        """Clear MQTT logs for a printer."""
        if printer_id in self._clients:
            self._clients[printer_id].clear_logs()
            return True
        return False

    def is_logging_enabled(self, printer_id: int) -> bool:
        """Check if logging is enabled for a printer."""
        if printer_id in self._clients:
            return self._clients[printer_id].logging_enabled
        return False

    def send_drying_command(
        self,
        printer_id: int,
        ams_id: int,
        temp: int,
        duration: int,
        mode: int = 1,
        filament: str = "",
        rotate_tray: bool = False,
    ) -> bool:
        """Send AMS drying command to printer."""
        if printer_id not in self._clients:
            return False
        return self._clients[printer_id].send_drying_command(ams_id, temp, duration, mode, filament, rotate_tray)

    def request_status_update(self, printer_id: int) -> bool:
        """Request a full status update from the printer.

        This sends a 'pushall' command to get the latest data including nozzle info.
        """
        if printer_id in self._clients:
            return self._clients[printer_id].request_status_update()
        return False

    # Probe budget for test_connection (#1445). Was a fixed 2s sleep, which was
    # too short for P1S firmware whose broker / TLS handshake routinely takes
    # 3–5s to surface a CONNACK on a cold MQTT session. We now poll up to
    # PROBE_TIMEOUT_SECONDS and early-return the moment we see connected=True,
    # so happy-path connections still finish in ~1–2s and slow brokers get the
    # headroom they need instead of getting falsely rejected.
    PROBE_TIMEOUT_SECONDS = 8.0
    PROBE_POLL_INTERVAL_SECONDS = 0.2

    async def test_connection(
        self,
        ip_address: str,
        serial_number: str,
        access_code: str,
    ) -> dict:
        """Test connection to a printer without persisting.

        Polls for up to PROBE_TIMEOUT_SECONDS and tears the probe client down
        off-loop. The teardown matters: `client.disconnect()` ends in paho's
        `loop_stop()` which `join()`s the network thread — if the thread is
        still mid-TLS-handshake to a slow printer, that join blocks the
        asyncio event loop and every other HTTP request queues behind it. The
        original synchronous teardown produced the #1445 "Docker container
        hangs" symptom on P1S when called from POST /printers/.
        """
        client = BambuMQTTClient(
            ip_address=ip_address,
            serial_number=serial_number,
            access_code=access_code,
        )

        try:
            client.connect()
            deadline = asyncio.get_running_loop().time() + self.PROBE_TIMEOUT_SECONDS
            while not client.state.connected and asyncio.get_running_loop().time() < deadline:
                await asyncio.sleep(self.PROBE_POLL_INTERVAL_SECONDS)

            result = {
                "success": client.state.connected,
                "state": client.state.state if client.state.connected else None,
                "model": client.state.raw_data.get("device_model"),
                # Why the probe failed, when the printer told us: one of the
                # CONNECT_ERROR_* slugs, else None. Lets the add-printer flow
                # and the connection diagnostic say "the printer rejected the
                # access code" instead of an unqualified failure (#2698).
                "reason": None if client.state.connected else client.last_connect_error,
            }
        finally:
            # Off-loop teardown — see docstring. paho's loop_stop() joins the
            # network thread which may still be in a slow TLS handshake.
            await asyncio.to_thread(client.disconnect)

        return result


def get_derived_status_name(state: PrinterState, model: str | None = None) -> str | None:
    """
    Compute a human-readable status name based on printer state.

    Uses stg_cur when available, otherwise derives status from temperature data
    when the printer is heating before a print starts.

    Args:
        state: The printer state to analyze
        model: Optional printer model for model-specific workarounds
    """
    # Firmware bug: some models (A1, P1P, P1S) report stg_cur=0 when not printing.
    # stg_cur=0 maps to "Printing" in STAGE_NAMES, which incorrectly overrides the
    # real state (IDLE, FINISH, FAILED, etc.). Only trust stg_cur when the printer
    # is actually in an active print state (RUNNING or PAUSE).
    if state.state not in ("RUNNING", "PAUSE") and state.stg_cur == 0 and has_stg_cur_idle_bug(model):
        return None

    # If we have a valid calibration stage, use it
    # X1 models use -1 for idle, A1/P1 models use 255 for idle
    # Valid stage numbers are 0-254
    if 0 <= state.stg_cur < 255:
        # A stage number the table does not cover is named "Preparing" rather
        # than "Unknown stage (72)". New models report stages before Bambuddy
        # learns their names -- the H2C still has several -- and the card is
        # the wrong place to say so: the number means nothing to the person
        # reading it, and every stage that has ever turned out to be unnamed
        # was part of the run-up to printing, so "Preparing" is both the more
        # useful answer and the more likely one.
        #
        # This is display only, and deliberately not pushed down into
        # `get_stage_name`. That function also feeds the stage-transition log
        # line and the once-per-session warning that exists precisely to
        # capture unnamed stages so they can be named later (bambu_mqtt.py
        # ~4100) -- there the number is the entire diagnostic value, and
        # replacing it with "Preparing" would hide the very thing that
        # reports these.
        if state.stg_cur not in STAGE_NAMES:
            return "Preparing"
        return get_stage_name(state.stg_cur)

    # If not in RUNNING state, no derived status needed
    if state.state != "RUNNING":
        return None

    # Check if we're in an early phase where temperatures are heating
    temps = state.temperatures or {}
    progress = state.progress or 0

    # Only derive heating status when progress is very low (< 2%)
    # This indicates we're in the preparation phase, not actually printing
    if progress >= 2:
        return None

    # Check bed temperature - if target is set and current is significantly below
    bed_temp = temps.get("bed", 0)
    bed_target = temps.get("bed_target", 0)

    # Check nozzle temperature
    nozzle_temp = temps.get("nozzle", 0)
    nozzle_target = temps.get("nozzle_target", 0)

    # Temperature thresholds: consider "heating" if more than 10°C below target
    TEMP_THRESHOLD = 10

    # Determine what's heating (prioritize bed since it takes longer)
    if bed_target > 30 and (bed_target - bed_temp) > TEMP_THRESHOLD:
        return "Heating heatbed"
    elif nozzle_target > 30 and (nozzle_target - nozzle_temp) > TEMP_THRESHOLD:
        return "Heating nozzle"

    # If targets are set but we're close to them, we might be in final prep
    if bed_target > 30 or nozzle_target > 30:
        if progress == 0 and state.layer_num == 0:
            return "Preparing"

    return None


_PLATE_ID_RE = re.compile(r"plate_(\d+)\.gcode")


def parse_plate_id(gcode_file: str | None) -> int | None:
    """Extract the 1-indexed plate number from a Bambu gcode_file path.

    Returns None when the path is missing or has no `plate_N.gcode` segment.
    Shared by the REST status route and the WebSocket push path so both agree
    on the value sent to the frontend (#881 follow-up).
    """
    if not gcode_file:
        return None
    match = _PLATE_ID_RE.search(gcode_file)
    return int(match.group(1)) if match else None


def resolve_plate_id(state) -> int | None:
    """Resolve the active plate number from a PrinterState.

    Some firmware versions (e.g. P1S 01.10.00.00, #1166) put only the .3mf
    filename in print.gcode_file, so parse_plate_id() returns None and the
    printer card falls back to plate 1 — wrong thumbnail. When Bambuddy
    dispatched the print itself we already know the right plate, so we prefer
    that over the gcode_file echo. The subtask check prevents stale values
    from a previous Bambuddy-dispatched print bleeding into a Studio-direct
    print on the same printer.
    """
    dispatched_plate = getattr(state, "dispatched_plate_id", None)
    dispatched_subtask = getattr(state, "dispatched_subtask", None)
    if (
        dispatched_plate is not None
        and dispatched_subtask is not None
        and state.subtask_name
        and dispatched_subtask == state.subtask_name
    ):
        return dispatched_plate
    return parse_plate_id(state.gcode_file)


def print_ams_mapping(state: PrinterState) -> list:
    """The running print's AMS mapping: the one captured from its print command,
    else whatever the printer's push carried (it rarely does)."""
    if isinstance(state.print_ams_mapping, list):
        return list(state.print_ams_mapping)
    raw = (state.raw_data or {}).get("ams_mapping")
    return list(raw) if isinstance(raw, list) else []


def nozzle_wiring(state: PrinterState) -> dict | None:
    """Which nozzle each AMS is wired to, for the runout hint; None with a
    Filament Track Switch fitted, where an AMS can feed either nozzle and the
    map can still hold an AMS's wiring from before it moved onto the switch."""
    if state.fila_switch and state.fila_switch.installed:
        return None
    return state.ams_extruder_map


def _trays_on_extruder(
    raw_slot: int,
    ams_layout: list[tuple[int, bool]],
    active_extruder: int | None,
    ams_extruder_map: dict | None,
) -> set[int] | None:
    """The global trays a local ``raw_slot`` can be on the AMS units wired to
    ``active_extruder`` (an AMS-HT's one tray for slot 0), or None when the
    wiring is unknown.

    An AMS on the Filament Track Switch has no entry in the map (it can feed
    either nozzle), so one without an entry makes the answer unknown too.
    """
    if active_extruder is None or not ams_extruder_map:
        return None
    trays: set[int] = set()
    for ams_id, is_ht in ams_layout:
        nozzle = ams_extruder_map.get(str(ams_id))
        if nozzle is None:
            return None
        if nozzle != active_extruder:
            continue
        if not is_ht:
            trays.add(ams_id * 4 + raw_slot)
        elif raw_slot == 0:
            trays.add(ams_id)
    return trays


def resolve_expected_tray(
    raw_slot: int | None,
    ams_layout: list[tuple[int, bool]],
    mapping_raw: object,
    active_extruder: int | None = None,
    ams_extruder_map: dict | None = None,
) -> int | None:
    """Globalise a raw firmware ``tray_tar``/``tray_pre`` value for the runout UI (#2587).

    The firmware reports the target/previous slot as a bare number whose meaning
    depends on the AMS layout (see ``PrinterState.tray_tar``). This mirrors the
    ``tray_now`` handling so the resolved ID lines up with what the AMS graphic
    already highlights via ``ams_id*4 + slot``.

    ``ams_layout`` is a list of ``(ams_id, is_ams_ht)`` for the connected units.

    - ``255``/``-1`` (none/idle) -> ``None``
    - ``254`` (external spool) -> ``254``
    - ``128``-``135`` (AMS-HT) -> already global, returned as-is
    - ``0``-``3`` local slot:
        * exactly one regular AMS -> ``ams_id*4 + slot``
        * several regular AMS -> resolved via the snow-encoded ``mapping`` field
          (each entry = ``ams_hw_id*256 + slot``; ``65535`` = unmapped), then
          narrowed to the AMS that feed ``active_extruder`` (``ams_extruder_map``,
          dual-nozzle printers), or ``None`` when it stays ambiguous (honest
          "can't determine")
        * no regular AMS -> ``None``
    - ``4``-``15`` -> already a global regular-AMS ID, returned as-is

    Returns ``None`` for anything it can't place, so the caller surfaces a
    "check the printer" message instead of pointing at the wrong slot.
    """
    if raw_slot is None or raw_slot in (255, -1):
        return None
    if raw_slot == 254:
        return 254
    if 128 <= raw_slot <= 135:
        return raw_slot
    if 0 <= raw_slot <= 3:
        regular = [ams_id for ams_id, is_ht in ams_layout if not is_ht]
        if len(regular) == 1:
            return regular[0] * 4 + raw_slot
        if len(regular) > 1:
            candidates: set[int] = set()
            for value in mapping_raw if isinstance(mapping_raw, list) else []:
                if not isinstance(value, int) or value >= 65535:
                    continue
                ams_hw_id = value >> 8
                slot = value & 0xFF
                if 0 <= ams_hw_id <= 3 and (slot & 0x03) == raw_slot:
                    candidates.add(ams_hw_id * 4 + raw_slot)
                elif 128 <= ams_hw_id <= 135 and raw_slot == 0:
                    candidates.add(ams_hw_id)
            if len(candidates) == 1:
                return candidates.pop()
            # The print paused on the active nozzle, so the slot is on an AMS
            # that feeds it.
            narrowed = _trays_on_extruder(raw_slot, ams_layout, active_extruder, ams_extruder_map)
            if narrowed is None:
                return None
            if candidates:
                narrowed &= candidates
            return narrowed.pop() if len(narrowed) == 1 else None
        return None
    if 4 <= raw_slot <= 15:
        return raw_slot
    # 24-27 = A2L AMS-Lite (normalised unit 6) global tray ids, already resolved.
    if 24 <= raw_slot <= 27:
        return raw_slot
    return None


def _ams_temp(raw: object) -> float | None:
    """The AMS unit temperature as a number, or None when it is not one."""
    if isinstance(raw, bool) or not isinstance(raw, (int, float, str)):
        return None
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return None
    # NaN or infinity would put a token in the WebSocket frame that the
    # browser's JSON.parse rejects, dropping the whole status push.
    return value if math.isfinite(value) else None


def ams_units_to_dicts(
    ams_list: list,
    resolve_slot_k: Callable,
    drying_targets: dict[int, dict] | None = None,
) -> list[dict]:
    """The AMS units of a printer's ``raw_data["ams"]``, as the printer card reads them.

    The one serialiser for both the WebSocket push and the REST status
    (routes/printers.py). The card is seeded from REST and shallow-merged with
    every push, so any field one of them lacks reads as missing on the page:
    ``exists`` (#2670), the stalled-countdown flag and the drying status were
    each lost that way while the two were separate copies.
    """
    ams_units: list[dict] = []
    for ams_data in ams_list:
        # Skip if ams_data is not a dict (defensive check)
        if not isinstance(ams_data, dict):
            continue
        ams_id = int(ams_data.get("id", 0))
        trays = []
        for tray in ams_data.get("tray", []):
            tag_uid = tray.get("tag_uid")
            if tag_uid in ("", "0000000000000000"):
                tag_uid = None
            tray_uuid = tray.get("tray_uuid")
            if tray_uuid in ("", "00000000000000000000000000000000"):
                tray_uuid = None

            # Get K value: first try tray's k field, then lookup from K-profiles
            k_value = tray.get("k")
            cali_idx = tray.get("cali_idx")
            if k_value is None:
                k_value = resolve_slot_k(cali_idx, ams_id, int(tray.get("id", 0)))

            # P1S / A1 Mini physically-empty-slot signal (#1322 follow-up by
            # @RosdasHH): for a truly empty slot the firmware sends only
            # {"id": N} — no state, no tray_type, no anything else. Treat
            # that as the firmware's "no spool" indicator (state=9) so the
            # assign-spool path in inventory.py can short-circuit a MQTT
            # publish the firmware would silently drop anyway. The
            # post-"Reset Slot" A1 Mini BMCU case sends a populated payload
            # (state=3, tray_type="") — different shape, doesn't match this
            # guard, still attempts the MQTT push per the #1322 fix.
            state_val = tray.get("state")
            if state_val is None and len(tray) == 1 and "id" in tray:
                state_val = 9

            trays.append(
                {
                    "id": int(tray.get("id", 0)),
                    "tray_color": tray.get("tray_color"),
                    "tray_type": tray.get("tray_type"),
                    "tray_sub_brands": tray.get("tray_sub_brands"),
                    "tray_id_name": tray.get("tray_id_name"),
                    "tray_info_idx": tray.get("tray_info_idx"),
                    "remain": tray.get("remain", 0),
                    "k": k_value,
                    "cali_idx": cali_idx,
                    "tag_uid": tag_uid,
                    "tray_uuid": tray_uuid,
                    "nozzle_temp_min": tray.get("nozzle_temp_min"),
                    "nozzle_temp_max": tray.get("nozzle_temp_max"),
                    "drying_temp": tray.get("drying_temp"),
                    "drying_time": tray.get("drying_time"),
                    "state": state_val,
                    # Firmware's authoritative presence bit (tray_exist_bits),
                    # set by apply_tray_exist_bits. Without it getEmptySlotKind
                    # falls back to the firmware-variant state 9/10 heuristic,
                    # wrong for AMS-HT in both directions (#2670).
                    "exists": tray.get("exists"),
                }
            )
        # Percentage only — the 1-5 index is inverted and must never stand
        # in for one (#3140). See utils/ams_humidity.
        humidity_pct = ams_humidity_percent(ams_data)
        humidity_value = int(round(humidity_pct)) if humidity_pct is not None else None

        # AMS-HT by its unit id: a regular AMS can report fewer than 4 trays.
        is_ams_ht = 128 <= ams_id <= 135

        # Active-cycle filament + target temperature for the badge.
        # Bambu does not echo the cycle's chosen filament/temp on the
        # per-tick AMS push, so prefer the cached target from the last
        # ``send_drying_command``. When we have no record (drying
        # started in a previous backend lifetime, or the cache was
        # never seeded), the loaded trays can still name the filament
        # if they agree — but never the temperature, which only the
        # cache knows. See uniform_tray_filament_hint.
        ams_id_int = int(ams_data.get("id", 0))
        target = (drying_targets or {}).get(ams_id_int)
        dry_target_temp: int | None = None
        dry_filament: str | None = None
        if target:
            temp_val = target.get("temp")
            fil_val = target.get("filament") or ""
            if temp_val is not None:
                try:
                    dry_target_temp = int(temp_val)
                except (TypeError, ValueError):
                    dry_target_temp = None
            if fil_val:
                dry_filament = str(fil_val)
        if not dry_filament:
            dry_filament = uniform_tray_filament_hint([tray.get("tray_type") or "" for tray in trays])

        ams_units.append(
            {
                "id": ams_id_int,
                "humidity": humidity_value,
                # A number, as the REST schema has always coerced it: the
                # firmware sends a string, and the card showed "24°C" after a
                # page load but "24.0°C" after the first push.
                "temp": _ams_temp(ams_data.get("temp")),
                "is_ams_ht": is_ams_ht,
                "tray": trays,
                # Serial number: Bambu MQTT uses "sn" key on AMS unit objects
                "serial_number": str(ams_data.get("sn") or ams_data.get("serial_number") or ""),
                # Firmware version: populated by _handle_version_info from get_version
                "sw_ver": str(ams_data.get("sw_ver") or ""),
                # Drying: dry_time > 0 means drying is active (minutes remaining)
                "dry_time": int(ams_data.get("dry_time") or 0),
                # Stall flag stamped onto raw_data by the MQTT layer: the
                # firmware set the countdown but it never ticked. Without it
                # the card falls back to the amber "active cycle" badge, so
                # the neutral "Drying not started" state is unreachable.
                "dry_countdown_stalled": bool(ams_data.get("dry_countdown_stalled") or False),
                # Drying status from info hex bits (0=Off, 1=Checking, 2=Drying, 3=Cooling, etc.)
                "dry_status": int(ams_data.get("dry_status") or 0),
                "dry_sub_status": int(ams_data.get("dry_sub_status") or 0),
                # Cannot-dry reasons from firmware (e.g. 1=InsufficientPower, 8=NeedPluginPower)
                "dry_sf_reason": list(ams_data.get("dry_sf_reason") or []),
                # Active-cycle filament name + target temperature
                "dry_target_temp": dry_target_temp,
                "dry_filament": dry_filament,
                # Module type: "ams", "n3f", "n3s" (from get_version)
                "module_type": str(ams_data.get("module_type") or ""),
            }
        )
    return ams_units


def vt_trays_to_dicts(raw: object, resolve_slot_k: Callable) -> list[dict]:
    """The external spools of a printer's ``raw_data["vt_tray"]``.

    Shared by the WebSocket push and the REST status, like ``ams_units_to_dicts``.
    """
    vt_tray: list[dict] = []
    vt_tray_raw = raw
    # Defensive: MQTT sends vt_tray as a dict; normalize to list
    if isinstance(vt_tray_raw, dict):
        vt_tray_raw = [vt_tray_raw]
    elif not isinstance(vt_tray_raw, list):
        vt_tray_raw = []
    for vt_data in vt_tray_raw:
        vt_tag_uid = vt_data.get("tag_uid")
        if vt_tag_uid in ("", "0000000000000000"):
            vt_tag_uid = None
        vt_tray_uuid = vt_data.get("tray_uuid")
        if vt_tray_uuid in ("", "00000000000000000000000000000000"):
            vt_tray_uuid = None

        # Get K value for vt_tray
        vt_k_value = vt_data.get("k")
        vt_cali_idx = vt_data.get("cali_idx")
        if vt_k_value is None:
            # External holder: id 254 is Ext-L, 255 is Ext-R. The resolver
            # takes the 0/1 tray index, so normalise before asking.
            vt_id = int(vt_data.get("id", 254))
            vt_k_value = resolve_slot_k(vt_cali_idx, 255, vt_id - 254 if vt_id >= 254 else vt_id)

        tray_id = int(vt_data.get("id", 254))
        vt_tray.append(
            {
                "id": tray_id,
                "tray_color": vt_data.get("tray_color"),
                "tray_type": vt_data.get("tray_type"),
                "tray_sub_brands": vt_data.get("tray_sub_brands"),
                "tray_id_name": vt_data.get("tray_id_name"),
                "tray_info_idx": vt_data.get("tray_info_idx"),
                "remain": vt_data.get("remain", 0),
                "k": vt_k_value,
                "cali_idx": vt_cali_idx,
                "tag_uid": vt_tag_uid,
                "tray_uuid": vt_tray_uuid,
                "nozzle_temp_min": vt_data.get("nozzle_temp_min"),
                "nozzle_temp_max": vt_data.get("nozzle_temp_max"),
            }
        )
    return vt_tray


def printer_state_to_dict(
    state: PrinterState,
    printer_id: int | None = None,
    model: str | None = None,
    drying_targets: dict[int, dict] | None = None,
) -> dict:
    """Convert PrinterState to a JSON-serializable dict.

    Args:
        state: The printer state to convert
        printer_id: Optional printer ID for generating cover URLs
        model: Optional printer model for filtering unsupported features
        drying_targets: Optional per-AMS active-cycle params
            (``{ams_id: {"filament": str, "temp": int}}``) sourced from the
            BambuMQTTClient cache so the badge can display "PETG @ 65°C".
    """
    # Parse AMS data from raw_data
    ams_units = []
    vt_tray = []
    raw_data = state.raw_data or {}

    # K value for a slot's bound profile. Shared with the REST serializer of
    # the same card (routes/printers.py) so the two cannot answer differently:
    # this one used to key on cali_idx alone, which on a dual-nozzle machine
    # meant whichever nozzle's table was listed last won the slot.
    resolve_slot_k = build_slot_k_resolver(state)

    if "ams" in raw_data and isinstance(raw_data["ams"], list):
        ams_units = ams_units_to_dicts(raw_data["ams"], resolve_slot_k, drying_targets)

    # Parse virtual tray (external spool) — now a list
    if "vt_tray" in raw_data:
        vt_tray = vt_trays_to_dicts(raw_data["vt_tray"], resolve_slot_k)

    # Get ams_extruder_map from raw_data (populated by MQTT handler from AMS info field)
    ams_extruder_map = raw_data.get("ams_extruder_map", {})

    # Filter out chamber temp for models that don't have a real sensor
    # P1P, P1S, A1, A1Mini report meaningless chamber_temper values
    temperatures = state.temperatures
    if not supports_chamber_temp(model):
        temperatures = {
            k: v for k, v in temperatures.items() if k not in ("chamber", "chamber_target", "chamber_heating")
        }

    result = {
        "connected": state.connected,
        "state": state.state,
        "current_print": state.current_print,
        "subtask_name": state.subtask_name,
        "gcode_file": state.gcode_file,
        "progress": state.progress,
        "remaining_time": state.remaining_time,
        "layer_num": state.layer_num,
        "total_layers": state.total_layers,
        "temperatures": temperatures,
        "hms_errors": [
            {
                "code": e.code,
                "attr": e.attr,
                "module": e.module,
                "severity": e.severity,
                "actions": e.actions,
                "job_id": e.job_id,
                "full_code": e.full_code,
                # Same field as the status response carries (#2926) — a relay
                # watching the stream should not have to poll REST to find out
                # what a fault means.
                "description": e.description,
            }
            for e in (state.hms_errors or [])
        ],
        # AMS data for filament colors
        "ams": ams_units if ams_units else None,
        "vt_tray": vt_tray,
        # AMS status for filament change tracking
        "ams_status_main": state.ams_status_main,
        "ams_status_sub": state.ams_status_sub,
        "tray_now": state.tray_now,
        # Grows only when tray_now changes, which already triggers a push.
        "tray_change_log": [[tray, layer] for tray, layer in getattr(state, "tray_change_log", None) or []],
        # Runout / filament-replacement guidance (#2587). Only meaningful while
        # PAUSED — resolve the firmware's target/previous slot to a global tray ID
        # so the AMS graphic can highlight the slot the print now expects and name
        # the one that ran out. None when idle, not paused, or unresolvable.
        "expected_tray": (
            resolve_expected_tray(
                state.tray_tar,
                [(u["id"], u.get("is_ams_ht", False)) for u in ams_units],
                raw_data.get("mapping"),
                active_extruder=state.active_extruder,
                ams_extruder_map=nozzle_wiring(state),
            )
            if state.state == "PAUSE"
            else None
        ),
        "previous_tray": (
            resolve_expected_tray(
                state.tray_pre,
                [(u["id"], u.get("is_ams_ht", False)) for u in ams_units],
                raw_data.get("mapping"),
                active_extruder=state.active_extruder,
                ams_extruder_map=nozzle_wiring(state),
            )
            if state.state == "PAUSE"
            else None
        ),
        # Per-AMS extruder map: {ams_id: extruder_id} where 0=right, 1=left
        "ams_extruder_map": ams_extruder_map,
        # The running print's mapping, for the slot pills (#1762).
        "ams_mapping": print_ams_mapping(state),
        # Pushed so a change reaches the card at once rather than with the
        # 30 s REST poll; the frontend shallow-merges each push over its status.
        "nozzles": [
            {"nozzle_type": n.nozzle_type, "nozzle_diameter": n.nozzle_diameter} for n in (state.nozzles or [])
        ],
        "firmware_version": state.firmware_version,
        "developer_mode": state.developer_mode,
        # Filament Track Switch. Both fields have to travel on the WebSocket, not
        # only on the REST status: the frontend shallow-merges each push over its
        # cached status, so a field that is absent here keeps whatever the last
        # full fetch left behind. Omitting them meant the AMS inlet badges only
        # ever changed on a page reload.
        "fila_switch": (
            {
                "installed": True,
                "in_slots": list(state.fila_switch.in_slots),
                "out_extruders": list(state.fila_switch.out_extruders),
                "stat": state.fila_switch.stat,
                "info": state.fila_switch.info,
                # Mirrors BambuStudio's DevFilaSwitch::IsReady — every AMS has to
                # be bound to an inlet before the switch can route anything. Until
                # the operator has done that on the printer's Manual AMS Setup
                # screen, Studio refuses a load outright rather than sending a
                # command the firmware cannot act on, and so do we.
                # An empty AMS list is "ready", as it is in Studio: there is then
                # no slot to load from, so nothing can reach the check anyway, and
                # reporting not-ready would only mean a confusing toast on a
                # payload that has not carried the AMS block yet.
                #
                # An AMS still reporting a real extruder id rather than 0xE has no
                # inlet entry, so a machine with one hard-wired unit reads as not
                # ready. That looks harsh but is exactly Studio's own rule —
                # IsReady() requires a switcher position on *every* AMS, and only
                # the 0xE branch ever sets one (DevFilaSystem.cpp:596-615).
                "ready": all(str(u["id"]) in state.ams_switch_inlet for u in ams_units),
            }
            if state.fila_switch and state.fila_switch.installed
            else None
        ),
        # Per-AMS FTS inlet binding: {ams_id: "A" | "B"}. Gated on the accessory
        # so a stale binding cannot outlive it being unplugged.
        "ams_switch_inlet": (dict(state.ams_switch_inlet) if state.fila_switch and state.fila_switch.installed else {}),
        # Which AMS slot each hotend is fed from: {extruder_id: {...}}. Travels on
        # the WebSocket for the same reason as fila_switch above — the frontend
        # shallow-merges pushes over its cached status, so an absent field keeps a
        # stale value forever. Empty on printers that do not report it.
        "extruder_slots": {
            str(ext_id): {
                "ams_id": slot.ams_id,
                "slot_id": slot.slot_id,
                "has_filament": slot.has_filament,
            }
            for ext_id, slot in state.extruder_slots.items()
        },
        # WiFi signal strength
        "wifi_signal": state.wifi_signal,
        "wired_network": state.wired_network,
        "door_open": state.door_open,
        # AMS Filament Backup state (auto-switch to second spool). Tri-state:
        # True / False / None. None = unknown or unsupported (A1 family). UI
        # uses this to drive the small status icon next to the AMS drying icon.
        "ams_filament_backup": state.ams_filament_backup,
        # Calibration stage tracking
        "stg_cur": state.stg_cur,
        "stg_cur_name": get_derived_status_name(state, model),
        # Printable objects count for skip objects feature
        "printable_objects_count": len(state.printable_objects),
        # Fan speeds (0-100 percentage, None if not available)
        "cooling_fan_speed": state.cooling_fan_speed,
        "big_fan1_speed": state.big_fan1_speed,
        "big_fan2_speed": state.big_fan2_speed,
        "heatbreak_fan_speed": state.heatbreak_fan_speed,
        "left_aux_fan_speed": state.left_aux_fan_speed,
        "exhaust_fan_present": state.exhaust_fan_present,
        # Chamber light state
        "chamber_light": state.chamber_light,
        # Active extruder for dual-nozzle printers (0=right, 1=left)
        "active_extruder": state.active_extruder,
        # Print speed mode (1=silent, 2=standard, 3=sport, 4=ludicrous)
        "speed_level": state.speed_level,
        # Switched on the printer's screen too; REST-only, the card showed the
        # old mode until the next poll (PSF-4).
        "airduct_mode": state.airduct_mode,
        # H2C nozzle rack (tool-changer dock positions)
        # Map raw MQTT field names (type/diameter) to schema names (nozzle_type/nozzle_diameter)
        "nozzle_rack": [
            {
                "id": n.get("id", 0),
                "nozzle_type": n.get("type", ""),
                "nozzle_diameter": n.get("diameter", ""),
                "wear": n.get("wear"),
                "stat": n.get("stat"),
                "max_temp": n.get("max_temp", 0),
                "serial_number": n.get("serial_number", ""),
                "filament_color": n.get("filament_color", ""),
                "filament_id": n.get("filament_id", ""),
                # The push replaces the whole rack: without it a slot lost its
                # material label on every push (PSF-3).
                "filament_type": n.get("filament_type", ""),
            }
            for n in (state.nozzle_rack or [])
        ],
        # AMS drying support
        "supports_drying": supports_drying(model, state.firmware_version),
        "supports_drying_while_printing": supports_drying_while_printing(model, state.firmware_version),
        "drying_screen_only": drying_screen_only(model),
        # 1-indexed plate number parsed from gcode_file (e.g. /Metadata/plate_2.gcode).
        # Pushed via WebSocket so the printer card picks up plate transitions within
        # a multi-plate 3MF without waiting for the 30 s REST poll (#881 follow-up).
        # current_archive_id is intentionally REST-only — it's stable for the life
        # of a print and needs a DB lookup the WebSocket path shouldn't pay for.
        "current_plate_id": resolve_plate_id(state),
        # Plate-clear gate (#939). Lives on the PrinterManager rather than PrinterState,
        # so surface it here — without this, WebSocket merges drop the flag and the
        # "Clear Plate" button only appears when the 30 s REST fallback poll runs.
        "awaiting_plate_clear": printer_manager.is_awaiting_plate_clear(printer_id) if printer_id else False,
    }
    # Add cover URL if there's an active print and printer_id is provided
    # Include PAUSE state so skip objects modal can show cover
    if printer_id and state.state in ("RUNNING", "PAUSE") and state.gcode_file:
        result["cover_url"] = f"/api/v1/printers/{printer_id}/cover"
    else:
        result["cover_url"] = None
    # The display name + model, so a WS consumer can label the printer from the
    # snapshot alone (#963 follow-up; the SPA itself takes them from
    # /api/v1/printers). PrinterInfo only carries name/serial_number; the model
    # comes through via the `model` arg.
    if printer_id:
        _printer_info = printer_manager.get_printer(printer_id)
        if _printer_info is not None:
            result["name"] = _printer_info.name
    if model:
        result["model"] = model
    return result


# Global printer manager instance
printer_manager = PrinterManager()


async def init_printer_connections(db: AsyncSession):
    """Initialize connections to all active printers.

    Connections are started concurrently. ``connect_printer()`` is non-blocking
    apart from a fixed 1-second settle wait — ``BambuMQTTClient.connect()`` only
    calls ``connect_async()`` + ``loop_start()``, so the handshake happens on a
    background thread and the coroutine's only real cost is that ``sleep(1)``. A
    serial loop therefore spent one whole second per printer inside the FastAPI
    lifespan *before* the ASGI server begins serving: on a large farm that was
    ~100s of dead air before port 8000 responded (issue #2572, reporter's
    93-printer farm). Gathering overlaps the settle waits so the whole step takes
    ~1s regardless of fleet size. Exceptions are isolated per printer with
    ``return_exceptions=True`` so one unreachable row can't abort the rest — or
    startup itself, which the old serial loop's un-caught await would have done.

    All columns ``connect_printer`` reads are eagerly loaded by the SELECT above
    and touched synchronously before its trailing ``await``, so no concurrent
    lazy-load is triggered on the shared session.
    """
    result = await db.execute(select(Printer).where(Printer.is_active.is_(True)))
    printers = result.scalars().all()

    outcomes = await asyncio.gather(
        *(printer_manager.connect_printer(printer) for printer in printers),
        return_exceptions=True,
    )
    for printer, outcome in zip(printers, outcomes, strict=True):
        if isinstance(outcome, Exception):
            logger.warning(
                "Failed to connect printer %s (%s) at startup: %s",
                printer.id,
                printer.name,
                outcome,
            )