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": (
            "开始记录路线。GPS 模式下，标记风险点会保存当前 GPS 位置；"
            "结束录制后会进入预警状态。"
            "当用户说“开始记录路线”“开始录制路线”“开始记录”时，优先调用这个工具。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "route_name": {
                    "type": "string",
                    "description": "路线名称，可选",
                },
                "sample_hz": {
                    "type": "integer",
                    "description": "采样频率，默认 10Hz",
                },
                "mode": {
                    "type": "string",
                    "description": "路线模式。通常使用 gps",
                    "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 = "gps",
):
    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.monitor_started:
        if result.mode == "counter":
            response = (
                f"路线录制已结束，路线 ID 是 {result.route_id}。"
                f"已记录 {result.hazard_count} 个风险点，并且已经开始监听预警。"
            )
        else:
            response = (
                f"路线记录已结束，路线 ID 是 {result.route_id}。"
                f"已记录 {result.hazard_count} 个风险点，并且已经进入预警状态。"
            )
    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": (
            "在当前位置标记一个风险点。"
            "GPS 模式下会保存当前 GPS 数据，后续进入该风险点范围内时触发预警。"
            "典型口令包括“标记水坑”“标记急转弯”“标记障碍”。"
        ),
        "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}。"
    elif result.mode == "gps" and result.lat is not None and result.lon is not None:
        response = f"已标记风险点，GPS 数据是 {result.lat:.6f},{result.lon:.6f}。"
    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)
    if svc.run_mode == "counter":
        response = f"已经开始监听预警，使用路线 ID={active_route_id}。"
    else:
        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 ""
        monitoring = "（预警中）" if route.monitoring else ""
        lines.append(
            f"路线 {route.id}{active}{monitoring}：{route.name}，模式 {route.mode}，点数 {route.points}，风险点 {route.hazards}"
        )
    return ActionResponse(action=Action.RESPONSE, response="\n".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)}。",
    )