from __future__ import annotations from typing import Any from .base import ProtocolSpec BACNET_OBJECT_TYPE_ALIASES = { "analoginput": "AnalogInput", "analogoutput": "AnalogOutput", "analogvalue": "AnalogValue", "binaryinput": "BinaryInput", "binaryoutput": "BinaryOutput", "binaryvalue": "BinaryValue", "multistateinput": "MultiStateInput", "multistateoutput": "MultiStateOutput", "multistatevalue": "MultiStateValue", } def normalize_bacnet_object_type(value: Any) -> str: raw_value = str(value or "").strip() if not raw_value: raise ValueError("payload.object_type is required") lookup_key = "".join(char for char in raw_value if char.isalnum()).lower() return BACNET_OBJECT_TYPE_ALIASES.get(lookup_key, raw_value) BACNET_SPEC = ProtocolSpec( protocol="bacnet", create_device_path="/api/collector/device", create_point_path="/api/collector/bacnet/point/add_collect_point", point_test_path="/api/dc-gateway/bacnet/read_points", search_points_path="/api/dc-gateway/bacnet/search_points", bbmd_whois_path="/api/dc-gateway/bacnet/bbmd/whois", device_defaults={ "type": "bacnet", "device_type": 1, "port": 47808, "bacnet_net": 0, "asp_ip": "", "timeout": 3, "is_persistent": False, "group_id": 0, "alarm_interval": 90, "collect_interval": 5, }, point_defaults={ "point_id": "", "priority": None, "object_name": "", "units": "", "value_type": 0, "group_id": 0, "scale_ratio": 1, "value_offset": 0, "describe": "", "invalid_values": "", "valid_range_start": None, "valid_range_end": None, }, )