diff --git a/talkingq-url/banban/dao/device_alarm.py b/talkingq-url/banban/dao/device_alarm.py index a0adcb2..30231ac 100644 --- a/talkingq-url/banban/dao/device_alarm.py +++ b/talkingq-url/banban/dao/device_alarm.py @@ -59,6 +59,39 @@ class DeviceAlarmDAO(BaseDAO): await self.commit() return int(result.lastrowid) + async def get_by_id(self, *, alarm_id: int, device_id: str | None = None) -> Mapping[str, Any] | None: + where_device = "AND device_id = :device_id" if device_id is not None else "" + params: dict[str, Any] = {"alarm_id": alarm_id} + if device_id is not None: + params["device_id"] = device_id + + result = await self.execute( + text( + f""" + SELECT + alarm_id, + device_id, + owner_user_id, + child_id, + source_msg_id, + coord_type, + lat, + lng, + location_updated_at, + address, + address_resolved_at, + address_resolve_status, + created_at + FROM device_alarm_events + WHERE alarm_id = :alarm_id + {where_device} + LIMIT 1 + """ + ), + params, + ) + return result.mappings().first() + async def get_current_location_snapshot(self, *, child_id: int) -> Mapping[str, Any] | None: result = await self.execute( text( @@ -103,6 +136,40 @@ class DeviceAlarmDAO(BaseDAO): await self.commit() return bool(result.rowcount) + async def update_location_snapshot( + self, + *, + alarm_id: int, + location: Mapping[str, Any], + ) -> bool: + lat = location.get("lat") + lng = location.get("lng") + result = await self.execute( + text( + """ + UPDATE device_alarm_events + SET coord_type = :coord_type, + lat = :lat, + lng = :lng, + location_updated_at = :location_updated_at, + address = NULL, + address_resolved_at = NULL, + address_resolve_status = :address_resolve_status + WHERE alarm_id = :alarm_id + """ + ), + { + "alarm_id": alarm_id, + "coord_type": location.get("coord_type"), + "lat": lat, + "lng": lng, + "location_updated_at": location.get("updated_at"), + "address_resolve_status": 0 if lat is not None and lng is not None else None, + }, + ) + await self.commit() + return bool(result.rowcount) + async def get_active_binding_context(self, *, device_id: str) -> Mapping[str, Any] | None: result = await self.execute( text( diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index d7ca96e..a732909 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -342,8 +342,7 @@ def _row_to_alarm_item(row: Mapping) -> DeviceAlarmItem: location_age_seconds = _seconds_between(created_at, location_updated_at) location_stale = ( location_age_seconds is None - or location_age_seconds < 0 - or location_age_seconds > settings.alarm_location_stale_seconds + or abs(location_age_seconds) > settings.alarm_location_stale_seconds ) return DeviceAlarmItem( diff --git a/talkingq-url/banban/service/device_alarm.py b/talkingq-url/banban/service/device_alarm.py index 88d9591..ff70cb9 100644 --- a/talkingq-url/banban/service/device_alarm.py +++ b/talkingq-url/banban/service/device_alarm.py @@ -17,7 +17,13 @@ class DeviceAlarmService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="device_alarm_service") - async def record_alarm_event(self, *, device_id: str, source_msg_id: str = "010") -> int | None: + async def record_alarm_event( + self, + *, + device_id: str, + source_msg_id: str = "010", + resolve_address: bool = True, + ) -> int | None: db_session = await self.get_session() try: dao = DeviceAlarmDAO(db_session) @@ -36,7 +42,7 @@ class DeviceAlarmService(DatabaseServiceBase): source_msg_id=source_msg_id, location=location, ) - if alarm_id and location and location.get("lat") is not None and location.get("lng") is not None: + if resolve_address and alarm_id and location and location.get("lat") is not None and location.get("lng") is not None: await task_manager.create_task( self.resolve_alarm_address( alarm_id=alarm_id, @@ -51,6 +57,76 @@ class DeviceAlarmService(DatabaseServiceBase): finally: await db_session.close() + async def get_alarm_event( + self, + *, + alarm_id: int, + device_id: str | None = None, + ) -> Mapping[str, Any] | None: + db_session = await self.get_session() + try: + dao = DeviceAlarmDAO(db_session) + return await dao.get_by_id(alarm_id=alarm_id, device_id=device_id) + finally: + await db_session.close() + + async def update_alarm_location( + self, + *, + alarm_id: int, + location: Mapping[str, Any], + ) -> bool: + db_session = await self.get_session() + try: + dao = DeviceAlarmDAO(db_session) + return await dao.update_location_snapshot(alarm_id=alarm_id, location=location) + finally: + await db_session.close() + + async def apply_alarm_location( + self, + *, + alarm_id: int, + device_id: str, + location: Mapping[str, Any], + ) -> bool: + if location.get("lat") is None or location.get("lng") is None: + return False + + updated = await self.update_alarm_location(alarm_id=alarm_id, location=location) + if not updated: + return False + + address = str(location.get("address") or "").strip() + if address: + await self.update_alarm_address( + alarm_id=alarm_id, + address=address, + status=ADDRESS_RESOLVE_SUCCESS, + ) + return True + + await self.resolve_alarm_address( + alarm_id=alarm_id, + device_id=device_id, + lat=float(location["lat"]), + lng=float(location["lng"]), + ) + return True + + async def resolve_alarm_current_location(self, *, alarm_id: int, device_id: str) -> bool: + alarm = await self.get_alarm_event(alarm_id=alarm_id, device_id=device_id) + if not alarm or alarm.get("lat") is None or alarm.get("lng") is None: + return False + + await self.resolve_alarm_address( + alarm_id=alarm_id, + device_id=device_id, + lat=float(alarm["lat"]), + lng=float(alarm["lng"]), + ) + return True + async def resolve_alarm_address( self, *, diff --git a/talkingq-url/config.py b/talkingq-url/config.py index ffe9152..946f094 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -48,6 +48,7 @@ class Settings(BaseSettings): ) amap_http_timeout_seconds: float = Field(default=5.0, validation_alias="AMAP_HTTP_TIMEOUT_SECONDS") alarm_location_stale_seconds: int = Field(default=600, validation_alias="ALARM_LOCATION_STALE_SECONDS") + alarm_gps_query_timeout_seconds: float = Field(default=5.0, validation_alias="ALARM_GPS_QUERY_TIMEOUT_SECONDS") aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY") aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID") diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 667c4f6..deabbc3 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -51,6 +51,8 @@ class TalkingQMQTTService: self._volume_command_lock = asyncio.Lock() self._volume_command_states: Dict[tuple[str, int], dict] = {} self._last_volume_command_time_ms = 0 + self._gps_waiter_lock = asyncio.Lock() + self._gps_waiters: Dict[str, list[asyncio.Future]] = {} self._msg_handlers: Dict[str, Callable[[str, dict], Awaitable[None]]] = { "000": self._handle_device_info, @@ -189,10 +191,8 @@ class TalkingQMQTTService: lon = data.get("longitude") logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}") - await self._schedule_persistence( - device_id, - "gps", - location_service.report_mqtt_device_location( + async def _persist_gps_and_notify(): + row = await location_service.report_mqtt_device_location( device_id=device_id, latitude=data.get("latitude"), longitude=data.get("longitude"), @@ -204,8 +204,10 @@ class TalkingQMQTTService: source=data.get("source"), battery_pct=data.get("battery_pct"), device_time=parsed_device_time, - ), - ) + ) + await self._notify_gps_waiters(device_id, row) + + await self._schedule_persistence(device_id, "gps", _persist_gps_and_notify()) async def _handle_volume_response(self, device_id: str, payload: dict): command_time = self._parse_command_time(payload.get("time")) @@ -399,7 +401,27 @@ class TalkingQMQTTService: async def _handle_alarm_report(self, device_id: str, payload: dict): logger.info(device_id, "", f"[告警] 设备 {device_id} 发送紧急报警") async def _record_alarm_and_notify(): - alarm_id = await device_alarm_service.record_alarm_event(device_id=device_id, source_msg_id="010") + alarm_id = await device_alarm_service.record_alarm_event( + device_id=device_id, + source_msg_id="010", + resolve_address=False, + ) + if alarm_id: + location = await self.query_gps_and_wait_for_location( + device_id, + timeout_seconds=settings.alarm_gps_query_timeout_seconds, + ) + if location and location.get("lat") is not None and location.get("lng") is not None: + await device_alarm_service.apply_alarm_location( + alarm_id=alarm_id, + device_id=device_id, + location=location, + ) + else: + await device_alarm_service.resolve_alarm_current_location( + alarm_id=alarm_id, + device_id=device_id, + ) sms_notification_service.schedule_alarm_notification(device_id=device_id, alarm_id=alarm_id) await self._schedule_persistence(device_id, "alarm_event", _record_alarm_and_notify()) @@ -647,6 +669,36 @@ class TalkingQMQTTService: await self._publish(f"device/{device_id}/command", {"msg_id": "001"}) return "001" + async def query_gps_and_wait_for_location(self, device_id: str, *, timeout_seconds: float) -> dict | None: + timeout = max(0.1, float(timeout_seconds or 0)) + loop = asyncio.get_running_loop() + future = loop.create_future() + async with self._gps_waiter_lock: + self._gps_waiters.setdefault(device_id, []).append(future) + + try: + await self.send_gps_query(device_id) + return await asyncio.wait_for(future, timeout=timeout) + except asyncio.TimeoutError: + return None + finally: + async with self._gps_waiter_lock: + waiters = self._gps_waiters.get(device_id) + if waiters and future in waiters: + waiters.remove(future) + if waiters == []: + self._gps_waiters.pop(device_id, None) + + async def _notify_gps_waiters(self, device_id: str, row: dict | None) -> None: + if row is None: + return + async with self._gps_waiter_lock: + waiters = self._gps_waiters.pop(device_id, []) + + for future in waiters: + if not future.done(): + future.set_result(dict(row)) + def _parse_command_time(self, value) -> int | None: if value is None: return None