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

from ipaddress import ip_address
import re

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_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


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

    bbmd_ip: str = Field(min_length=1)
    bbmd_port: int = Field(default=47808, ge=1, le=65535)
    local_ip: str | None = Field(default=None, min_length=1)
    local_port: int | None = Field(default=None, ge=1, le=65535)
    ttl: int = Field(default=60, ge=1, le=65535)
    timeout: float = Field(default=5, gt=0, le=120)
    low_limit: int | None = Field(default=0, ge=0, le=OBJECT_ID_MAX)
    high_limit: int | None = Field(default=OBJECT_ID_MAX, ge=0, le=OBJECT_ID_MAX)
    local_device_id: int | None = Field(default=None, ge=0, le=OBJECT_ID_MAX)

    @field_validator("bbmd_ip", "local_ip")
    @classmethod
    def validate_optional_ip(cls, value: str | None) -> str | None:
        if value is None:
            return value
        try:
            ip_address(value)
        except ValueError as exc:
            raise ValueError("must be a valid IP address") from exc
        return value

    @model_validator(mode="after")
    def validate_whois_range(self) -> "BacnetBBMDWhoIsRequest":
        if (self.low_limit is None) != (self.high_limit is None):
            raise ValueError("low_limit and high_limit must be provided together or both omitted")
        if self.low_limit is not None and self.high_limit is not None and self.low_limit > self.high_limit:
            raise ValueError("low_limit must be less than or equal to high_limit")
        return self