From 368999a23492a9c0d9bd255bdddebe0ce62b5fe5 Mon Sep 17 00:00:00 2001 From: maziggy Date: Tue, 13 Jan 2026 12:31:45 +0100 Subject: [PATCH] Add MQTT publishing for external automation integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New feature: Publish BamBuddy events to external MQTT brokers for integration with Home Assistant, Node-RED, and other automation platforms. Backend changes: - New MQTTRelayService (backend/app/services/mqtt_relay.py) with paho-mqtt - MQTT settings added to AppSettings schema (enabled, broker, port, auth, TLS) - New /settings/mqtt/status endpoint for connection status - Event hooks in main.py, print_queue.py, print_scheduler.py, maintenance.py, smart_plugs.py - PrinterInfo class and get_printer() method added to PrinterManager - MQTT reconfiguration added to backup/restore flow - Printer status throttled to 1 update/sec to avoid flooding Frontend changes: - New "Network" tab in Settings (between Filament and API Keys) - FTP Retry settings moved from General to Network tab - MQTT configuration card with broker, port, TLS toggle, auth fields - Port auto-populates when TLS toggled (1883 ↔ 8883) - Connection status indicator (green/red dot) in tab and card header - Real-time status polling when on Network tab Published topics: - bambuddy/status - Online/offline - bambuddy/printers/{serial}/status - Printer state (throttled) - bambuddy/printers/{serial}/print/* - Print lifecycle - bambuddy/printers/{serial}/ams/changed - AMS changes - bambuddy/queue/* - Queue events - bambuddy/maintenance/* - Maintenance alerts - bambuddy/smart_plugs/* - Plug state/energy - bambuddy/archive/* - Archive events Tests: - Added MQTT settings and status endpoint tests Closes #78 --- CHANGELOG.md | 22 + README.md | 1 + backend/app/api/routes/maintenance.py | 12 + backend/app/api/routes/print_queue.py | 14 + backend/app/api/routes/settings.py | 63 ++ backend/app/api/routes/smart_plugs.py | 22 + backend/app/main.py | 122 +++- backend/app/schemas/settings.py | 16 + backend/app/services/mqtt_relay.py | 650 ++++++++++++++++++ backend/app/services/print_scheduler.py | 14 + backend/app/services/printer_manager.py | 15 + .../tests/integration/test_settings_api.py | 71 ++ frontend/src/api/client.ts | 18 + frontend/src/pages/SettingsPage.tsx | 397 ++++++++--- .../{index-rSeEpV8h.js => index-DvXWbCbD.js} | 184 ++--- static/index.html | 2 +- static/sw.js | 4 +- 17 files changed, 1432 insertions(+), 195 deletions(-) create mode 100644 backend/app/services/mqtt_relay.py rename static/assets/{index-rSeEpV8h.js => index-DvXWbCbD.js} (61%) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4b004b01..4bfd513e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,28 @@ All notable changes to Bambuddy will be documented in this file. +## [0.1.6b11] - 2026-01-13 + +### Added +- **MQTT Publishing** - Publish BamBuddy events to external MQTT brokers for integration with Home Assistant, Node-RED, and other automation platforms: + - New "Network" tab in Settings (between Filament and API Keys) + - Configure broker hostname, port, username/password, TLS, and topic prefix + - Auto-populate port when toggling TLS (1883 ↔ 8883) + - Real-time connection status indicator in tab and settings card + - Published topics include: + - `bambuddy/status` - Online/offline status + - `bambuddy/printers/{serial}/status` - Real-time printer state (throttled to 1/sec) + - `bambuddy/printers/{serial}/print/started|completed|failed` - Print lifecycle events + - `bambuddy/printers/{serial}/ams/changed` - AMS filament changes + - `bambuddy/queue/job_added|job_started|job_completed` - Print queue events + - `bambuddy/maintenance/alert|reset` - Maintenance notifications + - `bambuddy/smart_plugs/on|off|energy` - Smart plug state changes + - `bambuddy/archive/created|updated` - Archive events + - Supports TLS/SSL encryption with self-signed certificates + - Auto-reconnect when settings change + - Settings included in backup/restore +- **FTP Retry moved to Network tab** - FTP retry settings relocated from General to Network tab for better organization + ## [0.1.6b10] - 2026-01-11 ### Added diff --git a/README.md b/README.md index f973dbc0d..8f7c420e2 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,7 @@ ### 🔧 Integrations - [Spoolman](https://github.com/Donkie/Spoolman) filament sync +- MQTT publishing for Home Assistant, Node-RED, etc. - Bambu Cloud profile management - K-profiles (pressure advance) - External sidebar links diff --git a/backend/app/api/routes/maintenance.py b/backend/app/api/routes/maintenance.py index 782521247..5e2f5115d 100644 --- a/backend/app/api/routes/maintenance.py +++ b/backend/app/api/routes/maintenance.py @@ -496,6 +496,18 @@ async def perform_maintenance( await db.commit() + # MQTT relay - publish maintenance reset + try: + from backend.app.services.mqtt_relay import mqtt_relay + + await mqtt_relay.on_maintenance_reset( + printer_id=item.printer_id, + printer_name=printer.name, + maintenance_type=item.maintenance_type.name, + ) + except Exception: + pass # Don't fail if MQTT fails + # Calculate status interval = item.custom_interval_hours or item.maintenance_type.default_interval_hours interval_type = getattr(item.maintenance_type, "interval_type", "hours") or "hours" diff --git a/backend/app/api/routes/print_queue.py b/backend/app/api/routes/print_queue.py index d76f9b30e..4843f0525 100644 --- a/backend/app/api/routes/print_queue.py +++ b/backend/app/api/routes/print_queue.py @@ -128,6 +128,20 @@ async def add_to_queue( await db.refresh(item, ["archive", "printer"]) logger.info(f"Added archive {data.archive_id} to queue for printer {data.printer_id}") + + # MQTT relay - publish queue job added + try: + from backend.app.services.mqtt_relay import mqtt_relay + + await mqtt_relay.on_queue_job_added( + job_id=item.id, + filename=item.archive.filename if item.archive else "", + printer_id=item.printer_id, + printer_name=item.printer.name if item.printer else None, + ) + except Exception: + pass # Don't fail queue add if MQTT fails + return _enrich_response(item) diff --git a/backend/app/api/routes/settings.py b/backend/app/api/routes/settings.py index 3034cae43..e7a5abb9e 100644 --- a/backend/app/api/routes/settings.py +++ b/backend/app/api/routes/settings.py @@ -73,6 +73,8 @@ async def get_settings(db: AsyncSession = Depends(get_db)): "telemetry_enabled", "virtual_printer_enabled", "ftp_retry_enabled", + "mqtt_enabled", + "mqtt_use_tls", ]: settings_dict[setting.key] = setting.value.lower() == "true" elif setting.key in ["default_filament_cost", "energy_cost_per_kwh", "ams_temp_good", "ams_temp_fair"]: @@ -83,6 +85,7 @@ async def get_settings(db: AsyncSession = Depends(get_db)): "ams_history_retention_days", "ftp_retry_count", "ftp_retry_delay", + "mqtt_port", ]: settings_dict[setting.key] = int(setting.value) elif setting.key == "default_printer_id": @@ -102,6 +105,18 @@ async def update_settings( """Update application settings.""" update_data = settings_update.model_dump(exclude_unset=True) + # Check if any MQTT settings are being updated + mqtt_keys = { + "mqtt_enabled", + "mqtt_broker", + "mqtt_port", + "mqtt_username", + "mqtt_password", + "mqtt_topic_prefix", + "mqtt_use_tls", + } + mqtt_updated = bool(mqtt_keys & set(update_data.keys())) + for key, value in update_data.items(): # Convert value to string for storage if isinstance(value, bool): @@ -114,6 +129,24 @@ async def update_settings( await db.commit() + # Reconfigure MQTT relay if any MQTT settings changed + if mqtt_updated: + try: + from backend.app.services.mqtt_relay import mqtt_relay + + mqtt_settings = { + "mqtt_enabled": (await get_setting(db, "mqtt_enabled") or "false") == "true", + "mqtt_broker": await get_setting(db, "mqtt_broker") or "", + "mqtt_port": int(await get_setting(db, "mqtt_port") or "1883"), + "mqtt_username": await get_setting(db, "mqtt_username") or "", + "mqtt_password": await get_setting(db, "mqtt_password") or "", + "mqtt_topic_prefix": await get_setting(db, "mqtt_topic_prefix") or "bambuddy", + "mqtt_use_tls": (await get_setting(db, "mqtt_use_tls") or "false") == "true", + } + await mqtt_relay.configure(mqtt_settings) + except Exception: + pass # Don't fail the settings update if MQTT reconfiguration fails + # Return updated settings return await get_settings(db) @@ -1623,6 +1656,23 @@ async def import_backup( except Exception: pass # Virtual printer config failed, but don't fail the restore + # Reconfigure MQTT relay if settings were restored + try: + from backend.app.services.mqtt_relay import mqtt_relay + + mqtt_settings = { + "mqtt_enabled": (await get_setting(db, "mqtt_enabled") or "false") == "true", + "mqtt_broker": await get_setting(db, "mqtt_broker") or "", + "mqtt_port": int(await get_setting(db, "mqtt_port") or "1883"), + "mqtt_username": await get_setting(db, "mqtt_username") or "", + "mqtt_password": await get_setting(db, "mqtt_password") or "", + "mqtt_topic_prefix": await get_setting(db, "mqtt_topic_prefix") or "bambuddy", + "mqtt_use_tls": (await get_setting(db, "mqtt_use_tls") or "false") == "true", + } + await mqtt_relay.configure(mqtt_settings) + except Exception: + pass # MQTT relay config failed, but don't fail the restore + # Build summary message restored_parts = [] for key, count in restored.items(): @@ -1780,3 +1830,16 @@ async def update_virtual_printer_settings( ) return await get_virtual_printer_settings(db) + + +# ============================================================================= +# MQTT Relay Settings +# ============================================================================= + + +@router.get("/mqtt/status") +async def get_mqtt_status(): + """Get MQTT relay connection status.""" + from backend.app.services.mqtt_relay import mqtt_relay + + return mqtt_relay.get_status() diff --git a/backend/app/api/routes/smart_plugs.py b/backend/app/api/routes/smart_plugs.py index 5102d80d7..0ffb5568f 100644 --- a/backend/app/api/routes/smart_plugs.py +++ b/backend/app/api/routes/smart_plugs.py @@ -298,6 +298,28 @@ async def control_smart_plug( plug.last_checked = datetime.utcnow() await db.commit() + # MQTT relay - publish smart plug state change + if expected_state: + 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, + ) + except Exception: + pass # Don't fail if MQTT fails + return {"success": True, "action": control.action} diff --git a/backend/app/main.py b/backend/app/main.py index e376fad9b..d6b3c44a3 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -85,6 +85,7 @@ from backend.app.models.smart_plug import SmartPlug from backend.app.services.archive import ArchiveService from backend.app.services.bambu_ftp import download_file_async, get_ftp_retry_settings, with_ftp_retry from backend.app.services.bambu_mqtt import PrinterState +from backend.app.services.mqtt_relay import mqtt_relay from backend.app.services.notification_service import notification_service from backend.app.services.print_scheduler import scheduler as print_scheduler from backend.app.services.printer_manager import ( @@ -244,8 +245,17 @@ async def on_printer_status_change(printer_id: int, state: PrinterState): f"{state.cooling_fan_speed}:{state.big_fan1_speed}:{state.big_fan2_speed}:" f"{state.chamber_light}" ) + + # MQTT relay - publish status (before dedup check - always publish to MQTT) + try: + printer_info = printer_manager.get_printer(printer_id) + if printer_info: + await mqtt_relay.on_printer_status(printer_id, state, printer_info.name, printer_info.serial_number) + except Exception: + pass # Don't fail status callback if MQTT fails + if _last_status_broadcast.get(printer_id) == status_key: - return # No change, skip broadcast + return # No change, skip WebSocket broadcast _last_status_broadcast[printer_id] = status_key @@ -261,6 +271,14 @@ async def on_ams_change(printer_id: int, ams_data: list): logger = logging.getLogger(__name__) + # MQTT relay - publish AMS change + try: + printer_info = printer_manager.get_printer(printer_id) + if printer_info: + await mqtt_relay.on_ams_change(printer_id, printer_info.name, printer_info.serial_number, ams_data) + except Exception: + pass # Don't fail AMS callback if MQTT fails + try: async with async_session() as db: from backend.app.api.routes.settings import get_setting @@ -379,6 +397,20 @@ async def on_print_start(printer_id: int, data: dict): await ws_manager.send_print_start(printer_id, data) + # MQTT relay - publish print start + try: + printer_info = printer_manager.get_printer(printer_id) + if printer_info: + await mqtt_relay.on_print_start( + printer_id, + printer_info.name, + printer_info.serial_number, + data.get("filename", ""), + data.get("subtask_name", ""), + ) + except Exception: + pass # Don't fail print start callback if MQTT fails + # Track if notification was sent (to avoid sending twice) notification_sent = False @@ -751,6 +783,17 @@ async def on_print_start(printer_id: int, data: dict): } ) + # MQTT relay - publish archive created + try: + await mqtt_relay.on_archive_created( + archive_id=fallback_archive.id, + print_name=fallback_archive.print_name, + printer_name=printer.name, + status=fallback_archive.status, + ) + except Exception: + pass # Don't fail if MQTT fails + # Send notification without archive data (file not found) if not notification_sent: await _send_print_start_notification(printer_id, data, logger=logger) @@ -813,6 +856,17 @@ async def on_print_start(printer_id: int, data: dict): } ) + # MQTT relay - publish archive created + try: + await mqtt_relay.on_archive_created( + archive_id=archive.id, + print_name=archive.print_name, + printer_name=printer.name, + status=archive.status, + ) + except Exception: + pass # Don't fail if MQTT fails + # Send notification with archive data (new archive created) if not notification_sent: archive_data = {"print_time_seconds": archive.print_time_seconds} @@ -985,6 +1039,21 @@ async def on_print_complete(printer_id: int, data: dict): except Exception as e: logger.warning(f"[CALLBACK] WebSocket send_print_complete failed: {e}") + # MQTT relay - publish print complete + try: + printer_info = printer_manager.get_printer(printer_id) + if printer_info: + await mqtt_relay.on_print_complete( + printer_id, + printer_info.name, + printer_info.serial_number, + data.get("filename", ""), + data.get("subtask_name", ""), + data.get("status", "completed"), + ) + except Exception: + pass # Don't fail print complete callback if MQTT fails + filename = data.get("filename", "") subtask_name = data.get("subtask_name", "") @@ -1140,6 +1209,16 @@ async def on_print_complete(printer_id: int, data: dict): } ) logger.info(f"[ARCHIVE] WebSocket notification sent for archive {archive_id}") + + # MQTT relay - publish archive updated + try: + await mqtt_relay.on_archive_updated( + archive_id=archive_id, + print_name=filename or subtask_name, + status=status, + ) + except Exception: + pass # Don't fail if MQTT fails except Exception as e: logger.error(f"[ARCHIVE] Failed to update archive {archive_id} status: {e}", exc_info=True) # Continue with other operations even if archive update fails @@ -1340,6 +1419,19 @@ async def on_print_complete(printer_id: int, data: dict): if items_needing_attention: await notification_service.on_maintenance_due(printer_id, printer_name, items_needing_attention, db) logger.info(f"[MAINT-BG] Sent notification: {len(items_needing_attention)} items need attention") + + # MQTT relay - publish maintenance alerts + for item in items_needing_attention: + 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 + ) + except Exception: + pass # Don't fail if MQTT fails else: logger.info("[MAINT-BG] Completed (no items need attention)") except Exception as e: @@ -1379,6 +1471,19 @@ async def on_print_complete(printer_id: int, data: dict): await db.commit() logger.info(f"Updated queue item {queue_item.id} status to {status}") + # MQTT relay - publish queue job completed + try: + printer_info = printer_manager.get_printer(printer_id) + await mqtt_relay.on_queue_job_completed( + job_id=queue_item.id, + filename=filename or subtask_name, + printer_id=printer_id, + printer_name=printer_info.name if printer_info else "Unknown", + status=status, + ) + except Exception: + pass # Don't fail if MQTT fails + # Handle auto_off_after - power off printer if requested (after cooldown) if queue_item.auto_off_after: result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id)) @@ -1734,6 +1839,21 @@ async def lifespan(app: FastAPI): printer_manager.set_print_complete_callback(on_print_complete) printer_manager.set_ams_change_callback(on_ams_change) + # Initialize MQTT relay from settings + async with async_session() as db: + from backend.app.api.routes.settings import get_setting + + mqtt_settings = { + "mqtt_enabled": (await get_setting(db, "mqtt_enabled") or "false") == "true", + "mqtt_broker": await get_setting(db, "mqtt_broker") or "", + "mqtt_port": int(await get_setting(db, "mqtt_port") or "1883"), + "mqtt_username": await get_setting(db, "mqtt_username") or "", + "mqtt_password": await get_setting(db, "mqtt_password") or "", + "mqtt_topic_prefix": await get_setting(db, "mqtt_topic_prefix") or "bambuddy", + "mqtt_use_tls": (await get_setting(db, "mqtt_use_tls") or "false") == "true", + } + await mqtt_relay.configure(mqtt_settings) + # Connect to all active printers async with async_session() as db: await init_printer_connections(db) diff --git a/backend/app/schemas/settings.py b/backend/app/schemas/settings.py index bd71bf53d..c03f4f851 100644 --- a/backend/app/schemas/settings.py +++ b/backend/app/schemas/settings.py @@ -73,6 +73,15 @@ class AppSettings(BaseModel): ftp_retry_count: int = Field(default=3, description="Number of retry attempts for FTP operations (1-10)") ftp_retry_delay: int = Field(default=2, description="Seconds to wait between FTP retry attempts (1-30)") + # MQTT Relay settings for publishing events to external broker + mqtt_enabled: bool = Field(default=False, description="Enable MQTT event publishing to external broker") + mqtt_broker: str = Field(default="", description="MQTT broker hostname or IP address") + mqtt_port: int = Field(default=1883, description="MQTT broker port (default 1883, TLS typically 8883)") + mqtt_username: str = Field(default="", description="MQTT username for authentication (optional)") + mqtt_password: str = Field(default="", description="MQTT password for authentication (optional)") + mqtt_topic_prefix: str = Field(default="bambuddy", description="Topic prefix for all published messages") + mqtt_use_tls: bool = Field(default=False, description="Use TLS/SSL encryption for MQTT connection") + class AppSettingsUpdate(BaseModel): """Schema for updating settings (all fields optional).""" @@ -110,3 +119,10 @@ class AppSettingsUpdate(BaseModel): ftp_retry_enabled: bool | None = None ftp_retry_count: int | None = None ftp_retry_delay: int | None = None + mqtt_enabled: bool | None = None + mqtt_broker: str | None = None + mqtt_port: int | None = None + mqtt_username: str | None = None + mqtt_password: str | None = None + mqtt_topic_prefix: str | None = None + mqtt_use_tls: bool | None = None diff --git a/backend/app/services/mqtt_relay.py b/backend/app/services/mqtt_relay.py new file mode 100644 index 000000000..de0b78d8c --- /dev/null +++ b/backend/app/services/mqtt_relay.py @@ -0,0 +1,650 @@ +"""MQTT Relay Service for publishing BamBuddy events to external MQTT brokers. + +This service enables integration with external automation systems like +Node-RED, Home Assistant, and other MQTT-based platforms. +""" + +import asyncio +import json +import logging +import ssl +import threading +import time +from datetime import datetime +from typing import Any + +import paho.mqtt.client as mqtt + +logger = logging.getLogger(__name__) + + +class MQTTRelayService: + """Publishes BamBuddy events to an external MQTT broker.""" + + # Minimum interval between status updates per printer (seconds) + STATUS_THROTTLE_SECONDS = 1.0 + + def __init__(self): + self.client: mqtt.Client | None = None + self.enabled = False + self.connected = False + self.topic_prefix = "bambuddy" + self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._broker = "" + self._port = 1883 + self._last_printer_status: dict[int, float] = {} # printer_id -> last publish timestamp + + async def configure(self, settings: dict) -> bool: + """Configure MQTT connection from settings. + + Returns True if connection was successful or MQTT is disabled. + """ + self.enabled = settings.get("mqtt_enabled", False) + + if not self.enabled: + await self.disconnect() + logger.info("MQTT relay disabled") + return True + + broker = settings.get("mqtt_broker", "") + port = settings.get("mqtt_port", 1883) + username = settings.get("mqtt_username", "") + password = settings.get("mqtt_password", "") + self.topic_prefix = settings.get("mqtt_topic_prefix", "bambuddy") + use_tls = settings.get("mqtt_use_tls", False) + + if not broker: + logger.warning("MQTT enabled but no broker configured") + return False + + # Store for status endpoint + self._broker = broker + self._port = port + + # Disconnect existing connection if settings changed + if self.client: + await self.disconnect() + + # Create and connect client + return await self._connect(broker, port, username, password, use_tls) + + async def _connect(self, broker: str, port: int, username: str, password: str, use_tls: bool) -> bool: + """Establish MQTT connection.""" + try: + # Create client with callback API version 2 (use MQTTv311 for broader compatibility) + self.client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id=f"bambuddy-{id(self)}", + protocol=mqtt.MQTTv311, + ) + + # Set up callbacks + self.client.on_connect = self._on_connect + self.client.on_disconnect = self._on_disconnect + + # Configure authentication + if username: + self.client.username_pw_set(username, password) + + # Configure TLS (allow self-signed certs for testing) + if use_tls: + self.client.tls_set(cert_reqs=ssl.CERT_NONE) + self.client.tls_insecure_set(True) # Allow self-signed certs + + # Connect (non-blocking with loop_start) + self.client.connect_async(broker, port, keepalive=60) + self.client.loop_start() + + # Wait briefly for connection + await asyncio.sleep(1.0) + + if self.connected: + logger.info(f"MQTT relay connected to {broker}:{port}") + # Publish online status + self._publish_status("online") + return True + else: + logger.warning(f"MQTT relay connection pending to {broker}:{port}") + return True # Connection is async, may succeed later + + except Exception as e: + logger.error(f"MQTT relay connection failed: {e}") + self.connected = False + return False + + def _on_connect( + self, + client: mqtt.Client, + userdata: Any, + flags: dict, + reason_code: int | mqtt.ReasonCode, + properties: mqtt.Properties | None = None, + ): + """Callback when connected to broker.""" + # Handle both MQTTv311 (int) and MQTTv5 (ReasonCode) return codes + rc = reason_code if isinstance(reason_code, int) else reason_code.value + if rc == 0: + self.connected = True + logger.info("MQTT relay connected successfully") + # Publish online status + self._publish_status("online") + else: + self.connected = False + logger.error(f"MQTT relay connection failed: {reason_code}") + + def _on_disconnect( + self, + client: mqtt.Client, + userdata: Any, + flags_or_rc: dict | int | mqtt.ReasonCode, + reason_code: int | mqtt.ReasonCode | None = None, + properties: mqtt.Properties | None = None, + ): + """Callback when disconnected from broker.""" + self.connected = False + # Handle both MQTTv311 (rc as 3rd param) and MQTTv5 (flags, rc, props) + rc = reason_code if reason_code is not None else flags_or_rc + rc_val = rc if isinstance(rc, int) else getattr(rc, "value", 0) + if rc_val != 0: + logger.warning(f"MQTT relay disconnected: {rc}") + else: + logger.info("MQTT relay disconnected cleanly") + + async def disconnect(self): + """Disconnect from MQTT broker.""" + if self.client: + try: + # Publish offline status before disconnecting + self._publish_status("offline") + self.client.loop_stop() + self.client.disconnect() + except Exception as e: + logger.debug(f"MQTT disconnect error (ignored): {e}") + finally: + self.client = None + self.connected = False + + def _publish_status(self, status: str): + """Publish BamBuddy status (online/offline).""" + self._publish( + f"{self.topic_prefix}/status", + {"status": status, "timestamp": datetime.utcnow().isoformat()}, + retain=True, + ) + + def _publish(self, topic: str, payload: dict, retain: bool = False): + """Publish message to MQTT broker.""" + if not self.client or not self.connected: + return + + try: + with self._lock: + self.client.publish(topic, json.dumps(payload, default=str), qos=1, retain=retain) + except Exception as e: + logger.debug(f"MQTT publish error: {e}") + + def get_status(self) -> dict: + """Get current MQTT relay status for API.""" + return { + "enabled": self.enabled, + "connected": self.connected, + "broker": self._broker if self.enabled else "", + "port": self._port if self.enabled else 0, + "topic_prefix": self.topic_prefix, + } + + # ========================================================================= + # Printer Events + # ========================================================================= + + async def on_printer_status(self, printer_id: int, state: Any, printer_name: str, printer_serial: str): + """Publish printer status change (throttled to 1 update/sec per printer).""" + if not self.enabled or not self.connected: + return + + # Throttle status updates to avoid flooding MQTT broker + now = time.time() + last_publish = self._last_printer_status.get(printer_id, 0) + if now - last_publish < self.STATUS_THROTTLE_SECONDS: + return # Skip this update, too soon since last publish + self._last_printer_status[printer_id] = now + + # Build status payload from PrinterState + payload = { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "timestamp": datetime.utcnow().isoformat(), + "connected": state.connected, + "state": state.state, + "progress": state.progress, + "remaining_time": state.remaining_time, + "layer_num": state.layer_num, + "total_layers": state.total_layers, + "current_print": state.current_print, + "subtask_name": state.subtask_name, + "gcode_file": state.gcode_file, + "temperatures": state.temperatures, + "wifi_signal": state.wifi_signal, + "chamber_light": state.chamber_light, + "speed_level": state.speed_level, + "cooling_fan_speed": state.cooling_fan_speed, + "big_fan1_speed": state.big_fan1_speed, + "big_fan2_speed": state.big_fan2_speed, + "heatbreak_fan_speed": state.heatbreak_fan_speed, + } + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/status", + payload, + retain=True, + ) + + async def on_printer_online(self, printer_id: int, printer_name: str, printer_serial: str): + """Publish printer came online event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/online", + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_printer_offline(self, printer_id: int, printer_name: str, printer_serial: str): + """Publish printer went offline event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/offline", + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_print_start( + self, + printer_id: int, + printer_name: str, + printer_serial: str, + filename: str, + subtask_name: str, + ): + """Publish print started event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/print/started", + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "filename": filename, + "subtask_name": subtask_name, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_print_complete( + self, + printer_id: int, + printer_name: str, + printer_serial: str, + filename: str, + subtask_name: str, + status: str, + ): + """Publish print completed event.""" + if not self.enabled or not self.connected: + return + + # Determine topic based on status + if status == "completed": + topic = f"{self.topic_prefix}/printers/{printer_serial}/print/completed" + else: + topic = f"{self.topic_prefix}/printers/{printer_serial}/print/failed" + + self._publish( + topic, + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "filename": filename, + "subtask_name": subtask_name, + "status": status, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_ams_change( + self, + printer_id: int, + printer_name: str, + printer_serial: str, + ams_data: list, + ): + """Publish AMS filament change event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/ams/changed", + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "ams_units": ams_data, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_printer_error( + self, + printer_id: int, + printer_name: str, + printer_serial: str, + errors: list, + ): + """Publish printer HMS error event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/printers/{printer_serial}/error", + { + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "errors": errors, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + # ========================================================================= + # Print Queue Events + # ========================================================================= + + async def on_queue_job_added( + self, + job_id: int, + filename: str, + printer_id: int | None, + printer_name: str | None, + ): + """Publish job added to queue event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/queue/job_added", + { + "job_id": job_id, + "filename": filename, + "printer_id": printer_id, + "printer_name": printer_name, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_queue_job_started( + self, + job_id: int, + filename: str, + printer_id: int, + printer_name: str, + printer_serial: str, + ): + """Publish queued job started printing event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/queue/job_started", + { + "job_id": job_id, + "filename": filename, + "printer_id": printer_id, + "printer_name": printer_name, + "printer_serial": printer_serial, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_queue_job_completed( + self, + job_id: int, + filename: str, + printer_id: int, + printer_name: str, + status: str, + ): + """Publish queued job finished event.""" + if not self.enabled or not self.connected: + return + + topic = ( + f"{self.topic_prefix}/queue/job_completed" + if status == "completed" + else f"{self.topic_prefix}/queue/job_failed" + ) + + self._publish( + topic, + { + "job_id": job_id, + "filename": filename, + "printer_id": printer_id, + "printer_name": printer_name, + "status": status, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + # ========================================================================= + # Maintenance Events + # ========================================================================= + + async def on_maintenance_alert( + self, + printer_id: int, + printer_name: str, + maintenance_type: str, + current_value: float, + threshold: float, + ): + """Publish maintenance alert triggered event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/maintenance/alert", + { + "printer_id": printer_id, + "printer_name": printer_name, + "maintenance_type": maintenance_type, + "current_value": current_value, + "threshold": threshold, + "timestamp": datetime.utcnow().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, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_maintenance_reset( + self, + printer_id: int, + printer_name: str, + maintenance_type: str, + ): + """Publish maintenance counter reset event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/maintenance/reset", + { + "printer_id": printer_id, + "printer_name": printer_name, + "maintenance_type": maintenance_type, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + # ========================================================================= + # Archive Events + # ========================================================================= + + async def on_archive_created( + self, + archive_id: int, + print_name: str, + printer_name: str, + status: str, + ): + """Publish print archived event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/archive/created", + { + "archive_id": archive_id, + "print_name": print_name, + "printer_name": printer_name, + "status": status, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_archive_updated( + self, + archive_id: int, + print_name: str, + status: str, + ): + """Publish archive record updated event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/archive/updated", + { + "archive_id": archive_id, + "print_name": print_name, + "status": status, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + # ========================================================================= + # Filament/Spoolman Events + # ========================================================================= + + async def on_filament_low( + self, + spool_id: int, + spool_name: str, + remaining_weight: float, + remaining_percent: float, + ): + """Publish filament inventory low event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/filament/low", + { + "spool_id": spool_id, + "spool_name": spool_name, + "remaining_weight": remaining_weight, + "remaining_percent": remaining_percent, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + # ========================================================================= + # Smart Plug Events + # ========================================================================= + + async def on_smart_plug_state( + self, + plug_id: int, + plug_name: str, + state: str, + printer_id: int | None = None, + printer_name: str | None = None, + ): + """Publish smart plug state change event.""" + if not self.enabled or not self.connected: + return + + topic = f"{self.topic_prefix}/smart_plugs/on" if state == "on" else f"{self.topic_prefix}/smart_plugs/off" + + self._publish( + topic, + { + "plug_id": plug_id, + "plug_name": plug_name, + "state": state, + "printer_id": printer_id, + "printer_name": printer_name, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + async def on_smart_plug_energy( + self, + plug_id: int, + plug_name: str, + power: float, + energy_today: float, + energy_total: float, + ): + """Publish smart plug energy update event.""" + if not self.enabled or not self.connected: + return + + self._publish( + f"{self.topic_prefix}/smart_plugs/energy", + { + "plug_id": plug_id, + "plug_name": plug_name, + "power_watts": power, + "energy_today_kwh": energy_today, + "energy_total_kwh": energy_total, + "timestamp": datetime.utcnow().isoformat(), + }, + ) + + +# Global instance +mqtt_relay = MQTTRelayService() diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py index bd6f5c128..f5d036268 100644 --- a/backend/app/services/print_scheduler.py +++ b/backend/app/services/print_scheduler.py @@ -337,6 +337,20 @@ class PrintScheduler: item.started_at = datetime.utcnow() await db.commit() logger.info(f"Queue item {item.id}: Print started - {archive.filename}") + + # 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=archive.filename, + printer_id=printer.id, + printer_name=printer.name, + printer_serial=printer.serial_number, + ) + except Exception: + pass # Don't fail if MQTT fails else: item.status = "failed" item.error_message = "Failed to send print command" diff --git a/backend/app/services/printer_manager.py b/backend/app/services/printer_manager.py index 0aa17ef13..60a9ae10d 100644 --- a/backend/app/services/printer_manager.py +++ b/backend/app/services/printer_manager.py @@ -37,18 +37,31 @@ def supports_chamber_temp(model: str | None) -> bool: return model_upper in CHAMBER_TEMP_SUPPORTED_MODELS +class PrinterInfo: + """Basic printer info for callbacks.""" + + def __init__(self, name: str, serial_number: str): + self.name = name + self.serial_number = serial_number + + class PrinterManager: """Manager for multiple printer connections.""" def __init__(self): self._clients: dict[int, BambuMQTTClient] = {} self._models: dict[int, str | None] = {} # Cache printer models for feature detection + self._printer_info: dict[int, PrinterInfo] = {} # Cache printer name/serial for callbacks self._on_print_start: Callable[[int, dict], None] | None = None self._on_print_complete: Callable[[int, dict], None] | None = None self._on_status_change: Callable[[int, PrinterState], None] | None = None self._on_ams_change: Callable[[int, list], None] | None = None self._loop: asyncio.AbstractEventLoop | None = None + def get_printer(self, printer_id: int) -> PrinterInfo | None: + """Get printer info by ID.""" + return self._printer_info.get(printer_id) + def set_event_loop(self, loop: asyncio.AbstractEventLoop): """Set the event loop for async callbacks.""" self._loop = loop @@ -125,6 +138,7 @@ class PrinterManager: client.connect() self._clients[printer_id] = client self._models[printer_id] = printer.model # Cache model for feature detection + self._printer_info[printer_id] = PrinterInfo(printer.name, printer.serial_number) # Wait a moment for connection await asyncio.sleep(1) @@ -136,6 +150,7 @@ class PrinterManager: self._clients[printer_id].disconnect() del self._clients[printer_id] self._models.pop(printer_id, None) # Clean up model cache + self._printer_info.pop(printer_id, None) # Clean up printer info cache def disconnect_all(self): """Disconnect from all printers.""" diff --git a/backend/tests/integration/test_settings_api.py b/backend/tests/integration/test_settings_api.py index 5e308ebff..9a9c09fbd 100644 --- a/backend/tests/integration/test_settings_api.py +++ b/backend/tests/integration/test_settings_api.py @@ -214,3 +214,74 @@ class TestSettingsAPI: result = response.json() assert result["currency"] == "JPY" assert result["check_updates"] is False + + # ======================================================================== + # MQTT settings tests + # ======================================================================== + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_update_mqtt_settings(self, async_client: AsyncClient): + """Verify MQTT settings can be updated.""" + response = await async_client.put( + "/api/v1/settings/", + json={ + "mqtt_enabled": True, + "mqtt_broker": "mqtt.example.com", + "mqtt_port": 8883, + "mqtt_username": "testuser", + "mqtt_password": "testpass", + "mqtt_topic_prefix": "myprefix", + "mqtt_use_tls": True, + }, + ) + + assert response.status_code == 200 + result = response.json() + assert result["mqtt_enabled"] is True + assert result["mqtt_broker"] == "mqtt.example.com" + assert result["mqtt_port"] == 8883 + assert result["mqtt_username"] == "testuser" + assert result["mqtt_password"] == "testpass" + assert result["mqtt_topic_prefix"] == "myprefix" + assert result["mqtt_use_tls"] is True + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_mqtt_status_endpoint(self, async_client: AsyncClient): + """Verify MQTT status endpoint returns expected fields.""" + response = await async_client.get("/api/v1/settings/mqtt/status") + + assert response.status_code == 200 + result = response.json() + assert "enabled" in result + assert "connected" in result + assert "broker" in result + assert "port" in result + assert "topic_prefix" in result + + @pytest.mark.asyncio + @pytest.mark.integration + async def test_mqtt_defaults(self, async_client: AsyncClient): + """Verify MQTT has correct default values.""" + # Reset MQTT settings to defaults + await async_client.put( + "/api/v1/settings/", + json={ + "mqtt_enabled": False, + "mqtt_broker": "", + "mqtt_port": 1883, + "mqtt_username": "", + "mqtt_password": "", + "mqtt_topic_prefix": "bambuddy", + "mqtt_use_tls": False, + }, + ) + + response = await async_client.get("/api/v1/settings/") + result = response.json() + + assert result["mqtt_enabled"] is False + assert result["mqtt_port"] == 1883 + assert result["mqtt_topic_prefix"] == "bambuddy" + assert result["mqtt_use_tls"] is False diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 742bdd5d8..a2bc7b7f6 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -569,10 +569,27 @@ export interface AppSettings { ftp_retry_count: number; ftp_retry_delay: number; ftp_timeout: number; + // MQTT relay settings + mqtt_enabled: boolean; + mqtt_broker: string; + mqtt_port: number; + mqtt_username: string; + mqtt_password: string; + mqtt_topic_prefix: string; + mqtt_use_tls: boolean; } export type AppSettingsUpdate = Partial; +// MQTT relay status +export interface MQTTStatus { + enabled: boolean; + connected: boolean; + broker: string; + port: number; + topic_prefix: string; +} + // Cloud types export interface CloudAuthStatus { is_authenticated: boolean; @@ -1783,6 +1800,7 @@ export const api = { method: 'PUT', body: JSON.stringify(data), }), + getMQTTStatus: () => request('/settings/mqtt/status'), resetSettings: () => request('/settings/reset', { method: 'POST' }), exportBackup: async (categories?: Record): Promise<{ blob: Blob; filename: string }> => { diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx index 6e128ed5a..0be43125e 100644 --- a/frontend/src/pages/SettingsPage.tsx +++ b/frontend/src/pages/SettingsPage.tsx @@ -1,5 +1,5 @@ import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; -import { Loader2, Plus, Plug, AlertTriangle, RotateCcw, Bell, Download, RefreshCw, ExternalLink, Globe, Droplets, Thermometer, FileText, Edit2, Send, CheckCircle, XCircle, History, Trash2, Upload, Zap, TrendingUp, Calendar, DollarSign, Power, PowerOff, Key, Copy, Database, Info, X, Shield, Printer, Cylinder } from 'lucide-react'; +import { Loader2, Plus, Plug, AlertTriangle, RotateCcw, Bell, Download, RefreshCw, ExternalLink, Globe, Droplets, Thermometer, FileText, Edit2, Send, CheckCircle, XCircle, History, Trash2, Upload, Zap, TrendingUp, Calendar, DollarSign, Power, PowerOff, Key, Copy, Database, Info, X, Shield, Printer, Cylinder, Wifi } from 'lucide-react'; import { useTranslation } from 'react-i18next'; import { api } from '../api/client'; import { formatDateOnly } from '../utils/date'; @@ -46,7 +46,7 @@ export function SettingsPage() { const [editingTemplate, setEditingTemplate] = useState(null); const [showLogViewer, setShowLogViewer] = useState(false); const [defaultView, setDefaultViewState] = useState(getDefaultView()); - const [activeTab, setActiveTab] = useState<'general' | 'plugs' | 'notifications' | 'filament' | 'apikeys' | 'virtual-printer'>('general'); + const [activeTab, setActiveTab] = useState<'general' | 'network' | 'plugs' | 'notifications' | 'filament' | 'apikeys' | 'virtual-printer'>('general'); const [showCreateAPIKey, setShowCreateAPIKey] = useState(false); const [newAPIKeyName, setNewAPIKeyName] = useState(''); const [newAPIKeyPermissions, setNewAPIKeyPermissions] = useState({ @@ -216,6 +216,13 @@ export function SettingsPage() { }, }); + // MQTT status for Network tab + const { data: mqttStatus } = useQuery({ + queryKey: ['mqtt-status'], + queryFn: api.getMQTTStatus, + refetchInterval: activeTab === 'network' ? 5000 : false, // Poll every 5s when on Network tab + }); + const applyUpdateMutation = useMutation({ mutationFn: api.applyUpdate, onSuccess: (data) => { @@ -349,7 +356,14 @@ export function SettingsPage() { settings.ftp_retry_enabled !== localSettings.ftp_retry_enabled || settings.ftp_retry_count !== localSettings.ftp_retry_count || settings.ftp_retry_delay !== localSettings.ftp_retry_delay || - settings.ftp_timeout !== localSettings.ftp_timeout; + settings.ftp_timeout !== localSettings.ftp_timeout || + settings.mqtt_enabled !== localSettings.mqtt_enabled || + settings.mqtt_broker !== localSettings.mqtt_broker || + settings.mqtt_port !== localSettings.mqtt_port || + settings.mqtt_username !== localSettings.mqtt_username || + settings.mqtt_password !== localSettings.mqtt_password || + settings.mqtt_topic_prefix !== localSettings.mqtt_topic_prefix || + settings.mqtt_use_tls !== localSettings.mqtt_use_tls; if (!hasChanges) { return; @@ -386,6 +400,13 @@ export function SettingsPage() { ftp_retry_count: localSettings.ftp_retry_count, ftp_retry_delay: localSettings.ftp_retry_delay, ftp_timeout: localSettings.ftp_timeout, + mqtt_enabled: localSettings.mqtt_enabled, + mqtt_broker: localSettings.mqtt_broker, + mqtt_port: localSettings.mqtt_port, + mqtt_username: localSettings.mqtt_username, + mqtt_password: localSettings.mqtt_password, + mqtt_topic_prefix: localSettings.mqtt_topic_prefix, + mqtt_use_tls: localSettings.mqtt_use_tls, }; updateMutation.mutate(settingsToSave); }, 500); @@ -472,6 +493,18 @@ export function SettingsPage() { Filament +