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)