rothmanandClaude Opus 5.5
19595aaf39
Dark grey background, name top left, date and clock top right, a flat panel with a blue top edge. The pages show the kiosk link (token masked) and the management address; the clock code is shared in kiosk.js. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
256 lines
8.0 KiB
Python
256 lines
8.0 KiB
Python
"""Kiosk configuration: defaults, validation and atomic persistence.
|
|
|
|
``load_config`` never raises: a missing, corrupt or partially invalid file
|
|
falls back to the ``.bak`` copy and then to defaults per key, so a bad edit
|
|
can never keep the kiosk from booting. ``save_config`` validates the whole
|
|
result before anything touches the disk.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import fcntl
|
|
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from . import validation as v
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
DEFAULTS: dict[str, Any] = {
|
|
"device_name": "VoxIP Kiosk",
|
|
"url": "https://schermen-ijsselflat.voxip.nl",
|
|
"refresh_interval": 0,
|
|
"screen_always_on": True,
|
|
"screen_off_minutes": 10,
|
|
"rotation": 0,
|
|
"resolution": "auto",
|
|
"zoom": 1.0,
|
|
"browser_restart": True,
|
|
"browser_profile": "persistent",
|
|
"admin_port": 8080,
|
|
"admin_https": False,
|
|
"admin_allow_public": False,
|
|
}
|
|
|
|
|
|
def _bool(value: Any) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
raise v.ValidationError("Verwacht ja/nee.")
|
|
|
|
|
|
def _int_range(lo: int, hi: int, message: str, allow_zero: bool = False) -> Callable[[Any], int]:
|
|
def check(value: Any) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
|
|
raise v.ValidationError(message)
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
raise v.ValidationError(message) from None
|
|
if allow_zero and number == 0:
|
|
return 0
|
|
if not lo <= number <= hi:
|
|
raise v.ValidationError(message)
|
|
return number
|
|
|
|
return check
|
|
|
|
|
|
def _rotation(value: Any) -> int:
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError):
|
|
number = -1
|
|
if isinstance(value, bool) or number not in (0, 90, 180, 270):
|
|
raise v.ValidationError("Rotatie moet 0, 90, 180 of 270 graden zijn.")
|
|
return number
|
|
|
|
|
|
def _zoom(value: Any) -> float:
|
|
try:
|
|
number = round(float(value), 2)
|
|
except (TypeError, ValueError):
|
|
number = -1.0
|
|
if isinstance(value, bool) or not 0.25 <= number <= 4.0:
|
|
raise v.ValidationError("Zoomfactor moet tussen 0,25 en 4 liggen.")
|
|
return number
|
|
|
|
|
|
def _choice(*options: str) -> Callable[[Any], str]:
|
|
def check(value: Any) -> str:
|
|
if value not in options:
|
|
raise v.ValidationError("Kies een van: " + ", ".join(options) + ".")
|
|
return value
|
|
|
|
return check
|
|
|
|
|
|
def _str(fn: Callable[[str], Any]) -> Callable[[Any], Any]:
|
|
def check(value: Any) -> Any:
|
|
if not isinstance(value, str):
|
|
raise v.ValidationError("Verwacht tekst.")
|
|
return fn(value)
|
|
|
|
return check
|
|
|
|
|
|
VALIDATORS: dict[str, Callable[[Any], Any]] = {
|
|
"device_name": _str(v.device_name),
|
|
"url": _str(v.kiosk_url),
|
|
"refresh_interval": _int_range(30, 86400, "Verversinterval: 0 (uit) of 30 tot 86400 seconden.", allow_zero=True),
|
|
"screen_always_on": _bool,
|
|
"screen_off_minutes": _int_range(1, 240, "Scherm uit na: 1 tot 240 minuten."),
|
|
"rotation": _rotation,
|
|
"resolution": _str(v.resolution),
|
|
"zoom": _zoom,
|
|
"browser_restart": _bool,
|
|
"browser_profile": _choice("persistent", "ephemeral"),
|
|
"admin_port": _int_range(1024, 65535, "Poort moet tussen 1024 en 65535 liggen."),
|
|
"admin_https": _bool,
|
|
"admin_allow_public": _bool,
|
|
}
|
|
|
|
|
|
class ConfigError(Exception):
|
|
def __init__(self, errors: dict[str, str]):
|
|
super().__init__("; ".join(f"{k}: {m}" for k, m in errors.items()))
|
|
self.errors = errors
|
|
|
|
|
|
def admin_url(config: dict[str, Any], hostname: str) -> str:
|
|
"""Address of the web interface as shown on the kiosk screen."""
|
|
scheme = "https" if config["admin_https"] else "http"
|
|
return f"{scheme}://{hostname}.local:{config['admin_port']}"
|
|
|
|
|
|
def validate(updates: dict[str, Any]) -> dict[str, Any]:
|
|
"""Validate known keys in ``updates``; raise ConfigError listing every problem."""
|
|
clean: dict[str, Any] = {}
|
|
errors: dict[str, str] = {}
|
|
for key, value in updates.items():
|
|
check = VALIDATORS.get(key)
|
|
if check is None:
|
|
errors[key] = "Onbekende instelling."
|
|
continue
|
|
try:
|
|
clean[key] = check(value)
|
|
except v.ValidationError as exc:
|
|
errors[key] = str(exc)
|
|
if errors:
|
|
raise ConfigError(errors)
|
|
return clean
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any] | None:
|
|
try:
|
|
with path.open(encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
except FileNotFoundError:
|
|
return None
|
|
except (OSError, ValueError) as exc:
|
|
log.warning("Kan %s niet lezen: %s", path, exc)
|
|
return None
|
|
if not isinstance(data, dict):
|
|
log.warning("%s bevat geen JSON-object", path)
|
|
return None
|
|
return data
|
|
|
|
|
|
def _read_raw(path: Path) -> dict[str, Any]:
|
|
data = _read_json(path)
|
|
if data is None and path.exists():
|
|
data = _read_json(path.with_suffix(path.suffix + ".bak"))
|
|
if data is not None:
|
|
log.warning("Configuratie hersteld uit %s.bak", path.name)
|
|
return data or {}
|
|
|
|
|
|
def load_config(path: Path) -> dict[str, Any]:
|
|
"""Effective configuration: defaults overlaid with every valid stored value."""
|
|
raw = _read_raw(path)
|
|
config = copy.deepcopy(DEFAULTS)
|
|
for key, value in raw.items():
|
|
check = VALIDATORS.get(key)
|
|
if check is None:
|
|
continue
|
|
try:
|
|
config[key] = check(value)
|
|
except v.ValidationError as exc:
|
|
log.warning("Ongeldige waarde voor %s in %s (%s); standaard gebruikt", key, path, exc)
|
|
return config
|
|
|
|
|
|
def atomic_write(path: Path, content: bytes, mode: int = 0o640, owner: tuple[int, int] | None = None) -> None:
|
|
"""Write ``content`` to ``path`` so readers see either the old or the new file, never a partial one."""
|
|
path = Path(path)
|
|
if owner is None and path.exists():
|
|
st = path.stat()
|
|
owner = (st.st_uid, st.st_gid)
|
|
fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
|
|
try:
|
|
with os.fdopen(fd, "wb") as fh:
|
|
fh.write(content)
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
os.chmod(tmp, mode)
|
|
if owner is not None and os.geteuid() == 0:
|
|
os.chown(tmp, *owner)
|
|
os.replace(tmp, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp)
|
|
except FileNotFoundError:
|
|
pass
|
|
raise
|
|
dir_fd = os.open(str(path.parent), os.O_RDONLY)
|
|
try:
|
|
os.fsync(dir_fd)
|
|
finally:
|
|
os.close(dir_fd)
|
|
|
|
|
|
def atomic_write_json(path: Path, data: Any, mode: int = 0o640, owner: tuple[int, int] | None = None) -> None:
|
|
content = (json.dumps(data, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
|
|
atomic_write(path, content, mode=mode, owner=owner)
|
|
|
|
|
|
_thread_lock = threading.Lock()
|
|
|
|
|
|
def save_config(path: Path, updates: dict[str, Any]) -> dict[str, Any]:
|
|
"""Validate ``updates``, merge them into the stored file and write it atomically.
|
|
|
|
Unknown keys already in the file are preserved; the previous version is
|
|
kept as ``config.json.bak``. Returns the new effective configuration.
|
|
"""
|
|
clean = validate(updates)
|
|
path = Path(path)
|
|
lock_path = path.with_name(".config.lock")
|
|
with _thread_lock:
|
|
# Read-only is enough for flock and works for every writer (the web
|
|
# interface as voxip, the helper as root), whoever created the file.
|
|
lock_fd = os.open(str(lock_path), os.O_RDONLY | os.O_CREAT, 0o640)
|
|
try:
|
|
fcntl.flock(lock_fd, fcntl.LOCK_EX)
|
|
return _save_locked(path, clean)
|
|
finally:
|
|
os.close(lock_fd)
|
|
|
|
|
|
def _save_locked(path: Path, clean: dict[str, Any]) -> dict[str, Any]:
|
|
raw = _read_raw(path)
|
|
if path.exists():
|
|
try:
|
|
atomic_write(path.with_suffix(path.suffix + ".bak"), path.read_bytes())
|
|
except OSError as exc:
|
|
log.warning("Kon geen back-up van de configuratie maken: %s", exc)
|
|
raw.update(clean)
|
|
atomic_write_json(path, raw)
|
|
return load_config(path)
|