data-collector-mcp/data_collector_mcp/bacnet_server.py

from __future__ import annotations

from typing import Annotated, Any

from pydantic import BaseModel, ConfigDict, Field

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


JsonScalar = str | int | float | bool | None

ProjectKeyParam = Annotated[str, Field(description="项目标识，来自 project.list 返回的 project_key。")]
BacnetIpParam = Annotated[str, Field(description="BACnet/IP 设备地址。")]
BacnetDeviceIdParam = Annotated[int, Field(description="BACnet 设备对象实例号，范围 0..4194303。")]
BacnetPortParam = Annotated[int, Field(description="BACnet/IP UDP 端口，默认 47808。")]
BacnetObjectTypeParam = Annotated[
    str,
    Field(description="BACnet 对象类型；支持 AnalogInput、analogInput、analog-input 等常见写法。"),
]
BacnetObjectIdParam = Annotated[int, Field(description="BACnet 对象实例号，范围 0..4194303。")]


class FlexibleModel(BaseModel):
    model_config = ConfigDict(extra="allow")


class BacnetDeviceCreateItem(BaseModel):
    name: str = Field(description="设备名称。")
    ip: str = Field(description="BACnet/IP 设备地址。")
    bacnet_device_id: int = Field(description="BACnet 设备对象实例号，范围 0..4194303。")
    port: int = Field(default=47808, description="BACnet/IP UDP 端口。")
    bacnet_net: int = Field(default=0, description="BACnet 网络号；ASP 设备通常为 0。")
    asp_ip: str = Field(default="", description="RP 设备关联 ASP 地址；ASP 设备通常为空。")
    timeout: int = Field(default=3, description="连接超时时间。")
    is_persistent: bool = Field(default=False, description="是否持久化连接。")
    group_id: int = Field(default=0, description="设备分组 id。")
    alarm_interval: int = Field(default=90, description="告警间隔，单位秒。")
    collect_interval: int = Field(default=5, description="采集周期，单位秒。")


class BacnetPointSpec(BaseModel):
    object_type: BacnetObjectTypeParam
    object_id: BacnetObjectIdParam
    name: str | None = Field(default=None, description="点位名称；读取当前值时可不传。")
    description: str | None = Field(default=None, description="点位描述；读取当前值时可不传。")


class BacnetPointCreateItem(BaseModel):
    device_id: int = Field(description="所属 BACnet 设备 id，来自 collector.device_list 或设备创建结果。")
    object_type: BacnetObjectTypeParam
    object_id: BacnetObjectIdParam
    name: str = Field(default="", description="点位名称；为空时可由 object_name 补齐。")
    object_name: str = Field(default="", description="BACnet 对象名；为空时可由 name 补齐。")
    point_id: str = Field(default="", description="外部点位编码。")
    priority: int | None = Field(default=None, description="写值优先级；读取点位通常为 null。")
    value_type: int = Field(
        default=0,
        description="BACnet PresentValue 类型；常见值：1=Boolean，2=Unsigned，3=Signed，4=Real，5=Double，6=Enumerated。",
    )
    scale_ratio: float = Field(default=1, description="缩放系数。")
    value_offset: float = Field(default=0, description="值偏移。")
    group_id: int = Field(default=0, description="点位分组 id。")
    invalid_values: str = Field(default="", description="无效值列表，多个值用逗号分隔。")
    valid_range_start: float | None = Field(default=None, description="合法范围最小值。")
    valid_range_end: float | None = Field(default=None, description="合法范围最大值。")
    describe: str = Field(default="", description="点位描述。")


class GatewayBacnetDevice(FlexibleModel):
    device_type: str | None = Field(default=None, description="设备类型，通常为 BACnet/IP。")
    ip: str | None = Field(default=None, description="BACnet/IP 设备地址。")
    port: int | None = Field(default=None, description="BACnet/IP UDP 端口。")
    bacnet_device_id: int | None = Field(default=None, description="BACnet 设备对象实例号。")


class GatewayBacnetPoint(FlexibleModel):
    name: str | None = Field(default=None, description="点位名称；点位搜索时由 object-name 读取。")
    description: str | None = Field(default=None, description="点位描述；点位搜索时由 description 读取。")
    object_type: str | None = Field(default=None, description="BACnet 对象类型。")
    object_id: int | None = Field(default=None, description="BACnet 对象实例号。")
    present_value: JsonScalar = Field(default=None, description="当前值；读取 BACnet present-value 得到。")


class GatewayBacnetPointsData(FlexibleModel):
    device: GatewayBacnetDevice | None = Field(default=None, description="本次读取或搜索的 BACnet 设备信息。")
    points: list[GatewayBacnetPoint] = Field(default_factory=list, description="BACnet 点位列表。")


class BacnetGatewayPointsOutput(FlexibleModel):
    code: int | str | None = Field(default=None, description="网关状态码；0 表示成功。")
    msg: str | None = Field(default=None, description="网关状态说明或错误信息。")
    data: GatewayBacnetPointsData | None = Field(default=None, description="BACnet 点位读取或搜索结果。")


