Improve GPS streaming and risk alert delivery

main/xingxing-server/core/api/app_gps_handler.py

@@ -32,6 +32,18 @@ class AppGpsHandler(BaseHandler):
         self._tool_timeout_s = 8
         self._nav_db = NavDB.default()
 
+    async def _attach_recent_alert(self, device_id: str, payload: dict | None) -> dict | None:
+        if payload is None:
+            return None
+        merged = dict(payload)
+        try:
+            recent_alert = await self.ws_server.get_recent_device_alert(device_id)
+        except Exception:
+            recent_alert = None
+        if recent_alert is not None:
+            merged["risk_alert_event"] = recent_alert
+        return merged
+
     async def _resolve_conn(self, request):
         requested_device_id = request.query.get("device_id")
         conn = await self.ws_server.get_connection(device_id=requested_device_id)
@@ -136,7 +148,12 @@ class AppGpsHandler(BaseHandler):
             "coord_type": "WGS84",
             "source": "device_mcp_gnss",
         }
-        self._latest_payloads[self._device_key(conn)] = (time.monotonic(), payload)
+        device_key = self._device_key(conn)
+        self._latest_payloads[device_key] = (time.monotonic(), payload)
+        try:
+            await self.ws_server.update_device_gps(device_key, payload)
+        except Exception:
+            pass
         return payload
 
     async def _get_or_start_inflight(self, device_key: str, conn, tool_name: str):
@@ -221,6 +238,9 @@ class AppGpsHandler(BaseHandler):
                 },
                 status=503,
             )
+        conn, _ = await self._resolve_conn(request)
+        if conn is not None:
+            payload = await self._attach_recent_alert(self._device_key(conn), payload)
         return self._json_response(payload)
 
     async def handle_get_risk_alerts(self, request):
@@ -368,7 +388,12 @@ class AppGpsHandler(BaseHandler):
                         subscribed_device_key, max_age_s=3.0, allow_stale=True
                     )
                     if current_payload is not None:
-                        await ws.send_json(current_payload)
+                        await ws.send_json(
+                            await self._attach_recent_alert(
+                                subscribed_device_key,
+                                current_payload,
+                            )
+                        )
                     else:
                         payload, error_code = await self._get_fix_payload(request)
                         if error_code:
@@ -380,11 +405,21 @@ class AppGpsHandler(BaseHandler):
                                 }
                             )
                         else:
-                            await ws.send_json(payload)
+                            await ws.send_json(
+                                await self._attach_recent_alert(
+                                    subscribed_device_key,
+                                    payload,
+                                )
+                            )
 
                 try:
                     payload = await asyncio.wait_for(gps_queue.get(), timeout=30.0)
-                    await ws.send_json(payload)
+                    await ws.send_json(
+                        await self._attach_recent_alert(
+                            subscribed_device_key,
+                            payload,
+                        )
+                    )
                 except asyncio.TimeoutError:
                     payload, error_code = await self._get_fix_payload(request)
                     if error_code:
@@ -401,7 +436,12 @@ class AppGpsHandler(BaseHandler):
                             subscribed_device_key = None
                             await asyncio.sleep(interval_s)
                     else:
-                        await ws.send_json(payload)
+                        await ws.send_json(
+                            await self._attach_recent_alert(
+                                subscribed_device_key,
+                                payload,
+                            )
+                        )
 
         except asyncio.CancelledError:
             raise

main/xingxing-server/core/websocket_server.py

@@ -50,6 +50,7 @@ class WebSocketServer:
         self.active_connections = {}
         self.gps_lock = asyncio.Lock()
         self.latest_gps_payloads = {}
+        self.latest_app_alert_payloads = {}
         self.gps_subscribers = {}
         modules = initialize_modules(
             self.logger,
@@ -144,6 +145,48 @@ class WebSocketServer:
             except asyncio.QueueEmpty:
                 pass
 
+    async def push_device_alert(self, device_id: str, payload: dict) -> None:
+        if not device_id or not isinstance(payload, dict):
+            return
+
+        outbound_payload = {
+            "ok": True,
+            "device_id": device_id,
+            "risk_alert_event": dict(payload),
+        }
+
+        async with self.gps_lock:
+            self.latest_app_alert_payloads[device_id] = (time.monotonic(), dict(payload))
+            latest_gps = self.latest_gps_payloads.get(device_id)
+            subscribers = list(self.gps_subscribers.get(device_id, []))
+
+        if latest_gps and isinstance(latest_gps[1], dict):
+            outbound_payload.update(dict(latest_gps[1]))
+            outbound_payload["risk_alert_event"] = dict(payload)
+
+        for subscriber in subscribers:
+            try:
+                if subscriber.full():
+                    subscriber.get_nowait()
+                subscriber.put_nowait(outbound_payload)
+            except asyncio.QueueFull:
+                pass
+            except asyncio.QueueEmpty:
+                pass
+
+    async def get_recent_device_alert(self, device_id: str, *, max_age_s: float = 20.0):
+        async with self.gps_lock:
+            cached = self.latest_app_alert_payloads.get(device_id)
+            if not cached:
+                return None
+
+            age_s = time.monotonic() - cached[0]
+            if age_s > max_age_s:
+                self.latest_app_alert_payloads.pop(device_id, None)
+                return None
+
+            return dict(cached[1])
+
     async def get_latest_gps(self, device_id: str, *, max_age_s: float = 3.0, allow_stale: bool = False):
         async with self.gps_lock:
             cached = self.latest_gps_payloads.get(device_id)

main/xingxing-server/plugins_func/nav_copilot/service.py

@@ -25,7 +25,9 @@ TAG = __name__
 logger = setup_logging()
 
 MONITOR_TICK_HZ = 5
-MIN_MOVEMENT_M = 1.5
+# Only ignore effectively identical fixes. A larger threshold misses
+# slow crossings that are visually obvious on the app map.
+MIN_MOVEMENT_M = 0.05
 LINE_BUFFER_M = 2.5
 TRIGGER_COOLDOWN_S = 15.0
 NOTE_MAX_LEN = 120
@@ -239,6 +241,26 @@ class NavCopilotService:
         except Exception as exc:
             logger.bind(tag=TAG).warning(f"TTS speak failed: {exc}")
 
+    async def _push_app_alert(self, *, alert_id: int, note: str) -> None:
+        server = getattr(self.conn, "server", None)
+        device_id = self.device_id()
+        if server is None or not device_id:
+            return
+
+        triggered_at_epoch_s = time.time()
+        payload = {
+            "type": "risk_line_triggered",
+            "event_id": uuid.uuid4().hex,
+            "alert_id": int(alert_id),
+            "note": str(note or "").strip(),
+            "triggered_at": float(triggered_at_epoch_s),
+            "triggered_at_ms": int(triggered_at_epoch_s * 1000),
+        }
+        try:
+            await server.push_device_alert(device_id, payload)
+        except Exception as exc:
+            logger.bind(tag=TAG).warning(f"push app alert failed: {exc}")
+
     async def _get_fix(self) -> GnssFix:
         server = getattr(self.conn, "server", None)
         device_id = getattr(self.conn, "device_id", None)
@@ -425,6 +447,10 @@ class NavCopilotService:
                         f"device_id={device_id}, alert_id={alert.id}, note={alert.note!r}"
                     )
                     self._speak(alert.note)
+                    await self._push_app_alert(
+                        alert_id=alert.id,
+                        note=alert.note,
+                    )
                     self._last_trigger_at[alert.id] = now
 
                 await asyncio.sleep(1.0 / MONITOR_TICK_HZ)