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)