from __future__ import annotations from plugins_func.register import Action, ActionResponse, ToolType, register_function from plugins_func.nav_copilot.service import get_service nav_start_record_desc = { "type": "function", "function": { "name": "nav_start_record", "description": ( "开始记录路线。" "如果设备接了外部 STM32 串口计数器，就默认使用计数路线模式；" "计数路线模式下，这一步只进入准备状态，不要求立刻收到串口值。" "后续用户说“标记急转弯”时，再抓取当前串口值作为标记点；说“结束录制”后自动进入监听预警。" "否则退回 GPS 路线模式。" "当用户说“开始记录路线”“开始录制路线”“录制路线”“开始记录”“开始记路线”时，" "不要只做自然语言回应，必须优先直接调用这个工具。" ), "parameters": { "type": "object", "properties": { "route_name": { "type": "string", "description": "路线名称，可选", }, "sample_hz": { "type": "integer", "description": "采样频率，计数模式建议 10Hz 左右", }, "mode": { "type": "string", "description": "auto 表示自动选择；counter 表示串口计数；gps 表示 GNSS 轨迹", "enum": ["auto", "counter", "gps"], }, }, "required": [], }, }, } @register_function("nav_start_record", nav_start_record_desc, ToolType.SYSTEM_CTL) async def nav_start_record( conn, route_name: str | None = None, sample_hz: int = 10, mode: str = "auto", ): svc = get_service(conn) route_id = await svc.start_recording( route_name=route_name, sample_hz=sample_hz, mode=mode, ) route_mode = svc.record_mode or "auto" if route_mode == "counter": text = ( f"已经进入计数路线录制准备状态，路线 ID 是 {route_id}。" "你到目标点时直接说“标记急转弯”或“标记障碍”，说“结束录制”后会自动开始监听预警。" ) else: text = ( f"已经开始记录 GPS 路线，路线 ID 是 {route_id}。" "你到目标点时可以直接说“标记急转弯”或“标记障碍”。" ) return ActionResponse(action=Action.RESPONSE, result=str(route_id), response=text) nav_stop_record_desc = { "type": "function", "function": { "name": "nav_stop_record", "description": "结束当前路线记录并保存。计数路线模式下，如果已经标记过风险点，会自动进入监听预警。", "parameters": {"type": "object", "properties": {}, "required": []}, }, } @register_function("nav_stop_record", nav_stop_record_desc, ToolType.SYSTEM_CTL) async def nav_stop_record(conn): svc = get_service(conn) result = await svc.stop_recording() if result is None: return ActionResponse(action=Action.RESPONSE, response="当前没有正在记录的路线。") if result.mode == "counter": if result.monitor_started: response = ( f"路线录制已结束，路线 ID 是 {result.route_id}。" f"已记录 {result.hazard_count} 个风险点，并且已经自动开始监听预警。" ) else: response = ( f"路线录制已结束，路线 ID 是 {result.route_id}。" "但你还没有标记风险点，所以还没有启动监听预警。" ) else: response = f"路线记录已结束，路线 ID 是 {result.route_id}。" return ActionResponse( action=Action.RESPONSE, result=str(result.route_id), response=response, ) nav_mark_hazard_desc = { "type": "function", "function": { "name": "nav_mark_hazard", "description": ( "在当前位置标记一个风险点。" "计数路线模式下，会把当前 STM32 串口传来的最新计数值保存成标记点。" "典型口令包括“标记急转弯”“标记起转弯”“标记大坑”“标记障碍”。" ), "parameters": { "type": "object", "properties": { "hazard_type": { "type": "string", "description": "风险点类型", "enum": [ "sharp_turn", "bump", "jump", "slope", "water", "obstacle", "other", ], }, "note": { "type": "string", "description": "补充说明，可选", }, }, "required": ["hazard_type"], }, }, } @register_function("nav_mark_hazard", nav_mark_hazard_desc, ToolType.SYSTEM_CTL) async def nav_mark_hazard(conn, hazard_type: str, note: str | None = None): svc = get_service(conn) result = await svc.mark_hazard(hazard_type=hazard_type, note=note or "") if result.mode == "counter" and result.counter_value is not None: response = f"已标记风险点，当前串口值是 {result.counter_value}。" else: response = "已标记风险点。" return ActionResponse( action=Action.RESPONSE, result=str(result.hazard_id), response=response, ) nav_mark_hazard_reject_desc = { "type": "function", "function": { "name": "nav_mark_hazard_reject", "description": "当用户进入风险点录制口令但没有说出有效风险类型时，提示其保持安静并等待重新录入。", "parameters": {"type": "object", "properties": {}, "required": []}, }, } @register_function("nav_mark_hazard_reject", nav_mark_hazard_reject_desc, ToolType.SYSTEM_CTL) async def nav_mark_hazard_reject(conn): return ActionResponse( action=Action.RESPONSE, response="请保持安静，等待录入风险点。", ) nav_start_run_desc = { "type": "function", "function": { "name": "nav_start_run", "description": ( "开始预警播报。" "对于串口计数路线，会在计数快接近标记点时播报“注意急转弯”，到达标记点时播报“当前急转弯”。" "通常计数路线在“结束录制”后会自动开始监听预警，这个工具主要用于手动补启。" "当用户说“开始预警”“开始领航”“开始播报”时，不要只做自然语言回应，必须优先直接调用这个工具。" ), "parameters": { "type": "object", "properties": { "route_id": { "type": "integer", "description": "指定路线 ID，可选", }, "tick_hz": { "type": "integer", "description": "刷新频率，计数模式建议 10Hz 左右", }, "mode": { "type": "string", "description": "可选。通常不必传，服务端会按路线类型自动选择", "enum": ["auto", "counter", "gps"], }, }, "required": [], }, }, } @register_function("nav_start_run", nav_start_run_desc, ToolType.SYSTEM_CTL) async def nav_start_run( conn, route_id: int | None = None, tick_hz: int = 10, mode: str = "auto", ): svc = get_service(conn) active_route_id = await svc.start_run(route_id=route_id, tick_hz=tick_hz, mode=mode) response = f"已经开始预警，使用路线 ID={active_route_id}。" if svc.run_mode == "counter": response = f"已经开始监听预警，使用路线 ID={active_route_id}。" return ActionResponse( action=Action.RESPONSE, result=str(active_route_id), response=response, ) nav_stop_run_desc = { "type": "function", "function": { "name": "nav_stop_run", "description": "停止当前预警播报。", "parameters": {"type": "object", "properties": {}, "required": []}, }, } @register_function("nav_stop_run", nav_stop_run_desc, ToolType.SYSTEM_CTL) async def nav_stop_run(conn): svc = get_service(conn) route_id = await svc.stop_run() if route_id is None: return ActionResponse(action=Action.RESPONSE, response="当前没有正在运行的预警。") return ActionResponse( action=Action.RESPONSE, result=str(route_id), response="预警已停止。", ) nav_status_desc = { "type": "function", "function": { "name": "nav_status", "description": "查询路线记录和预警状态。", "parameters": {"type": "object", "properties": {}, "required": []}, }, } @register_function("nav_status", nav_status_desc, ToolType.SYSTEM_CTL) async def nav_status(conn): svc = get_service(conn) return ActionResponse(action=Action.RESPONSE, response=svc.status_text()) nav_list_routes_desc = { "type": "function", "function": { "name": "nav_list_routes", "description": "列出最近保存的路线。", "parameters": { "type": "object", "properties": { "limit": { "type": "integer", "description": "返回数量，默认 10", } }, "required": [], }, }, } @register_function("nav_list_routes", nav_list_routes_desc, ToolType.SYSTEM_CTL) async def nav_list_routes(conn, limit: int = 10): svc = get_service(conn) routes = svc.db.list_routes(limit=int(limit or 10)) if not routes: return ActionResponse(action=Action.RESPONSE, response="暂无路线，请先开始记录路线。") lines = [] for route in routes: active = "（激活）" if route.active else "" lines.append( f"路线 {route.id}{active}：{route.name}，模式 {route.mode}，点数 {route.points}，风险点 {route.hazards}" ) return ActionResponse(action=Action.RESPONSE, response="；".join(lines)) nav_set_active_route_desc = { "type": "function", "function": { "name": "nav_set_active_route", "description": "将某条路线设置为当前激活路线。", "parameters": { "type": "object", "properties": { "route_id": { "type": "integer", "description": "路线 ID", } }, "required": ["route_id"], }, }, } @register_function("nav_set_active_route", nav_set_active_route_desc, ToolType.SYSTEM_CTL) async def nav_set_active_route(conn, route_id: int): svc = get_service(conn) svc.db.set_active_route(int(route_id)) return ActionResponse( action=Action.RESPONSE, response=f"已激活路线 ID={int(route_id)}。", )