按设备回执确认音量调整完成

talkingq-url/handlers/mqtt_handler.py

@@ -26,6 +26,10 @@ from services.offline_audio_cache import offline_audio_cache
 from services.task_manager import task_manager
 from utils.logger import session_logger as logger
 
+VOLUME_COMMAND_TIMEOUT_MS = 10 * 1000
+VOLUME_COMMAND_RETENTION_MS = 10 * 60 * 1000
+
+
 class TalkingQMQTTService:
     _instance = None
     _lock = asyncio.Lock()
@@ -44,6 +48,9 @@ class TalkingQMQTTService:
         self._connected = False
         self._connect_lock = asyncio.Lock()
         self._message_task: Optional[asyncio.Task] = None
+        self._volume_command_lock = asyncio.Lock()
+        self._volume_command_states: Dict[tuple[str, int], dict] = {}
+        self._last_volume_command_time_ms = 0
 
         self._msg_handlers: Dict[str, Callable[[str, dict], Awaitable[None]]] = {
             "000": self._handle_device_info,
@@ -201,15 +208,38 @@ class TalkingQMQTTService:
         )
 
     async def _handle_volume_response(self, device_id: str, payload: dict):
+        command_time = self._parse_command_time(payload.get("time"))
         if payload.get("status") != "success":
+            await self._mark_volume_command_response(
+                device_id=device_id,
+                command_time=command_time,
+                status="failed",
+                response_status=payload.get("status"),
+                error=str(payload),
+            )
             logger.warning(device_id, "", f"[volume] command failed: {payload}")
             return
 
         data = payload.get("data", {})
         current_level = data.get("current_level")
         if current_level is None:
+            await self._mark_volume_command_response(
+                device_id=device_id,
+                command_time=command_time,
+                status="failed",
+                response_status=payload.get("status"),
+                error="missing current_level",
+            )
             return
         logger.info(device_id, "", f"[音量] 设备 {device_id} 当前音量: {current_level}")
+        await self._mark_volume_command_response(
+            device_id=device_id,
+            command_time=command_time,
+            status="completed",
+            current_level=current_level,
+            response_status=payload.get("status"),
+            error=None,
+        )
             
         await self._schedule_persistence(
             device_id,
@@ -617,12 +647,113 @@ class TalkingQMQTTService:
         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(
+    def _parse_command_time(self, value) -> int | None:
+        if value is None:
+            return None
+        try:
+            command_time = int(value)
+        except (TypeError, ValueError):
+            return None
+        return command_time if command_time > 0 else None
+
+    def _current_utc_timestamp_ms(self) -> int:
+        return int(time.time() * 1000)
+
+    async def _next_volume_command_time(self) -> int:
+        async with self._volume_command_lock:
+            command_time = max(self._current_utc_timestamp_ms(), self._last_volume_command_time_ms + 1)
+            self._last_volume_command_time_ms = command_time
+            return command_time
+
+    async def _record_volume_command(self, *, device_id: str, level: int, command_time: int) -> dict:
+        state = {
+            "device_id": device_id,
+            "msg_id": "002",
+            "level": level,
+            "time": command_time,
+            "status": "sent",
+            "current_level": None,
+            "response_status": None,
+            "error": None,
+            "created_at": command_time,
+            "updated_at": command_time,
+        }
+        async with self._volume_command_lock:
+            self._volume_command_states[(device_id, command_time)] = dict(state)
+            self._cleanup_volume_command_states_locked()
+        return state
+
+    def _cleanup_volume_command_states_locked(self):
+        cutoff = self._current_utc_timestamp_ms() - VOLUME_COMMAND_RETENTION_MS
+        expired_keys = [
+            key
+            for key, state in self._volume_command_states.items()
+            if int(state.get("updated_at") or state.get("created_at") or 0) < cutoff
+        ]
+        for key in expired_keys:
+            self._volume_command_states.pop(key, None)
+
+    async def _mark_volume_command_response(
+        self,
+        *,
+        device_id: str,
+        command_time: int | None,
+        status: str,
+        current_level=None,
+        response_status=None,
+        error: str | None,
+    ) -> None:
+        if command_time is None:
+            return
+        async with self._volume_command_lock:
+            state = self._volume_command_states.get((device_id, command_time))
+            if not state:
+                return
+            state.update(
+                {
+                    "status": status,
+                    "current_level": current_level,
+                    "response_status": response_status,
+                    "error": error,
+                    "updated_at": self._current_utc_timestamp_ms(),
+                }
+            )
+
+    async def get_volume_command_status(self, device_id: str, command_time: int) -> dict | None:
+        async with self._volume_command_lock:
+            self._cleanup_volume_command_states_locked()
+            state = self._volume_command_states.get((device_id, command_time))
+            if state and state.get("status") == "sent":
+                now = self._current_utc_timestamp_ms()
+                if now - int(state.get("created_at") or command_time) > VOLUME_COMMAND_TIMEOUT_MS:
+                    state.update(
+                        {
+                            "status": "timeout",
+                            "error": "device response timeout",
+                            "updated_at": now,
+                        }
+                    )
+            return dict(state) if state else None
+
+    async def send_volume_command(self, device_id: str, level: int) -> dict:
+        command_time = await self._next_volume_command_time()
+        state = await self._record_volume_command(device_id=device_id, level=level, command_time=command_time)
+        published = await self._publish(
             f"device/{device_id}/command",
-            {"msg_id": "002", "params": {"level": level}},
+            {"msg_id": "002", "params": {"level": level}, "time": command_time},
         )
-        return "002"
+        if not published:
+            await self._mark_volume_command_response(
+                device_id=device_id,
+                command_time=command_time,
+                status="failed",
+                response_status=None,
+                error="publish failed",
+            )
+            state["status"] = "failed"
+            state["error"] = "publish failed"
+            state["updated_at"] = self._current_utc_timestamp_ms()
+        return state
 
     async def send_ota_command(self, device_id: str, url: str, version: str) -> str:
         await self._publish(