from __future__ import annotations import asyncio import os import socket import threading import warnings from contextlib import closing from typing import Any BAC0_REGISTERED_WARNING = r"object type .* for vendor identifier 842 already registered.*" warnings.filterwarnings("ignore", message=BAC0_REGISTERED_WARNING) import BAC0 from app.response import response_payload from app.schemas.bacnet import ( BacnetPointReadRequest, BacnetPointSearchRequest, BacnetPointSpec, bac0_object_type, normalize_object_type, ) POINT_OBJECT_TYPES = { "AnalogInput", "AnalogOutput", "AnalogValue", "BinaryInput", "BinaryOutput", "BinaryValue", "MultiStateInput", "MultiStateOutput", "MultiStateValue", } _BACNET_LOCK = threading.Lock() try: BAC0.log_level("silence") except Exception: pass def device_payload(request: BacnetPointReadRequest | BacnetPointSearchRequest) -> dict[str, Any]: return { "device_type": "BACnet/IP", "ip": request.ip, "port": request.port, "bacnet_device_id": request.bacnet_device_id, } def bacnet_address(ip: str, port: int) -> str: return f"{ip}:{port}" def property_read_args(address: str, object_type: str, object_id: int, property_name: str) -> str: return f"{address} {bac0_object_type(object_type)} {object_id} {property_name}" def get_local_ip(remote_ip: str, remote_port: int) -> str: configured_ip = os.getenv("BACNET_LOCAL_IP") if configured_ip: return configured_ip with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as sock: sock.connect((remote_ip, remote_port)) return str(sock.getsockname()[0]) def get_free_udp_port() -> int: configured_port = os.getenv("BACNET_LOCAL_PORT") if configured_port: return int(configured_port) with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as sock: sock.bind(("", 0)) return int(sock.getsockname()[1]) def local_bacnet_ip(remote_ip: str, remote_port: int) -> str: local_ip = get_local_ip(remote_ip, remote_port) local_mask = int(os.getenv("BACNET_LOCAL_MASK", "24")) return f"{local_ip}/{local_mask}" def json_value(value: Any) -> Any: if value is None or isinstance(value, bool | int | float | str): return value if isinstance(value, list | tuple): return [json_value(item) for item in value] if isinstance(value, dict): return {str(key): json_value(item) for key, item in value.items()} if hasattr(value, "value") and isinstance(value.value, bool | int | float | str): return value.value return str(value) async def read_property_or_none(bacnet: Any, address: str, object_type: str, object_id: int, property_name: str) -> Any: try: value = await bacnet.read(property_read_args(address, object_type, object_id, property_name)) except Exception: return None return json_value(value) async def read_present_value(bacnet: Any, address: str, point: BacnetPointSpec) -> Any: return json_value(await bacnet.read(property_read_args(address, point.object_type, point.object_id, "presentValue"))) def split_object_identifier(value: Any) -> tuple[str, int] | None: if isinstance(value, list | tuple) and len(value) == 2: raw_type, raw_id = value try: return normalize_object_type(str(raw_type).replace("ObjectType.", "")), int(raw_id) except (TypeError, ValueError): return None text = str(value).strip().strip("()") if "," in text: raw_type, raw_id = text.split(",", 1) elif ":" in text: raw_type, raw_id = text.split(":", 1) else: return None raw_type = raw_type.strip().strip("'\"").replace("ObjectType.", "") raw_id = raw_id.strip().strip("'\"") try: return normalize_object_type(raw_type), int(raw_id) except (TypeError, ValueError): return None async def read_points_async(request: BacnetPointReadRequest) -> list[dict[str, Any]]: address = bacnet_address(request.ip, request.port) local_ip = local_bacnet_ip(request.ip, request.port) local_port = get_free_udp_port() points: list[dict[str, Any]] = [] with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=BAC0_REGISTERED_WARNING) async with BAC0.start(ip=local_ip, port=local_port, ping=False) as bacnet: for point in request.points: points.append( { "object_type": point.object_type, "object_id": point.object_id, "present_value": await read_present_value(bacnet, address, point), } ) return points async def search_points_async(request: BacnetPointSearchRequest) -> list[dict[str, Any]]: address = bacnet_address(request.ip, request.port) local_ip = local_bacnet_ip(request.ip, request.port) local_port = get_free_udp_port() points: list[dict[str, Any]] = [] with warnings.catch_warnings(): warnings.filterwarnings("ignore", message=BAC0_REGISTERED_WARNING) async with BAC0.start(ip=local_ip, port=local_port, ping=False) as bacnet: object_list = await bacnet.read( property_read_args(address, "Device", request.bacnet_device_id, "objectList") ) for item in object_list or []: parsed = split_object_identifier(item) if parsed is None: continue object_type, object_id = parsed if object_type not in POINT_OBJECT_TYPES: continue points.append( { "name": await read_property_or_none(bacnet, address, object_type, object_id, "objectName"), "description": await read_property_or_none(bacnet, address, object_type, object_id, "description"), "object_type": object_type, "object_id": object_id, "present_value": await read_property_or_none(bacnet, address, object_type, object_id, "presentValue"), } ) return points def read_points(request: BacnetPointReadRequest) -> dict[str, Any]: try: with _BACNET_LOCK: points = asyncio.run(read_points_async(request)) except Exception as exc: return response_payload(1, str(exc), {"device": device_payload(request), "points": []}) return response_payload(0, "success", {"device": device_payload(request), "points": points}) def search_points(request: BacnetPointSearchRequest) -> dict[str, Any]: try: with _BACNET_LOCK: points = asyncio.run(search_points_async(request)) except Exception as exc: return response_payload(1, str(exc), {"device": device_payload(request), "points": []}) return response_payload(0, "success", {"device": device_payload(request), "points": points})