"""Runtime configuration. Secrets are read from the environment, not JSON.""" from dataclasses import dataclass import json import os from pathlib import Path @dataclass(frozen=True) class Config: system_id: str uav_type: int serial_port: str serial_baudrate: int mqtt_host: str mqtt_port: int = 1883 mqtt_client_id: str = "" mqtt_username: str | None = None mqtt_password_env: str | None = None mqtt_tls: bool = False mqtt_ca_file: str | None = None mqtt_cert_file: str | None = None mqtt_key_file: str | None = None mqtt_keepalive: int = 60 max_chunk_bytes: int = 1024 max_command_bytes: int = 4096 reconnect_seconds: float = 2.0 publish_raw_mavlink: bool = True def __post_init__(self) -> None: if not self.system_id or any(c in self.system_id for c in "/+#\x00"): raise ValueError("system_id must be a nonempty single MQTT topic level") if type(self.uav_type) is not int or self.uav_type < 0: raise ValueError("uav_type must be a nonnegative integer assigned by the platform") if not self.serial_port or not self.mqtt_host: raise ValueError("serial_port and mqtt_host are required") if not 1 <= self.serial_baudrate or not 1 <= self.mqtt_port <= 65535: raise ValueError("invalid serial_baudrate or mqtt_port") if not 1 <= self.max_chunk_bytes <= self.max_command_bytes: raise ValueError("max_chunk_bytes must be positive and <= max_command_bytes") if self.mqtt_keepalive < 1 or self.reconnect_seconds <= 0: raise ValueError("invalid keepalive or reconnect interval") if type(self.publish_raw_mavlink) is not bool: raise ValueError("publish_raw_mavlink must be a boolean") if self.mqtt_password_env and not self.mqtt_username: raise ValueError("mqtt_username is required when mqtt_password_env is set") if (self.mqtt_cert_file or self.mqtt_key_file or self.mqtt_ca_file) and not self.mqtt_tls: raise ValueError("TLS certificate settings require mqtt_tls=true") if bool(self.mqtt_cert_file) != bool(self.mqtt_key_file): raise ValueError("mqtt_cert_file and mqtt_key_file must be set together") @property def obtain_topic(self) -> str: return f"/OBTAIN/{self.system_id}" @property def publish_topic(self) -> str: return f"/PUBLISH/{self.system_id}" @property def receive_topic(self) -> str: return f"/RECEIVE/{self.system_id}" @property def client_id(self) -> str: return self.mqtt_client_id or f"mavlink-bridge-{self.system_id}" @classmethod def from_file(cls, path: str | Path) -> "Config": with open(path, encoding="utf-8") as file: values = json.load(file) if not isinstance(values, dict): raise ValueError("configuration must be a JSON object") return cls(**values) def mqtt_password(self) -> str | None: if self.mqtt_password_env is None: return None value = os.environ.get(self.mqtt_password_env) if value is None: raise ValueError(f"missing MQTT password environment variable: {self.mqtt_password_env}") return value