data-collector-mcp/data_collector_mcp/bacnet_server.py

from __future__ import annotations

from typing import Any, NotRequired, TypedDict

from pydantic import BaseModel

from .collector_api import (
    create_bacnet_devices as api_create_bacnet_devices,
    create_bacnet_points as api_create_bacnet_points,
    edit_bacnet_device as api_edit_bacnet_device,
    edit_bacnet_point as api_edit_bacnet_point,
)
from .gateway_api import (
    bacnet_point_collect_test as api_bacnet_point_collect_test,
    bacnet_point_search as api_bacnet_point_search,
)
from .mcp_app import mcp


class BacnetDeviceCreateItem(BaseModel):
    name: str
    ip: str
    bacnet_device_id: int
    port: int = 47808
    bacnet_net: int = 0
    asp_ip: str = ""
    timeout: int = 3
    is_persistent: bool = False
    group_id: int = 0
    alarm_interval: int = 90
    collect_interval: int = 5


class BacnetPointSpec(TypedDict):
    object_type: str
    object_id: int
    name: NotRequired[str]
    description: NotRequired[str]


class BacnetPointCreateItem(TypedDict):
    device_id: int
    object_type: str
    object_id: int
    name: NotRequired[str]
    object_name: NotRequired[str]
    point_id: NotRequired[str]
    priority: NotRequired[int | None]
    units: NotRequired[str]
    value_type: NotRequired[int]
    group_id: NotRequired[int]
    scale_ratio: NotRequired[float]
    value_offset: NotRequired[float]
    describe: NotRequired[str]


@mcp.tool(
    name="bacnet.point_collect_test",
    description=(
        "通过采集网关读取 BACnet/IP 点位 present-value。调用 "
        "{base_url}/api/dc-gateway/bacnet/read_points。ip 为 BACnet/IP 设备地址，"
        "bacnet_device_id 为 BACnet 设备对象实例号，范围 0..4194303；port 默认 47808。"
        "points 每项必须包含 object_type 和 object_id；object_type 支持 AnalogInput 这类格式。"
        "如果传入 analogInput、analog-input 等常见写法，需要先转换为 AnalogInput 这类格式；"
        "网关响应也会规范为 AnalogInput 形式。"
        "响应 data.points[].present_value 为当前值；响应透传上游 JSON，code=0 表示业务成功。"
    ),
)
def bacnet_point_collect_test(
    project_key: str,
    ip: str,
    bacnet_device_id: int,
    points: list[BacnetPointSpec],
    port: int = 47808,
) -> dict[str, Any]:
    return api_bacnet_point_collect_test(
        project_key,
        ip=ip,
        bacnet_device_id=bacnet_device_id,
        port=port,
        points=points,
    )


@mcp.tool(
    name="bacnet.point_search",
    description=(
        "通过采集网关搜索 BACnet/IP 设备点位。调用 "
        "{base_url}/api/dc-gateway/bacnet/search_points。ip 为 BACnet/IP 设备地址，"
        "bacnet_device_id 为 BACnet 设备对象实例号，范围 0..4194303；port 默认 47808。"
        "网关读取设备对象的 object-list，并过滤常见点位对象类型：AnalogInput、AnalogOutput、"
        "AnalogValue、BinaryInput、BinaryOutput、BinaryValue、MultiStateInput、MultiStateOutput、"
        "MultiStateValue。响应 data.points[] 包含 name、description、object_type、object_id、"
        "present_value；单点 description 或 present_value 读取失败时字段为 null。"
        "响应透传上游 JSON，code=0 表示业务成功。"
    ),
)
def bacnet_point_search(
    project_key: str,
    ip: str,
    bacnet_device_id: int,
    port: int = 47808,
) -> dict[str, Any]:
    return api_bacnet_point_search(
        project_key,
        ip=ip,
        bacnet_device_id=bacnet_device_id,
        port=port,
    )


@mcp.tool(
    name="collector.bacnet_device_create",
    description=(
        "汇采-批量创建 BACnet 设备。依次调用 {data_collector_base_url}/api/collector/device，"
        "请求体固定 type=bacnet 且 device_type=1(ASP 设备)。全部设备创建调用完成后会内部调用"
        "设备列表并匹配每个设备 id；如果匹配到多个设备，选择 id 最大的。devices 每项必须传 "
        "name、ip、bacnet_device_id；port 默认 47808，bacnet_net 默认 0，asp_ip 默认空字符串。"
        "默认参数: timeout=3, is_persistent=false, group_id=0, alarm_interval=90, collect_interval=5。"
        "响应返回批量结果，state=0 表示全部创建并匹配成功。"
    ),
)
def collector_bacnet_device_create(
    project_key: str,
    devices: list[BacnetDeviceCreateItem],
) -> dict[str, Any]:
    return api_create_bacnet_devices(project_key, [_dump_model_or_dict(item) for item in devices])


