From a6882e8371a207a89b028fa8eb6a1564daee4885 Mon Sep 17 00:00:00 2001 From: stu2not Date: Mon, 25 May 2026 08:39:41 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=E8=AE=BE=E5=A4=87=E6=9C=9F?= =?UTF-8?q?=E6=9C=9B=E9=9F=B3=E9=87=8F=E5=B9=B6=E8=A1=A5=E5=8F=91=E9=9F=B3?= =?UTF-8?q?=E9=87=8F=E6=8C=87=E4=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- talkingq-url/banban/routers/mqtt_router.py | 2 + talkingq-url/banban/service/device.py | 2 + talkingq-url/handlers/mqtt_handler.py | 26 +++++- .../services/device_volume_manager.py | 23 ++++- talkingq-url/tests/test_device_volume_sync.py | 89 +++++++++++++++++++ 5 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 talkingq-url/tests/test_device_volume_sync.py diff --git a/talkingq-url/banban/routers/mqtt_router.py b/talkingq-url/banban/routers/mqtt_router.py index 6dd134c..7000fb6 100644 --- a/talkingq-url/banban/routers/mqtt_router.py +++ b/talkingq-url/banban/routers/mqtt_router.py @@ -3,6 +3,7 @@ from typing import Optional from fastapi import APIRouter, HTTPException, Query from banban.service.device_audio_cache import device_audio_cache_service from banban.service.pending_voice_message import pending_voice_message_service +from services.device_volume_manager import device_volume_manager from services.offline_audio_cache import offline_audio_cache from banban.schemas.mqtt_models import ( GPSQueryRequest, @@ -52,6 +53,7 @@ async def set_volume( req: VolumeRequest, # current_user_id: int = Depends(get_current_user_id) ): + await device_volume_manager.set_volume(req.device_id, req.level) service = await _get_service() msg_id = await service.send_volume_command(req.device_id, req.level) return CommandResponse(msg_id=msg_id, device_id=req.device_id) diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 0cfe1e0..89453dc 100644 --- a/talkingq-url/banban/service/device.py +++ b/talkingq-url/banban/service/device.py @@ -7,6 +7,7 @@ from fastapi import HTTPException from banban.dao.device import DeviceDAO from banban.service.device_setting import device_setting_service +from services.device_volume_manager import device_volume_manager from services.device_update_manager import device_firmware_update_manager from services.system_config_manager import system_config_manager @@ -84,6 +85,7 @@ class DeviceService(DatabaseServiceBase): level: int, ) -> str: await self.ensure_device_access(device_id=device_id, user_id=user_id) + await device_volume_manager.set_volume(device_id, level) from handlers.mqtt_handler import TalkingQMQTTService diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 9bc6672..87fe80f 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -20,6 +20,7 @@ from services.device_identity_initializer import ( device_identity_initializer, ) from services.device_update_manager import device_firmware_update_manager +from services.device_volume_manager import device_volume_manager from services.offline_audio_cache import offline_audio_cache from services.task_manager import task_manager from utils.logger import session_logger as logger @@ -120,8 +121,29 @@ class TalkingQMQTTService: task_type="persistence", ) + async def _sync_desired_volume(self, device_id: str, current_level) -> None: + if current_level is None: + return + try: + current_level_int = int(current_level) + except (TypeError, ValueError): + logger.warning(device_id, "volume", f"invalid current volume level: {current_level}") + return + + desired_level = await device_volume_manager.get_configured_volume(device_id) + if desired_level is None or desired_level == current_level_int: + return + + logger.info( + device_id, + "volume", + f"当前音量 {current_level_int} 与期望音量 {desired_level} 不一致，补发音量指令", + ) + await self.send_volume_command(device_id, desired_level) + async def _handle_device_info(self, device_id: str, payload: dict): data = payload.get("data", {}) + current_volume = data.get("voice") await self._schedule_persistence( device_id, "device_info", @@ -130,9 +152,10 @@ class TalkingQMQTTService: power=data.get("power"), signal_strength=data.get("signal"), version_str=data.get("version"), - volume=data.get("voice"), + volume=current_volume, ), ) + await self._sync_desired_volume(device_id, current_volume) 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): @@ -187,6 +210,7 @@ class TalkingQMQTTService: version_str=None, ), ) + await self._sync_desired_volume(device_id, current_level) async def _handle_ota_response(self, device_id: str, payload: dict): status = payload.get("status") diff --git a/talkingq-url/services/device_volume_manager.py b/talkingq-url/services/device_volume_manager.py index be2229b..47c80e8 100644 --- a/talkingq-url/services/device_volume_manager.py +++ b/talkingq-url/services/device_volume_manager.py @@ -39,6 +39,27 @@ class DeviceVolumeManager(DatabaseServiceBase): return self.default_volume # 出错时返回默认值 finally: await db_session.close() + + async def get_configured_volume(self, device_id: str) -> Optional[int]: + """获取已持久化的期望音量；未配置时返回 None。""" + await self._init_database() + + db_session = await self.get_session() + try: + query = select(DeviceConfig.volume).where(DeviceConfig.device_id == device_id) + result = await db_session.execute(query) + volume = result.scalar_one_or_none() + if volume is None: + return None + + normalized_volume = int(volume) + self.device_volumes[device_id] = normalized_volume + return normalized_volume + except Exception as e: + session_logger.error(device_id, "volume", f"获取期望音量失败: {str(e)}") + raise + finally: + await db_session.close() async def set_volume(self, device_id: str, volume: int) -> bool: """设置设备音量并保存到数据库""" @@ -59,7 +80,7 @@ class DeviceVolumeManager(DatabaseServiceBase): update_stmt = ( update(DeviceConfig) .where(DeviceConfig.device_id == device_id) - .values(volume=volume) + .values(volume=volume, last_update_time=time.time()) ) await db_session.execute(update_stmt) else: diff --git a/talkingq-url/tests/test_device_volume_sync.py b/talkingq-url/tests/test_device_volume_sync.py new file mode 100644 index 0000000..44b5ef2 --- /dev/null +++ b/talkingq-url/tests/test_device_volume_sync.py @@ -0,0 +1,89 @@ +import pytest + +from handlers.mqtt_handler import TalkingQMQTTService + + +@pytest.mark.asyncio +async def test_sync_desired_volume_republishes_when_reported_volume_differs(monkeypatch): + service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) + sent_commands = [] + + class FakeDeviceVolumeManager: + async def get_configured_volume(self, device_id): + assert device_id == "TalkingQ_XQSN00001001" + return 70 + + async def fake_send_volume_command(device_id, level): + sent_commands.append((device_id, level)) + return "002" + + monkeypatch.setattr("handlers.mqtt_handler.device_volume_manager", FakeDeviceVolumeManager()) + monkeypatch.setattr(service, "send_volume_command", fake_send_volume_command) + + await service._sync_desired_volume("TalkingQ_XQSN00001001", 30) + + assert sent_commands == [("TalkingQ_XQSN00001001", 70)] + + +@pytest.mark.asyncio +async def test_sync_desired_volume_skips_when_reported_volume_matches(monkeypatch): + service = TalkingQMQTTService({"device_prefix": "TalkingQ"}) + sent_commands = [] + + class FakeDeviceVolumeManager: + async def get_configured_volume(self, device_id): + return 70 + + async def fake_send_volume_command(device_id, level): + sent_commands.append((device_id, level)) + return "002" + + monkeypatch.setattr("handlers.mqtt_handler.device_volume_manager", FakeDeviceVolumeManager()) + monkeypatch.setattr(service, "send_volume_command", fake_send_volume_command) + + await service._sync_desired_volume("TalkingQ_XQSN00001001", "70") + + assert sent_commands == [] + + +@pytest.mark.asyncio +async def test_device_service_persists_desired_volume_before_mqtt(monkeypatch): + from banban.service.device import DeviceService + + service = DeviceService() + calls = [] + + async def fake_ensure_device_access(*, device_id, user_id): + calls.append(("access", device_id, user_id)) + + class FakeDeviceVolumeManager: + async def set_volume(self, device_id, level): + calls.append(("persist", device_id, level)) + return True + + class FakeMQTTService: + async def send_volume_command(self, device_id, level): + calls.append(("mqtt", device_id, level)) + return "002" + + class FakeTalkingQMQTTService: + @classmethod + async def get_instance(cls): + return FakeMQTTService() + + monkeypatch.setattr(service, "ensure_device_access", fake_ensure_device_access) + monkeypatch.setattr("banban.service.device.device_volume_manager", FakeDeviceVolumeManager()) + monkeypatch.setattr("handlers.mqtt_handler.TalkingQMQTTService", FakeTalkingQMQTTService) + + msg_id = await service.set_device_volume( + device_id="TalkingQ_XQSN00001001", + user_id=1, + level=70, + ) + + assert msg_id == "002" + assert calls == [ + ("access", "TalkingQ_XQSN00001001", 1), + ("persist", "TalkingQ_XQSN00001001", 70), + ("mqtt", "TalkingQ_XQSN00001001", 70), + ]