692 lines
21 KiB
Python
692 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Integrator REST API, prefix /api/v1.
|
|
|
|
Read methods are open. Writes require the X-Api-Key header.
|
|
Sensor math matches collector/protocol.py (tenths of a volt, 2xAA curve).
|
|
"""
|
|
import hmac
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
from werkzeug.exceptions import HTTPException
|
|
|
|
from api import _apply_button_action, _apply_button_latch_reset
|
|
from config_store import load_config, save_config
|
|
from paths import run_dir
|
|
from sensors import (
|
|
_load_names,
|
|
_load_settings,
|
|
_normalize_addr,
|
|
_read_settings_raw,
|
|
_read_status,
|
|
_save_names,
|
|
_save_settings,
|
|
_write_cmd,
|
|
_write_settings_raw,
|
|
)
|
|
from system_info import collect_stats, trigger_reboot
|
|
|
|
def _load_protocol():
|
|
"""collector/protocol.py lives beside webconfig both in the repo and on device."""
|
|
path = Path(__file__).resolve().parent.parent / "collector" / "protocol.py"
|
|
spec = importlib.util.spec_from_file_location("collector_protocol", path)
|
|
if spec is None or spec.loader is None:
|
|
raise ImportError("collector protocol.py not found")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
_PROTOCOL = _load_protocol()
|
|
BATT_UNKNOWN = _PROTOCOL.BATT_UNKNOWN
|
|
charge_percent = _PROTOCOL.charge_percent
|
|
|
|
rest_bp = Blueprint("rest_api", __name__, url_prefix="/api/v1")
|
|
|
|
STALE_PERIODS = 3 # collector/config.py: silence of k periods -> stale
|
|
STATUS_FRESH_S = 15
|
|
HISTORY_NAME = "sensors-history.json"
|
|
MAC_RE = re.compile(r"^([0-9A-F]{2}:){5}[0-9A-F]{2}$")
|
|
|
|
|
|
def _err(code, status):
|
|
return jsonify({"ok": False, "error": code}), status
|
|
|
|
|
|
def _api_key_ok() -> bool:
|
|
expected = str(load_config().get("settings", {}).get("apiKey") or "")
|
|
got = request.headers.get("X-Api-Key") or ""
|
|
if not expected or not got or len(got) != len(expected):
|
|
return False
|
|
return hmac.compare_digest(got.encode("utf-8"), expected.encode("utf-8"))
|
|
|
|
|
|
def require_api_key(fn):
|
|
@wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
if not _api_key_ok():
|
|
return _err("unauthorized", 401)
|
|
return fn(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def _timing():
|
|
minutes = int(_load_settings()["report_interval_min"])
|
|
return minutes, minutes * 60 * STALE_PERIODS
|
|
|
|
|
|
def _level_pct(raw_mm, cal):
|
|
if raw_mm is None or not isinstance(cal, dict):
|
|
return None
|
|
try:
|
|
empty = int(cal["min"])
|
|
full = int(cal["max"])
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
if empty == full:
|
|
return None
|
|
t = (float(raw_mm) - empty) / float(full - empty)
|
|
if t < 0:
|
|
t = 0.0
|
|
elif t > 1:
|
|
t = 1.0
|
|
return round(t * 100.0, 1)
|
|
|
|
|
|
def _power(battery):
|
|
try:
|
|
batt = int(battery)
|
|
except (TypeError, ValueError):
|
|
return None, None
|
|
if batt == BATT_UNKNOWN:
|
|
return None, None
|
|
return batt * 100, charge_percent(batt / 10.0)
|
|
|
|
|
|
def _raw_mm(slot):
|
|
reading = slot.get("reading")
|
|
if reading is not None:
|
|
try:
|
|
return int(reading)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
dist = slot.get("dist")
|
|
try:
|
|
dist = int(dist)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if dist == 0xFFFF:
|
|
return None
|
|
return dist
|
|
|
|
|
|
def _card(slot, names, calibration):
|
|
mac = _normalize_addr(slot.get("addr") or "")
|
|
raw = _raw_mm(slot)
|
|
cal = calibration.get(mac) or {}
|
|
voltage_mv, charge = _power(slot.get("battery"))
|
|
stored_name = names.get(mac)
|
|
if stored_name is None:
|
|
stored_name = slot.get("name") or ""
|
|
return {
|
|
"mac": mac,
|
|
"name": stored_name,
|
|
"raw_mm": raw,
|
|
"level_pct": _level_pct(raw, cal),
|
|
"voltage_mv": voltage_mv,
|
|
"charge_pct": charge,
|
|
"age_s": slot.get("age_s"),
|
|
"state": slot.get("state") or "",
|
|
"rssi": slot.get("rssi"),
|
|
"last_seen": slot.get("last_seen"),
|
|
}
|
|
|
|
|
|
def _occupied_slots(status):
|
|
out = []
|
|
for slot in status.get("slots") or []:
|
|
if not isinstance(slot, dict) or slot.get("empty"):
|
|
continue
|
|
if not _normalize_addr(slot.get("addr") or ""):
|
|
continue
|
|
out.append(slot)
|
|
return out
|
|
|
|
|
|
def _slot_by_mac(status, mac):
|
|
for slot in _occupied_slots(status):
|
|
if _normalize_addr(slot.get("addr") or "") == mac:
|
|
return slot
|
|
return None
|
|
|
|
|
|
def _check_mac(mac):
|
|
addr = _normalize_addr(mac)
|
|
if not MAC_RE.match(addr):
|
|
return None
|
|
return addr
|
|
|
|
|
|
def _read_history():
|
|
path = run_dir() / HISTORY_NAME
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
return raw if isinstance(raw, dict) else {}
|
|
|
|
|
|
def _history_view(entries, cal, limit, since):
|
|
rows = []
|
|
for entry in entries:
|
|
if not isinstance(entry, dict) or "ts" not in entry:
|
|
continue
|
|
try:
|
|
ts = float(entry["ts"])
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if since is not None and ts < since:
|
|
continue
|
|
raw = entry.get("raw_mm")
|
|
if raw is not None:
|
|
try:
|
|
raw = int(raw)
|
|
except (TypeError, ValueError):
|
|
raw = None
|
|
voltage_mv, charge = _power(entry.get("battery"))
|
|
rows.append(
|
|
{
|
|
"ts": ts,
|
|
"raw_mm": raw,
|
|
"level_pct": _level_pct(raw, cal),
|
|
"voltage_mv": voltage_mv,
|
|
"charge_pct": charge,
|
|
"rssi": entry.get("rssi"),
|
|
}
|
|
)
|
|
rows.reverse()
|
|
return rows[:limit]
|
|
|
|
|
|
@rest_bp.route("/health", methods=["GET"])
|
|
def api_health():
|
|
status = _read_status()
|
|
now = time.time()
|
|
try:
|
|
ts = float(status.get("ts") or 0)
|
|
except (TypeError, ValueError):
|
|
ts = 0.0
|
|
fresh = bool(status.get("available")) and ts > 0 and (now - ts) <= STATUS_FRESH_S
|
|
live = stale = 0
|
|
for slot in _occupied_slots(status):
|
|
if slot.get("state") == "stale":
|
|
stale += 1
|
|
else:
|
|
live += 1
|
|
started = status.get("started_ts")
|
|
uptime = None
|
|
if started is not None:
|
|
try:
|
|
uptime = round(now - float(started), 1)
|
|
except (TypeError, ValueError):
|
|
uptime = None
|
|
stats = collect_stats()
|
|
version = "unknown"
|
|
try:
|
|
from updater import current_version
|
|
|
|
version = current_version().get("version") or "unknown"
|
|
except Exception:
|
|
pass
|
|
healthy = fresh
|
|
return jsonify(
|
|
{
|
|
"ok": healthy,
|
|
"status": "healthy" if healthy else "degraded",
|
|
"uptime_s": uptime,
|
|
"version": version,
|
|
"ts": now,
|
|
"collector": {
|
|
"available": fresh,
|
|
"sensors_live": live,
|
|
"sensors_stale": stale,
|
|
"sensors_total": live + stale,
|
|
},
|
|
"system": {
|
|
"cpu_percent": stats.get("cpu_percent"),
|
|
"memory_used_percent": (stats.get("memory") or {}).get("used_percent"),
|
|
"load_1m": (stats.get("load") or {}).get("1m"),
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@rest_bp.route("/sensors", methods=["GET"])
|
|
def api_sensors():
|
|
minutes, stale_s = _timing()
|
|
names = _load_names()
|
|
calibration = _load_settings().get("calibration") or {}
|
|
status = _read_status()
|
|
cards = [_card(slot, names, calibration) for slot in _occupied_slots(status)]
|
|
return jsonify(
|
|
{
|
|
"report_interval_min": minutes,
|
|
"stale_timeout_s": stale_s,
|
|
"sensors": cards,
|
|
}
|
|
)
|
|
|
|
|
|
@rest_bp.route("/sensors/<mac>", methods=["GET"])
|
|
def api_sensor(mac):
|
|
addr = _check_mac(mac)
|
|
if addr is None:
|
|
return _err("sensor_not_found", 404)
|
|
slot = _slot_by_mac(_read_status(), addr)
|
|
if slot is None:
|
|
return _err("sensor_not_found", 404)
|
|
names = _load_names()
|
|
calibration = _load_settings().get("calibration") or {}
|
|
return jsonify(_card(slot, names, calibration))
|
|
|
|
|
|
@rest_bp.route("/sensors/<mac>/history", methods=["GET"])
|
|
def api_sensor_history(mac):
|
|
addr = _check_mac(mac)
|
|
if addr is None:
|
|
return _err("sensor_not_found", 404)
|
|
status = _read_status()
|
|
history = _read_history()
|
|
entries = history.get(addr)
|
|
if entries is None:
|
|
entries = history.get(mac)
|
|
known = _slot_by_mac(status, addr) is not None or isinstance(entries, list)
|
|
if not known and addr not in _load_names():
|
|
return _err("sensor_not_found", 404)
|
|
raw_limit = request.args.get("limit")
|
|
if raw_limit is None or raw_limit == "":
|
|
limit = 100
|
|
else:
|
|
try:
|
|
limit = int(raw_limit)
|
|
except (TypeError, ValueError):
|
|
return _err("invalid_limit", 400)
|
|
if limit < 1 or limit > 1000:
|
|
return _err("invalid_limit", 400)
|
|
since = None
|
|
raw_since = request.args.get("since")
|
|
if raw_since not in (None, ""):
|
|
try:
|
|
since = float(raw_since)
|
|
except (TypeError, ValueError):
|
|
return _err("invalid_since", 400)
|
|
cal = (_load_settings().get("calibration") or {}).get(addr) or {}
|
|
return jsonify(
|
|
{
|
|
"mac": addr,
|
|
"entries": _history_view(entries or [], cal, limit, since),
|
|
}
|
|
)
|
|
|
|
|
|
@rest_bp.route("/sensors/<mac>/name", methods=["PUT"])
|
|
@require_api_key
|
|
def api_sensor_name(mac):
|
|
addr = _check_mac(mac)
|
|
if addr is None:
|
|
return _err("invalid_mac", 400)
|
|
body = request.get_json(silent=True) or {}
|
|
name = body.get("name")
|
|
if not isinstance(name, str):
|
|
return _err("invalid_name", 400)
|
|
name = name.strip()
|
|
names = _load_names()
|
|
if name:
|
|
names[addr] = name
|
|
else:
|
|
names.pop(addr, None)
|
|
try:
|
|
_save_names(names)
|
|
except OSError as e:
|
|
return _err(str(e), 500)
|
|
_write_cmd({"op": "reload_settings"})
|
|
return jsonify({"ok": True, "mac": addr, "name": name})
|
|
|
|
|
|
@rest_bp.route("/settings", methods=["GET"])
|
|
def api_settings_get():
|
|
minutes, stale_s = _timing()
|
|
return jsonify({"report_interval_min": minutes, "stale_timeout_s": stale_s})
|
|
|
|
|
|
@rest_bp.route("/settings", methods=["PUT"])
|
|
@require_api_key
|
|
def api_settings_put():
|
|
body = request.get_json(silent=True) or {}
|
|
if "report_interval_min" not in body or isinstance(body.get("report_interval_min"), bool):
|
|
return _err("invalid_interval", 400)
|
|
try:
|
|
minutes = int(body.get("report_interval_min"))
|
|
except (TypeError, ValueError):
|
|
return _err("invalid_interval", 400)
|
|
if minutes < 1 or minutes > 65535:
|
|
return _err("invalid_interval", 400)
|
|
try:
|
|
_save_settings(minutes)
|
|
except OSError as e:
|
|
return _err(str(e), 500)
|
|
return jsonify(
|
|
{
|
|
"report_interval_min": minutes,
|
|
"stale_timeout_s": minutes * 60 * STALE_PERIODS,
|
|
}
|
|
)
|
|
|
|
|
|
def _bound(body, key):
|
|
if key not in body or isinstance(body.get(key), bool):
|
|
return None
|
|
val = body.get(key)
|
|
if isinstance(val, float) and not val.is_integer():
|
|
return None
|
|
try:
|
|
ival = int(val)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if ival < 0 or ival > 65535:
|
|
return None
|
|
return ival
|
|
|
|
|
|
@rest_bp.route("/sensors/<mac>/calibration", methods=["GET"])
|
|
def api_calibration_get(mac):
|
|
addr = _check_mac(mac)
|
|
if addr is None:
|
|
return _err("sensor_not_found", 404)
|
|
cal = (_load_settings().get("calibration") or {}).get(addr)
|
|
if not isinstance(cal, dict):
|
|
return jsonify({"empty_mm": None, "full_mm": None})
|
|
return jsonify({"empty_mm": cal.get("min"), "full_mm": cal.get("max")})
|
|
|
|
|
|
@rest_bp.route("/sensors/<mac>/calibration", methods=["PUT"])
|
|
@require_api_key
|
|
def api_calibration_put(mac):
|
|
addr = _check_mac(mac)
|
|
if addr is None:
|
|
return _err("invalid_mac", 400)
|
|
body = request.get_json(silent=True) or {}
|
|
empty = _bound(body, "empty_mm")
|
|
full = _bound(body, "full_mm")
|
|
if empty is None or full is None or empty == full:
|
|
return _err("invalid_calibration", 400)
|
|
data = _read_settings_raw()
|
|
if "report_interval_min" not in data:
|
|
data["report_interval_min"] = _load_settings()["report_interval_min"]
|
|
cal = data.get("calibration")
|
|
if not isinstance(cal, dict):
|
|
cal = {}
|
|
cal[addr] = {"min": empty, "max": full}
|
|
data["calibration"] = cal
|
|
try:
|
|
_write_settings_raw(data)
|
|
except OSError as e:
|
|
return _err(str(e), 500)
|
|
return jsonify({"empty_mm": empty, "full_mm": full})
|
|
|
|
|
|
def _buttons():
|
|
return [b for b in (load_config().get("buttons") or []) if isinstance(b, dict) and b.get("id")]
|
|
|
|
|
|
def _find_button(buttons, button_id):
|
|
for btn in buttons:
|
|
if btn.get("id") == button_id:
|
|
return btn
|
|
return None
|
|
|
|
|
|
def _invalid(exc, code):
|
|
message = exc.description if isinstance(getattr(exc, "description", None), str) else code
|
|
body = {"ok": False, "error": code}
|
|
if message and message != code:
|
|
body["message"] = message
|
|
return jsonify(body), getattr(exc, "code", None) or 400
|
|
|
|
|
|
_BG_TYPES = {"gradient", "solid", "image"}
|
|
_COLOR_RE = re.compile(r"^#?[0-9A-Fa-f]{6}$")
|
|
|
|
|
|
def _norm_color(value):
|
|
if not isinstance(value, str) or not _COLOR_RE.match(value.strip()):
|
|
return None
|
|
text = value.strip()
|
|
if not text.startswith("#"):
|
|
text = "#" + text
|
|
return text.lower()
|
|
|
|
|
|
@rest_bp.route("/buttons", methods=["GET"])
|
|
def api_buttons():
|
|
return jsonify([{"id": b.get("id"), "label": b.get("label") or ""} for b in _buttons()])
|
|
|
|
|
|
@rest_bp.route("/buttons", methods=["POST"])
|
|
@require_api_key
|
|
def api_button_create():
|
|
data = request.get_json(silent=True) or {}
|
|
if not isinstance(data, dict):
|
|
return _err("invalid_button", 400)
|
|
try:
|
|
btn = {
|
|
"id": uuid.uuid4().hex,
|
|
"label": data.get("label") or "Кнопка",
|
|
"iconPath": data.get("iconPath") or "",
|
|
"action": _apply_button_action(
|
|
{
|
|
"type": data.get("actionType", "http_get"),
|
|
"url": data.get("url", ""),
|
|
"headers": data.get("headers") or {},
|
|
"body": data.get("body") or "",
|
|
},
|
|
data,
|
|
),
|
|
"feedback": {
|
|
"successText": data.get("successText") or "",
|
|
"errorText": data.get("errorText") or "Ошибка",
|
|
"pendingText": data.get("pendingText") or "...",
|
|
"fadeMs": int(data.get("fadeMs", 5000)),
|
|
},
|
|
"trigger": {
|
|
"mode": "click" if data.get("triggerMode") == "click" else "hold",
|
|
"holdMs": int(data.get("holdMs", 800)),
|
|
},
|
|
"color": data.get("color") or "#7b007b",
|
|
}
|
|
_apply_button_latch_reset(btn, data)
|
|
except HTTPException as exc:
|
|
return _invalid(exc, "invalid_button")
|
|
except (TypeError, ValueError):
|
|
return _err("invalid_button", 400)
|
|
cfg = load_config()
|
|
cfg.setdefault("buttons", []).append(btn)
|
|
save_config(cfg)
|
|
return jsonify(btn), 201
|
|
|
|
|
|
@rest_bp.route("/buttons/reorder", methods=["POST"])
|
|
@require_api_key
|
|
def api_button_reorder():
|
|
data = request.get_json(silent=True) or {}
|
|
order = data.get("order")
|
|
if not isinstance(order, list) or not all(isinstance(i, str) for i in order):
|
|
return _err("invalid_order", 400)
|
|
cfg = load_config()
|
|
by_id = {b.get("id"): b for b in cfg.get("buttons") or [] if isinstance(b, dict)}
|
|
new_list = [by_id[i] for i in order if i in by_id]
|
|
for btn in cfg.get("buttons") or []:
|
|
if btn not in new_list:
|
|
new_list.append(btn)
|
|
cfg["buttons"] = new_list
|
|
save_config(cfg)
|
|
return jsonify({"ok": True, "order": [b.get("id") for b in new_list]})
|
|
|
|
|
|
@rest_bp.route("/buttons/<button_id>", methods=["GET"])
|
|
@require_api_key
|
|
def api_button_get(button_id):
|
|
btn = _find_button(_buttons(), button_id)
|
|
if btn is None:
|
|
return _err("button_not_found", 404)
|
|
return jsonify(btn)
|
|
|
|
|
|
@rest_bp.route("/buttons/<button_id>", methods=["PUT"])
|
|
@require_api_key
|
|
def api_button_update(button_id):
|
|
data = request.get_json(silent=True) or {}
|
|
if not isinstance(data, dict):
|
|
return _err("invalid_button", 400)
|
|
cfg = load_config()
|
|
for btn in cfg.get("buttons") or []:
|
|
if not isinstance(btn, dict) or btn.get("id") != button_id:
|
|
continue
|
|
try:
|
|
if "label" in data:
|
|
btn["label"] = data.get("label") or ""
|
|
if "iconPath" in data:
|
|
btn["iconPath"] = data.get("iconPath") or ""
|
|
action = btn.setdefault("action", {})
|
|
_apply_button_action(action, data)
|
|
feedback = btn.setdefault("feedback", {})
|
|
if "successText" in data:
|
|
feedback["successText"] = data.get("successText") or ""
|
|
if "errorText" in data:
|
|
feedback["errorText"] = data.get("errorText") or ""
|
|
if "pendingText" in data:
|
|
feedback["pendingText"] = data.get("pendingText") or ""
|
|
if "fadeMs" in data:
|
|
feedback["fadeMs"] = int(data["fadeMs"])
|
|
trigger = btn.setdefault("trigger", {})
|
|
if "triggerMode" in data:
|
|
trigger["mode"] = "click" if data.get("triggerMode") == "click" else "hold"
|
|
if "holdMs" in data:
|
|
trigger["holdMs"] = int(data["holdMs"])
|
|
if "color" in data:
|
|
btn["color"] = data.get("color") or ""
|
|
_apply_button_latch_reset(btn, data)
|
|
except HTTPException as exc:
|
|
return _invalid(exc, "invalid_button")
|
|
except (TypeError, ValueError):
|
|
return _err("invalid_button", 400)
|
|
save_config(cfg)
|
|
return jsonify(btn)
|
|
return _err("button_not_found", 404)
|
|
|
|
|
|
@rest_bp.route("/buttons/<button_id>", methods=["DELETE"])
|
|
@require_api_key
|
|
def api_button_delete(button_id):
|
|
cfg = load_config()
|
|
before = cfg.get("buttons") or []
|
|
kept = [b for b in before if not (isinstance(b, dict) and b.get("id") == button_id)]
|
|
if len(kept) == len(before):
|
|
return _err("button_not_found", 404)
|
|
cfg["buttons"] = kept
|
|
save_config(cfg)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@rest_bp.route("/background", methods=["GET"])
|
|
def api_background_get():
|
|
bg = load_config().get("background") or {}
|
|
return jsonify(
|
|
{
|
|
"type": bg.get("type") or "gradient",
|
|
"color1": bg.get("color1") or "#00007b",
|
|
"color2": bg.get("color2") or "#7b007b",
|
|
"animated": bool(bg.get("animated", True)),
|
|
"imagePath": bg.get("imagePath") or "",
|
|
}
|
|
)
|
|
|
|
|
|
@rest_bp.route("/background", methods=["PUT"])
|
|
@require_api_key
|
|
def api_background_put():
|
|
data = request.get_json(silent=True) or {}
|
|
if not isinstance(data, dict):
|
|
return _err("invalid_background", 400)
|
|
cfg = load_config()
|
|
bg = cfg.setdefault("background", {})
|
|
if "type" in data:
|
|
kind = data.get("type")
|
|
if kind not in _BG_TYPES:
|
|
return _err("invalid_background", 400)
|
|
bg["type"] = kind
|
|
for key in ("color1", "color2"):
|
|
if key not in data:
|
|
continue
|
|
color = _norm_color(data.get(key))
|
|
if color is None:
|
|
return _err("invalid_background", 400)
|
|
bg[key] = color
|
|
if "animated" in data:
|
|
if not isinstance(data.get("animated"), bool):
|
|
return _err("invalid_background", 400)
|
|
bg["animated"] = data["animated"]
|
|
if "imagePath" in data:
|
|
if not isinstance(data.get("imagePath"), str):
|
|
return _err("invalid_background", 400)
|
|
bg["imagePath"] = data["imagePath"]
|
|
save_config(cfg)
|
|
return jsonify(
|
|
{
|
|
"type": bg.get("type") or "gradient",
|
|
"color1": bg.get("color1") or "#00007b",
|
|
"color2": bg.get("color2") or "#7b007b",
|
|
"animated": bool(bg.get("animated", True)),
|
|
"imagePath": bg.get("imagePath") or "",
|
|
}
|
|
)
|
|
|
|
|
|
@rest_bp.route("/buttons/<button_id>/press", methods=["POST"])
|
|
@require_api_key
|
|
def api_button_press(button_id):
|
|
buttons = load_config().get("buttons") or []
|
|
known = any(isinstance(b, dict) and b.get("id") == button_id for b in buttons)
|
|
if not known:
|
|
return _err("button_not_found", 404)
|
|
path = run_dir() / "ui-cmd"
|
|
try:
|
|
path.write_text(
|
|
json.dumps({"op": "invoke", "id": button_id}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
except OSError as e:
|
|
return _err(str(e), 500)
|
|
return jsonify({"ok": True, "id": button_id}), 202
|
|
|
|
|
|
@rest_bp.route("/system/reboot", methods=["POST"])
|
|
@require_api_key
|
|
def api_reboot():
|
|
try:
|
|
trigger_reboot()
|
|
except OSError as e:
|
|
return _err(str(e), 500)
|
|
return jsonify({"ok": True})
|