fix(smart-plugs): select printer power deterministically

backend/app/api/routes/smart_plugs.py

@@ -41,6 +41,11 @@
 from backend.app.services.printer_manager import printer_manager
 from backend.app.services.rest_smart_plug import rest_smart_plug_service
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import (
+    can_be_switched,
+    pick_power_plug,
+    plugs_for_printer,
+)
 from backend.app.services.tasmota import tasmota_service
 from backend.app.utils.local_time import utcnow_naive
 
@@ -183,25 +188,8 @@ async def get_smart_plug_by_printer(
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_READ),
 ):
-    """Get the main smart plug assigned to a printer.
-
-    When multiple plugs are assigned (e.g., a regular plug + script),
-    returns the main (non-script) plug for power control.
-    """
-    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-    plugs = result.scalars().all()
-
-    if not plugs:
-        return None
-
-    # If multiple plugs, prefer the non-script one (main power plug)
-    for plug in plugs:
-        is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
-        if not is_script:
-            return plug
-
-    # All are scripts, return the first one
-    return plugs[0]
+    """Get the linked row that best represents the printer's power supply."""
+    return pick_power_plug(await plugs_for_printer(db, printer_id))
 
 
 @router.get("/by-printer/{printer_id}/scripts", response_model=list[SmartPlugResponse])
