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