Files
ledgrab/server/src/ledgrab/config.py
T
alexei.dolgolyov 530316c2c3 feat(mqtt): multi-broker MQTT + Zigbee2MQTT light target
- New Z2MLightOutputTarget storage, processor, editor and routes for
  Zigbee2MQTT light entities (shares the HA-Light editor UI via the new
  light-target-editor module)
- Replace global MQTTService/MQTTConfig with per-source MQTTManager +
  MQTTRuntime; thread mqtt_source_id through Z2M targets, DIY MQTT
  devices, and the automation engine
- Migrate legacy single-broker YAML/env config to a "Default Broker"
  MQTTSource on startup (core/mqtt/legacy_migration.py) and drop the
  obsolete core/mqtt/mqtt_service.py
- Refresh /api/v1/system integration status to surface every MQTT source
- Extract shared light-target editor and refactor OutputTargetStore +
  output_targets routes around typed factories / auto-registry
- Modal CSS polish, locale strings, and storage/bindable test coverage
2026-05-12 18:06:09 +03:00

244 lines
7.6 KiB
Python

"""Configuration management for LedGrab."""
import os
import sys
from pathlib import Path
from typing import List, Literal
import yaml
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from ledgrab import paths as _paths
# Evaluate once at import time so every StorageConfig/AssetsConfig instance
# sees the same default across the process. Use POSIX separators so the
# default value is stable across platforms (SQLite and Python both accept
# forward slashes on Windows).
_DEFAULT_DATA_DIR_STR = _paths.default_data_dir().as_posix()
# ── Legacy env var migration ─────────────────────────────────
# Warn users who still have WLED_ env vars from pre-rename installs.
_OLD_PREFIX = "WLED_"
_NEW_PREFIX = "LEDGRAB_"
_ENV_MIGRATION_MAP = {
"WLED_CONFIG_PATH": "LEDGRAB_CONFIG_PATH",
"WLED_DEMO": "LEDGRAB_DEMO",
"WLED_RESTART": "LEDGRAB_RESTART",
"WLED_TRAY": "LEDGRAB_TRAY",
}
def _migrate_legacy_env_vars() -> None:
"""Detect old WLED_ env vars and auto-forward them to LEDGRAB_ equivalents."""
migrated = []
for old_key, new_key in _ENV_MIGRATION_MAP.items():
old_val = os.environ.get(old_key)
if old_val is not None and os.environ.get(new_key) is None:
os.environ[new_key] = old_val
migrated.append(f" {old_key} -> {new_key}")
# Also forward any WLED_<nested> vars (e.g. WLED_SERVER__PORT)
for key in list(os.environ):
if key.startswith(_OLD_PREFIX) and key not in _ENV_MIGRATION_MAP:
new_key = _NEW_PREFIX + key[len(_OLD_PREFIX) :]
if os.environ.get(new_key) is None:
os.environ[new_key] = os.environ[key]
migrated.append(f" {key} -> {new_key}")
if migrated:
print(
"WARNING: Detected legacy WLED_ environment variables. "
"The app was renamed to LedGrab — please update your env vars:\n" + "\n".join(migrated),
file=sys.stderr,
)
_migrate_legacy_env_vars()
class ServerConfig(BaseSettings):
"""Server configuration."""
host: str = "0.0.0.0"
port: int = 8080
log_level: str = "INFO"
cors_origins: List[str] = ["http://localhost:8080"]
class AuthConfig(BaseSettings):
"""Authentication configuration."""
api_keys: dict[str, str] = {} # label: key mapping (empty = auth disabled)
class AssetsConfig(BaseSettings):
"""Assets configuration."""
max_file_size_mb: int = 50 # Max upload size in MB
assets_dir: str = f"{_DEFAULT_DATA_DIR_STR}/assets"
class StorageConfig(BaseSettings):
"""Storage configuration."""
database_file: str = f"{_DEFAULT_DATA_DIR_STR}/ledgrab.db"
# The legacy single-broker ``MQTTConfig`` block has been removed. Brokers
# are now first-class :class:`MQTTSource` entries managed through the UI;
# see :mod:`ledgrab.core.mqtt.legacy_migration` for the one-shot upgrade
# path that seeds an MQTTSource from any pre-existing ``mqtt:`` YAML block.
class LoggingConfig(BaseSettings):
"""Logging configuration."""
format: Literal["json", "text"] = "json"
file: str = "logs/ledgrab.log"
max_size_mb: int = 100
backup_count: int = 5
class UpdatesConfig(BaseSettings):
"""Auto-update configuration.
``allow_unchecked`` enables installs of update artifacts that have no
published sha256 checksum. Default is False — leave it that way unless
you control the release server end-to-end.
"""
allow_unchecked: bool = False
class Config(BaseSettings):
"""Main application configuration."""
model_config = SettingsConfigDict(
env_prefix="LEDGRAB_",
env_nested_delimiter="__",
case_sensitive=False,
# ``extra="ignore"`` lets pre-existing YAML files (with the now-removed
# ``mqtt:`` block, etc.) load without raising. The legacy MQTT block
# is handled by ``core.mqtt.legacy_migration`` on first startup.
extra="ignore",
)
demo: bool = False
server: ServerConfig = Field(default_factory=ServerConfig)
auth: AuthConfig = Field(default_factory=AuthConfig)
storage: StorageConfig = Field(default_factory=StorageConfig)
assets: AssetsConfig = Field(default_factory=AssetsConfig)
logging: LoggingConfig = Field(default_factory=LoggingConfig)
updates: UpdatesConfig = Field(default_factory=UpdatesConfig)
def model_post_init(self, __context: object) -> None:
"""Override storage and assets paths when demo mode is active.
Inserts a ``demo`` segment before the final path component so that
``<data_dir>/ledgrab.db`` becomes ``<data_dir>/demo/ledgrab.db``.
Works for both absolute platform paths and legacy relative ones.
"""
if not self.demo:
return
def _demo_path(value: str) -> str:
p = Path(value)
if "demo" in p.parts:
return value
return str(p.parent / "demo" / p.name)
for field_name in StorageConfig.model_fields:
value = getattr(self.storage, field_name)
if isinstance(value, str) and value:
setattr(self.storage, field_name, _demo_path(value))
for field_name in AssetsConfig.model_fields:
value = getattr(self.assets, field_name)
if isinstance(value, str) and value:
setattr(self.assets, field_name, _demo_path(value))
@classmethod
def from_yaml(cls, config_path: str | Path) -> "Config":
"""Load configuration from YAML file.
Args:
config_path: Path to YAML configuration file
Returns:
Config instance
"""
config_path = Path(config_path)
if not config_path.exists():
raise FileNotFoundError(f"Configuration file not found: {config_path}")
with open(config_path, "r", encoding="utf-8") as f:
config_data = yaml.safe_load(f)
return cls(**config_data)
@classmethod
def load(cls) -> "Config":
"""Load configuration from default locations.
Tries to load from:
1. Environment variable LEDGRAB_CONFIG_PATH
2. LEDGRAB_DEMO=true → ./config/demo_config.yaml (if it exists)
3. ./config/default_config.yaml
4. Default values
Returns:
Config instance
"""
config_path = os.getenv("LEDGRAB_CONFIG_PATH")
if config_path:
return cls.from_yaml(config_path)
# Demo mode: try dedicated demo config first
if os.getenv("LEDGRAB_DEMO", "").lower() in ("true", "1", "yes"):
demo_path = Path("config/demo_config.yaml")
if demo_path.exists():
return cls.from_yaml(demo_path)
# Try default location (guard against permission errors on Android)
try:
default_path = Path("config/default_config.yaml")
if default_path.exists():
return cls.from_yaml(default_path)
except (PermissionError, OSError):
pass
# Use defaults
return cls()
# Global configuration instance
config: Config | None = None
def get_config() -> Config:
"""Get global configuration instance.
Returns:
Config instance
"""
global config
if config is None:
config = Config.load()
return config
def reload_config() -> Config:
"""Reload configuration from file.
Returns:
New Config instance
"""
global config
config = Config.load()
return config
def is_demo_mode() -> bool:
"""Check whether the application is running in demo mode."""
return get_config().demo