class BacnetBbmdConfig(FlexibleModel):
    bbmd_ip: str | None = Field(default=None, description="BBMD 地址。")
    bbmd_port: int | None = Field(default=None, description="BBMD UDP 端口。")
    ttl: int | None = Field(default=None, description="Foreign Device 注册 TTL，单位秒。")
    timeout: int | None = Field(default=None, description="收集 I-Am 响应的等待时间，单位秒。")
    low_limit: int | None = Field(default=None, description="Who-Is 设备号下限。")
    high_limit: int | None = Field(default=None, description="Who-Is 设备号上限。")


class BacnetDiscoveredDevice(FlexibleModel):
    bacnet_device_id: int | None = Field(default=None, description="发现的 BACnet 设备对象实例号。")
    ip: str | None = Field(default=None, description="发现的 BACnet/IP 设备地址。")
    port: int | None = Field(default=None, description="发现的 BACnet/IP UDP 端口。")
    max_apdu: int | None = Field(default=None, description="设备支持的最大 APDU 长度。")
    segmentation: str | None = Field(default=None, description="设备分段能力。")
    vendor_id: int | None = Field(default=None, description="BACnet 厂商 id。")


class BacnetBbmdWhoisData(FlexibleModel):
    bbmd: BacnetBbmdConfig | None = Field(default=None, description="本次 BBMD Who-Is 使用的配置。")
    devices: list[BacnetDiscoveredDevice] = Field(default_factory=list, description="发现的 BACnet 设备列表。")


class BacnetBbmdWhoisOutput(FlexibleModel):
    code: int | str | None = Field(default=None, description="网关状态码；0 表示成功。")
    msg: str | None = Field(default=None, description="网关状态说明或错误信息。")
    data: BacnetBbmdWhoisData | None = Field(default=None, description="BBMD Who-Is 设备发现结果。")


class CollectorBaseOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="汇采状态码；0 表示成功。")
    state_info: str | None = Field(default=None, description="汇采状态说明。")


class BatchError(FlexibleModel):
    index: int | None = Field(default=None, description="输入数组中的下标。")
    name: str | None = Field(default=None, description="输入项名称。")
    stage: str | None = Field(default=None, description="失败阶段。")
    error: str | None = Field(default=None, description="失败原因。")


class BacnetDeviceCreateSummary(FlexibleModel):
    total: int | None = Field(default=None, description="输入设备总数。")
    created: int | None = Field(default=None, description="汇采创建设备成功数量。")
    matched: int | None = Field(default=None, description="创建后从设备列表匹配到设备 id 的数量。")
    failed: int | None = Field(default=None, description="失败数量。")


class BacnetDeviceCreateResult(FlexibleModel):
    index: int | None = Field(default=None, description="输入 devices 数组中的下标。")
    name: str | None = Field(default=None, description="设备名称。")
    device_id: int | None = Field(default=None, description="匹配到的汇采设备 id。")
    create_response: dict[str, Any] | None = Field(default=None, description="汇采创建设备接口原始响应。")
    matched_device: dict[str, Any] | None = Field(default=None, description="设备列表中匹配到的设备详情。")


class BacnetDeviceCreateOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="批量创建状态；0 表示全部成功，1 表示存在失败。")
    summary: BacnetDeviceCreateSummary | None = Field(default=None, description="批量创建设备汇总。")
    results: list[BacnetDeviceCreateResult] = Field(default_factory=list, description="每个设备的创建和匹配结果。")
    errors: list[BatchError] = Field(default_factory=list, description="失败项列表。")


class BacnetPointCreateSummary(FlexibleModel):
    total: int | None = Field(default=None, description="输入点位总数。")
    success: int | None = Field(default=None, description="汇采创建点位成功数量。")
    failed: int | None = Field(default=None, description="失败数量。")


class BacnetPointCreateResult(FlexibleModel):
    index: int | None = Field(default=None, description="输入 points 数组中的下标。")
    name: str | None = Field(default=None, description="点位名称。")
    device_id: int | None = Field(default=None, description="所属 BACnet 设备 id。")
    response: dict[str, Any] | None = Field(default=None, description="汇采创建点位接口原始响应。")


class BacnetPointCreateOutput(FlexibleModel):
    state: int | str | None = Field(default=None, description="批量创建状态；0 表示全部成功，1 表示存在失败。")
    summary: BacnetPointCreateSummary | None = Field(default=None, description="批量创建点位汇总。")
    results: list[BacnetPointCreateResult] = Field(default_factory=list, description="每个点位的创建结果。")
    errors: list[BatchError] = Field(default_factory=list, description="失败项列表。")