@mcp.tool(
    name="collector.bacnet_device_edit",
    description=(
        "汇采-编辑 BACnet 设备。调用 {data_collector_base_url}/api/collector/bacnet/device/edit。"
        "必须传 ori_id 原设备 id、name、ip、bacnet_device_id；MCP 会把 bacnet_device_id 映射为"
        "汇采旧接口的 device_id 字符串，并固定 type=1(ASP 设备)。编辑前设备不能处于已连接状态；"
        "若已连接，请先调用 collector.device_disconnect，并传 device_type=bacnet。该接口是全量更新语义。"
        "默认参数: port=47808, net=0, asp_ip='', timeout=3, is_persistent=false, "
        "device_group_id=0, alarm_interval=90, collect_interval=5。响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_bacnet_device_edit(
    project_key: str,
    ori_id: int,
    name: str,
    ip: str,
    bacnet_device_id: int,
    port: int = 47808,
    bacnet_net: int = 0,
    asp_ip: str = "",
    is_persistent: bool = False,
    device_group_id: int = 0,
    timeout: int = 3,
    alarm_interval: int = 90,
    collect_interval: int = 5,
) -> dict[str, Any]:
    return api_edit_bacnet_device(
        project_key,
        {
            "ori_id": ori_id,
            "name": name,
            "ip": ip,
            "bacnet_device_id": bacnet_device_id,
            "port": port,
            "bacnet_net": bacnet_net,
            "asp_ip": asp_ip,
            "is_persistent": is_persistent,
            "device_group_id": device_group_id,
            "timeout": timeout,
            "alarm_interval": alarm_interval,
            "collect_interval": collect_interval,
        },
    )


@mcp.tool(
    name="collector.bacnet_point_create",
    description=(
        "汇采-批量创建 BACnet 采集点位。依次调用 "
        "{data_collector_base_url}/api/collector/bacnet/point/add_collect_point。"
        "points 每项必须传 device_id、object_type、object_id，并传 name 或 object_name；"
        "object_type 支持 AnalogInput 这类格式；如果传入 analogInput、analog-input 等常见写法，"
        "需要先转换为 AnalogInput 这类格式；"
        "MCP 会将单个点位包装成汇采接口要求的 {device_id, points:[...]}。"
        "每个点位会补齐默认值: point_id='', priority=null, units='', value_type=0, "
        "group_id=0, scale_ratio=1, value_offset=0, describe=''。"
        "常见 value_type: 1=Boolean, 2=Unsigned, 3=Signed, 4=Real, 5=Double, 6=Enumerated。"
        "响应返回批量结果，state=0 表示全部创建成功。"
    ),
)
def collector_bacnet_point_create(
    project_key: str,
    points: list[BacnetPointCreateItem],
) -> dict[str, Any]:
    return api_create_bacnet_points(project_key, points)


@mcp.tool(
    name="collector.bacnet_point_edit",
    description=(
        "汇采-编辑 BACnet 采集点位。调用 {data_collector_base_url}/api/collector/bacnet/point/edit。"
        "必须传 ori_id 原点位 id、object_type、object_id，并传 name 或 object_name；ori_id 对应 "
        "collector.device_points 返回的 data.point[].id，MCP 会映射为汇采接口字段 id。"
        "object_type 支持 AnalogInput 这类格式；如果传入 analogInput、analog-input 等常见写法，"
        "MCP 会先转换为 AnalogInput 这类格式再调用汇采接口。"
        "该接口是全量更新语义，未传字段可能被默认值覆盖。默认参数: point_id='', priority=null, "
        "units='', value_type=0, group_id=0, scale_ratio=1, value_offset=0, describe='', "
        "invalid_values='', valid_range_start=null, valid_range_end=null。"
        "响应透传上游 JSON，state=0 表示业务成功。"
    ),
)
def collector_bacnet_point_edit(
    project_key: str,
    ori_id: int,
    object_type: str,
    object_id: int,
    name: str = "",
    object_name: str = "",
    point_id: str = "",
    priority: int | None = None,
    units: str = "",
    value_type: int = 0,
    group_id: int = 0,
    scale_ratio: float = 1,
    value_offset: float = 0,
    describe: str = "",
    invalid_values: str = "",
    valid_range_start: float | None = None,
    valid_range_end: float | None = None,
) -> dict[str, Any]:
    return api_edit_bacnet_point(
        project_key,
        {
            "ori_id": ori_id,
            "name": name,
            "object_type": object_type,
            "object_id": object_id,
            "object_name": object_name,
            "point_id": point_id,
            "priority": priority,
            "units": units,
            "value_type": value_type,
            "group_id": group_id,
            "scale_ratio": scale_ratio,
            "value_offset": value_offset,
            "describe": describe,
            "invalid_values": invalid_values,
            "valid_range_start": valid_range_start,
            "valid_range_end": valid_range_end,
        },
    )


def _dump_model_or_dict(item: Any) -> dict[str, Any]:
    if isinstance(item, BaseModel):
        return item.model_dump(exclude_none=True)
    return dict(item)