Button/winDeployScripts/software/webconfig/sensors.py

#!/usr/bin/env python3
"""BLE sensor status / log / names / settings (collector UI bridge)."""
import json

from flask import Blueprint, jsonify, request

from auth import login_required
from paths import install_root, run_dir

sensors_bp = Blueprint("sensors", __name__)

STATUS_NAME = "sensors-status.json"
LOG_NAME = "sensors-ui.log"
CMD_NAME = "sensors-cmd"
NAMES_NAME = "sensors-names.json"
SETTINGS_NAME = "sensors-settings.json"
LOG_LIMIT = 200
DEFAULT_INTERVAL_MIN = 30


def _config_dir():
    p = install_root() / "config"
    try:
        p.mkdir(parents=True, exist_ok=True)
    except OSError:
        pass
    return p


def _normalize_addr(addr: str) -> str:
    return (addr or "").strip().upper()


def _read_status():
    path = run_dir() / STATUS_NAME
    if not path.exists():
        return {
            "n": 4,
            "pool_full": False,
            "period_s": DEFAULT_INTERVAL_MIN * 60,
            "report_interval_min": DEFAULT_INTERVAL_MIN,
            "slots": [],
            "ts": 0,
            "available": False,
        }
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        data["available"] = True
        if "report_interval_min" not in data and "period_s" in data:
            try:
                data["report_interval_min"] = max(1, int(data["period_s"]) // 60)
            except (TypeError, ValueError):
                data["report_interval_min"] = DEFAULT_INTERVAL_MIN
        return data
    except (OSError, ValueError):
        return {"available": False, "slots": [], "error": "bad status file"}


def _read_log(limit=LOG_LIMIT):
    path = run_dir() / LOG_NAME
    if not path.exists():
        return []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError:
        return []
    entries = []
    for line in lines[-limit:]:
        line = line.strip()
        if not line:
            continue
        try:
            entries.append(json.loads(line))
        except ValueError:
            continue
    entries.reverse()
    return entries


def _write_cmd(payload: dict) -> bool:
    path = run_dir() / CMD_NAME
    try:
        path.write_text(json.dumps(payload), encoding="utf-8")
        return True
    except OSError:
        return False


def _load_names():
    path = _config_dir() / NAMES_NAME
    if not path.exists():
        return {}
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    if not isinstance(raw, dict):
        return {}
    out = {}
    for k, v in raw.items():
        addr = _normalize_addr(str(k))
        name = str(v).strip()
        if addr and name:
            out[addr] = name
    return out


def _save_names(names: dict) -> dict:
    cleaned = {}
    for k, v in (names or {}).items():
        addr = _normalize_addr(str(k))
        name = str(v).strip()
        if addr and name:
            cleaned[addr] = name
    path = _config_dir() / NAMES_NAME
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(cleaned, indent=2, ensure_ascii=False), encoding="utf-8")
    tmp.replace(path)
    return cleaned


def _load_settings():
    path = _config_dir() / SETTINGS_NAME
    if not path.exists():
        return {"report_interval_min": DEFAULT_INTERVAL_MIN}
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
        v = int(raw.get("report_interval_min", DEFAULT_INTERVAL_MIN))
    except (OSError, ValueError, TypeError):
        return {"report_interval_min": DEFAULT_INTERVAL_MIN}
    return {"report_interval_min": max(1, min(65535, v))}


def _save_settings(minutes: int) -> dict:
    v = max(1, min(65535, int(minutes)))
    data = {"report_interval_min": v}
    path = _config_dir() / SETTINGS_NAME
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2), encoding="utf-8")
    tmp.replace(path)
    _write_cmd({"op": "reload_settings", "report_interval_min": v})
    return data


@sensors_bp.route("/api/sensors", methods=["GET"])
@login_required
def api_sensors():
    return jsonify(_read_status())


@sensors_bp.route("/api/sensors/log", methods=["GET"])
@login_required
def api_sensors_log():
    return jsonify({"entries": _read_log()})


@sensors_bp.route("/api/sensors/advertise", methods=["POST"])
@login_required
def api_sensors_advertise():
    if not _write_cmd({"op": "advertise"}):
        return jsonify({"ok": False, "error": "cannot write command"}), 500
    return jsonify({"ok": True})


@sensors_bp.route("/api/sensors/names", methods=["GET"])
@login_required
def api_sensors_names_get():
    return jsonify({"names": _load_names()})


@sensors_bp.route("/api/sensors/names", methods=["PUT"])
@login_required
def api_sensors_names_put():
    body = request.get_json(silent=True) or {}
    names = body.get("names", body)
    if not isinstance(names, dict):
        return jsonify({"ok": False, "error": "names must be an object"}), 400
    try:
        cleaned = _save_names(names)
    except OSError as e:
        return jsonify({"ok": False, "error": str(e)}), 500
    _write_cmd({"op": "reload_settings"})
    return jsonify({"ok": True, "names": cleaned})


@sensors_bp.route("/api/sensors/settings", methods=["GET"])
@login_required
def api_sensors_settings_get():
    return jsonify(_load_settings())


@sensors_bp.route("/api/sensors/settings", methods=["PUT"])
@login_required
def api_sensors_settings_put():
    body = request.get_json(silent=True) or {}
    try:
        minutes = int(body.get("report_interval_min", DEFAULT_INTERVAL_MIN))
    except (TypeError, ValueError):
        return jsonify({"ok": False, "error": "report_interval_min must be int"}), 400
    if minutes < 1 or minutes > 65535:
        return jsonify({"ok": False, "error": "report_interval_min out of range"}), 400
    try:
        data = _save_settings(minutes)
    except OSError as e:
        return jsonify({"ok": False, "error": str(e)}), 500
    return jsonify({"ok": True, **data})