"""Persistent Notify! Live Activities, independent of push and digest selection. Only the worker performs network I/O. MQTT callbacks copy their mutable state and wake it; database sessions are closed before every HTTP call. A durable intent is written *before* starting a tile: after a crash or an ambiguous network failure we never blindly repeat a push-to-start and consume another of the five device slots. """ import asyncio import hashlib import json import logging from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from uuid import uuid4 import httpx from sqlalchemy import delete, select, update from backend.app.core.database import async_session from backend.app.models.notification import NotificationProvider from backend.app.models.notification_live_activity import NotificationLiveActivity from backend.app.models.printer import Printer from backend.app.services.notify_client import NotifyClient, NotifyError, _https_url from backend.app.services.notify_worker import NotifyWorkerLifecycle logger = logging.getLogger(__name__) _RUNNING = {"RUNNING", "PRINTING", "PAUSE", "PREPARE", "SLICING"} _TERMINAL = {"IDLE", "FINISH", "FAILED", "COMPLETED", "CANCELLED", "ABORTED", "STOPPED"} _STOPPED = {"ended", "suppressed"} _ROLLOVER = {"expired", "overdue", "abandoned"} _INTERVAL = 60 _STARTUP_GRACE = timedelta(minutes=2) def _now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def _date(value) -> datetime | None: if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc).replace(tzinfo=None) except (ValueError, AttributeError): return None def _config(provider) -> dict: return json.loads(provider.config) if isinstance(provider.config, str) else provider.config or {} def _credential_key(config: dict) -> str: return hashlib.sha256(f"{config.get('device_id', '')}:{config.get('token', '')}".encode()).hexdigest() def _identity(state) -> str: job_id = str(getattr(state, "subtask_id", "") or "") if job_id and job_id != "0": return f"job:{job_id}"[:160] # raw_data contains MQTT deltas, so gcode_start_time may disappear on the # next push. The stable job ID is preferred; the filename fallback is given # a fresh generation by real start callbacks and adopted on restart. filename = getattr(state, "subtask_name", None) or getattr(state, "current_print", None) or "" digest = hashlib.sha256(str(filename).encode()).hexdigest()[:32] return f"file:{digest}" def _printer_fault(state) -> str | None: """Match the printer card's actionable HMS filtering, including print_error. Bambu's MQTT parser folds print_error into hms_errors. Level-3 sixteen-digit HMS notices without actions (for example an open cover) are informational; eight-digit task-stopping print_error prompts at that level still count. """ for error in getattr(state, "hms_errors", None) or []: severity = getattr(error, "severity", 0) description = str(getattr(error, "description", "") or "") actions = bool(getattr(error, "actions", None)) notice = severity == 3 and len(str(getattr(error, "full_code", "") or "")) == 16 if severity >= 1 and (actions or (description and not notice)): code = getattr(error, "full_code", "") or getattr(error, "code", "") return (description or f"Printer needs attention ({code})").replace("\x00", "")[:170] return None @dataclass(frozen=True) class PrintSnapshot: key: str connected: bool state: str filename: str progress: float remaining_seconds: int layer: int layers: int job_key: str | None = None fault: str | None = None temperatures: tuple = () stage: int = -1 @classmethod def from_state(cls, state): return cls( key=_identity(state), job_key=_identity(state) if _identity(state).startswith("job:") else None, connected=bool(state.connected), state=str(state.state or "").upper(), filename=str(getattr(state, "subtask_name", None) or getattr(state, "current_print", None) or "Print"), progress=max(0, min(100, float(state.progress or 0))), remaining_seconds=max(0, int(getattr(state, "remaining_time", 0) or 0) * 60), layer=int(state.layer_num or 0), layers=int(state.total_layers or 0), temperatures=tuple((getattr(state, "temperatures", {}) or {}).items()), stage=getattr(state, "stg_cur", -1), fault=_printer_fault(state), ) def content(self, printer_name: str, config: dict | None = None) -> dict: config = config or {} phase = "Printing" if not self.connected: phase = "Printer offline" elif self.state == "PAUSE": phase = "Paused" elif self.state in ("PREPARE", "SLICING") or ( self.state in {"RUNNING", "PRINTING"} and self.layer < 1 and (self.layers > 0 or self.stage not in {0, -1, 255}) ): phase = "Preparing" elif self.state in _TERMINAL: phase = { "FINISH": "Complete", "COMPLETED": "Complete", "FAILED": "Failed", "CANCELLED": "Stopped", "ABORTED": "Stopped", "STOPPED": "Stopped", "IDLE": "Finished", }[self.state] if self.connected and self.fault and self.state in _RUNNING: lower = self.fault.lower() runout = "filament" in lower and any( word in lower for word in ("run out", "ran out", "runout", "exhaust", "empty") ) phase = "Filament runout" if runout else "Printer error" # Character and UTF-8 limits leave ample room for server-owned fields # within Notify's 2048-byte merged-content budget, including emoji names. filename = ( "Print in progress" if config.get("live_activity_privacy") is True else self.filename.replace("\x00", "")[:120] ) layer = f" · Layer {self.layer}/{self.layers}" if self.layers else "" progress = 0 if phase == "Preparing" else self.progress content = { "title": printer_name.replace("\x00", "")[:80] or "Bambuddy", "body": self.fault if self.connected and self.fault and self.state in _RUNNING else f"{filename}{layer}"[:170], "symbol": str(config.get("live_activity_symbol") or "printer.fill")[:64], "tint": config.get("live_activity_tint") or "#00AE42", "progress": 100 if phase == "Complete" else progress, "status": phase, "endsIn": self.remaining_seconds if self.connected and not self.fault and self.state in {"RUNNING", "PRINTING"} and 0 < self.remaining_seconds <= 86400 else None, } style = config.get("live_activity_style", "bar") content.update( steps=10 if style == "segments" else None, step=int(progress // 10) if style == "segments" else None ) if style == "none": content["progress"] = None if ( config.get("live_activity_stage") is True and self.connected and not self.fault and self.state in {"RUNNING", "PRINTING"} ): from backend.app.services.bambu_mqtt import get_stage_name content["status"] = get_stage_name(self.stage)[:40] if self.stage >= 0 else phase eta = f"{self.remaining_seconds // 3600}h {(self.remaining_seconds // 60) % 60}m" content["trailing"] = eta if self.remaining_seconds > 86400 and phase == "Printing" else None # The default tile uses the native countdown. Metric chips are opt-in; # send null when disabled so merge-patch clears previously selected chips. metrics = [] temperatures = dict(self.temperatures) for field in (config.get("live_activity_metrics") or [])[:6]: label, value, unit = "", None, "" if field == "progress": label, value = "Progress", f"{progress:.0f}%" elif field == "eta" and phase == "Printing" and self.remaining_seconds > 0: label, value = "Remaining", eta elif field == "layers" and self.layers: label, value = "Layer", f"{self.layer}/{self.layers}" elif field in {"nozzle", "bed", "chamber"} and temperatures.get(field) is not None: label, value, unit = field.title(), f"{temperatures[field]:.0f}", "°C" if value is not None: metrics.append({"label": label, "value": value[:16], "unit": unit}) content["metrics"] = None if self.fault else metrics or None button = _https_url(config.get("live_activity_button_url")) content["button"] = ( {"title": "Open Bambuddy", "url": button, "open": True} if button and len(button) <= 512 else None ) # Notify adds timestamps when merging. Reserve 200 bytes for those; # shorten display text first, then drop optional cells/button if needed. def size(): return len(json.dumps(content, ensure_ascii=False, separators=(",", ":")).encode()) while size() > 1800 and len(content["body"]) > 20: content["body"] = content["body"][:-10] if size() > 1800: content["button"] = None while size() > 1800 and content["metrics"]: content["metrics"].pop() return content class NotifyLiveActivityService(NotifyWorkerLifecycle): ownership_model = NotificationLiveActivity feature_field = "live_activities" worker_interval = 30 worker_name = "notify-live-activities" stopped_states = ("ended", "suppressed") _cleanup_key = staticmethod(_credential_key) def __init__(self, session_factory=None, client=None): self._session = session_factory or async_session self._client = client self._http: httpx.AsyncClient | None = None self._snapshots: dict[int, PrintSnapshot] = {} self._new_fallbacks: dict[int, str] = {} self._started_at: dict[int, datetime] = {} self._progress_resets: dict[int, tuple[float, datetime]] = {} self._finished: list[tuple[int, str | None, str]] = [] self._lock = asyncio.Lock() self._wake = asyncio.Event() self._task: asyncio.Task | None = None self._init_worker() self._status_grace_until: datetime | None = None def _on_activated(self): # No MQTT observations are accepted while dormant. Anything retained # from before opt-out is stale until a real status arrives after opt-in. self._snapshots.clear() self._finished.clear() self._new_fallbacks.clear() self._started_at.clear() self._progress_resets.clear() self._status_grace_until = None def observe(self, printer_id: int, state) -> None: if self._provider_enabled is False: return snapshot = PrintSnapshot.from_state(state) if snapshot.state in {"", "UNKNOWN"}: previous = self._snapshots.get(printer_id) if previous is None or snapshot.connected: return # Broker connection alone is not a real printer status. snapshot = replace(previous, connected=False) reset = self._progress_resets.get(printer_id) if reset: initial, until = reset if ( snapshot.state not in _RUNNING or snapshot.progress < initial or snapshot.progress <= 5 or _now() >= until ): self._progress_resets.pop(printer_id, None) else: snapshot = replace(snapshot, progress=0, remaining_seconds=0) previous = self._snapshots.get(printer_id) if ( previous and previous.state in _RUNNING and snapshot.state in _RUNNING and previous.filename == snapshot.filename ): # A delta may omit or belatedly supply a job ID. Retain the print's # established identity until an explicit start or a different ID. if snapshot.key.startswith("file:") or previous.key.startswith("file:"): snapshot = replace(snapshot, key=previous.key) self._snapshots[printer_id] = snapshot if self._provider_enabled and ( previous is None or (previous.key, previous.state, previous.connected, previous.fault) != ( snapshot.key, snapshot.state, snapshot.connected, snapshot.fault, ) ): self._wake.set() def print_started(self, printer_id: int, state, data: dict | None = None) -> None: if state is None or self._provider_enabled is False: return snapshot = PrintSnapshot.from_state(state) job_id = (data or {}).get("subtask_id") or ((data or {}).get("raw_data") or {}).get("subtask_id") if job_id and str(job_id) != "0": snapshot = replace(snapshot, key=f"job:{job_id}", job_key=f"job:{job_id}") self._progress_resets[printer_id] = (snapshot.progress, _now() + timedelta(minutes=2)) snapshot = replace(snapshot, progress=0, remaining_seconds=0) self._started_at[printer_id] = _now() self._new_fallbacks.pop(printer_id, None) if snapshot.key.startswith("file:"): self._new_fallbacks[printer_id] = f"{snapshot.key}:{uuid4().hex}" self._snapshots[printer_id] = snapshot self._wake.set() def print_finished(self, printer_id: int, data: dict) -> None: if self._provider_enabled is False: return job = data.get("subtask_id") or (data.get("raw_data") or {}).get("subtask_id") snapshot = self._snapshots.get(printer_id) filename = data.get("subtask_name") or data.get("filename") key = f"job:{job}" if job and str(job) != "0" else None if key is None and snapshot and (not filename or snapshot.filename == filename): key = ( self._new_fallbacks.get(printer_id, snapshot.key) if snapshot.key.startswith("file:") else snapshot.key ) if key is None: # No unambiguous ownership: the real terminal state can reconcile # it, but a delayed event must never end the printer's next job. return self._finished.append((printer_id, key, str(data.get("status", "completed")).upper())) snapshot = self._snapshots.get(printer_id) if snapshot and key in (snapshot.key, snapshot.job_key): self._snapshots[printer_id] = replace(snapshot, state=self._finished[-1][2], remaining_seconds=0) self._wake.set() async def _api(self): if self._client is None: self._http = httpx.AsyncClient( timeout=httpx.Timeout(30, connect=5), follow_redirects=False, headers={"User-Agent": "Bambuddy/1.0"} ) self._client = NotifyClient(self._http) return self._client async def _save(self, row) -> None: async with self._session() as db: values = {column.name: getattr(row, column.name) for column in row.__table__.columns if column.name != "id"} # Never resurrect ownership after provider deletion. A queued cleanup # still holds this object's newly returned remote ID. await db.execute( update(NotificationLiveActivity).where(NotificationLiveActivity.id == row.id).values(**values) ) await db.commit() async def tick(self) -> None: async with self._lock: async with self._session() as db: providers = ( await db.scalars(select(NotificationProvider).where(NotificationProvider.provider_type == "notify")) ).all() any_enabled = any(self._eligible(provider) for provider in providers) previously_enabled = self._provider_enabled self._provider_enabled = any_enabled if any_enabled and previously_enabled is False: self._on_activated() if any_enabled and self._status_grace_until is None: # Start when reconciliation begins, not at module import. self._status_grace_until = _now() + _STARTUP_GRACE if not any_enabled and not previously_enabled and not self._cleanup_once and not self._cleanup_pending: return rows = (await db.scalars(select(NotificationLiveActivity))).all() self._working_rows = rows names = dict((await db.execute(select(Printer.id, Printer.name))).all()) if any_enabled else {} finished, self._finished = self._finished, [] snapshots = self._snapshots.copy() for provider in providers: config = _config(provider) owned = [r for r in rows if r.provider_id == provider.id] enabled = bool( provider.enabled and config.get("live_activities") is True and not str(config.get("device_id", "")).upper().startswith(("GRP", "MC", "WB")) ) for row in owned: snapshot = snapshots.get(row.printer_id) selected = provider.printer_id is None or provider.printer_id == row.printer_id if ( row.state == "ended" and row.end_reason == "disabled" and enabled and selected and snapshot and snapshot.connected and snapshot.state in _RUNNING and self._matches(row, snapshot, row.printer_id) ): # An explicit re-enable may resume this job's tile. A # user-dismissed/failed-to-appear tile remains suppressed. row.state, row.activity_id, row.end_reason = "pending", None, None row.failures, row.next_attempt_at, row.last_sent_at = 0, None, None await self._save(row) if ( snapshot and snapshot.job_key and not row.job_key and self._matches(row, snapshot, row.printer_id) ): row.job_key = snapshot.job_key await self._save(row) if self._retired_row(row): continue if row.state in _STOPPED: continue if row.credential_key != _credential_key(config): # The edit hook owns cleanup using the old credentials. continue if row.state == "ending": await self._end(row, config, row.end_reason or "STOPPED") continue snapshot = snapshots.get(row.printer_id) final = next( ( status for pid, key, status in finished if pid == row.printer_id and key in (row.print_key, row.job_key) ), None, ) selected = provider.printer_id is None or provider.printer_id == row.printer_id same_print = snapshot and self._matches(row, snapshot, row.printer_id) if final or not enabled or not selected or row.printer_id not in names: await self._end( row, config, final or ("DISABLED" if not enabled or not selected else "STOPPED") ) elif snapshot and snapshot.connected and snapshot.state in _TERMINAL: await self._end(row, config, snapshot.state) elif snapshot and snapshot.connected and snapshot.state in _RUNNING and not same_print: await self._end(row, config, "STOPPED") elif snapshot is None and self._status_grace_until and _now() < self._status_grace_until: # Allow reconnect time without leaving a switched-off # printer's saved countdown running indefinitely. continue else: # A missing/offline printer is not evidence the print ended. # Clear its countdown and keep its last known progress. content = ( snapshot.content(names[row.printer_id], config) if snapshot else json.loads(row.content) ) if snapshot is None or not snapshot.connected: content.update(status="Printer offline", endsIn=None, trailing=None) if snapshot is None: content["metrics"] = None # Saved ETA/temperature chips are stale after a restart. await self._sync( row, config, content, can_start=bool( snapshot and snapshot.connected and snapshot.state in _RUNNING and enabled and not self._quiet(provider) ), ) if not enabled or self._quiet(provider): continue for printer_id, snapshot in snapshots.items(): if ( not snapshot.connected or snapshot.state not in _RUNNING or printer_id not in names or provider.printer_id not in (None, printer_id) ): continue # Finish the old tile before using another device slot for # this printer. A failed DELETE remains durable and retries. if any(r.printer_id == printer_id and r.state == "ending" and r.activity_id for r in owned): continue if any(self._matches(r, snapshot, printer_id) for r in owned): continue key = ( self._new_fallbacks.get(printer_id, snapshot.key) if snapshot.key.startswith("file:") else snapshot.key ) row = NotificationLiveActivity( provider_id=provider.id, printer_id=printer_id, print_key=key, print_name=snapshot.filename[:255], job_key=snapshot.job_key, credential_key=_credential_key(config), state="pending", content=json.dumps(snapshot.content(names[printer_id], config)), created_at=_now(), ) if self._retired_row(row): continue self._working_rows.append(row) async with self._session() as db: db.add(row) await db.commit() await self._sync(row, config, json.loads(row.content), can_start=True) self._cleanup_pending = any( row.state == "ending" and row.provider_id in {p.id for p in providers} for row in rows ) # Tombstones need no maintenance while the integration is dormant. if not any_enabled: return # Tombstones must outlive any plausible print, but need not grow forever. async with self._session() as db: await db.execute( delete(NotificationLiveActivity).where( NotificationLiveActivity.state.in_(_STOPPED), NotificationLiveActivity.created_at < _now() - timedelta(days=30), NotificationLiveActivity.printer_id.not_in( [pid for pid, snapshot in snapshots.items() if snapshot.state in _RUNNING] ), ) ) await db.commit() def _matches(self, row, snapshot, printer_id): if row.printer_id != printer_id: return False if row.job_key and snapshot.job_key: return row.job_key == snapshot.job_key fallback = self._new_fallbacks.get(printer_id) if fallback and snapshot.key.startswith("file:"): return row.print_key == fallback if ( row.print_key.startswith("file:") and snapshot.key.startswith("job:") and row.print_name == snapshot.filename[:255] and row.job_key is None and self._started_at.get(printer_id, row.created_at) <= row.created_at ): return True # Restart after firmware finally supplied its job ID. return row.print_key == snapshot.key or ( snapshot.key.startswith("file:") and row.print_key.startswith(snapshot.key + ":") ) @staticmethod def _quiet(provider) -> bool: # Share Bambuddy's configured/local-time quiet-hours semantics. Digest # and event toggles deliberately never participate in this lifecycle. from backend.app.services.notification_service import notification_service return notification_service._is_in_quiet_hours(provider) async def _sync(self, row, config, content, *, can_start): if self._retired_row(row): return now = _now() urgent = json.loads(row.content).get("status") != content.get("status") if row.next_attempt_at and row.next_attempt_at > now and not (urgent and row.last_sent_at and not row.failures): return # Store an absolute deadline rather than repeatedly resetting a frozen # MQTT minute estimate. Refresh only when that estimate or phase changes. seconds = content.get("endsIn") if seconds is None: row.eta_seconds, row.eta_deadline = None, None elif seconds != row.eta_seconds or row.eta_deadline is None: row.eta_seconds = seconds row.eta_deadline = now + timedelta(seconds=seconds) if row.eta_deadline: remaining = int((row.eta_deadline - now).total_seconds()) content["endsIn"] = remaining if remaining > 0 else None api = await self._api() token = config.get("token", "") try: if row.activity_id: remote = await api.get_activity(row.activity_id, token) state, reason = remote.get("state"), remote.get("endReason") if state in {"ended", "dismissed"}: if reason not in _ROLLOVER or not can_start: row.state = "suppressed" if reason not in _ROLLOVER else "pending" row.end_reason = reason or state if row.state == "pending": row.activity_id = None await self._save(row) return row.activity_id = None row.state = "pending" await self._save(row) else: row.state = state or "active" row.expires_at = _date(remote.get("expiresAt")) or row.expires_at if can_start and row.expires_at and row.expires_at <= now: # Poll first: a dismissed tile must never be rolled over. await api.end_activity(row.activity_id, token) row.activity_id = None row.state = "pending" await self._save(row) else: if self._retired_row(row): return await api.update_activity(row.activity_id, token, content) row.content = json.dumps(content) row.failures = 0 row.last_sent_at = now row.next_attempt_at = now + timedelta(seconds=_INTERVAL) await self._save(row) return if row.state == "uncertain": # No usable ID after a timeout/crash. The API has no client # idempotency key; do not adopt an unrelated device tile. row.state = "suppressed" row.end_reason = "unconfirmed-start" await self._save(row) return if not can_start or row.state in _STOPPED: return row.state = "uncertain" row.content = json.dumps(content) await self._save(row) if self._retired_row(row): return result = await api.start_activity(config.get("device_id", ""), token, content) row.activity_id = result["activityId"] row.state = "starting" row.failures = 0 row.expires_at = _date(result.get("expiresAt")) or now + timedelta(hours=8) row.last_sent_at = now row.next_attempt_at = now + timedelta(seconds=_INTERVAL) await self._save(row) except NotifyError as error: await self._failure(row, error) async def _failure(self, row, error): ending = row.state == "ending" if error.activity_id: row.activity_id = error.activity_id row.state = "starting" elif not row.activity_id and error.delivery_state == "unknown": row.state = "suppressed" row.end_reason = "unconfirmed-start" elif error.status_code in (401, 403): row.state = "suppressed" row.end_reason = "credentials" elif error.status_code == 410 and row.activity_id: # GET on the next pass determines whether this was a dismissal or # the eight-hour ceiling; never infer rollover from HTTP 410 alone. pass elif not row.activity_id: if error.status_code in (409, 429, 503) or error.retry_after_seconds is not None: row.state = "pending" elif error.status_code == 400 and "5" in str(error.payload.get("message", "")): row.state = "pending" # Device cap: another printer may free a slot. else: row.state = "suppressed" row.end_reason = "start-rejected" if ending: row.state = "ending" row.failures = min((row.failures or 0) + 1, 5) row.next_attempt_at = _now() + timedelta( seconds=max(60 * 2 ** (row.failures - 1), error.retry_after_seconds or 0) ) # Do not log upstream bodies, URLs, or credentials. logger.warning( "Notify Live Activity request failed for provider %s (HTTP %s)", row.provider_id, error.status_code ) await self._save(row) message = str(error) if error.retry_after_seconds: message += f" Retry after {error.retry_after_seconds} seconds." async with self._session() as db: await db.execute( update(NotificationProvider) .where(NotificationProvider.id == row.provider_id) .values( last_error=message, last_error_at=_now(), ) ) await db.commit() async def _end(self, row, config, status): content = json.loads(row.content) content.update( status={"COMPLETED": "Complete", "FINISH": "Complete", "FAILED": "Failed"}.get(status, "Stopped"), endsIn=None, ) if status in {"COMPLETED", "FINISH"}: content["progress"] = 100 # Persist the terminal decision before I/O. A failed end must still # finish this exact job after restart or after the next job begins. if row.state != "ending": row.state = "ending" row.end_reason = status row.content = json.dumps(content) await self._save(row) if row.failures and row.next_attempt_at and row.next_attempt_at > _now(): return if row.activity_id: try: await (await self._api()).end_activity(row.activity_id, config.get("token", ""), content) except NotifyError as error: if error.status_code not in (403, 404, 410): await self._failure(row, error) return row.state = "ended" row.end_reason = status.lower() row.content = json.dumps(content) await self._save(row) async def cleanup_provider(self, provider_id: int, old_config: dict, *, captured_rows=None) -> None: """Called after saving a credential/scope change or deleting a provider. Cleanup is best effort (the remote eight-hour ceiling bounds an outage). No old credential is retained in a second database location. """ async with self._lock: async with self._session() as db: rows = ( await db.scalars( select(NotificationLiveActivity).where( NotificationLiveActivity.provider_id == provider_id, NotificationLiveActivity.credential_key == _credential_key(old_config), ) ) ).all() if captured_rows is not None: merged = {row.id: row for row in rows} merged.update( {row.id: row for row in captured_rows if row.credential_key == _credential_key(old_config)} ) rows = list(merged.values()) for row in rows: if row.activity_id and row.state not in _STOPPED: try: await (await self._api()).end_activity(row.activity_id, old_config.get("token", "")) except NotifyError: logger.warning("Notify Live Activity cleanup failed for provider %s", provider_id) async with self._session() as db: await db.execute( delete(NotificationLiveActivity).where( NotificationLiveActivity.provider_id == provider_id, NotificationLiveActivity.credential_key == _credential_key(old_config), ) ) await db.commit() for row in rows: row.state = "ended" self._wake.set() notify_live_activities = NotifyLiveActivityService()