@mcp.tool(
    name="bacnet.point_collect_test",
    description=(
        "采集网关-读取 BACnet/IP 设备点位的当前值。"
    ),
)
def bacnet_point_collect_test(
    project_key: ProjectKeyParam,
    ip: BacnetIpParam,
    bacnet_device_id: BacnetDeviceIdParam,
    points: Annotated[list[BacnetPointSpec], Field(description="要读取的 BACnet 点位列表。")],
    port: BacnetPortParam = 47808,
) -> BacnetGatewayPointsOutput:
    return api_bacnet_point_collect_test(
        project_key,
        ip=ip,
        bacnet_device_id=bacnet_device_id,
        port=port,
        points=[_dump_model_or_dict(item) for item in points],
    )


@mcp.tool(
    name="bacnet.point_search",
    description=(
        "采集网关-搜索 BACnet/IP 设备点位"
    ),
)
def bacnet_point_search(
    project_key: ProjectKeyParam,
    ip: BacnetIpParam,
    bacnet_device_id: BacnetDeviceIdParam,
    port: BacnetPortParam = 47808,
) -> BacnetGatewayPointsOutput:
    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 设备发现。"
    ),
)
def bacnet_bbmd_whois(project_key: ProjectKeyParam) -> BacnetBbmdWhoisOutput:
    return api_bacnet_bbmd_whois(project_key)


@mcp.tool(
    name="collector.bacnet_device_create",
    description=(
        "汇采-批量创建 BACnet 设备。创建后会读取设备列表并匹配返回设备 id。"
    ),
)
def collector_bacnet_device_create(
    project_key: ProjectKeyParam,
    devices: Annotated[list[BacnetDeviceCreateItem], Field(description="要创建的 BACnet 设备列表。")],
) -> BacnetDeviceCreateOutput:
    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 设备。编辑前设备不能处于已连接状态；"
        "若已连接请先调用 collector.device_disconnect 且 device_type=bacnet。"
    ),
)
def collector_bacnet_device_edit(
    project_key: ProjectKeyParam,
    id: Annotated[int, Field(description="要编辑的原汇采设备 id。")],
    name: Annotated[str, Field(description="设备名称。")],
    ip: BacnetIpParam,
    bacnet_device_id: BacnetDeviceIdParam,
    port: BacnetPortParam = 47808,
    bacnet_net: Annotated[int, Field(description="BACnet 网络号；ASP 设备通常为 0。")] = 0,
    asp_ip: Annotated[str, Field(description="RP 设备关联 ASP 地址；ASP 设备通常为空。")] = "",
    is_persistent: Annotated[bool, Field(description="是否持久化连接。")] = False,
    device_group_id: Annotated[int, Field(description="设备分组 id。")] = 0,
    timeout: Annotated[int, Field(description="连接超时时间，默认3秒。")] = 3,
    alarm_interval: Annotated[int, Field(description="告警间隔，默认90秒。")] = 90,
    collect_interval: Annotated[int, Field(description="采集周期，默认5秒。")] = 5,
) -> CollectorBaseOutput:
    return api_edit_bacnet_device(
        project_key,
        {
            "ori_id": 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 采集点位。"
    ),
)
def collector_bacnet_point_create(
    project_key: ProjectKeyParam,
    points: Annotated[list[BacnetPointCreateItem], Field(description="要创建的 BACnet 采集点位列表。")],
) -> BacnetPointCreateOutput:
    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 采集点位。对象类型会在调用汇采接口前规范化。"
    ),
)
def collector_bacnet_point_edit(
    project_key: ProjectKeyParam,
    id: Annotated[int, Field(description="要编辑的原采集点位 id，对应 collector.device_points 返回的点位 id。")],
    object_type: BacnetObjectTypeParam,
    object_id: BacnetObjectIdParam,
    name: Annotated[str, Field(description="点位名称；为空时可由 object_name 补齐。")] = "",
    object_name: Annotated[str, Field(description="BACnet 对象名；为空时可由 name 补齐。")] = "",
    point_id: Annotated[str, Field(description="外部点位编码。")] = "",
    priority: Annotated[int | None, Field(description="写值优先级；读取点位通常为 null。")] = None,
    value_type: Annotated[
        int,
        Field(description="BACnet PresentValue 类型；常见值：1=Boolean，2=Unsigned，3=Signed，4=Real，5=Double，6=Enumerated。"),
    ] = 0,
    scale_ratio: Annotated[float, Field(description="缩放系数。")] = 1,
    value_offset: Annotated[float, Field(description="值偏移。")] = 0,
    group_id: Annotated[int, Field(description="点位分组 id。")] = 0,
    invalid_values: Annotated[str, Field(description="无效值列表，多个值用逗号分隔。")] = "",
    valid_range_start: Annotated[float | None, Field(description="合法范围最小值。")] = None,
    valid_range_end: Annotated[float | None, Field(description="合法范围最大值。")] = None,
    describe: Annotated[str, Field(description="点位描述。")] = "",
) -> CollectorBaseOutput:
    return api_edit_bacnet_point(
        project_key,
        {
            "id": id,
            "name": name,
            "object_type": object_type,
            "object_id": object_id,
            "object_name": object_name,
            "point_id": point_id,
            "priority": priority,
            "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)