diff --git a/banban-mini/src/pages/device/index.tsx b/banban-mini/src/pages/device/index.tsx index 60ab715..f69c48e 100644 --- a/banban-mini/src/pages/device/index.tsx +++ b/banban-mini/src/pages/device/index.tsx @@ -1,11 +1,11 @@ import { View, Text, Slider, Image } from '@tarojs/components' -import { useState } from 'react' +import { useRef, useState } from 'react' import Taro, { useDidShow } from '@tarojs/taro' import { getToken } from '@/services/auth' import { DEVICE_UNAVAILABLE_MESSAGE } from '@/services/api' import { Binding, loadCurrentChildBindingContext } from '@/services/binding' import { Child } from '@/services/child' -import { DeviceAlarmItem, DeviceStatus, getDeviceAlarms, getDeviceStatus, setDeviceRemoteSleepWake, setDeviceVolume } from '@/services/device' +import { DeviceAlarmItem, DeviceStatus, getDeviceAlarms, getDeviceStatus, getDeviceVolumeCommandStatus, setDeviceRemoteSleepWake, setDeviceVolume } from '@/services/device' import { useSystemBanner } from '@/components/system-banner/use-system-banner' import './index.scss' @@ -41,6 +41,12 @@ function getSignalLabel(value?: number | null): string { return '弱' } +function sleep(ms: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, ms) + }) +} + function formatBatteryDisplay(value?: number | null): { text: string; showPercent: boolean } { if (value === null || value === undefined) return { text: '--', showPercent: false } const normalized = Number(value) @@ -123,6 +129,7 @@ export default function Device() { const [isSavingVolume, setIsSavingVolume] = useState(false) const [sleepWakePending, setSleepWakePending] = useState<'on' | 'off' | null>(null) const [volumeValue, setVolumeValue] = useState(0) + const volumeCommandInFlightRef = useRef(false) useDidShow(() => { void loadDeviceInfo() @@ -173,21 +180,42 @@ export default function Device() { } const handleVolumeChanging = (e: any) => { + if (volumeCommandInFlightRef.current) return setVolumeValue(Number(e.detail?.value || 0)) } const handleVolumeCommit = async (e: any) => { if (!binding?.device_id) return + if (volumeCommandInFlightRef.current) return const nextLevel = Number(e.detail?.value || 0) const fallbackLevel = deviceStatus?.volume ?? volumeValue ?? 0 + volumeCommandInFlightRef.current = true setVolumeValue(nextLevel) setIsSavingVolume(true) try { - await setDeviceVolume(nextLevel, binding.device_id) - setDeviceStatus((current) => (current ? { ...current, volume: nextLevel } : current)) + const command = await setDeviceVolume(nextLevel, binding.device_id) + let completedLevel: number | null = null + let failedMessage = '' + for (let attempt = 0; attempt < 13; attempt += 1) { + await sleep(800) + const status = await getDeviceVolumeCommandStatus(binding.device_id, command.time) + if (status.status === 'completed') { + completedLevel = status.current_level ?? nextLevel + break + } + if (status.status === 'failed' || status.status === 'timeout') { + failedMessage = status.error || (status.status === 'timeout' ? '设备未确认音量调整,请稍后查看设备状态' : '设备音量调整失败') + break + } + } + if (completedLevel === null) { + throw new Error(failedMessage || '设备未确认音量调整,请稍后查看设备状态') + } + setVolumeValue(completedLevel) + setDeviceStatus((current) => (current ? { ...current, volume: completedLevel } : current)) Taro.showToast({ - title: '音量指令已发送', + title: '音量已成功调整', icon: 'success', }) } catch (error: any) { @@ -200,6 +228,7 @@ export default function Device() { icon: 'none', }) } finally { + volumeCommandInFlightRef.current = false setIsSavingVolume(false) } } @@ -402,7 +431,7 @@ export default function Device() { 外放音量 - {isSavingVolume ? '发送中...' : `当前 ${volumeLabel}`} + {isSavingVolume ? '调整中...' : `当前 ${volumeLabel}`} @@ -417,6 +446,7 @@ export default function Device() { min={0} max={100} value={volumeValue} + disabled={isSavingVolume} onChanging={handleVolumeChanging} onChange={handleVolumeCommit} activeColor='#FF8C42' diff --git a/banban-mini/src/services/device.ts b/banban-mini/src/services/device.ts index 327237b..d429eee 100644 --- a/banban-mini/src/services/device.ts +++ b/banban-mini/src/services/device.ts @@ -65,6 +65,19 @@ export interface DeviceVolumeUpdateResponse { device_id: string level: number msg_id: string + time: number + status: string +} + +export interface DeviceVolumeCommandStatusResponse { + device_id: string + level: number + msg_id: string + time: number + status: 'sent' | 'completed' | 'failed' | string + current_level?: number | null + response_status?: string | null + error?: string | null } export interface DeviceSleepScheduleUpdateResponse { @@ -179,6 +192,13 @@ export async function setDeviceVolume(level: number, deviceId?: string): Promise }) } +export async function getDeviceVolumeCommandStatus( + deviceId: string, + commandTime: number, +): Promise { + return request(`/banban/devices/${deviceId}/volume/${commandTime}`) +} + export async function setDeviceSleepSchedule( start: string, end: string, diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index f7752b6..d7ca96e 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -138,6 +138,19 @@ class DeviceVolumeUpdateResponse(BaseModel): device_id: str level: int msg_id: str + time: int + status: str + + +class DeviceVolumeCommandStatusResponse(BaseModel): + device_id: str + level: int + msg_id: str + time: int + status: str + current_level: int | None = None + response_status: str | None = None + error: str | None = None class DeviceSleepScheduleUpdateRequest(BaseModel): @@ -549,7 +562,7 @@ async def set_device_volume( request: Request, current_user_id: int = Depends(get_current_user_id), ) -> DeviceVolumeUpdateResponse: - msg_id = await device_service.set_device_volume( + command_state = await device_service.set_device_volume( device_id=device_id, user_id=current_user_id, level=payload.level, @@ -563,10 +576,34 @@ async def set_device_volume( "user_id": current_user_id, "device_id": device_id, "level": payload.level, - "msg_id": msg_id, + "msg_id": command_state["msg_id"], + "time": command_state["time"], + "status": command_state["status"], }, ) - return DeviceVolumeUpdateResponse(device_id=device_id, level=payload.level, msg_id=msg_id) + return DeviceVolumeUpdateResponse( + device_id=device_id, + level=payload.level, + msg_id=command_state["msg_id"], + time=command_state["time"], + status=command_state["status"], + ) + + +@router.get("/{device_id}/volume/{command_time}", response_model=DeviceVolumeCommandStatusResponse) +async def get_device_volume_command_status( + device_id: str, + command_time: int, + current_user_id: int = Depends(get_current_user_id), +) -> DeviceVolumeCommandStatusResponse: + command_state = await device_service.get_device_volume_command_status( + device_id=device_id, + user_id=current_user_id, + command_time=command_time, + ) + if not command_state: + raise HTTPException(status_code=404, detail="volume command not found") + return DeviceVolumeCommandStatusResponse(**command_state) @router.post("/{device_id}/sleep-schedule", response_model=DeviceSleepScheduleResponse) diff --git a/talkingq-url/banban/routers/mqtt_router.py b/talkingq-url/banban/routers/mqtt_router.py index 7000fb6..8597396 100644 --- a/talkingq-url/banban/routers/mqtt_router.py +++ b/talkingq-url/banban/routers/mqtt_router.py @@ -55,8 +55,12 @@ async def set_volume( ): 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) + command_state = await service.send_volume_command(req.device_id, req.level) + return CommandResponse( + msg_id=command_state["msg_id"], + device_id=req.device_id, + time=command_state["time"], + ) @router.post("/ota", response_model=CommandResponse, summary="OTA升级") diff --git a/talkingq-url/banban/schemas/mqtt_models.py b/talkingq-url/banban/schemas/mqtt_models.py index 24b11d7..6e2eb64 100644 --- a/talkingq-url/banban/schemas/mqtt_models.py +++ b/talkingq-url/banban/schemas/mqtt_models.py @@ -59,6 +59,7 @@ class CommandResponse(BaseModel): message: str = "success" msg_id: str = Field(..., description="消息ID") device_id: str = Field(..., description="设备ID") + time: Optional[int] = Field(None, description="命令时间戳") class DeviceResponseData(BaseModel): diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 4efda27..43b00b2 100644 --- a/talkingq-url/banban/service/device.py +++ b/talkingq-url/banban/service/device.py @@ -185,7 +185,7 @@ class DeviceService(DatabaseServiceBase): device_id: str, user_id: int, level: int, - ) -> str: + ) -> Mapping[str, Any]: await self.ensure_device_access(device_id=device_id, user_id=user_id) from handlers.mqtt_handler import TalkingQMQTTService @@ -195,6 +195,22 @@ class DeviceService(DatabaseServiceBase): raise HTTPException(status_code=503, detail="MQTT 服务未初始化") return await service.send_volume_command(device_id, level) + async def get_device_volume_command_status( + self, + *, + device_id: str, + user_id: int, + command_time: int, + ) -> Mapping[str, Any] | None: + await self.ensure_device_access(device_id=device_id, user_id=user_id) + + from handlers.mqtt_handler import TalkingQMQTTService + + service = await TalkingQMQTTService.get_instance() + if service is None: + raise HTTPException(status_code=503, detail="MQTT 服务未初始化") + return await service.get_volume_command_status(device_id, command_time) + async def set_sleep_schedule( self, *, diff --git a/talkingq-url/docs/mqtt协议_V1.3.txt b/talkingq-url/docs/mqtt协议_V1.3.txt index 279c450..de15c69 100644 --- a/talkingq-url/docs/mqtt协议_V1.3.txt +++ b/talkingq-url/docs/mqtt协议_V1.3.txt @@ -46,21 +46,23 @@ device/TalkingQ_xxx/response 2. 音量调节 device/TalkingQ_xxx/command 服务器下发: -{ - "msg_id": "002", - "params": { - "level": 70 - } -} -device/TalkingQ_xxx/response -4G模块响应: -{ - "msg_id": "002", - "status": "success", - "data": { - "current_level": 70 - } -} +{ + "msg_id": "002", + "params": { + "level": 70 + }, + "time": 1717560000000 +} +device/TalkingQ_xxx/response +4G模块响应: +{ + "msg_id": "002", + "status": "success", + "data": { + "current_level": 70 + }, + "time": 1717560000000 +} 3. OTA升级 device/TalkingQ_xxx/command diff --git a/talkingq-url/docs/mqtt协议_V1.6.txt b/talkingq-url/docs/mqtt协议_V1.6.txt index 560618f..49dee52 100644 --- a/talkingq-url/docs/mqtt协议_V1.6.txt +++ b/talkingq-url/docs/mqtt协议_V1.6.txt @@ -46,21 +46,23 @@ device/TalkingQ_xxx/response 2. 音量调节 device/TalkingQ_xxx/command 服务器下发: -{ - "msg_id": "002", - "params": { - "level": 70 - } -} -device/TalkingQ_xxx/response -4G模块响应: -{ - "msg_id": "002", - "status": "success", - "data": { - "current_level": 70 - } -} +{ + "msg_id": "002", + "params": { + "level": 70 + }, + "time": 1717560000000 +} +device/TalkingQ_xxx/response +4G模块响应: +{ + "msg_id": "002", + "status": "success", + "data": { + "current_level": 70 + }, + "time": 1717560000000 +} 3. OTA升级 device/TalkingQ_xxx/command @@ -288,4 +290,4 @@ emqx_DPWQyk { "device_id": "TalkingQ_XQSN00001005", "url": "http://175.24.73.253:80/assets/audio/test_zh.mp3" -} \ No newline at end of file +} diff --git a/talkingq-url/docs/mqtt协议_V1.7.txt b/talkingq-url/docs/mqtt协议_V1.7.txt index 0fd5dd4..93c2e14 100644 --- a/talkingq-url/docs/mqtt协议_V1.7.txt +++ b/talkingq-url/docs/mqtt协议_V1.7.txt @@ -46,21 +46,23 @@ device/imei/response 2. 音量调节 device/imei/command 服务器下发: -{ - "msg_id": "002", - "params": { - "level": 70 - } -} -device/imei/response -4G模块响应: -{ - "msg_id": "002", - "status": "success", - "data": { - "current_level": 70 - } -} +{ + "msg_id": "002", + "params": { + "level": 70 + }, + "time": 1717560000000 +} +device/imei/response +4G模块响应: +{ + "msg_id": "002", + "status": "success", + "data": { + "current_level": 70 + }, + "time": 1717560000000 +} 3. OTA升级 device/imei/command diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 2a123b6..667c4f6 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/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( diff --git a/talkingq-url/tests/test_volume_command_status.py b/talkingq-url/tests/test_volume_command_status.py new file mode 100644 index 0000000..e641e58 --- /dev/null +++ b/talkingq-url/tests/test_volume_command_status.py @@ -0,0 +1,98 @@ +import pytest + +from handlers.mqtt_handler import TalkingQMQTTService + + +@pytest.mark.asyncio +async def test_volume_command_includes_time_and_completes_on_matching_response(monkeypatch): + service = TalkingQMQTTService({"broker": "127.0.0.1"}) + published = [] + + async def fake_publish(topic, payload): + published.append((topic, payload)) + return True + + async def fake_schedule_persistence(device_id, label, coro): + del device_id, label + coro.close() + + monkeypatch.setattr(service, "_publish", fake_publish) + monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence) + monkeypatch.setattr(service, "_current_utc_timestamp_ms", lambda: 1717560000000) + + command = await service.send_volume_command("TalkingQ_XQSN00001005", 70) + + assert command["msg_id"] == "002" + assert command["time"] == 1717560000000 + assert command["status"] == "sent" + assert published == [ + ( + "device/TalkingQ_XQSN00001005/command", + {"msg_id": "002", "params": {"level": 70}, "time": 1717560000000}, + ) + ] + + await service._handle_volume_response( + "TalkingQ_XQSN00001005", + { + "msg_id": "002", + "status": "success", + "data": {"current_level": 70}, + "time": 1717560000000, + }, + ) + + status = await service.get_volume_command_status("TalkingQ_XQSN00001005", 1717560000000) + assert status["status"] == "completed" + assert status["current_level"] == 70 + + +@pytest.mark.asyncio +async def test_volume_command_response_without_time_does_not_complete_command(monkeypatch): + service = TalkingQMQTTService({"broker": "127.0.0.1"}) + + async def fake_publish(topic, payload): + del topic, payload + return True + + async def fake_schedule_persistence(device_id, label, coro): + del device_id, label + coro.close() + + monkeypatch.setattr(service, "_publish", fake_publish) + monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence) + monkeypatch.setattr(service, "_current_utc_timestamp_ms", lambda: 1717560000000) + + command = await service.send_volume_command("TalkingQ_XQSN00001005", 70) + await service._handle_volume_response( + "TalkingQ_XQSN00001005", + { + "msg_id": "002", + "status": "success", + "data": {"current_level": 70}, + }, + ) + + status = await service.get_volume_command_status("TalkingQ_XQSN00001005", command["time"]) + assert status["status"] == "sent" + assert status["current_level"] is None + + +@pytest.mark.asyncio +async def test_volume_command_status_times_out(monkeypatch): + service = TalkingQMQTTService({"broker": "127.0.0.1"}) + now = 1717560000000 + + async def fake_publish(topic, payload): + del topic, payload + return True + + monkeypatch.setattr(service, "_publish", fake_publish) + monkeypatch.setattr(service, "_current_utc_timestamp_ms", lambda: now) + + command = await service.send_volume_command("TalkingQ_XQSN00001005", 70) + now += 10001 + + status = await service.get_volume_command_status("TalkingQ_XQSN00001005", command["time"]) + assert status["status"] == "timeout" + assert status["error"] == "device response timeout"