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_bbmd_whois as api_bacnet_bbmd_whois,
    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(BaseModel):
    device_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
    scale_ratio: float = 1
    value_offset: float = 0
    group_id: int = 0
    invalid_values: str = ""
    valid_range_start: float | None = None
    valid_range_end: float | None = None
    describe: str = ""


@mcp.tool(
    name="bacnet.point_collect_test",
    description=(
        "采集网关-读取 BACnet/IP 设备点位的值，只需传 project_key、ip、"
        "bacnet_device_id 和 points；port 默认 47808。points 每项包含 object_type、object_id；"
        "object_type 支持 AnalogInput、analogInput、analog-input"
        "不需要传name字段"
        "响应 data.points[].present_value 为当前值，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 设备点位。必须传 project_key、ip、bacnet_device_id；port 默认 47808。"
    ),
)
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="bacnet.bbmd_whois",
    description=(
        "采集网关-查询设备列表，通过 BBMD 执行 BACnet/IP Who-Is 设备发现。只需传 project_key。"
        "响应 data.devices 为发现的 BACnet 设备，"
    ),
)
def bacnet_bbmd_whois(project_key: str) -> dict[str, Any]:
    return api_bacnet_bbmd_whois(project_key)


@mcp.tool(
    name="collector.bacnet_device_create",
    description=(
        "汇采-批量创建 BACnet 设备。devices 每项必须传 name、ip、bacnet_device_id；"
        "返回批量创建结果和对应的设备 id"
    ),
)
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 设备。必须传 ori_id、name、ip、bacnet_device_id；"
        "ori_id是设备id，bacnet_device_id是设备号"
        "编辑前设备不能处于已连接状态，"
        "若已连接请先调用 collector.device_disconnect 且 device_type=bacnet。"
    ),
)
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 采集点位。points 每项必须传 device_id、object_type、object_id， name，"
        "object_type 支持 AnalogInput、analogInput、analog-input"
    ),
)
def collector_bacnet_point_create(
    project_key: str,
    points: list[BacnetPointCreateItem],
) -> dict[str, Any]:
    return api_create_bacnet_points(project_key, [_dump_model_or_dict(item) for item in points])


@mcp.tool(
    name="collector.bacnet_point_edit",
    description=(
        "汇采-编辑 BACnet 采集点位。必须传 ori_id、object_type、object_id，name"
        "ori_id 是设备id，object_type 支持 AnalogInput、analogInput、analog-input"
    ),
)
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,
    scale_ratio: float = 1,
    value_offset: float = 0,
    group_id: int = 0,
    invalid_values: str = "",
    valid_range_start: float | None = None,
    valid_range_end: float | None = None,
    describe: str = "",
) -> 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,
            "scale_ratio": scale_ratio,
            "value_offset": value_offset,
            "group_id": group_id,
            "invalid_values": invalid_values,
            "valid_range_start": valid_range_start,
            "valid_range_end": valid_range_end,
            "describe": describe,
        },
    )


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