motian_mavbridge/motian_bridge/config.py

"""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