introduced config.yaml
This commit is contained in:
+80
-40
@@ -1,51 +1,91 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _bool(name: str, default: bool) -> bool:
|
||||
value = os.getenv(name)
|
||||
return default if value is None else value.lower() in {"1", "true", "yes", "on"}
|
||||
import yaml
|
||||
|
||||
|
||||
def _prefix(value: str) -> str:
|
||||
value = value.strip()
|
||||
if not value or value == "/":
|
||||
return ""
|
||||
return "/" + value.strip("/")
|
||||
value = str(value).strip()
|
||||
return "" if not value or value == "/" else "/" + value.strip("/")
|
||||
|
||||
|
||||
class Config:
|
||||
URL_PREFIX = _prefix(os.getenv("URL_PREFIX", ""))
|
||||
RRD_FOLDER = Path(os.getenv("RRD_FOLDER", "./rrd")).expanduser().resolve()
|
||||
LOG_FILE = Path(
|
||||
os.getenv("LOG_FILE", str(RRD_FOLDER / "netatmo_service.log"))
|
||||
).expanduser().resolve()
|
||||
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
|
||||
LOG_MAX_BYTES = int(os.getenv("LOG_MAX_BYTES", str(5 * 1024 * 1024)))
|
||||
LOG_BACKUP_COUNT = int(os.getenv("LOG_BACKUP_COUNT", "5"))
|
||||
LOG_CONSOLE = _bool("LOG_CONSOLE", True)
|
||||
NETATMO_TOKEN_FILE = Path(
|
||||
os.getenv("NETATMO_TOKEN_FILE", str(RRD_FOLDER / "netatmo_tokens.json"))
|
||||
).expanduser().resolve()
|
||||
GRAPH_WIDTH = int(os.getenv("GRAPH_WIDTH", "900"))
|
||||
GRAPH_HEIGHT = int(os.getenv("GRAPH_HEIGHT", "240"))
|
||||
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "600"))
|
||||
START_COLLECTOR = _bool("START_COLLECTOR", True)
|
||||
def _path(value: str | Path, base: Path) -> Path:
|
||||
path = Path(value).expanduser()
|
||||
return (base / path).resolve() if not path.is_absolute() else path.resolve()
|
||||
|
||||
NETATMO_CLIENT_ID = os.getenv("NETATMO_CLIENT_ID", "")
|
||||
NETATMO_CLIENT_SECRET = os.getenv("NETATMO_CLIENT_SECRET", "")
|
||||
NETATMO_REFRESH_TOKEN = os.getenv("NETATMO_REFRESH_TOKEN", "")
|
||||
NETATMO_ACCESS_TOKEN = os.getenv("NETATMO_ACCESS_TOKEN", "")
|
||||
NETATMO_DEVICE_ID = os.getenv("NETATMO_DEVICE_ID", "")
|
||||
NETATMO_TOKEN_URL = os.getenv("NETATMO_TOKEN_URL", "https://api.netatmo.com/oauth2/token")
|
||||
NETATMO_STATIONS_URL = os.getenv(
|
||||
"NETATMO_STATIONS_URL", "https://api.netatmo.com/api/getstationsdata"
|
||||
|
||||
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, "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": 240, "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_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"])),
|
||||
"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
|
||||
)
|
||||
|
||||
MODULE_OUTDOOR = os.getenv("MODULE_OUTDOOR", "Outdoor")
|
||||
MODULE_WIND = os.getenv("MODULE_WIND", "Wind")
|
||||
MODULE_BEDROOM = os.getenv("MODULE_BEDROOM", "Bedroom")
|
||||
MODULE_STUDY = os.getenv("MODULE_STUDY", "Study")
|
||||
MODULE_LIVING = os.getenv("MODULE_LIVING", "Living")
|
||||
for name in ("outdoor", "wind", "bedroom", "study", "living"):
|
||||
result[f"MODULE_{name.upper()}"] = str(modules.get(name, result[f"MODULE_{name.upper()}"]))
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user