mirror of
https://github.com/BigBodyCobain/Shadowbroker.git
synced 2026-05-08 10:24:48 +02:00
BigBodyCobain
6ffd54931c
Ship the 0.9.75 source update with improved startup/runtime hardening, operator API key onboarding, Meshtastic MQTT controls, Infonet/MeshChat separation, desktop package versioning, and aircraft telemetry refinements. Also updates focused backend/frontend tests for node settings, Meshtastic MQTT settings, and desktop runtime behavior.
72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = Path(__file__).parent.parent / "data"
|
|
NODE_FILE = DATA_DIR / "node.json"
|
|
_cache: dict | None = None
|
|
_cache_ts: float = 0.0
|
|
_CACHE_TTL = 5.0
|
|
_DEFAULTS = {
|
|
"enabled": True,
|
|
"operator_disabled": False,
|
|
"timemachine_enabled": False,
|
|
}
|
|
|
|
|
|
def _safe_int(value: object, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def read_node_settings() -> dict:
|
|
global _cache, _cache_ts
|
|
now = time.monotonic()
|
|
if _cache is not None and (now - _cache_ts) < _CACHE_TTL:
|
|
return _cache
|
|
if not NODE_FILE.exists():
|
|
result = {**_DEFAULTS, "updated_at": 0}
|
|
else:
|
|
try:
|
|
data = json.loads(NODE_FILE.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
result = {**_DEFAULTS, "updated_at": 0}
|
|
else:
|
|
operator_disabled = bool(data.get("operator_disabled", False))
|
|
raw_enabled = data.get("enabled", _DEFAULTS["enabled"])
|
|
# v0.9.7 initially wrote enabled:false as a default/offline state,
|
|
# which accidentally blocked InfoNet participation. Treat legacy
|
|
# false-without-marker as auto-enabled; only an explicit operator
|
|
# disable should keep the participant sync loop off.
|
|
enabled = False if operator_disabled else bool(raw_enabled or "operator_disabled" not in data)
|
|
result = {
|
|
"enabled": enabled,
|
|
"operator_disabled": operator_disabled,
|
|
"timemachine_enabled": bool(data.get("timemachine_enabled", _DEFAULTS["timemachine_enabled"])),
|
|
"updated_at": _safe_int(data.get("updated_at", 0) or 0),
|
|
}
|
|
_cache = result
|
|
_cache_ts = now
|
|
return result
|
|
|
|
|
|
def write_node_settings(*, enabled: bool | None = None, timemachine_enabled: bool | None = None) -> dict:
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
existing = read_node_settings()
|
|
next_enabled = bool(existing.get("enabled", _DEFAULTS["enabled"])) if enabled is None else bool(enabled)
|
|
payload = {
|
|
"enabled": next_enabled,
|
|
"operator_disabled": bool(existing.get("operator_disabled", _DEFAULTS["operator_disabled"])) if enabled is None else not next_enabled,
|
|
"timemachine_enabled": bool(existing.get("timemachine_enabled", _DEFAULTS["timemachine_enabled"])) if timemachine_enabled is None else bool(timemachine_enabled),
|
|
"updated_at": int(time.time()),
|
|
}
|
|
NODE_FILE.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
global _cache, _cache_ts
|
|
_cache = payload
|
|
_cache_ts = time.monotonic()
|
|
return payload
|