Add app GPS endpoint and websocket keepalive support

main/xiaozhi-server/app.py

@@ -72,7 +72,7 @@ async def main():
     ws_server = WebSocketServer(config)
     ws_task = asyncio.create_task(ws_server.start())
     # 启动 Simple http 服务器
-    ota_server = SimpleHttpServer(config)
+    ota_server = SimpleHttpServer(config, ws_server=ws_server)
     ota_task = asyncio.create_task(ota_server.start())
 
     read_config_from_api = config.get("read_config_from_api", False)

main/xiaozhi-server/config.yaml

@@ -72,7 +72,7 @@ enable_stop_tts_notify: false
 # 说完话是否开启提示音，音效地址
 stop_tts_notify_voice: "config/assets/tts_notify.mp3"
 # 是否启用WebSocket心跳保活机制
-enable_websocket_ping: false
+enable_websocket_ping: true
 
 
 # TTS音频发送延迟配置

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

@@ -0,0 +1,162 @@
+import asyncio
+import time
+from aiohttp import web
+
+from config.logger import setup_logging
+from core.api.base_handler import BaseHandler
+from core.providers.tools.device_mcp.mcp_handler import call_mcp_tool
+from plugins_func.nav_copilot.service import _parse_fix, _pick_gnss_tool_name
+
+TAG = __name__
+
+
+class AppGpsHandler(BaseHandler):
+    def __init__(self, config: dict, ws_server):
+        super().__init__(config)
+        self.ws_server = ws_server
+        self.logger = setup_logging()
+        self._inflight_lock = asyncio.Lock()
+        self._inflight_tasks: dict[str, asyncio.Task] = {}
+        self._latest_payloads: dict[str, tuple[float, dict]] = {}
+        self._cache_ttl_s = 1.5
+        self._tool_timeout_s = 8
+
+    async def _resolve_conn(self, request):
+        device_id = request.query.get("device_id")
+        conn = await self.ws_server.get_connection(device_id=device_id)
+        if conn is None:
+            return None, "no_active_device"
+        if not hasattr(conn, "mcp_client") or conn.mcp_client is None:
+            return None, "device_mcp_not_initialized"
+        if not await conn.mcp_client.is_ready():
+            return None, "device_mcp_not_ready"
+        tool_name = _pick_gnss_tool_name(conn)
+        if not tool_name:
+            return None, "device_gnss_tool_not_found"
+        return (conn, tool_name), None
+
+    @staticmethod
+    def _device_key(conn) -> str:
+        return str(getattr(conn, "device_id", "") or "default")
+
+    async def _fetch_fix_payload(self, conn, tool_name: str) -> dict:
+        raw = await call_mcp_tool(
+            conn,
+            conn.mcp_client,
+            tool_name,
+            "{}",
+            timeout=self._tool_timeout_s,
+        )
+
+        fix = _parse_fix(raw)
+        if not fix.ok:
+            return {
+                "ok": False,
+                "device_id": getattr(conn, "device_id", None),
+                "reason": "no_fix",
+            }
+
+        payload = {
+            "ok": True,
+            "device_id": getattr(conn, "device_id", None),
+            "lat": fix.lat,
+            "lon": fix.lon,
+            "speed_mps": fix.speed_mps,
+            "heading_deg": fix.heading_deg,
+            "timestamp": fix.t,
+            "source": "device_mcp_gnss",
+        }
+        self._latest_payloads[self._device_key(conn)] = (time.monotonic(), payload)
+        return payload
+
+    async def _get_or_start_inflight(self, device_key: str, conn, tool_name: str):
+        async with self._inflight_lock:
+            task = self._inflight_tasks.get(device_key)
+            if task is None or task.done():
+                task = asyncio.create_task(self._fetch_fix_payload(conn, tool_name))
+                self._inflight_tasks[device_key] = task
+            return task
+
+    async def _get_fix_payload(self, request):
+        resolved, error_code = await self._resolve_conn(request)
+        if error_code:
+            return None, error_code
+
+        conn, tool_name = resolved
+        device_key = self._device_key(conn)
+
+        cached = self._latest_payloads.get(device_key)
+        now = time.monotonic()
+        if cached and (now - cached[0]) <= self._cache_ttl_s:
+            return cached[1], None
+
+        try:
+            task = await self._get_or_start_inflight(device_key, conn, tool_name)
+            payload = await task
+            return payload, None
+        except Exception as exc:
+            self.logger.bind(tag=TAG).warning(f"app gps tool call failed: {exc}")
+            cached = self._latest_payloads.get(device_key)
+            if cached:
+                stale_payload = dict(cached[1])
+                stale_payload["stale"] = True
+                return stale_payload, None
+            return None, "device_gnss_call_failed"
+        finally:
+            async with self._inflight_lock:
+                task = self._inflight_tasks.get(device_key)
+                if task is not None and task.done():
+                    self._inflight_tasks.pop(device_key, None)
+
+    def _json_response(self, payload: dict, status: int = 200):
+        response = web.json_response(payload, status=status)
+        self._add_cors_headers(response)
+        return response
+
+    async def handle_get_latest(self, request):
+        payload, error_code = await self._get_fix_payload(request)
+        if error_code:
+            return self._json_response(
+                {
+                    "ok": False,
+                    "error": error_code,
+                    "active_devices": await self.ws_server.list_active_device_ids(),
+                },
+                status=503,
+            )
+        return self._json_response(payload)
+
+    async def handle_ws(self, request):
+        ws = web.WebSocketResponse(heartbeat=30)
+        await ws.prepare(request)
+
+        interval_s = 1.0
+        interval_arg = request.query.get("interval")
+        if interval_arg:
+            try:
+                interval_s = max(0.2, float(interval_arg))
+            except ValueError:
+                interval_s = 1.0
+
+        try:
+            while True:
+                payload, error_code = await self._get_fix_payload(request)
+                if error_code:
+                    await ws.send_json(
+                        {
+                            "ok": False,
+                            "error": error_code,
+                            "active_devices": await self.ws_server.list_active_device_ids(),
+                        }
+                    )
+                else:
+                    await ws.send_json(payload)
+                await asyncio.sleep(interval_s)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            self.logger.bind(tag=TAG).warning(f"app gps websocket closed: {exc}")
+        finally:
+            await ws.close()
+
+        return ws

