337 lines
13 KiB
Python
337 lines
13 KiB
Python
import asyncio
|
|
import json
|
|
import time
|
|
from typing import Awaitable, Callable, Dict, Optional
|
|
|
|
import aiomqtt
|
|
|
|
from banban.service.binding import BindingService
|
|
from banban.service.device_setting import device_setting_service
|
|
from banban.service.location import location_service
|
|
from config import settings
|
|
from database.models import ChildLocationCurrent
|
|
from services.card_service import card_service
|
|
from services.device_target_cache import device_target_cache
|
|
from services.offline_audio_cache import offline_audio_cache
|
|
from utils.logger import session_logger as logger
|
|
|
|
|
|
class TalkingQMQTTService:
|
|
_instance = None
|
|
_lock = asyncio.Lock()
|
|
|
|
def __init__(self, config: dict):
|
|
self.broker = config.get("broker", "broker.emqx.io")
|
|
self.port = config.get("port", 1883)
|
|
self.username = config.get("username")
|
|
self.password = config.get("password")
|
|
self.device_prefix = config.get("device_prefix", "TalkingQ")
|
|
self.qos = config.get("qos", 0)
|
|
self.keepalive = config.get("keepalive", 60)
|
|
self.nfc_notice_interval = config.get("nfc_notice_interval", 600)
|
|
|
|
self._client: Optional[aiomqtt.Client] = None
|
|
self._connected = False
|
|
self._connect_lock = asyncio.Lock()
|
|
self._message_task: Optional[asyncio.Task] = None
|
|
|
|
self._msg_handlers: Dict[str, Callable[[str, dict], Awaitable[None]]] = {
|
|
"000": self._handle_device_info,
|
|
"001": self._handle_gps_response,
|
|
"002": self._handle_volume_response,
|
|
"003": self._handle_ota_response,
|
|
"004": self._handle_nfc_notice_response,
|
|
"005": self._handle_nfc_listen_report,
|
|
"006": self._handle_bind_response,
|
|
"007": self._handle_open_response,
|
|
}
|
|
|
|
@classmethod
|
|
async def get_instance(cls, config: dict = None) -> Optional["TalkingQMQTTService"]:
|
|
async with cls._lock:
|
|
if cls._instance is None and config is not None:
|
|
cls._instance = cls(config)
|
|
return cls._instance
|
|
|
|
@classmethod
|
|
async def reset_instance(cls):
|
|
async with cls._lock:
|
|
if cls._instance is not None:
|
|
await cls._instance.disconnect()
|
|
cls._instance = None
|
|
|
|
async def _message_loop(self):
|
|
try:
|
|
async for message in self._client.messages:
|
|
try:
|
|
topic = str(message.topic)
|
|
parts = topic.split("/")
|
|
if not parts or parts[0] != "device":
|
|
continue
|
|
|
|
device_id = parts[1] if len(parts) >= 2 else "unknown"
|
|
if not device_id.startswith(f"{self.device_prefix}_"):
|
|
continue
|
|
|
|
payload = json.loads(message.payload.decode("utf-8"))
|
|
msg_id = payload.get("msg_id")
|
|
handler = self._msg_handlers.get(msg_id)
|
|
if handler:
|
|
await handler(device_id, payload)
|
|
else:
|
|
logger.warning(device_id, "", f"unknown msg_id={msg_id}")
|
|
except json.JSONDecodeError as exc:
|
|
logger.error("", "", f"invalid mqtt payload: {exc}")
|
|
except Exception as exc:
|
|
logger.error("", "", f"mqtt message handling failed: {exc}")
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except Exception as exc:
|
|
logger.error("", "", f"mqtt loop failed: {exc}")
|
|
self._connected = False
|
|
|
|
async def _handle_device_info(self, device_id: str, payload: dict):
|
|
data = payload.get("data", {})
|
|
await device_setting_service.insert_or_update(
|
|
device_id=device_id,
|
|
power=data.get("power"),
|
|
signal_strength=data.get("signal"),
|
|
version_str=data.get("version"),
|
|
volume=data.get("voice"),
|
|
)
|
|
await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"})
|
|
|
|
async def _handle_gps_response(self, device_id: str, payload: dict):
|
|
if payload.get("status") != "success":
|
|
logger.warning(device_id, "", f"[GPS] query failed: {payload}")
|
|
return
|
|
|
|
data = payload.get("data", {})
|
|
location = ChildLocationCurrent(
|
|
device_id=device_id,
|
|
lat=data.get("latitude"),
|
|
lon=data.get("longitude"),
|
|
)
|
|
await location_service.insert_or_update(device_id=device_id, location=location)
|
|
|
|
async def _handle_volume_response(self, device_id: str, payload: dict):
|
|
if payload.get("status") != "success":
|
|
logger.warning(device_id, "", f"[volume] command failed: {payload}")
|
|
return
|
|
|
|
data = payload.get("data", {})
|
|
await device_setting_service.insert_or_update(device_id=device_id, volume=data.get("current_level"))
|
|
|
|
async def _handle_ota_response(self, device_id: str, payload: dict):
|
|
status = payload.get("status")
|
|
data = payload.get("data", {})
|
|
if status == "success":
|
|
await device_setting_service.insert_or_update(device_id=device_id, version_str=data.get("new_version"))
|
|
elif status != "accepted":
|
|
logger.warning(device_id, "", f"[OTA] command failed: {payload}")
|
|
|
|
async def _handle_nfc_notice_response(self, device_id: str, payload: dict):
|
|
status = payload.get("status")
|
|
if status != "success":
|
|
logger.warning(device_id, "", f"[NFC notice] command failed: {payload}")
|
|
|
|
async def _handle_nfc_listen_report(self, device_id: str, payload: dict):
|
|
params = payload.get("params", {})
|
|
nfc_uuid = params.get("uuid")
|
|
logger.info(device_id, "", f"[NFC收听] 设备 {device_id} 请求收听留言, UUID={nfc_uuid}")
|
|
await self._send_nfc_listen_response(device_id, nfc_uuid)
|
|
|
|
async def _handle_bind_response(self, device_id: str, payload: dict):
|
|
params = payload.get("params", {})
|
|
data = payload.get("data", {})
|
|
status = payload.get("status") or params.get("status")
|
|
logger.info(device_id, "", f"[NFC bind] device={device_id} status={status}")
|
|
|
|
if status not in (None, "success", "accepted"):
|
|
logger.warning(device_id, "", f"[NFC bind] bind failed: {payload}")
|
|
return
|
|
|
|
nfc_uuid = data.get("uuid") or params.get("uuid")
|
|
if not nfc_uuid:
|
|
logger.warning(device_id, "", f"[NFC bind] missing uuid: {payload}")
|
|
return
|
|
service = BindingService()
|
|
result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid)
|
|
if result is None:
|
|
logger.warning(device_id, "", "[NFC bind] no pending bind session found")
|
|
return
|
|
|
|
logger.info(device_id, "", f"[NFC bind] bind completed, uuid={nfc_uuid}, result={result}")
|
|
|
|
async def _handle_open_response(self, device_id: str, payload: dict):
|
|
params = payload.get("params", {})
|
|
logger.info(device_id, "", f"[device open] response={params}")
|
|
|
|
async def _send_nfc_listen_response(self, device_id: str, nfc_uuid: str):
|
|
topic = f"device/{device_id}/event_resp"
|
|
card = await card_service.get_card_by_uuid(nfc_uuid)
|
|
if not card:
|
|
payload = {
|
|
"msg_id": "005",
|
|
"type": 0,
|
|
"params": {
|
|
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/error_card.mp3"
|
|
},
|
|
}
|
|
await self._publish(topic, payload)
|
|
return
|
|
|
|
is_owner = await card_service.check_card_ownership(nfc_uuid, device_id)
|
|
if is_owner:
|
|
has_pending = await offline_audio_cache.has_pending_audio(device_id)
|
|
if nfc_uuid == "53D92B6DA20001":
|
|
payload = {
|
|
"msg_id": "005",
|
|
"type": 0,
|
|
"params": {
|
|
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3"
|
|
},
|
|
}
|
|
await self._publish(topic, payload)
|
|
return
|
|
|
|
if has_pending:
|
|
audio_urls = await offline_audio_cache.get_audio_urls(device_id)
|
|
if len(audio_urls) == 0:
|
|
payload = {
|
|
"msg_id": "005",
|
|
"type": 0,
|
|
"params": {
|
|
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3"
|
|
},
|
|
}
|
|
else:
|
|
params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)}
|
|
payload = {"msg_id": "005", "type": 0, "params": params}
|
|
await offline_audio_cache.clear_audio_urls(device_id)
|
|
await self._publish(topic, payload)
|
|
return
|
|
|
|
payload = {
|
|
"msg_id": "005",
|
|
"type": 0,
|
|
"params": {
|
|
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3"
|
|
},
|
|
}
|
|
await self._publish(topic, payload)
|
|
return
|
|
|
|
existing_card = await card_service.get_card_by_uuid(nfc_uuid)
|
|
if existing_card:
|
|
target_device_id = existing_card.device_id
|
|
else:
|
|
await card_service.activate_card(nfc_uuid, device_id)
|
|
return
|
|
|
|
await device_target_cache.set_target(device_id, target_device_id)
|
|
payload = {
|
|
"msg_id": "005",
|
|
"type": 1,
|
|
"params": {
|
|
"url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3"
|
|
},
|
|
}
|
|
await self._publish(topic, payload)
|
|
|
|
async def connect(self):
|
|
async with self._connect_lock:
|
|
if self._connected:
|
|
return
|
|
try:
|
|
self._client = aiomqtt.Client(
|
|
hostname=self.broker,
|
|
port=self.port,
|
|
username=self.username,
|
|
password=self.password,
|
|
identifier=f"talkingq_server_{int(time.time())}",
|
|
keepalive=self.keepalive,
|
|
)
|
|
await self._client.__aenter__()
|
|
self._connected = True
|
|
|
|
await self._client.subscribe("device/+/response", qos=self.qos)
|
|
await self._client.subscribe("device/+/event", qos=self.qos)
|
|
|
|
self._message_task = asyncio.create_task(self._message_loop())
|
|
logger.info("", "", f"mqtt connected: {self.broker}:{self.port}")
|
|
except Exception as exc:
|
|
self._connected = False
|
|
logger.error("", "", f"mqtt connect failed: {exc}")
|
|
|
|
async def disconnect(self):
|
|
if self._message_task and not self._message_task.done():
|
|
self._message_task.cancel()
|
|
try:
|
|
await self._message_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._message_task = None
|
|
|
|
if self._client is not None:
|
|
try:
|
|
await self._client.__aexit__(None, None, None)
|
|
except Exception:
|
|
pass
|
|
self._client = None
|
|
self._connected = False
|
|
|
|
async def _ensure_connected(self):
|
|
if not self._connected:
|
|
await self.connect()
|
|
|
|
async def _publish(self, topic: str, payload: dict):
|
|
await self._ensure_connected()
|
|
try:
|
|
await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos)
|
|
logger.info("", "", f"mqtt publish topic={topic} payload={payload}")
|
|
except Exception as exc:
|
|
logger.error("", "", f"mqtt publish failed: {exc}")
|
|
|
|
async def send_gps_query(self, device_id: str) -> str:
|
|
await self._publish(f"device/{device_id}/command", {"msg_id": "001"})
|
|
return "001"
|
|
|
|
async def send_volume_command(self, device_id: str, level: int) -> str:
|
|
await self._publish(
|
|
f"device/{device_id}/command",
|
|
{"msg_id": "002", "params": {"level": level}},
|
|
)
|
|
return "002"
|
|
|
|
async def send_ota_command(self, device_id: str, url: str, version: str) -> str:
|
|
await self._publish(
|
|
f"device/{device_id}/command",
|
|
{"msg_id": "003", "params": {"url": url, "version": version}},
|
|
)
|
|
return "003"
|
|
|
|
async def send_nfc_notice(self, device_id: str, url: str) -> str:
|
|
await self._publish(
|
|
f"device/{device_id}/command",
|
|
{"msg_id": "004", "params": {"url_1": url}},
|
|
)
|
|
return "004"
|
|
|
|
async def send_bind_nfc_command(self, device_id: str, uuid: Optional[str] = None) -> str:
|
|
del uuid
|
|
await self._publish(f"device/{device_id}/command", {"msg_id": "006"})
|
|
return "006"
|
|
|
|
async def send_open_command(self, device_id: str, open_type: int, is_open: int) -> str:
|
|
if open_type not in [0, 1]:
|
|
raise ValueError("open_type must be 0 or 1")
|
|
|
|
if open_type == 0:
|
|
payload = {"msg_id": "007", "type": open_type}
|
|
else:
|
|
payload = {"msg_id": "007", "type": open_type, "status": is_open}
|
|
|
|
await self._publish(f"device/{device_id}/command", payload)
|
|
return "007"
|