import json
|
import logging
|
import time
|
import uuid
|
from typing import Any
|
|
import paho.mqtt.client as mqtt
|
|
from app.config import settings
|
from app.services.device_store import DeviceStore
|
|
logger = logging.getLogger(__name__)
|
|
TOPIC_RPT_STATE = "snlp/v1/+/rpt/state"
|
TOPIC_RPT_EVENT = "snlp/v1/+/rpt/event"
|
TOPIC_RPT_ACK = "snlp/v1/+/rpt/ack"
|
TOPIC_RPT_PROGRESS = "snlp/v1/+/rpt/progress"
|
TOPIC_RPT_LOG = "snlp/v1/+/rpt/log"
|
|
|
class MqttService:
|
def __init__(self, device_store: DeviceStore, auto_init: object | None = None) -> None:
|
self.device_store = device_store
|
self.auto_init = auto_init
|
self._client: mqtt.Client | None = None
|
self._connected = False
|
|
def start(self) -> None:
|
if not settings.mqtt_enabled:
|
logger.info("MQTT disabled by configuration")
|
return
|
|
self._client = mqtt.Client(
|
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
|
client_id=settings.mqtt_client_id,
|
)
|
if settings.mqtt_username:
|
self._client.username_pw_set(settings.mqtt_username, settings.mqtt_password)
|
|
self._client.on_connect = self._on_connect
|
self._client.on_message = self._on_message
|
self._client.on_disconnect = self._on_disconnect
|
|
try:
|
self._client.connect(settings.mqtt_broker, settings.mqtt_port, keepalive=60)
|
self._client.loop_start()
|
logger.info("MQTT client connecting to %s:%s", settings.mqtt_broker, settings.mqtt_port)
|
except Exception:
|
logger.exception("Failed to connect MQTT broker; running in API-only mode")
|
|
def stop(self) -> None:
|
if self._client:
|
self._client.loop_stop()
|
self._client.disconnect()
|
|
def _on_connect(self, client: mqtt.Client, userdata: Any, flags: Any, reason_code: Any, properties: Any = None) -> None:
|
if reason_code == 0 or str(reason_code) == "Success":
|
self._connected = True
|
for topic in (TOPIC_RPT_STATE, TOPIC_RPT_EVENT, TOPIC_RPT_ACK, TOPIC_RPT_PROGRESS, TOPIC_RPT_LOG):
|
client.subscribe(topic, qos=1 if "ack" in topic else 0)
|
logger.info("MQTT connected and subscribed")
|
else:
|
logger.error("MQTT connect failed: %s", reason_code)
|
|
def _on_disconnect(self, client: mqtt.Client, userdata: Any, flags: Any, reason_code: Any, properties: Any = None) -> None:
|
self._connected = False
|
logger.warning("MQTT disconnected: %s", reason_code)
|
|
def _on_message(self, client: mqtt.Client, userdata: Any, msg: mqtt.MQTTMessage) -> None:
|
try:
|
payload = json.loads(msg.payload.decode("utf-8"))
|
parts = msg.topic.split("/")
|
device_id = parts[2] if len(parts) > 2 else payload.get("header", {}).get("device_id", "unknown")
|
msg_type = parts[-1] if len(parts) > 4 else payload.get("msg_type", "unknown")
|
header = payload.get("header", {})
|
timestamp = int(header.get("timestamp", time.time()))
|
body = payload.get("payload", {})
|
|
if msg_type == "state":
|
self.device_store.update_state(device_id, body, timestamp)
|
if self.auto_init:
|
self.auto_init.on_device_state(device_id, body, timestamp)
|
elif msg_type == "ack":
|
self.device_store.record_ack(device_id, body)
|
elif msg_type in ("event", "progress"):
|
self.device_store.record_event(device_id, {**body, "_type": msg_type})
|
elif msg_type == "log":
|
logger.info("Device %s log: %s", device_id, body)
|
except Exception:
|
logger.exception("Failed to handle MQTT message on %s", msg.topic)
|
|
def publish_ctl(self, device_id: str, action: str, payload: dict[str, Any], ttl: int = 60) -> dict[str, Any]:
|
command = self._build_command(device_id, action, payload, ttl)
|
topic = f"snlp/v1/{device_id}/cmd/ctl"
|
self._publish(topic, command, qos=1)
|
return command
|
|
def publish_task(self, device_id: str, action: str, payload: dict[str, Any], ttl: int = 3600) -> dict[str, Any]:
|
command = self._build_command(device_id, action, payload, ttl)
|
topic = f"snlp/v1/{device_id}/cmd/task"
|
self._publish(topic, command, qos=1)
|
return command
|
|
def _build_command(self, device_id: str, action: str, payload: dict[str, Any], ttl: int) -> dict[str, Any]:
|
return {
|
"header": {
|
"version": "1.0",
|
"device_id": device_id,
|
"req_id": str(uuid.uuid4()),
|
"timestamp": int(time.time()),
|
"ttl": ttl,
|
},
|
"action": action,
|
"payload": payload,
|
}
|
|
def _publish(self, topic: str, payload: dict[str, Any], qos: int = 1) -> None:
|
if not self._client or not self._connected:
|
logger.warning("MQTT not connected; command queued locally only: %s", topic)
|
return
|
self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=qos)
|
|
@property
|
def connected(self) -> bool:
|
return self._connected
|