@@ -216,12 +204,20 @@ async def get_script_plugs_by_printer(
     show_on_printer_card enabled.
     Used to display action buttons alongside the main power plug.
     """
-    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-    plugs = result.scalars().all()
+    plugs = await plugs_for_printer(db, printer_id)
+    main_plug = pick_power_plug(plugs)
+    duplicate_id = main_plug.id if main_plug and can_be_switched(main_plug) else None
 
-    # Filter to HA entities with show_on_printer_card enabled
+    # A switchable main entity is already rendered in the Power row. Scripts
+    # remain here even in the script-only fallback so their one-click action is
+    # preserved.
     ha_entities = [
-        plug for plug in plugs if plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.show_on_printer_card
+        plug
+        for plug in plugs
+        if plug.plug_type == "homeassistant"
+        and plug.ha_entity_id
+        and plug.show_on_printer_card
+        and plug.id != duplicate_id
     ]
     return ha_entities
 

backend/app/core/database.py

@@ -2356,10 +2356,9 @@ async def run_migrations(conn):
     else:
         await _safe_execute(conn, "ALTER TABLE users ALTER COLUMN password_hash DROP NOT NULL")
 
-    # Migration: Add energy_start_kwh to print_archives (#941)
-    # Persists the smart plug lifetime counter captured at print start, so per-print
-    # energy tracking survives a backend restart mid-print.
+    # Persist the starting counter and its exact plug across restarts (#941, #140).
     await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN energy_start_kwh REAL")
+    await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN energy_start_plug_id INTEGER")
     # Migration: Add subtask_id to print_archives (#972)
     # MQTT-provided task identifier used to resume the same archive row across a
     # backend restart mid-print. Without it, a long print (e.g. 13h) triggers

backend/app/main.py

@@ -84,7 +84,6 @@
 from backend.app.core.database import async_session, engine, init_db
 from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
-from backend.app.models.smart_plug import SmartPlug
 from backend.app.services import (
     active_print_provenance as print_provenance,
     business_runtime,
@@ -123,6 +122,7 @@
     resolve_plate_id,
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import read_printer_energy
 from backend.app.services.spool_assignment_notifications import (
     notify_missing_spool_assignments_on_print_start,
 )
@@ -669,21 +669,19 @@ async def _record_energy_start(archive, printer_id: int, db, *, context: str = "
     """
     _logger = logging.getLogger(__name__)
     try:
-        plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-        plug = plug_result.scalar_one_or_none()
-        if not plug:
-            _logger.info("[ENERGY] No smart plug for printer %s (archive %s)", printer_id, archive.id)
-            return False
-        energy = await _get_plug_energy(plug, db)
-        if not energy or energy.get("total") is None:
-            _logger.warning("[ENERGY] No 'total' in energy response for archive %s", archive.id)
+        selected = await read_printer_energy(
+            db, printer_id, _get_plug_energy, log_prefix="ENERGY", context=f"archive {archive.id}"
+        )
+        if selected is None:
             return False
-        archive.energy_start_kwh = float(energy["total"])
+        plug, energy = selected
+        archive.energy_start_kwh, archive.energy_start_plug_id = float(energy["total"]), plug.id
         await db.commit()
         _logger.info(
-            "[ENERGY] Recorded starting energy%s for archive %s: %s kWh",
+            "[ENERGY] Recorded starting energy%s for archive %s from plug '%s': %s kWh",
             f" ({context})" if context else "",
             archive.id,
+            plug.name,
             energy["total"],
         )
         return True
@@ -4485,15 +4483,10 @@ async def _notify_no_archive():
 
     log_timing("Print log entry")
 
-    # Run slow operations as background tasks to avoid blocking the event loop
-    # These operations can take 5-10+ seconds and would freeze the UI if awaited
+    # Slow operations run in the background so completion does not freeze the UI.
 
     async def _background_energy_calculation():
-        """Calculate and save energy usage in background.
-
-        Reads the starting kWh from the archive row (#941: persisted so a mid-print
-        backend restart no longer loses per-print energy data).
-        """
+        """Calculate energy from the persisted start reading without blocking."""
         try:
             logger.info("[ENERGY-BG] Starting energy calculation for archive %s", archive_id)
             async with async_session() as db:
@@ -4503,22 +4496,23 @@ async def _background_energy_calculation():
                 if archive is None:
                     logger.warning("[ENERGY-BG] Archive %s no longer exists", archive_id)
                     return
-                starting_kwh = archive.energy_start_kwh
-                if starting_kwh is None:
-                    logger.info("[ENERGY-BG] No start kWh recorded for archive %s", archive_id)
+                starting_kwh, starting_plug_id = archive.energy_start_kwh, archive.energy_start_plug_id
+                if starting_kwh is None or not isinstance(starting_plug_id, int):
+                    logger.info("[ENERGY-BG] No start energy source recorded for archive %s", archive_id)
                     return
 
-                plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-                plug = plug_result.scalar_one_or_none()
-                if plug is None:
-                    logger.info("[ENERGY-BG] No smart plug for printer %s", printer_id)
-                    return
-
-                energy = await _get_plug_energy(plug, db)
-                logger.info("[ENERGY-BG] Energy response: %s", energy)
-                if not energy or energy.get("total") is None:
-                    logger.warning("[ENERGY-BG] No 'total' in energy response")
+                selected = await read_printer_energy(
+                    db,
+                    printer_id,
+                    _get_plug_energy,
+                    log_prefix="ENERGY-BG",
+                    context=f"archive {archive_id}",
+                    plug_id=starting_plug_id,
+                )
+                if selected is None:
                     return
+                plug, energy = selected
+                logger.info("[ENERGY-BG] Energy response from plug '%s': %s", plug.name, energy)
 
                 energy_used = round(energy["total"] - starting_kwh, 4)
                 logger.info("[ENERGY-BG] Per-print energy: %s kWh", energy_used)
@@ -4537,9 +4531,7 @@ async def _background_energy_calculation():
                 cost_per_kwh = float(energy_cost_per_kwh) if energy_cost_per_kwh else 0.15
                 energy_cost_value = round(energy_used * cost_per_kwh, 3)
 
-                # First-run-only overwrite of archive.energy_kwh / energy_cost so a
-                # reprint doesn't visually clobber the source archive's energy data
-                # (#1378). Reprint energy lives in the matching PrintLogEntry below.
+                # Do not let reprints clobber source-archive energy (#1378).
                 from sqlalchemy import func
 
                 from backend.app.models.print_log import PrintLogEntry

backend/app/models/archive.py

@@ -85,6 +85,7 @@ class PrintArchive(Base):
     # Plug lifetime counter captured at print start; delta at print end becomes energy_kwh.
     # Persisted so per-print tracking survives backend restarts mid-print (#941).
     energy_start_kwh: Mapped[float | None] = mapped_column(Float)
+    energy_start_plug_id: Mapped[int | None] = mapped_column(Integer)
 
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

backend/app/services/print_scheduler.py

@@ -53,6 +53,10 @@
     supports_drying_while_printing,
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import (
+    pick_power_plug as select_power_plug,
+    plugs_for_printer,
+)
 from backend.app.utils.filename import derive_remote_filename
 from backend.app.utils.printer_models import is_gcode_compatible, normalize_printer_model
 from backend.app.utils.threemf_tools import extract_bed_temperature_from_3mf
@@ -600,11 +604,11 @@ def _claim_library_row(candidate: PrintQueueItem) -> None:
 
                     # If printer not connected, try to power on via smart plug
                     if not printer_connected:
-                        plugs = await self._get_smart_plugs(db, item.printer_id)
+                        plugs = await plugs_for_printer(db, item.printer_id)
                         auto_on_plugs = [p for p in plugs if p.auto_on and p.enabled]
                         if auto_on_plugs:
                             logger.info("Printer %s offline, attempting to power on via smart plug(s)", item.printer_id)
-                            primary_plug = self._pick_power_plug(auto_on_plugs)
+                            primary_plug = select_power_plug(auto_on_plugs) or auto_on_plugs[0]
                             powered_on = await self._power_on_and_wait(primary_plug, item.printer_id, db)
                             if powered_on:
                                 # Also turn on any remaining auto_on plugs (e.g., filter)
@@ -2439,19 +2443,6 @@ async def _stop_drying(self, printer_id: int):
                 printer_manager.send_drying_command(printer_id, ams_id, 0, 0, mode=0)
         self._drying_in_progress.pop(printer_id, None)
 
-    async def _get_smart_plugs(self, db: AsyncSession, printer_id: int) -> list[SmartPlug]:
-        """Get all smart plugs associated with a printer."""
-        result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-        return list(result.scalars().all())
-
-    @staticmethod
-    def _pick_power_plug(auto_on_plugs: list[SmartPlug]) -> SmartPlug:
-        """Pick the plug that actually powers the printer, falling back to first."""
-        for plug in auto_on_plugs:
-            if plug.controls_printer_power:
-                return plug
-        return auto_on_plugs[0]
-
     # Bundled defaults for preheat_filament_targets (#1468). Values are the
     # chamber-temperature recommendations BambuStudio ships for the matching
     # filament profile; users can override via Settings → Workflow → Preheat

backend/app/services/smart_plug_selection.py

@@ -0,0 +1,147 @@
+"""Shared business rule for a printer's primary smart plug (#140).
+
+A printer may legitimately have several linked rows: its outlet, accessories,
+Home Assistant scripts, or a read-only MQTT meter.  Consumers must not depend
+on database row order when deciding which row represents printer power.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Awaitable, Callable, Iterable, Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.smart_plug import SmartPlug
+
+EnergyReader = Callable[[SmartPlug, AsyncSession | None], Awaitable[dict | None]]
+logger = logging.getLogger(__name__)
+
+
+def is_script_plug(plug: SmartPlug) -> bool:
+    """Return whether *plug* is a runnable HA script rather than a switch."""
+    entity_id = getattr(plug, "ha_entity_id", None)
+    return bool(getattr(plug, "plug_type", None) == "homeassistant" and entity_id and entity_id.startswith("script."))
+
+
+def can_be_switched(plug: SmartPlug) -> bool:
+    """Match the device kinds accepted by the manual control path."""
+    return not is_script_plug(plug) and getattr(plug, "plug_type", None) != "mqtt"
+
+
+def reports_power(plug: SmartPlug) -> bool:
+    """Whether configuration exposes a place to read current watts from."""
+    plug_type = getattr(plug, "plug_type", None)
+    if plug_type == "homeassistant":
+        return bool(getattr(plug, "ha_power_entity", None))
+    if plug_type == "mqtt":
+        return bool(getattr(plug, "mqtt_power_topic", None) or getattr(plug, "mqtt_topic", None))
+    if plug_type == "rest":
+        return bool(getattr(plug, "rest_power_path", None))
+    return True
+
+
+def power_plug_rank(plug: SmartPlug) -> tuple[bool, bool, bool, bool, bool, int]:
+    """Sort key for the row that represents the printer's power supply.
+
+    The explicit printer-power relationship comes first, including HA scripts
+    that start a printer. Switchability, display, and metering details only
+    break ties between equally assigned rows. Nullable legacy flags rank behind
+    explicit true values. Lowest id is the final stable tiebreaker.
+    """
+    plug_id = getattr(plug, "id", None)
+    return (
+        not bool(getattr(plug, "controls_printer_power", False)),
+        not can_be_switched(plug),
+        not bool(getattr(plug, "enabled", False)),
+        not bool(getattr(plug, "show_on_printer_card", False)),
+        not reports_power(plug),
+        plug_id if isinstance(plug_id, int) else 2**63 - 1,
+    )
+
+
+def _power_assignment(plug: SmartPlug) -> tuple[bool, bool, bool]:
+    """The identity-bearing part of the rank, excluding presentation details."""
+    return (
+        can_be_switched(plug),
+        bool(getattr(plug, "controls_printer_power", False)),
+        bool(getattr(plug, "enabled", False)),
+    )
+
+
+def rank_power_plugs(plugs: Iterable[SmartPlug]) -> list[SmartPlug]:
+    """Return linked plugs in deterministic primary-power order."""
+    return sorted(plugs, key=power_plug_rank)
+
+
+def pick_power_plug(plugs: Iterable[SmartPlug]) -> SmartPlug | None:
+    """Return the best printer-power row, retaining single-row fallbacks."""
+    return min(plugs, key=power_plug_rank, default=None)
+
+
+async def plugs_for_printer(
+    db: AsyncSession,
+    printer_id: int | None,
+) -> list[SmartPlug]:
+    """Load one printer's plugs and apply the shared business ordering."""
+    if printer_id is None:
+        return []
+    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id).order_by(SmartPlug.id))
+    return rank_power_plugs(result.scalars().all())
+
+
+async def select_energy_reading(
+    candidates: Sequence[SmartPlug],
+    read_energy: EnergyReader,
+    db: AsyncSession | None,
+) -> tuple[SmartPlug, dict] | None:
+    """Return the first ranked plug with a real lifetime counter.
+
+    A power-only response is not consumption.  Fallback is limited to rows
+    with the same printer-power assignment, so an accessory meter cannot be
+    billed merely because the actual outlet has no lifetime counter.
+    """
+    ranked = rank_power_plugs(candidates)
+    if not ranked:
+        return None
+
+    # Do not cross from the chosen printer-power relationship into an
+    # accessory merely because the accessory has a meter. Multiple rows with
+    # the same switchability/power/enabled role remain valid fallbacks for a
+    # temporarily unavailable counter.
+    assignment = _power_assignment(ranked[0])
+    for plug in ranked:
+        if _power_assignment(plug) != assignment:
+            break
+        energy = await read_energy(plug, db)
+        if energy and energy.get("total") is not None:
+            return plug, energy
+    return None
+
+
+async def read_printer_energy(
+    db: AsyncSession,
+    printer_id: int,
+    read_energy: EnergyReader,
+    *,
+    log_prefix: str,
+    context: str,
+    plug_id: int | None = None,
+) -> tuple[SmartPlug, dict] | None:
+    """Read the deterministically assigned printer meter with visible failures."""
+    candidates = await plugs_for_printer(db, printer_id)
+    if plug_id is not None:
+        candidates = [candidate for candidate in candidates if candidate.id == plug_id]
+    if not candidates:
+        logger.info("[%s] No smart plug for printer %s (%s)", log_prefix, printer_id, context)
+        return None
+    selected = await select_energy_reading(candidates, read_energy, db)
+    if selected is None:
+        logger.warning(
+            "[%s] No assigned plug reports a lifetime counter for %s (linked: %s)",
+            log_prefix,
+            context,
+            ", ".join(plug.name for plug in candidates),
+        )
+    return selected