data-collector-debugtool/app/schemas/bacnet.py

from ipaddress import ip_address
import re

from pydantic import BaseModel, ConfigDict, Field, field_validator


OBJECT_ID_MAX = 4_194_303

OBJECT_TYPE_ALIASES = {
    "analoginput": ("AnalogInput", "analogInput"),
    "analogoutput": ("AnalogOutput", "analogOutput"),
    "analogvalue": ("AnalogValue", "analogValue"),
    "binaryinput": ("BinaryInput", "binaryInput"),
    "binaryoutput": ("BinaryOutput", "binaryOutput"),
    "binaryvalue": ("BinaryValue", "binaryValue"),
    "multistateinput": ("MultiStateInput", "multiStateInput"),
    "multistateoutput": ("MultiStateOutput", "multiStateOutput"),
    "multistatevalue": ("MultiStateValue", "multiStateValue"),
    "device": ("Device", "device"),
    "calendar": ("Calendar", "calendar"),
    "schedule": ("Schedule", "schedule"),
    "trendlog": ("TrendLog", "trendLog"),
    "notificationclass": ("NotificationClass", "notificationClass"),
}


def object_type_key(value: str) -> str:
    return re.sub(r"[^0-9a-z]", "", value.lower())


def normalize_object_type(value: str) -> str:
    key = object_type_key(value)
    if key in OBJECT_TYPE_ALIASES:
        return OBJECT_TYPE_ALIASES[key][0]
    if not re.fullmatch(r"[A-Za-z][0-9A-Za-z_-]*", value):
        raise ValueError("object_type must be a valid BACnet object type")
    parts = re.split(r"[-_]", value)
    return "".join(part[:1].upper() + part[1:] for part in parts if part)


def bac0_object_type(value: str) -> str:
    key = object_type_key(value)
    if key in OBJECT_TYPE_ALIASES:
        return OBJECT_TYPE_ALIASES[key][1]
    normalized = normalize_object_type(value)
    return normalized[:1].lower() + normalized[1:]


class BacnetBaseRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    ip: str = Field(min_length=1)
    bacnet_device_id: int = Field(ge=0, le=OBJECT_ID_MAX)
    port: int = Field(default=47808, ge=1, le=65535)

    @field_validator("ip")
    @classmethod
    def validate_ip(cls, value: str) -> str:
        try:
            ip_address(value)
        except ValueError as exc:
            raise ValueError("ip must be a valid IP address") from exc
        return value


class BacnetPointSpec(BaseModel):
    model_config = ConfigDict(extra="forbid")

    object_type: str
    object_id: int = Field(ge=0, le=OBJECT_ID_MAX)

    @field_validator("object_type")
    @classmethod
    def validate_object_type(cls, value: str) -> str:
        return normalize_object_type(value)


class BacnetPointReadRequest(BacnetBaseRequest):
    points: list[BacnetPointSpec] = Field(min_length=1)


class BacnetPointSearchRequest(BacnetBaseRequest):
    pass