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

from __future__ import annotations

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

from plugins_func.nav_copilot.service import get_service


nav_start_record_desc = {
    "type": "function",
    "function": {
        "name": "nav_start_record",
        "description": (
            "开始录制越野路线（教练/带跑模式）。"
            "录制期间服务器会持续读取车载端GPS(GNSS)位置并保存为轨迹。"
            "你可以随时再调用 nav_mark_hazard 来标记障碍点/风险点。"
            "当用户说“开始录制路线/开始录路线/带我跑一遍”等，优先调用本工具。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "route_name": {"type": "string", "description": "路线名称（可选）"},
                "sample_hz": {
                    "type": "integer",
                    "description": "采样频率Hz（默认1，建议1~5）",
                },
            },
            "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 = 1):
    svc = get_service(conn)
    rid = await svc.start_recording(route_name=route_name, sample_hz=sample_hz)
    text = f"已开始录制路线，路线ID是{rid}。你可以说“标记急转弯/标记大坑/标记障碍”来添加风险点。"
    return ActionResponse(action=Action.RESPONSE, result=str(rid), 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)
    rid = await svc.stop_recording()
    if rid is None:
        return ActionResponse(action=Action.RESPONSE, response="当前没有在录制路线。")
    text = f"路线录制已结束，路线ID是{rid}。你可以说“开始领航”进行预警播报。"
    return ActionResponse(action=Action.RESPONSE, result=str(rid), response=text)


nav_mark_hazard_desc = {
    "type": "function",
    "function": {
        "name": "nav_mark_hazard",
        "description": (
            "在当前位置标记一个障碍点/风险点（例如急转弯、大坑、跳台、陡坡、涉水、障碍）。"
            "用于“带着AI跑一遍”时把赛道风险点记录下来，之后领航时自动提前播报。"
            "当用户说“标记急转弯/标记大坑/标记跳台/标记障碍点”等，优先调用本工具。"
        ),
        "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)
    hid = await svc.mark_hazard(hazard_type=hazard_type, note=note or "")
    text = "已标记风险点。"
    return ActionResponse(action=Action.RESPONSE, result=str(hid), response=text)


nav_start_run_desc = {
    "type": "function",
    "function": {
        "name": "nav_start_run",
        "description": (
            "开始领航播报（预警模式）。服务器会持续读取GPS定位，匹配到已录制路线，"
            "在接近前方障碍点时自动语音播报提示。"
            "当用户说“开始领航/开始预警/开始播报”等，优先调用本工具。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "route_id": {"type": "integer", "description": "指定路线ID（可选）"},
                "tick_hz": {
                    "type": "integer",
                    "description": "领航刷新频率Hz（默认1，建议1~10；频率越高越实时，但更耗资源）",
                },
            },
            "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 = 5):
    svc = get_service(conn)
    rid = await svc.start_run(route_id=route_id, tick_hz=tick_hz)
    text = f"已开始领航，使用路线ID={rid}。接近风险点时我会自动提醒。"
    return ActionResponse(action=Action.RESPONSE, result=str(rid), response=text)


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)
    rid = await svc.stop_run()
    if rid is None:
        return ActionResponse(action=Action.RESPONSE, response="当前没有在领航。")
    return ActionResponse(action=Action.RESPONSE, result=str(rid), 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 r in routes:
        active = "（激活）" if r.active else ""
        lines.append(f"路线{r.id}{active}：{r.name}，点{r.points}，风险点{r.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)}。")