93 lines
4.8 KiB
Python
93 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
def _prefix(value: str) -> str:
|
|
value = str(value).strip()
|
|
return "" if not value or value == "/" else "/" + value.strip("/")
|
|
|
|
|
|
def _path(value: str | Path, base: Path) -> Path:
|
|
path = Path(value).expanduser()
|
|
return (base / path).resolve() if not path.is_absolute() else path.resolve()
|
|
|
|
|
|
def default_config(base: Path | None = None) -> dict[str, Any]:
|
|
base = (base or Path.cwd()).resolve()
|
|
rrd = base / "rrd"
|
|
return {
|
|
"HOST": "0.0.0.0", "PORT": 5000, "THREADS": 4, "URL_PREFIX": "", "RRD_FOLDER": rrd,
|
|
"LOG_FILE": rrd / "netatmo_service.log", "LOG_LEVEL": "INFO",
|
|
"LOG_MAX_BYTES": 5242880, "LOG_BACKUP_COUNT": 5, "LOG_CONSOLE": True,
|
|
"NETATMO_TOKEN_FILE": rrd / "netatmo_tokens.json",
|
|
"NETATMO_REDIRECT_URI": "http://localhost:5000/",
|
|
"GRAPH_WIDTH": 900, "GRAPH_HEIGHT": 420, "POLL_INTERVAL": 600,
|
|
"START_COLLECTOR": True, "NETATMO_CLIENT_ID": "", "NETATMO_CLIENT_SECRET": "",
|
|
"NETATMO_REFRESH_TOKEN": "", "NETATMO_ACCESS_TOKEN": "", "NETATMO_DEVICE_ID": "",
|
|
"NETATMO_TOKEN_URL": "https://api.netatmo.com/oauth2/token",
|
|
"NETATMO_STATIONS_URL": "https://api.netatmo.com/api/getstationsdata",
|
|
"MODULE_OUTDOOR": "Outdoor", "MODULE_WIND": "Wind", "MODULE_RAIN": "Rain", "MODULE_BEDROOM": "Bedroom",
|
|
"MODULE_STUDY": "Study", "MODULE_LIVING": "Living",
|
|
}
|
|
|
|
|
|
def load_config(filename: str | Path | None = None) -> dict[str, Any]:
|
|
"""Load and normalize the service's single YAML configuration file."""
|
|
filename = filename or "config.yaml"
|
|
config_path = Path(filename).expanduser().resolve()
|
|
if not config_path.is_file():
|
|
raise RuntimeError(
|
|
f"Configuration file not found: {config_path}. "
|
|
"Copy config.example.yaml to config.yaml and edit it."
|
|
)
|
|
try:
|
|
raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
raise RuntimeError(f"Cannot load YAML configuration {config_path}: {exc}") from exc
|
|
if not isinstance(raw, dict):
|
|
raise RuntimeError(f"YAML configuration {config_path} must contain a mapping")
|
|
|
|
sections = {name: raw.get(name, {}) for name in
|
|
("server", "storage", "logging", "collector", "graphs", "netatmo", "modules")}
|
|
for name, section in sections.items():
|
|
if not isinstance(section, dict):
|
|
raise RuntimeError(f"YAML section '{name}' must be a mapping")
|
|
base, result = config_path.parent, default_config(config_path.parent)
|
|
server, storage, log = sections["server"], sections["storage"], sections["logging"]
|
|
collector, graphs = sections["collector"], sections["graphs"]
|
|
netatmo, modules = sections["netatmo"], sections["modules"]
|
|
result.update({
|
|
"HOST": str(server.get("host", result["HOST"])),
|
|
"PORT": int(server.get("port", result["PORT"])),
|
|
"THREADS": int(server.get("threads", result["THREADS"])),
|
|
"URL_PREFIX": _prefix(server.get("url_prefix", result["URL_PREFIX"])),
|
|
"GRAPH_WIDTH": int(graphs.get("width", result["GRAPH_WIDTH"])),
|
|
"GRAPH_HEIGHT": int(graphs.get("height", result["GRAPH_HEIGHT"])),
|
|
"POLL_INTERVAL": int(collector.get("interval_seconds", result["POLL_INTERVAL"])),
|
|
"START_COLLECTOR": bool(collector.get("enabled", result["START_COLLECTOR"])),
|
|
"LOG_LEVEL": str(log.get("level", result["LOG_LEVEL"])).upper(),
|
|
"LOG_MAX_BYTES": int(log.get("max_bytes", result["LOG_MAX_BYTES"])),
|
|
"LOG_BACKUP_COUNT": int(log.get("backup_count", result["LOG_BACKUP_COUNT"])),
|
|
"LOG_CONSOLE": bool(log.get("console", result["LOG_CONSOLE"])),
|
|
"NETATMO_CLIENT_ID": str(netatmo.get("client_id", "")),
|
|
"NETATMO_CLIENT_SECRET": str(netatmo.get("client_secret", "")),
|
|
"NETATMO_REFRESH_TOKEN": str(netatmo.get("refresh_token", "")),
|
|
"NETATMO_ACCESS_TOKEN": str(netatmo.get("access_token", "")),
|
|
"NETATMO_DEVICE_ID": str(netatmo.get("device_id", "")),
|
|
"NETATMO_REDIRECT_URI": str(netatmo.get("redirect_uri", result["NETATMO_REDIRECT_URI"])),
|
|
"NETATMO_TOKEN_URL": str(netatmo.get("token_url", result["NETATMO_TOKEN_URL"])),
|
|
"NETATMO_STATIONS_URL": str(netatmo.get("stations_url", result["NETATMO_STATIONS_URL"])),
|
|
})
|
|
result["RRD_FOLDER"] = _path(storage.get("rrd_folder", "rrd"), base)
|
|
result["LOG_FILE"] = _path(log.get("file", result["RRD_FOLDER"] / "netatmo_service.log"), base)
|
|
result["NETATMO_TOKEN_FILE"] = _path(
|
|
netatmo.get("token_file", result["RRD_FOLDER"] / "netatmo_tokens.json"), base
|
|
)
|
|
for name in ("outdoor", "wind", "rain", "bedroom", "study", "living"):
|
|
result[f"MODULE_{name.upper()}"] = str(modules.get(name, result[f"MODULE_{name.upper()}"]))
|
|
return result
|