ai_fwt_pro/main/xingxing-server/plugins_func/functions/nav_copilot.py

from __future__ import annotations

from plugins_func.register import Action, ActionResponse, ToolType, register_function

from plugins_func.nav_copilot.service import get_service


DISABLED_TEXT = "原来的语音风险点和路线录制功能已停用，请在手机 App 地图页管理风险线。"


def _disabled_response() -> ActionResponse:
    return ActionResponse(action=Action.RESPONSE, response=DISABLED_TEXT)


def _disabled_desc(name: str) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": DISABLED_TEXT,
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    }


@register_function("nav_start_record", _disabled_desc("nav_start_record"), ToolType.SYSTEM_CTL)
async def nav_start_record(conn, route_name: str | None = None, sample_hz: int = 10, mode: str = "gps"):
    return _disabled_response()


@register_function("nav_stop_record", _disabled_desc("nav_stop_record"), ToolType.SYSTEM_CTL)
async def nav_stop_record(conn):
    return _disabled_response()


@register_function("nav_mark_hazard", _disabled_desc("nav_mark_hazard"), ToolType.SYSTEM_CTL)
async def nav_mark_hazard(conn, hazard_type: str, note: str | None = None):
    return _disabled_response()


@register_function("nav_mark_hazard_reject", _disabled_desc("nav_mark_hazard_reject"), ToolType.SYSTEM_CTL)
async def nav_mark_hazard_reject(conn):
    return _disabled_response()


@register_function("nav_start_run", _disabled_desc("nav_start_run"), ToolType.SYSTEM_CTL)
async def nav_start_run(conn, route_id: int | None = None, tick_hz: int = 10, mode: str = "auto"):
    return _disabled_response()


@register_function("nav_stop_run", _disabled_desc("nav_stop_run"), ToolType.SYSTEM_CTL)
async def nav_stop_run(conn):
    return _disabled_response()


nav_status_desc = {
    "type": "function",
    "function": {
        "name": "nav_status",
        "description": "查询当前手机 App 风险线预警状态。",
        "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())


@register_function("nav_list_routes", _disabled_desc("nav_list_routes"), ToolType.SYSTEM_CTL)
async def nav_list_routes(conn, limit: int = 10):
    return _disabled_response()


@register_function("nav_set_active_route", _disabled_desc("nav_set_active_route"), ToolType.SYSTEM_CTL)
async def nav_set_active_route(conn, route_id: int):
    return _disabled_response()