Files
AidsMonitoring/backend/services/settings_store.py
T

96 lines
3.0 KiB
Python

"""
Runtime settings store. Persists the settings the user changes via the
SETTINGS modal to a sidecar JSON file so they survive restarts.
The single mutable `SETTINGS` dict is the source of truth at runtime —
import it, don't copy it. Endpoints should call `apply_patch()` (which
also writes the file) rather than mutating directly.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from threading import Lock
_PATH = Path(__file__).parent.parent / "settings.json"
_LOCK = Lock()
DEFAULTS: dict = {
# AIS source
"ais_source": os.getenv("AIS_SOURCE", "SIMULATOR"),
"ais_serial_port": "",
"ais_baud": 38400,
"ais_net_addr": "",
# GPS (mirrored from .env at startup; the GPS hot-change endpoint also writes here)
"gps_port": os.getenv("GPS_PORT", ""),
"gps_baud": int(os.getenv("GPS_BAUD", 9600) or 9600),
# Station / antenna (purely descriptive for now; antenna position is
# exposed to the frontend so it can centre the map and run proximity
# checks against the station)
"station_name": "",
"iala_region": "B",
"antenna_lat": None,
"antenna_lon": None,
"antenna_height_m": None,
"station_notes": "",
# Alert thresholds (live — alert_engine reads these every evaluation)
"displacement_warn_m": 10.0,
"displacement_alarm_m": 15.0,
"proximity_alert_meters": int(os.getenv("PROXIMITY_ALERT_METERS", 500)),
"projection_minutes": int(os.getenv("PROJECTION_MINUTES", 10)),
"projection_radius_meters": 800,
"auto_record_trigger_m": 200.0,
# ATON battery thresholds (volts)
"battery_warn_v": 11.5,
"battery_alarm_v": 10.8,
# SMTP — operator's organisation email account that sends report emails.
# When configured, /alerts/report sends EMAIL via SMTP instead of opening
# the operator's local mail client. Leave smtp_host blank to disable.
"smtp_host": "",
"smtp_port": 587,
"smtp_user": "",
"smtp_password": "",
"smtp_from": "",
"smtp_from_name": "AidsMonitoring",
"smtp_use_tls": True,
}
SETTINGS: dict = dict(DEFAULTS)
def _load_from_disk() -> dict:
if not _PATH.exists():
return {}
try:
return json.loads(_PATH.read_text())
except Exception:
return {}
def _flush_to_disk():
try:
_PATH.write_text(json.dumps(SETTINGS, indent=2))
except Exception:
pass
def init():
"""Call once at app startup — merges saved overrides into SETTINGS."""
SETTINGS.update(_load_from_disk())
return SETTINGS
def get_all() -> dict:
return dict(SETTINGS)
def apply_patch(patch: dict) -> dict:
"""Merge patch into SETTINGS, ignoring unknown keys, write to disk.
Returns the new full settings dict."""
with _LOCK:
for k, v in patch.items():
if k in DEFAULTS:
SETTINGS[k] = v
_flush_to_disk()
return get_all()