main/xiaozhi-server/core/connection.py

@@ -229,6 +229,8 @@ class ConnectionHandler:
 
             # 认证通过,继续处理
             self.websocket = ws
+            if self.server:
+                await self.server.register_connection(self)
 
             # 检查是否来自MQTT连接
             request_path = ws.request.path
@@ -1326,6 +1328,9 @@ class ConnectionHandler:
                 self.vad.release_conn_resources(self)
 
             # 清理音频缓冲区
+            if self.server:
+                await self.server.unregister_connection(self)
+
             if hasattr(self, "audio_buffer"):
                 self.audio_buffer.clear()
 

main/xiaozhi-server/core/handle/textMessageProcessor.py

@@ -1,44 +1,43 @@
 import json
+import time
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from core.connection import ConnectionHandler
+
 from core.handle.textMessageHandlerRegistry import TextMessageHandlerRegistry
 
 TAG = __name__
 
 
 class TextMessageProcessor:
-    """消息处理器主类"""
+    """Main dispatcher for inbound text websocket frames."""
 
     def __init__(self, registry: TextMessageHandlerRegistry):
         self.registry = registry
 
     async def process_message(self, conn: "ConnectionHandler", message: str) -> None:
-        """处理消息的主入口"""
+        """Route JSON text messages to the registered message handlers."""
         try:
-            # 解析JSON消息
+            # Count MCP replies and other text frames as live activity so the
+            # connection does not get closed while app-side GPS polling is active.
+            conn.last_activity_time = time.time() * 1000
+
             msg_json = json.loads(message)
 
-            # 处理JSON消息
             if isinstance(msg_json, dict):
                 message_type = msg_json.get("type")
-
-                # 记录日志
                 conn.logger.bind(tag=TAG).info(f"收到{message_type}消息：{message}")
 
-                # 获取并执行处理器
                 handler = self.registry.get_handler(message_type)
                 if handler:
                     await handler.handle(conn, msg_json)
                 else:
                     conn.logger.bind(tag=TAG).error(f"收到未知类型消息：{message}")
-            # 处理纯数字消息
             elif isinstance(msg_json, int):
                 conn.logger.bind(tag=TAG).info(f"收到数字消息：{message}")
                 await conn.websocket.send(message)
 
         except json.JSONDecodeError:
-            # 非JSON消息直接转发
             conn.logger.bind(tag=TAG).error(f"解析到错误的消息：{message}")
             await conn.websocket.send(message)

main/xiaozhi-server/core/http_server.py

@@ -1,6 +1,7 @@
 import asyncio
 from aiohttp import web
 from config.logger import setup_logging
