ai_fwt_pro/main/xiaozhi-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


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)}。",
    )