public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-22 12:11:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(mqtt-relay): publish the documented events, from every place they happen

Browse Source 
Several MQTT relay topics in the wiki were never sent or were wrong:

- smart_plugs/energy had no caller. A loop now publishes each enabled
  plug's power and energy once a minute while the relay is connected
  (no plug is read without the relay; an MQTT plug that stopped
  reporting is left out).
- smart_plugs/on and /off only came from switching a plug by hand with
  on/off. Every successful switch now publishes through one helper:
  toggle, auto-on, auto-off (delayed, by temperature, resumed),
  schedules, the queue powering a printer on and failure detection.
  Home Assistant scripts are not plugs and publish nothing; the printer
  name comes from the database.
- maintenance/alert always sent 0 for the value and the threshold; it
  now sends the hours or days since the item was done, its interval and
  a unit.

New: printers/{serial}/offline after a minute offline (the debounce of
the offline notification) and /online after it, never one without the
other; printers/{serial}/filament/low per AMS slot, once per crossing,
with its own record so the notification and the relay never silence
each other. on_maintenance_acknowledged is removed: nothing can be
acknowledged. Guard tests: every relay event has a caller, and every
function that switches a plug publishes it.
This commit is contained in:
maziggy
2026-10-11 12:03:13 +02:00
parent b5cd369f51
commit dbb560dfb6
13 changed files with 912 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+2
View File
@@ -5,6 +5,7 @@ All notable changes to Bambuddy will be documented in this file.
## [1.2.6b1] - Unreleased
### Added
- **The MQTT relay publishes printer offline/online and low filament** — Three new topics for Home Assistant, Node-RED and other subscribers. `printers/{serial}/offline` is sent once a printer has been disconnected for a minute, and `printers/{serial}/online` when it comes back after that, so a short blip sends neither. `printers/{serial}/filament/low` is sent once when the spool in an AMS slot drops below its low-stock threshold, at the same moment as the Low Filament notification, and works even when no notification provider has that event switched on. Wiki updated with the payloads.
- **Notify! notification provider with iOS Live Activities and Lock Screen widgets (#2919, requested and contributed by @simplytoast1 in #3265)** — Notify! used to need the separate notify-bambuddy companion container. Bambuddy now talks to it directly: add a provider of type Notify! with the device ID and token.
- **Alerts:** the usual events, printer filter, quiet hours and daily digest apply as for every provider, with the camera photo attached. Each printer gets its own notification thread. **Time Sensitive problem alerts** sends failures, stops, printer errors and failure detection as time-sensitive notifications.
- **Live Activities:** with **Live Activities (iOS)** on, each running print shows a tile on the Lock Screen with a countdown to the end. A fault, a pause, running out of filament or a lost connection replaces the countdown with the problem. Progress style, extra values (progress, ETA, layer, temperatures), print stage, a privacy mode that hides the file name, symbol, colour and a button that opens Bambuddy can be set under **Live Activity appearance**. A dismissed tile stays dismissed for that print, and tiles survive a Bambuddy restart.
@@ -161,6 +162,7 @@ All notable changes to Bambuddy will be documented in this file.
- **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
### Fixed
- **The MQTT relay publishes what the wiki says it does** — Three topics were missing or wrong. `smart_plugs/energy` was never sent: with the relay connected, each enabled plug's power, today's energy and total energy now go out once a minute (without the relay no plug is read for this, and an MQTT plug that stopped reporting is left out). `smart_plugs/on` and `/off` only came from switching a plug by hand; auto-on, auto-off, schedules, the queue powering a printer on and failure detection switching a plug off now publish them too. `maintenance/alert` always said 0 for the current value and the threshold; it now has the hours (or days) since the item was last done, its interval and a `unit` field.
- **PostgreSQL databases from older versions get the links a new install has** — A database created before v0.2.4.1 (or before v1.2.5.2 for the last one) was missing three links: from a print log entry to its archive and to the user who started it, and from an archive to the library file it was printed from. Deleting an archive, a user or a library file left their ID behind instead of clearing it, so such entries didn't show up under "no user", for example. The links are added on the next start, and IDs of things already deleted are cleared. A print log entry without an archive also stays that way now: every start used to attach it to the newest archive with the same name on the same printer, which could be someone else's. 51 columns, mostly settings flags, also accepted an empty value there or had a default only those databases had. They now match a new install, and an empty value gets the column's default, with two exceptions so nothing gains access: an empty API key permission becomes "off", and an OIDC provider without an email claim, which so far linked no accounts, gets auto-linking turned off instead of starting to link by email. Apart from that, SQLite installs are unchanged.
- **Spoolman: colour, material and cost from the spool weren't applied on older PostgreSQL installs** — On a PostgreSQL database created before v0.2.4.8, the start snapshot of the AMS levels was stored as text instead of JSON and read back as text. Reporting usage to Spoolman then failed after charging the spools from the 3MF, with "Spoolman usage reporting failed" in the log. Prints without a 3MF charged nothing, and no archive got the colour, material and cost of its Spoolman spools. The column is converted on the next start. Two number columns also only had single precision on PostgreSQL databases older than them and are widened, keeping the values as entered: a spool's "Reset usage" baseline (databases created before v0.2.4.2) and a REST smart plug's energy total multiplier (before v1.2.5). SQLite installs aren't affected.
- **Every install now gets the dependency versions that were tested** — Bambuddy's Python dependencies only had minimum versions, so each Docker build, install and update took whatever was newest that day: a fresh install could run library versions nobody had tested Bambuddy with (SQLAlchemy 2.1 arrived that way). They are now pinned in `requirements.lock`, which Docker, the installers, the update scripts and the tests all use, and newer versions are taken on deliberately, after testing. Nothing changes for an install today: the pins are the versions a fresh install gets now. A native install moves to the pinned versions with the update after the one that brings this change, since an update still runs the previous version of the update script.
backend/app/api/routes/maintenance.py
+11
View File
@@ -288,6 +288,17 @@ async def restore_default_maintenance_types(
# ============== Printer Maintenance ==============
def maintenance_alert_values(item: MaintenanceStatus) -> tuple[float, float, str]:
"""What the MQTT relay's maintenance/alert reports: (value, threshold, unit).
Print-hour items count hours since the last maintenance against the
interval; time-based ones count days, and their interval_hours holds days.
"""
if item.interval_type == "days":
return (item.days_since_maintenance or 0.0, item.interval_hours, "days")
return (item.hours_since_maintenance, item.interval_hours, "hours")
async def _get_printer_maintenance_internal(
printer_id: int,
db: AsyncSession,
backend/app/api/routes/smart_plugs.py
+11 -19
View File
@@ -42,6 +42,7 @@ from backend.app.services.mqtt_smart_plug import subscribe_plug_to_mqtt
from backend.app.services.plug_energy_history import fill_derived_energy
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 publish_plug_state
from backend.app.services.tasmota import tasmota_service
from backend.app.utils.local_time import utcnow_naive
@@ -695,27 +696,18 @@ async def control_smart_plug(
if is_main_plug and plug.printer_id and expected_state:
await trigger_associated_scripts(plug.printer_id, expected_state, db)
# MQTT relay - publish smart plug state change
if expected_state:
# MQTT relay - publish smart plug state change. A toggle doesn't say which
# way it went, so read the state it ended in.
published_state = expected_state
if published_state is None:
try:
from backend.app.services.mqtt_relay import mqtt_relay
# Get printer name if linked
printer_name = None
if plug.printer_id:
result = await db.execute(select(Printer).where(Printer.id == plug.printer_id))
printer = result.scalar_one_or_none()
printer_name = printer.name if printer else None
await mqtt_relay.on_smart_plug_state(
plug_id=plug.id,
plug_name=plug.name,
state="on" if expected_state == "ON" else "off",
printer_id=plug.printer_id,
printer_name=printer_name,
)
published_state = (await service.get_status(plug)).get("state")
except Exception:
pass # Don't fail if MQTT fails
published_state = None
if published_state in ("ON", "OFF"):
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=published_state == "ON", ha_entity_id=plug.ha_entity_id
)
return {"success": True, "action": control.action}
backend/app/main.py
+41 -9
View File
@@ -85,7 +85,11 @@ from backend.app.api.routes import (
webhook,
websocket,
)
from backend.app.api.routes.maintenance import _get_printer_maintenance_internal, ensure_default_types
from backend.app.api.routes.maintenance import (
_get_printer_maintenance_internal,
ensure_default_types,
maintenance_alert_values,
)
from backend.app.api.routes.support import init_debug_logging
from backend.app.core.config import APP_VERSION, settings as app_settings
from backend.app.core.database import async_session, engine, init_db
@@ -541,6 +545,9 @@ _user_stopped_printers: set[int] = set()
# after a debounce window — cancelled if the printer reconnects before the
# window elapses, so transient MQTT blips don't flood the user.
_printer_last_connected: dict[int, bool] = {}
# Printers whose relay offline event went out, so the reconnect publishes the
# matching online event and a blip shorter than the debounce publishes neither.
_printer_offline_published: set[int] = set()
_printer_offline_notify_tasks: dict[int, asyncio.Task] = {}
# Debounce: a printer must stay offline this long before we notify. Sized
# against the staleness path (`bambu_mqtt.py::STALE_RECONNECT_COOLDOWN = 30s`)
@@ -1448,6 +1455,14 @@ async def _maybe_notify_printer_offline(printer_id: int) -> None:
printer_id,
)
return
# Recorded only when the relay can publish it, so the reconnect's
# online event always has an offline before it.
if mqtt_relay.enabled and mqtt_relay.connected:
_printer_offline_published.add(printer_id)
try:
await mqtt_relay.on_printer_offline(printer_id, printer.name, printer.serial_number)
except Exception as e:
logger.warning("MQTT relay offline event failed for printer %s: %s", printer_id, e)
logger.info(
"[#1752] Dispatching on_printer_offline for printer %s (%s)",
printer_id,
@@ -1680,8 +1695,9 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
# Offline-notification edge (#1752): schedule `on_printer_offline` on
# connected → disconnected. The "back online" channel is already covered
# by the print-failure notification (firmware reports gcode_state=FAILED
# on reconnect of an interrupted print), so we don't add a symmetric
# online event here.
# on reconnect of an interrupted print), so there is no online
# notification. The MQTT relay gets both: offline after the debounce, and
# online on the reconnect that follows a published offline.
prev_connected = _printer_last_connected.get(printer_id)
_printer_last_connected[printer_id] = state.connected
if prev_connected is True and not state.connected:
@@ -1697,6 +1713,18 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
name=f"printer-offline-notify-{printer_id}",
)
elif state.connected:
# Kept until the relay can actually send online: with its broker down
# now, the next status update after the reconnect sends it.
if printer_id in _printer_offline_published and mqtt_relay.enabled and mqtt_relay.connected:
_printer_offline_published.discard(printer_id)
printer_info = printer_manager.get_printer(printer_id)
if printer_info:
try:
await mqtt_relay.on_printer_online(printer_id, printer_info.name, printer_info.serial_number)
except Exception as e:
logging.getLogger(__name__).warning(
"MQTT relay online event failed for printer %s: %s", printer_id, e
)
pending = _printer_offline_notify_tasks.pop(printer_id, None)
if pending is not None and not pending.done():
logging.getLogger(__name__).info(
@@ -8691,10 +8719,12 @@ async def on_print_complete(printer_id: int, data: dict):
await ensure_default_types(db)
overview = await _get_printer_maintenance_internal(printer_id, db, commit=True)
attention = [
item for item in overview.maintenance_items if item.enabled and (item.is_due or item.is_warning)
]
items_needing_attention = [
{"name": item.maintenance_type_name, "is_due": item.is_due, "is_warning": item.is_warning}
for item in overview.maintenance_items
if item.enabled and (item.is_due or item.is_warning)
for item in attention
]
if items_needing_attention:
@@ -8702,14 +8732,16 @@ async def on_print_complete(printer_id: int, data: dict):
logger.info("[MAINT-BG] Sent notification: %s items need attention", len(items_needing_attention))
# MQTT relay - publish maintenance alerts
for item in items_needing_attention:
for item in attention:
current_value, threshold, unit = maintenance_alert_values(item)
try:
await mqtt_relay.on_maintenance_alert(
printer_id=printer_id,
printer_name=printer_name,
maintenance_type=item["name"],
current_value=0, # Not easily available here
threshold=0, # Not easily available here
maintenance_type=item.maintenance_type_name,
current_value=current_value,
threshold=threshold,
unit=unit,
)
except Exception:
pass # Don't fail if MQTT fails
backend/app/services/mqtt_relay.py
+29 -33
View File
@@ -553,8 +553,13 @@ class MQTTRelayService:
maintenance_type: str,
current_value: float,
threshold: float,
unit: str,
):
"""Publish maintenance alert triggered event."""
"""Publish that a maintenance item is due or nearly due.
current_value is the hours (or days) since it was last done, threshold
its interval, and unit says which: "hours" or "days".
"""
if not self.enabled or not self.connected:
return
@@ -566,26 +571,7 @@ class MQTTRelayService:
"maintenance_type": maintenance_type,
"current_value": current_value,
"threshold": threshold,
"timestamp": datetime.now(timezone.utc).isoformat(),
},
)
async def on_maintenance_acknowledged(
self,
printer_id: int,
printer_name: str,
maintenance_type: str,
):
"""Publish maintenance alert acknowledged event."""
if not self.enabled or not self.connected:
return
self._publish(
f"{self.topic_prefix}/maintenance/acknowledged",
{
"printer_id": printer_id,
"printer_name": printer_name,
"maintenance_type": maintenance_type,
"unit": unit,
"timestamp": datetime.now(timezone.utc).isoformat(),
},
)
@@ -662,22 +648,32 @@ class MQTTRelayService:
async def on_filament_low(
self,
printer_id: int,
printer_name: str,
printer_serial: str,
slot: str,
remaining_percent: int,
color: str | None,
spool_id: int,
spool_name: str,
remaining_weight: float,
remaining_percent: float,
):
"""Publish filament inventory low event."""
"""Publish that the spool in an AMS slot crossed its low-stock threshold.
Sent once per crossing, at the same point as the Low Filament
notification (print_scheduler._emit_filament_low).
"""
if not self.enabled or not self.connected:
return
self._publish(
f"{self.topic_prefix}/filament/low",
f"{self.topic_prefix}/printers/{printer_serial}/filament/low",
{
"spool_id": spool_id,
"spool_name": spool_name,
"remaining_weight": remaining_weight,
"printer_id": printer_id,
"printer_name": printer_name,
"printer_serial": printer_serial,
"slot": slot,
"remaining_percent": remaining_percent,
"color": color,
"spool_id": spool_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
},
)
@@ -716,11 +712,11 @@ class MQTTRelayService:
self,
plug_id: int,
plug_name: str,
power: float,
energy_today: float,
energy_total: float,
power: float | None,
energy_today: float | None,
energy_total: float | None,
):
"""Publish smart plug energy update event."""
"""Publish a plug's power and energy; a value the plug doesn't report is null."""
if not self.enabled or not self.connected:
return
backend/app/services/obico_actions.py
+5 -2
View File
@@ -52,7 +52,7 @@ def _pause_print(printer_id: int) -> None:
async def _turn_off_linked_plugs(printer_id: int) -> None:
from backend.app.services.smart_plug_manager import smart_plug_manager
from backend.app.services.smart_plug_manager import publish_plug_state, smart_plug_manager
async with async_session() as db:
plugs = await smart_plug_manager._get_plugs_for_printer(printer_id, db)
@@ -61,7 +61,10 @@ async def _turn_off_linked_plugs(printer_id: int) -> None:
continue
try:
service = await smart_plug_manager.get_service_for_plug(plug, db)
await service.turn_off(plug)
if await service.turn_off(plug):
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=False, ha_entity_id=plug.ha_entity_id
)
logger.info("Obico action: turned off plug %s for printer %s", plug.name, printer_id)
except Exception as e:
logger.error("Obico action: failed to turn off plug %s: %s", plug.name, e)
backend/app/services/print_scheduler.py
+76 -26
View File
@@ -51,6 +51,7 @@ from backend.app.services.finance_budget import (
validate_print_budget,
)
from backend.app.services.ha_sensor_manager import ha_sensor_manager
from backend.app.services.mqtt_relay import mqtt_relay
from backend.app.services.notification_service import notification_service
from backend.app.services.print_cost_estimate import estimate_queue_source_cost
from backend.app.services.printer_manager import (
@@ -61,7 +62,7 @@ from backend.app.services.printer_manager import (
supports_drying,
supports_drying_while_printing,
)
from backend.app.services.smart_plug_manager import smart_plug_manager
from backend.app.services.smart_plug_manager import publish_plug_state, smart_plug_manager
from backend.app.services.spool_assignment_notifications import (
_global_tray_from_assignment,
_slot_label_from_global_tray,
@@ -1179,6 +1180,9 @@ class PrintScheduler:
# so it can never re-arm. Keyed with the spool, the new spool is simply a
# different key and the stale one is inert.
self._notified_filament_low: set[tuple[int, int, int, int]] = set()
# The same, for the MQTT relay's filament/low event: its own record, so
# one channel having been told never silences the other.
self._relayed_filament_low: set[tuple[int, int, int, int]] = set()
# Earliest monotonic time the next low-filament check may run (#2913).
self._filament_low_next_check: float = 0.0
# SKU key -> (condition, events told) for a SKU that is alerting (#2955): the
@@ -1970,7 +1974,14 @@ class PrintScheduler:
for extra_plug in [p for p in auto_on_plugs if p.id != primary_plug.id]:
try:
service = await smart_plug_manager.get_service_for_plug(extra_plug, db)
await service.turn_on(extra_plug)
if await service.turn_on(extra_plug):
await publish_plug_state(
extra_plug.id,
extra_plug.name,
extra_plug.printer_id,
on=True,
ha_entity_id=extra_plug.ha_entity_id,
)
logger.info(
"Also powered on plug '%s' for printer %s", extra_plug.name, item.printer_id
)
@@ -3030,7 +3041,14 @@ class PrintScheduler:
for extra_plug in [p for p in auto_on_plugs if p.id != primary_plug.id]:
try:
service = await smart_plug_manager.get_service_for_plug(extra_plug, db)
await service.turn_on(extra_plug)
if await service.turn_on(extra_plug):
await publish_plug_state(
extra_plug.id,
extra_plug.name,
extra_plug.printer_id,
on=True,
ha_entity_id=extra_plug.ha_entity_id,
)
logger.info("Also powered on plug '%s' for printer %s", extra_plug.name, printer.id)
except Exception as e:
logger.warning("Failed to power on extra plug '%s': %s", extra_plug.name, e)
@@ -5105,8 +5123,18 @@ class PrintScheduler:
# id in that time would stay silenced until a restart. Forgetting
# what was sent is right anyway: whoever switches the event back on
# is told about the spools that are low now.
if not await notification_service._get_providers_for_event(db, "on_filament_low"):
#
# The MQTT relay publishes the same event, so a connected relay is
# reason enough to run the check. Its record is forgotten when the
# relay is switched off, not when the broker drops: a reconnect must
# not re-send every low slot.
notify = bool(await notification_service._get_providers_for_event(db, "on_filament_low"))
relay = mqtt_relay.enabled and mqtt_relay.connected
if not notify:
self._notified_filament_low.clear()
if not mqtt_relay.enabled:
self._relayed_filament_low.clear()
if not notify and not relay:
return
global_threshold = await self._get_low_stock_threshold(db)
@@ -5147,7 +5175,7 @@ class PrintScheduler:
# otherwise re-alert on every spool as soon as it came back.
return
await self._emit_filament_low(db, slots)
await self._emit_filament_low(db, slots, notify=notify, relay=relay)
except Exception as e:
logger.warning("Low-filament check failed: %s", e, exc_info=True)
@@ -5223,19 +5251,29 @@ class PrintScheduler:
return slots
async def _emit_filament_low(
self, db: AsyncSession, slots: dict[tuple[int, int, int, int], tuple[float, float, str | None]]
self,
db: AsyncSession,
slots: dict[tuple[int, int, int, int], tuple[float, float, str | None]],
*,
notify: bool = True,
relay: bool = False,
) -> None:
"""Send one notification per slot that has newly crossed its threshold."""
printer_names: dict[int, str] = {}
"""Send one notification and/or MQTT relay event per slot that has newly crossed its threshold."""
printers: dict[int, Printer] = {}
for key, (pct, threshold, color) in slots.items():
printer_id, ams_id, tray_id, _spool_id = key
printer_id, ams_id, tray_id, spool_id = key
if pct >= threshold:
# Back above the line: re-arm rather than expire on a timer, so a
# refilled slot can alert again and a spool sitting just under the
# threshold stays quiet.
self._notified_filament_low.discard(key)
self._relayed_filament_low.discard(key)
continue
if key in self._notified_filament_low:
send_notification = notify and key not in self._notified_filament_low
# Only while connected, so a crossing the relay missed reaches it
# after the reconnect.
send_relay = relay and key not in self._relayed_filament_low
if not send_notification and not send_relay:
continue
# Marked before sending, which is a choice rather than the only
# option, and it is the opposite failure mode from the one the
@@ -5247,24 +5285,37 @@ class PrintScheduler:
# the two failures. A spool that is low stays low, so the next real
# signal is not far away; a provider stuck retrying is a signal that
# never stops.
self._notified_filament_low.add(key)
if send_notification:
self._notified_filament_low.add(key)
if send_relay:
self._relayed_filament_low.add(key)
if printer_id not in printer_names:
if printer_id not in printers:
printer = (await db.execute(select(Printer).where(Printer.id == printer_id))).scalar_one_or_none()
if printer is None:
continue
printer_names[printer_id] = printer.name
try:
await notification_service.on_filament_low(
printer_id,
printer_names[printer_id],
_ams_slot_label(ams_id, tray_id),
int(pct),
db,
color=color,
)
except Exception as e:
logger.warning("Low-filament notification failed for slot %s: %s", key, e)
printers[printer_id] = printer
printer = printers[printer_id]
slot = _ams_slot_label(ams_id, tray_id)
if send_notification:
try:
await notification_service.on_filament_low(
printer_id,
printer.name,
slot,
int(pct),
db,
color=color,
)
except Exception as e:
logger.warning("Low-filament notification failed for slot %s: %s", key, e)
if send_relay:
try:
await mqtt_relay.on_filament_low(
printer_id, printer.name, printer.serial_number, slot, int(pct), color, spool_id
)
except Exception as e:
logger.warning("Low-filament MQTT relay event failed for slot %s: %s", key, e)
async def _check_stock_forecast(self, db: AsyncSession) -> None:
"""Alert when a filament SKU reaches its reorder point or is about to break (#2955).
@@ -7401,6 +7452,7 @@ class PrintScheduler:
if not success:
logger.warning("Failed to turn on smart plug '%s'", plug.name)
return False
await publish_plug_state(plug.id, plug.name, plug.printer_id, on=True, ha_entity_id=plug.ha_entity_id)
logger.info("Powered on smart plug '%s' for printer %s", plug.name, printer_id)
# Get printer from database for connection
@@ -9113,8 +9165,6 @@ class PrintScheduler:
# MQTT relay - publish queue job started
try:
from backend.app.services.mqtt_relay import mqtt_relay
await mqtt_relay.on_queue_job_started(
job_id=item.id,
filename=filename,
backend/app/services/smart_plug_manager.py
+115
View File
@@ -10,6 +10,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.tasks import spawn_background_task
from backend.app.services.homeassistant import homeassistant_service
from backend.app.services.mqtt_relay import mqtt_relay
from backend.app.services.plug_energy import get_plug_energy
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.tasmota import tasmota_service
@@ -21,6 +23,43 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
async def publish_plug_state(
plug_id: int, plug_name: str, printer_id: int | None, *, on: bool, ha_entity_id: str | None = None
) -> None:
"""Tell the MQTT relay a plug was switched (smart_plugs/on or /off).
Called after every successful on or off command, whatever sent it, also when
the plug already was in that state. A Home Assistant script run through its
entity is not a switch and publishes nothing. The printer name is read from
the database: the connection cache keeps a renamed printer's old name and has
none for an inactive one. Never raises: the plug did switch, and a broker
problem must not turn that into an error.
"""
if ha_entity_id and ha_entity_id.startswith("script."):
return
if not mqtt_relay.enabled or not mqtt_relay.connected:
return
try:
printer_name = None
if printer_id:
from backend.app.core import database
from backend.app.models.printer import Printer
async with database.async_session() as db:
printer_name = (
await db.execute(select(Printer.name).where(Printer.id == printer_id))
).scalar_one_or_none()
await mqtt_relay.on_smart_plug_state(
plug_id=plug_id,
plug_name=plug_name,
state="on" if on else "off",
printer_id=printer_id,
printer_name=printer_name,
)
except Exception as e:
logger.debug("MQTT relay: failed to publish the state of plug %s: %s", plug_id, e)
class SmartPlugManager:
"""Manages smart plug automation and delayed turn-off."""
@@ -30,6 +69,7 @@ class SmartPlugManager:
self._scheduler_task: asyncio.Task | None = None
self._snapshot_task: asyncio.Task | None = None
self._power_alert_task: asyncio.Task | None = None
self._relay_energy_task: asyncio.Task | None = None
self._last_schedule_check: dict[int, str] = {} # plug_id -> "HH:MM" last executed
async def get_service_for_plug(self, plug: "SmartPlug", db: AsyncSession | None = None):
@@ -79,6 +119,9 @@ class SmartPlugManager:
if self._power_alert_task is None:
self._power_alert_task = asyncio.create_task(self._power_alert_loop())
logger.info("Smart plug power alert loop started")
if self._relay_energy_task is None:
self._relay_energy_task = asyncio.create_task(self._relay_energy_loop())
logger.info("Smart plug MQTT relay energy loop started")
def stop_scheduler(self):
"""Stop the background scheduler."""
@@ -94,6 +137,10 @@ class SmartPlugManager:
self._power_alert_task.cancel()
self._power_alert_task = None
logger.info("Smart plug power alert loop stopped")
if self._relay_energy_task:
self._relay_energy_task.cancel()
self._relay_energy_task = None
logger.info("Smart plug MQTT relay energy loop stopped")
async def _schedule_loop(self):
"""Background loop that checks scheduled on/off times every minute."""
@@ -152,6 +199,52 @@ class SmartPlugManager:
delay = (next_local_hour(now) - now).total_seconds()
await asyncio.sleep(max(delay, 60))
async def _relay_energy_loop(self):
"""Background loop that publishes every plug's energy to the MQTT relay, once a minute.
The relay's smart_plugs/energy topic has nothing else feeding it: the
plug status is otherwise only read for display, for power alerts and,
hourly, for the lifetime counter.
"""
from backend.app.core import database
await asyncio.sleep(30)
while True:
try:
# Looked up on each pass: a backup restore replaces the session factory.
async with database.async_session() as db:
await self._publish_energy_to_relay(db)
except Exception as e:
logger.error("Error publishing plug energy to the MQTT relay: %s", e)
await asyncio.sleep(60)
async def _publish_energy_to_relay(self, db: AsyncSession) -> None:
"""Read each enabled plug's power and energy and publish it to the MQTT relay.
Nothing is read while the relay is off or disconnected: on an install
without the relay this costs no request to any plug.
"""
from backend.app.models.smart_plug import SmartPlug
if not mqtt_relay.enabled or not mqtt_relay.connected:
return
plugs = (await db.execute(select(SmartPlug).where(SmartPlug.enabled.is_(True)))).scalars().all()
for plug in plugs:
# An MQTT plug's values are its last message; once it stops
# reporting they are stale, not a reading.
if plug.plug_type == "mqtt" and not mqtt_relay.smart_plug_service.is_reachable(plug.id):
continue
try:
energy = await get_plug_energy(plug, db)
except Exception as e:
logger.debug("MQTT relay: failed to read energy from plug %s: %s", plug.id, e)
continue
if not energy:
continue
await mqtt_relay.on_smart_plug_energy(
plug.id, plug.name, energy.get("power"), energy.get("today"), energy.get("total")
)
async def _capture_energy_snapshots(self):
"""Capture one energy snapshot row per plug with a usable lifetime counter."""
from backend.app.core.database import async_session
@@ -232,6 +325,9 @@ class SmartPlugManager:
plug.last_state = "ON"
plug.last_checked = utcnow_naive()
self._last_schedule_check[plug.id] = f"on:{current_time}"
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=True, ha_entity_id=plug.ha_entity_id
)
# Check if we should turn off
if plug.schedule_off_time == current_time:
@@ -243,6 +339,9 @@ class SmartPlugManager:
plug.last_state = "OFF"
plug.last_checked = utcnow_naive()
self._last_schedule_check[plug.id] = f"off:{current_time}"
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=False, ha_entity_id=plug.ha_entity_id
)
# Mark printer offline if this plug feeds it (#2629)
if plug.printer_id and plug.controls_printer_power:
printer_manager.mark_printer_offline(plug.printer_id)
@@ -288,6 +387,9 @@ class SmartPlugManager:
plug.last_state = "ON"
plug.last_checked = utcnow_naive()
plug.auto_off_executed = False # Reset flag when turning on
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=True, ha_entity_id=plug.ha_entity_id
)
except Exception as e:
logger.warning("Failed to turn on plug '%s' for printer %s: %s", plug.name, printer_id, e)
@@ -439,6 +541,7 @@ class SmartPlugManager:
printer_id,
delay_seconds,
controls_printer_power=plug.controls_printer_power,
plug_name=plug.name,
rest_off_url=plug.rest_off_url if plug.plug_type == "rest" else None,
rest_off_body=plug.rest_off_body if plug.plug_type == "rest" else None,
rest_method=plug.rest_method if plug.plug_type == "rest" else None,
@@ -459,6 +562,7 @@ class SmartPlugManager:
delay_seconds: int,
*,
controls_printer_power: bool = True,
plug_name: str = "",
rest_off_url: str | None = None,
rest_off_body: str | None = None,
rest_method: str | None = None,
@@ -506,6 +610,9 @@ class SmartPlugManager:
# Mark auto_off_executed in database and update printer status
if success:
await self._mark_auto_off_executed(plug_id)
await publish_plug_state(
plug_id, plug_name or f"plug_{plug_id}", printer_id, on=False, ha_entity_id=ha_entity_id
)
# Mark the printer as offline immediately — but only when this
# plug actually feeds the printer (#2629).
if controls_printer_power:
@@ -537,6 +644,7 @@ class SmartPlugManager:
printer_id,
temp_threshold,
controls_printer_power=plug.controls_printer_power,
plug_name=plug.name,
rest_off_url=plug.rest_off_url if plug.plug_type == "rest" else None,
rest_off_body=plug.rest_off_body if plug.plug_type == "rest" else None,
rest_method=plug.rest_method if plug.plug_type == "rest" else None,
@@ -557,6 +665,7 @@ class SmartPlugManager:
temp_threshold: int,
*,
controls_printer_power: bool = True,
plug_name: str = "",
rest_off_url: str | None = None,
rest_off_body: str | None = None,
rest_method: str | None = None,
@@ -637,6 +746,9 @@ class SmartPlugManager:
# Mark auto_off_executed in database and update printer status
if success:
await self._mark_auto_off_executed(plug_id)
await publish_plug_state(
plug_id, plug_name or f"plug_{plug_id}", printer_id, on=False, ha_entity_id=ha_entity_id
)
# Mark the printer as offline immediately — but only
# when this plug actually feeds the printer (#2629).
if controls_printer_power:
@@ -775,6 +887,9 @@ class SmartPlugManager:
success = await service.turn_off(plug)
if success:
await self._mark_auto_off_executed(plug.id)
await publish_plug_state(
plug.id, plug.name, plug.printer_id, on=False, ha_entity_id=plug.ha_entity_id
)
if plug.controls_printer_power:
printer_manager.mark_printer_offline(plug.printer_id)
backend/tests/integration/test_smart_plugs_api.py
+19
View File
@@ -284,6 +284,25 @@ class TestSmartPlugsAPI:
assert result["success"] is True
assert result["action"] == "toggle"
@pytest.mark.asyncio
@pytest.mark.integration
async def test_a_toggle_tells_the_mqtt_relay_the_resulting_state(
self, async_client: AsyncClient, smart_plug_factory, mock_tasmota_service, db_session
):
"""The SpoolBuddy quick menu and the printer card switch plugs with
toggle, so the state it ends in is read back and published."""
from unittest.mock import AsyncMock, patch
plug = await smart_plug_factory()
mock_tasmota_service.get_status.return_value = {"state": "OFF", "reachable": True}
with patch("backend.app.api.routes.smart_plugs.publish_plug_state", new_callable=AsyncMock) as publish:
response = await async_client.post(f"/api/v1/smart-plugs/{plug.id}/control", json={"action": "toggle"})
assert response.status_code == 200
publish.assert_awaited_once()
assert publish.await_args.kwargs["on"] is False
@pytest.mark.asyncio
@pytest.mark.integration
async def test_control_invalid_action(self, async_client: AsyncClient, smart_plug_factory, db_session):
backend/tests/unit/services/test_filament_low_producer_2913.py
+153
View File
@@ -625,3 +625,156 @@ async def test_the_event_reaches_a_provider_with_the_toggle_on():
assert variables["slot"] == "A1"
assert variables["remaining_percent"] == "15"
assert variables["color"] == "Jade White"
# -- the MQTT relay ------------------------------------------------------------
@pytest.fixture
def relay():
"""The MQTT relay, switched on and connected, with the publish replaced."""
with patch("backend.app.services.print_scheduler.mqtt_relay") as relay:
relay.enabled = True
relay.connected = True
relay.on_filament_low = AsyncMock()
yield relay
@pytest.mark.asyncio
async def test_the_relay_gets_the_slot_with_its_printer_and_spool(db_session, scheduler, notify, relay):
printer = await _printer(db_session)
spool = await _assigned_spool(db_session, printer, weight_used=850.0, color_name="Jade White")
await db_session.commit()
await _pass(scheduler, db_session)
relay.on_filament_low.assert_awaited_once_with(printer.id, "X2D", "S-X2D", "A1", 15, "Jade White", spool.id)
notify.on_filament_low.assert_awaited_once()
@pytest.mark.asyncio
async def test_the_relay_is_told_once_per_crossing(db_session, scheduler, notify, relay):
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
await _pass(scheduler, db_session)
await _pass(scheduler, db_session)
assert relay.on_filament_low.await_count == 1
@pytest.mark.asyncio
async def test_the_relay_alone_is_enough_to_run_the_check(db_session, scheduler, real_providers, relay):
"""No provider wants the alert, but the relay is on: the check runs for the
relay, and no notification is attempted."""
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
await _pass(scheduler, db_session)
relay.on_filament_low.assert_awaited_once()
real_providers.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_disconnected_relay_does_not_run_the_check(db_session, scheduler, real_providers, relay):
relay.connected = False
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
with patch.object(scheduler, "_get_low_stock_threshold", new_callable=AsyncMock) as work:
work.return_value = 20.0
await _pass(scheduler, db_session)
work.assert_not_awaited()
relay.on_filament_low.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_failing_relay_does_not_stop_the_notification(db_session, scheduler, notify, relay):
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
relay.on_filament_low.side_effect = RuntimeError("broker down")
await _pass(scheduler, db_session)
notify.on_filament_low.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_provider_switched_on_after_a_relay_only_send_is_still_notified(
db_session, scheduler, real_providers, relay
):
"""Each channel has its own once-per-crossing record: the relay having been
told must not silence the notification switched on afterwards."""
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
await _pass(scheduler, db_session)
assert relay.on_filament_low.await_count == 1
real_providers.assert_not_awaited()
await _provider(db_session, enabled=True, on_filament_low=True)
await db_session.commit()
await _pass(scheduler, db_session)
real_providers.assert_awaited_once()
assert relay.on_filament_low.await_count == 1
@pytest.mark.asyncio
async def test_a_crossing_while_the_relay_was_down_reaches_it_after_reconnecting(db_session, scheduler, notify, relay):
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
relay.connected = False
await _pass(scheduler, db_session)
relay.on_filament_low.assert_not_awaited()
relay.connected = True
await _pass(scheduler, db_session)
await _pass(scheduler, db_session)
assert relay.on_filament_low.await_count == 1
assert notify.on_filament_low.await_count == 1
@pytest.mark.asyncio
async def test_a_relay_reconnect_does_not_resend_every_low_slot(db_session, scheduler, real_providers, relay):
"""A broker blip is not the relay being switched off: what it was told stays told."""
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
await _pass(scheduler, db_session)
relay.connected = False
await _pass(scheduler, db_session)
relay.connected = True
await _pass(scheduler, db_session)
assert relay.on_filament_low.await_count == 1
@pytest.mark.asyncio
async def test_switching_the_relay_off_and_on_again_re_arms(db_session, scheduler, real_providers, relay):
"""The relay counterpart of re-arming the notification: switched off, it
forgets what it sent, and switched on again it reports what is low now."""
printer = await _printer(db_session)
await _assigned_spool(db_session, printer, weight_used=850.0)
await db_session.commit()
await _pass(scheduler, db_session)
relay.enabled = False
relay.connected = False
await _pass(scheduler, db_session)
relay.enabled = True
relay.connected = True
await _pass(scheduler, db_session)
assert relay.on_filament_low.await_count == 2
backend/tests/unit/services/test_smart_plug_manager.py
+11
View File
@@ -395,6 +395,7 @@ class TestSmartPlugManager:
patch.object(manager, "_schedule_loop") as mock_loop,
patch.object(manager, "_snapshot_loop"),
patch.object(manager, "_power_alert_loop") as mock_power_alert,
patch.object(manager, "_relay_energy_loop") as mock_relay_energy,
patch("asyncio.create_task") as mock_create,
):
mock_create.return_value = MagicMock()
@@ -405,6 +406,9 @@ class TestSmartPlugManager:
# Power alerts are checked in the background, not only while a page is open.
assert manager._power_alert_task is not None
mock_power_alert.assert_called_once()
# Feeds the MQTT relay's smart_plugs/energy topic.
assert manager._relay_energy_task is not None
mock_relay_energy.assert_called_once()
def test_stop_scheduler(self, manager):
"""Verify scheduler can be stopped."""
@@ -423,12 +427,14 @@ class TestSmartPlugManager:
manager._scheduler_task = mock_schedule_task
manager._snapshot_task = mock_snapshot_task
manager._power_alert_task = MagicMock()
manager._relay_energy_task = MagicMock()
# Mock the loop coroutines to avoid unawaited coroutine warnings
with (
patch.object(manager, "_schedule_loop") as mock_loop,
patch.object(manager, "_snapshot_loop") as mock_snapshot,
patch.object(manager, "_power_alert_loop") as mock_power_alert,
patch.object(manager, "_relay_energy_loop") as mock_relay_energy,
patch("asyncio.create_task") as mock_create,
):
manager.start_scheduler()
@@ -437,6 +443,7 @@ class TestSmartPlugManager:
mock_loop.assert_not_called()
mock_snapshot.assert_not_called()
mock_power_alert.assert_not_called()
mock_relay_energy.assert_not_called()
def test_stop_scheduler_cancels_snapshot_task(self, manager):
"""Verify stopping scheduler also cancels the snapshot loop (#941)."""
@@ -447,13 +454,17 @@ class TestSmartPlugManager:
mock_power_alert_task = MagicMock()
manager._power_alert_task = mock_power_alert_task
mock_relay_energy_task = MagicMock()
manager._relay_energy_task = mock_relay_energy_task
manager.stop_scheduler()
mock_schedule_task.cancel.assert_called_once()
mock_snapshot_task.cancel.assert_called_once()
mock_power_alert_task.cancel.assert_called_once()
mock_relay_energy_task.cancel.assert_called_once()
assert manager._power_alert_task is None
assert manager._relay_energy_task is None
assert manager._scheduler_task is None
assert manager._snapshot_task is None
backend/tests/unit/test_mqtt_relay_events.py
+276
View File
@@ -0,0 +1,276 @@
"""MQTT relay events reach the broker.
Five relay methods were written and never called, one of them for a topic the
wiki documents (``smart_plugs/energy``). The first test makes that impossible to
repeat: every public ``on_*`` method of the relay has a caller. The rest cover
the energy publisher; the printer online/offline and filament-low events are
tested where they are produced (test_printer_offline_notification.py,
test_filament_low_producer_2913.py).
"""
import ast
import re
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from backend.app.models.smart_plug import SmartPlug
from backend.app.services.mqtt_relay import MQTTRelayService
from backend.app.services.smart_plug_manager import SmartPlugManager
APP = Path(__file__).resolve().parents[2] / "app"
def test_every_relay_event_has_a_caller():
relay_source = (APP / "services" / "mqtt_relay.py").read_text(encoding="utf-8")
service = next(
node
for node in ast.parse(relay_source).body
if isinstance(node, ast.ClassDef) and node.name == MQTTRelayService.__name__
)
events = [
node.name
for node in service.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("on_")
]
callers = "\n".join(path.read_text(encoding="utf-8") for path in APP.rglob("*.py") if path.name != "mqtt_relay.py")
uncalled = [name for name in events if not re.search(rf"\.{name}\(", callers)]
assert events
assert uncalled == []
async def _plug(db_session, **kwargs) -> SmartPlug:
fields = {"name": "Printer plug", "ip_address": "192.168.1.50"}
fields.update(kwargs)
plug = SmartPlug(**fields)
db_session.add(plug)
await db_session.commit()
return plug
@pytest.fixture
def relay():
with patch("backend.app.services.smart_plug_manager.mqtt_relay") as relay:
relay.enabled = True
relay.connected = True
relay.on_smart_plug_energy = AsyncMock()
yield relay
@pytest.mark.asyncio
async def test_each_enabled_plug_publishes_its_energy(db_session, relay):
plug = await _plug(db_session)
await _plug(db_session, name="Off", enabled=False)
read = AsyncMock(return_value={"power": 120.5, "today": 1.25, "total": 310.0})
with patch("backend.app.services.smart_plug_manager.get_plug_energy", read):
await SmartPlugManager()._publish_energy_to_relay(db_session)
relay.on_smart_plug_energy.assert_awaited_once_with(plug.id, "Printer plug", 120.5, 1.25, 310.0)
@pytest.mark.asyncio
async def test_a_plug_that_cannot_be_read_publishes_nothing(db_session, relay):
await _plug(db_session)
with patch("backend.app.services.smart_plug_manager.get_plug_energy", AsyncMock(return_value=None)):
await SmartPlugManager()._publish_energy_to_relay(db_session)
relay.on_smart_plug_energy.assert_not_awaited()
@pytest.mark.asyncio
async def test_one_failing_plug_does_not_stop_the_others(db_session, relay):
await _plug(db_session, name="Broken")
good = await _plug(db_session, name="Good")
async def read(plug, _db):
if plug.name == "Broken":
raise OSError("unreachable")
return {"power": 5.0, "today": 0.1, "total": 2.0}
with patch("backend.app.services.smart_plug_manager.get_plug_energy", side_effect=read):
await SmartPlugManager()._publish_energy_to_relay(db_session)
relay.on_smart_plug_energy.assert_awaited_once_with(good.id, "Good", 5.0, 0.1, 2.0)
@pytest.mark.parametrize(("enabled", "connected"), [(False, False), (True, False)])
@pytest.mark.asyncio
async def test_no_plug_is_read_while_the_relay_is_off(db_session, relay, enabled, connected):
"""The loop runs on every install; only relay users pay for the reads."""
await _plug(db_session)
relay.enabled = enabled
relay.connected = connected
read = AsyncMock(return_value={"power": 1.0, "today": 0.0, "total": 0.0})
with patch("backend.app.services.smart_plug_manager.get_plug_energy", read):
await SmartPlugManager()._publish_energy_to_relay(db_session)
read.assert_not_awaited()
relay.on_smart_plug_energy.assert_not_awaited()
@pytest.mark.asyncio
async def test_an_mqtt_plug_that_stopped_reporting_publishes_nothing(db_session, relay):
"""An MQTT plug's values are the last message received; once the plug goes
quiet they are stale, and publishing them every minute would report a dead
plug as drawing power."""
await _plug(db_session, name="Quiet", plug_type="mqtt", ip_address=None)
live = await _plug(db_session, name="Live", plug_type="mqtt", ip_address=None)
relay.smart_plug_service.is_reachable.side_effect = lambda plug_id: plug_id == live.id
read = AsyncMock(return_value={"power": 50.0, "today": 0.5, "total": 0.5})
with patch("backend.app.services.smart_plug_manager.get_plug_energy", read):
await SmartPlugManager()._publish_energy_to_relay(db_session)
relay.on_smart_plug_energy.assert_awaited_once_with(live.id, "Live", 50.0, 0.5, 0.5)
# -- smart_plugs/on and /off: every switch, not only the manual one -------------
# The plug backends themselves (a REST toggle is a turn_on or turn_off).
PLUG_BACKENDS = {"tasmota.py", "homeassistant.py", "rest_smart_plug.py", "mqtt_smart_plug.py"}
# Runs a Home Assistant script through its entity's turn_on: no plug changes state.
NOT_A_SWITCH = {("api/routes/smart_plugs.py", "trigger_associated_scripts")}
def _called_names(node: ast.AST) -> set[str]:
names = set()
for call in ast.walk(node):
if isinstance(call, ast.Call):
if isinstance(call.func, ast.Attribute):
names.add(call.func.attr)
elif isinstance(call.func, ast.Name):
names.add(call.func.id)
return names
def test_every_function_that_switches_a_plug_tells_the_relay():
"""smart_plugs/on and /off used to come only from the manual switch; auto-on,
auto-off, schedules, the queue and failure detection switched plugs silently."""
switching, missing = [], []
for path in APP.rglob("*.py"):
if path.name in PLUG_BACKENDS:
continue
rel = path.relative_to(APP).as_posix()
for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) or (rel, node.name) in NOT_A_SWITCH:
continue
names = _called_names(node)
if names & {"turn_on", "turn_off", "toggle"}:
switching.append(f"{rel}:{node.name}")
if "publish_plug_state" not in names:
missing.append(f"{rel}:{node.name}")
assert switching
assert missing == []
@pytest.fixture
def session_for(db_session):
"""database.async_session() handing out the test session."""
class _Session:
async def __aenter__(self):
return db_session
async def __aexit__(self, *exc):
return False
with patch("backend.app.core.database.async_session", lambda: _Session()):
yield
@pytest.mark.asyncio
async def test_publish_plug_state_names_the_printer_from_the_database(db_session, session_for):
"""Not from the connection cache: that keeps a renamed printer's old name
and has none for an inactive printer."""
from backend.app.models.printer import Printer
from backend.app.services.smart_plug_manager import publish_plug_state
printer = Printer(name="Workshop", serial_number="SN1", ip_address="192.0.2.1", access_code="x", is_active=False)
db_session.add(printer)
await db_session.commit()
with patch("backend.app.services.smart_plug_manager.mqtt_relay") as relay:
relay.on_smart_plug_state = AsyncMock()
await publish_plug_state(3, "X1C plug", printer.id, on=False)
await publish_plug_state(4, "Lamp", None, on=True)
assert relay.on_smart_plug_state.await_args_list[0].kwargs == {
"plug_id": 3,
"plug_name": "X1C plug",
"state": "off",
"printer_id": printer.id,
"printer_name": "Workshop",
}
assert relay.on_smart_plug_state.await_args_list[1].kwargs["printer_name"] is None
assert relay.on_smart_plug_state.await_args_list[1].kwargs["state"] == "on"
@pytest.mark.asyncio
async def test_a_home_assistant_script_is_not_a_plug_switching(session_for):
"""Running a script through its entity changes no power state."""
from backend.app.services.smart_plug_manager import publish_plug_state
with patch("backend.app.services.smart_plug_manager.mqtt_relay") as relay:
relay.on_smart_plug_state = AsyncMock()
await publish_plug_state(5, "Lights script", None, on=True, ha_entity_id="script.printer_lights")
await publish_plug_state(6, "Switch", None, on=True, ha_entity_id="switch.printer")
relay.on_smart_plug_state.assert_awaited_once()
assert relay.on_smart_plug_state.await_args.kwargs["plug_id"] == 6
@pytest.mark.asyncio
async def test_publish_plug_state_never_raises():
"""Switching the plug worked; a broker problem must not turn that into an error."""
from backend.app.services.smart_plug_manager import publish_plug_state
with patch("backend.app.services.smart_plug_manager.mqtt_relay") as relay:
relay.on_smart_plug_state = AsyncMock(side_effect=RuntimeError("broker down"))
await publish_plug_state(3, "X1C plug", None, on=True)
@pytest.mark.asyncio
async def test_the_delayed_auto_off_publishes_with_the_plug_name():
manager = SmartPlugManager()
service = AsyncMock()
service.turn_off = AsyncMock(return_value=True)
with (
patch("backend.app.services.smart_plug_manager.printer_manager") as pm,
patch("backend.app.services.smart_plug_manager.publish_plug_state", new_callable=AsyncMock) as publish,
patch.object(manager, "get_service_for_plug", AsyncMock(return_value=service)),
patch.object(manager, "_mark_auto_off_executed", new_callable=AsyncMock),
):
pm.is_print_active.return_value = False
await manager._delayed_off(
3, "tasmota", "1.2.3.4", None, None, None, printer_id=1, delay_seconds=0, plug_name="X1C plug"
)
publish.assert_awaited_once_with(3, "X1C plug", 1, on=False, ha_entity_id=None)
# -- maintenance/alert carries the real values -----------------------------------
def test_a_print_hours_item_reports_hours_since_and_the_interval():
from types import SimpleNamespace
from backend.app.api.routes.maintenance import maintenance_alert_values
item = SimpleNamespace(
interval_type="hours", interval_hours=200.0, hours_since_maintenance=212.5, days_since_maintenance=None
)
assert maintenance_alert_values(item) == (212.5, 200.0, "hours")
def test_a_time_based_item_reports_days_since_and_the_interval():
from types import SimpleNamespace
from backend.app.api.routes.maintenance import maintenance_alert_values
item = SimpleNamespace(
interval_type="days", interval_hours=30.0, hours_since_maintenance=40.0, days_since_maintenance=31.2
)
assert maintenance_alert_values(item) == (31.2, 30.0, "days")
backend/tests/unit/test_printer_offline_notification.py
+163
View File
@@ -83,9 +83,11 @@ def _reset_edge_state():
if not task.done():
task.cancel()
main_module._printer_offline_notify_tasks.clear()
main_module._printer_offline_published.clear()
main_module._printer_reconciled_since_connect.clear()
main_module._last_status_broadcast.clear()
yield
main_module._printer_offline_published.clear()
main_module._printer_last_connected.clear()
for task in list(main_module._printer_offline_notify_tasks.values()):
if not task.done():
@@ -170,6 +172,167 @@ class TestMaybeNotifyPrinterOffline:
assert 1 not in main_module._printer_offline_notify_tasks
class TestRelayOfflineOnline:
"""The MQTT relay's printers/{serial}/offline and /online events: offline on
the same debounced edge as the notification, online only after an offline
was published, so they always come in pairs."""
@staticmethod
def _session(printer):
scalar = MagicMock()
scalar.scalar_one_or_none.return_value = printer
db = AsyncMock()
db.execute = AsyncMock(return_value=scalar)
session_cm = MagicMock()
session_cm.__aenter__ = AsyncMock(return_value=db)
session_cm.__aexit__ = AsyncMock(return_value=False)
return session_cm
@pytest.mark.asyncio
async def test_offline_is_published_when_still_offline_after_debounce(self):
printer = SimpleNamespace(id=1, name="Workshop", serial_number="SN1")
relay = MagicMock()
relay.on_printer_offline = AsyncMock()
with (
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.async_session", return_value=self._session(printer)),
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.mqtt_relay", relay),
):
mock_pm.is_connected.return_value = False
mock_notif.on_printer_offline = AsyncMock()
await main_module._maybe_notify_printer_offline(printer_id=1)
relay.on_printer_offline.assert_awaited_once_with(1, "Workshop", "SN1")
assert 1 in main_module._printer_offline_published
@pytest.mark.asyncio
async def test_a_failing_notification_still_publishes_offline(self):
printer = SimpleNamespace(id=1, name="Workshop", serial_number="SN1")
relay = MagicMock()
relay.on_printer_offline = AsyncMock()
with (
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.async_session", return_value=self._session(printer)),
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.mqtt_relay", relay),
):
mock_pm.is_connected.return_value = False
mock_notif.on_printer_offline = AsyncMock(side_effect=RuntimeError("provider down"))
await main_module._maybe_notify_printer_offline(printer_id=1)
relay.on_printer_offline.assert_awaited_once()
@pytest.mark.parametrize(("enabled", "connected"), [(False, False), (True, False)])
@pytest.mark.asyncio
async def test_an_offline_the_relay_could_not_send_is_not_paired_later(self, enabled, connected):
"""With the relay off or its broker down, no offline went out, so the
reconnect must not send an online on its own."""
printer = SimpleNamespace(id=1, name="Workshop", serial_number="SN1")
relay = MagicMock()
relay.enabled = enabled
relay.connected = connected
relay.on_printer_offline = AsyncMock()
with (
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.async_session", return_value=self._session(printer)),
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.mqtt_relay", relay),
):
mock_pm.is_connected.return_value = False
mock_notif.on_printer_offline = AsyncMock()
await main_module._maybe_notify_printer_offline(printer_id=1)
assert 1 not in main_module._printer_offline_published
mock_notif.on_printer_offline.assert_awaited_once()
@pytest.mark.asyncio
async def test_no_offline_when_reconnected_during_debounce(self):
relay = MagicMock()
relay.on_printer_offline = AsyncMock()
with (
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.mqtt_relay", relay),
):
mock_pm.is_connected.return_value = True
await main_module._maybe_notify_printer_offline(printer_id=1)
relay.on_printer_offline.assert_not_awaited()
assert 1 not in main_module._printer_offline_published
@staticmethod
def _handler_patches(relay, pm):
ws_mgr = MagicMock()
ws_mgr.send_printer_status = AsyncMock()
return (
patch("backend.app.main.ws_manager", ws_mgr),
patch("backend.app.main.mqtt_relay", relay),
patch("backend.app.main.printer_manager", pm),
_spawn_patch(),
patch("backend.app.main.printer_state_to_dict", return_value={}),
)
@staticmethod
def _relay_and_pm():
relay = MagicMock()
relay.on_printer_status = AsyncMock()
relay.on_printer_online = AsyncMock()
pm = MagicMock()
pm.get_printer.return_value = SimpleNamespace(name="Workshop", serial_number="SN1")
pm.get_model.return_value = ""
pm.is_awaiting_plate_clear.return_value = False
return relay, pm
@pytest.mark.asyncio
async def test_online_is_published_once_after_a_published_offline(self):
relay, pm = self._relay_and_pm()
main_module._printer_last_connected[1] = False
main_module._printer_offline_published.add(1)
p1, p2, p3, p4, p5 = self._handler_patches(relay, pm)
with p1, p2, p3, p4, p5:
await main_module.on_printer_status_change(1, _state(connected=True))
await main_module.on_printer_status_change(1, _state(connected=True))
relay.on_printer_online.assert_awaited_once_with(1, "Workshop", "SN1")
assert 1 not in main_module._printer_offline_published
@pytest.mark.asyncio
async def test_online_waits_for_the_relay_when_its_broker_is_down_at_reconnect(self):
relay, pm = self._relay_and_pm()
relay.enabled = True
relay.connected = False
main_module._printer_last_connected[1] = False
main_module._printer_offline_published.add(1)
p1, p2, p3, p4, p5 = self._handler_patches(relay, pm)
with p1, p2, p3, p4, p5:
await main_module.on_printer_status_change(1, _state(connected=True))
relay.on_printer_online.assert_not_awaited()
assert 1 in main_module._printer_offline_published
relay.connected = True
await main_module.on_printer_status_change(1, _state(connected=True))
relay.on_printer_online.assert_awaited_once_with(1, "Workshop", "SN1")
assert 1 not in main_module._printer_offline_published
@pytest.mark.asyncio
async def test_no_online_without_a_published_offline(self):
"""A blip shorter than the debounce published no offline, so no online."""
relay, pm = self._relay_and_pm()
p1, p2, p3, p4, p5 = self._handler_patches(relay, pm)
with p1, p2, p3, p4, p5, patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()):
await main_module.on_printer_status_change(1, _state(connected=True))
await main_module.on_printer_status_change(1, _state(connected=False))
await main_module.on_printer_status_change(1, _state(connected=True))
await asyncio.sleep(0)
relay.on_printer_online.assert_not_awaited()
class TestOfflineEdgeDetection:
"""Edge detection inside `on_printer_status_change` — only the
True → False transition schedules a task. Reconnects cancel pending