+from core.api.app_gps_handler import AppGpsHandler
 from core.api.ota_handler import OTAHandler
 from core.api.vision_handler import VisionHandler
 
@@ -8,9 +9,11 @@ TAG = __name__
 
 
 class SimpleHttpServer:
-    def __init__(self, config: dict):
+    def __init__(self, config: dict, ws_server=None):
         self.config = config
         self.logger = setup_logging()
+        self.ws_server = ws_server
+        self.app_gps_handler = AppGpsHandler(config, ws_server) if ws_server else None
         self.ota_handler = OTAHandler(config)
         self.vision_handler = VisionHandler(config)
 
@@ -76,6 +79,16 @@ class SimpleHttpServer:
                 )
 
                 # 运行服务
+                if self.app_gps_handler:
+                    app.add_routes(
+                        [
+                            web.get("/app/gps/latest", self.app_gps_handler.handle_get_latest),
+                            web.get("/app/gps/ws", self.app_gps_handler.handle_ws),
+                            web.options("/app/gps/latest", self.app_gps_handler.handle_options),
+                            web.options("/app/gps/ws", self.app_gps_handler.handle_options),
+                        ]
+                    )
+
                 runner = web.AppRunner(app)
                 await runner.setup()
                 site = web.TCPSite(runner, host, port)

main/xiaozhi-server/core/websocket_server.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+from typing import Optional
 
 import websockets
 from config.logger import setup_logging
@@ -44,6 +45,8 @@ class WebSocketServer:
         self.config = config
         self.logger = setup_logging()
         self.config_lock = asyncio.Lock()
+        self.connections_lock = asyncio.Lock()
+        self.active_connections = {}
         modules = initialize_modules(
             self.logger,
             self.config,
@@ -78,6 +81,64 @@ class WebSocketServer:
         ):
             await asyncio.Future()
 
+    async def register_connection(self, conn) -> None:
+        device_id = getattr(conn, "device_id", None)
+        if not device_id:
+            return
+        async with self.connections_lock:
+            self.active_connections[device_id] = conn
+        self.logger.bind(tag=TAG).info(f"Registered active device connection: {device_id}")
+
+    async def unregister_connection(self, conn) -> None:
+        device_id = getattr(conn, "device_id", None)
+        if not device_id:
+            return
+        removed = False
+        async with self.connections_lock:
+            current = self.active_connections.get(device_id)
+            if current is conn:
+                self.active_connections.pop(device_id, None)
+                removed = True
+        if removed:
+            self.logger.bind(tag=TAG).info(f"Unregistered active device connection: {device_id}")
+
+    async def get_connection(self, device_id: Optional[str] = None):
+        async with self.connections_lock:
+            if device_id:
+                conn = self.active_connections.get(device_id)
+                return conn if self._is_connection_alive(conn) else None
+
+            items = list(self.active_connections.items())
+            for _, conn in reversed(items):
+                if self._is_connection_alive(conn):
+                    return conn
+        return None
+
+    async def list_active_device_ids(self) -> list[str]:
+        async with self.connections_lock:
+            return [
+                device_id
+                for device_id, conn in self.active_connections.items()
+                if self._is_connection_alive(conn)
+            ]
+
+    @staticmethod
+    def _is_connection_alive(conn) -> bool:
+        if conn is None:
+            return False
+        stop_event = getattr(conn, "stop_event", None)
+        if stop_event is not None and stop_event.is_set():
+            return False
+
+        websocket = getattr(conn, "websocket", None)
+        if websocket is None:
+            return False
+        if hasattr(websocket, "closed"):
+            return not websocket.closed
+        if hasattr(websocket, "state"):
+            return websocket.state.name != "CLOSED"
+        return True
+
     async def _handle_connection(self, websocket: websockets.ServerConnection):
         headers = dict(websocket.request.headers)
         if headers.get("device-id", None) is None:

main/xiaozhi-server/deploy/baota/nginx_locations.conf

@@ -21,3 +21,16 @@ location /mcp/vision/explain {
   proxy_set_header Host $host;
 }
 
+location /app/gps/latest {
+  proxy_pass http://127.0.0.1:8003;
+  proxy_set_header Host $host;
+}
+
+location /app/gps/ws {
+  proxy_pass http://127.0.0.1:8003;
+  proxy_http_version 1.1;
+  proxy_set_header Upgrade $http_upgrade;
+  proxy_set_header Connection "upgrade";
+  proxy_set_header Host $host;
+  proxy_read_timeout 3600s;
+}