modbus-server-nd/app_config.py

"""YAML configuration loading."""

from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml


@dataclass(frozen=True)
class DbConfig:
    host: str
    port: int
    database: str
    user: str
    password: str


@dataclass(frozen=True)
class ModbusConfig:
    host: str = "0.0.0.0"
    port: int = 502
    interval: int = 5


@dataclass(frozen=True)
class HttpProviderConfig:
    url: str = "http://192.168.1.109:18503/data/get_points_real_value"
    timeout_seconds: int = 5


@dataclass(frozen=True)
class LoggingConfig:
    dir: str = "logs"
    retention_days: int = 3
    level: str = "INFO"


@dataclass(frozen=True)
class AppConfig:
    db: DbConfig
    modbus: ModbusConfig
    http_provider: HttpProviderConfig
    logging: LoggingConfig


def load_config(path: str | Path) -> AppConfig:
    config_path = Path(path)
    with config_path.open("r", encoding="utf-8") as file:
        raw = yaml.safe_load(file) or {}

    db = _require_mapping(raw, "db")
    return AppConfig(
        db=DbConfig(
            host=str(_require(db, "host")),
            port=int(_require(db, "port")),
            database=str(_require(db, "database")),
            user=str(_require(db, "user")),
            password=str(_require(db, "password")),
        ),
        modbus=_load_modbus(raw.get("modbus") or {}),
        http_provider=_load_http_provider(raw.get("http_provider") or {}),
        logging=_load_logging(raw.get("logging") or {}),
    )


def _load_modbus(raw: dict[str, Any]) -> ModbusConfig:
    return ModbusConfig(
        host=str(raw.get("host", "0.0.0.0")),
        port=int(raw.get("port", 502)),
        interval=int(raw.get("interval", 5)),
    )


def _load_http_provider(raw: dict[str, Any]) -> HttpProviderConfig:
    return HttpProviderConfig(
        url=str(raw.get("url", "http://192.168.1.109:18503/data/get_points_real_value")),
        timeout_seconds=int(raw.get("timeout_seconds", 5)),
    )


def _load_logging(raw: dict[str, Any]) -> LoggingConfig:
    return LoggingConfig(
        dir=str(raw.get("dir", "logs")),
        retention_days=int(raw.get("retention_days", 3)),
        level=str(raw.get("level", "INFO")),
    )


def _require_mapping(raw: dict[str, Any], key: str) -> dict[str, Any]:
    value = raw.get(key)
    if not isinstance(value, dict):
        raise ValueError(f"配置缺失或格式错误: {key}")
    return value


def _require(raw: dict[str, Any], key: str) -> Any:
    if key not in raw or raw[key] is None:
        raise ValueError(f"配置缺失: {key}")
    return raw[key]