scheianu
fbf5802c69
- /register endpoint no longer requires verifyAPIRequest() - it is the bootstrap handshake and api_secret cannot be set before it runs - Add api_secret field to the Add Board form (hardware boards only) with a cryptographic Generate button, same as the Edit form - Save api_secret from the add-board POST so the driver can sign requests immediately after registration
383 lines
15 KiB
Python
383 lines
15 KiB
Python
"""Board management routes."""
|
||
import json
|
||
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, jsonify
|
||
from flask_login import login_required, current_user
|
||
|
||
from app import db
|
||
from app.models.board import Board
|
||
from app.services.board_service import poll_board, register_webhook, set_relay
|
||
from app.drivers.registry import registry
|
||
from flask import current_app
|
||
|
||
_NFC_DRIVER_ID = "olimex_esp32_c6_evb_pn532"
|
||
|
||
boards_bp = Blueprint("boards", __name__)
|
||
|
||
|
||
def _board_types():
|
||
"""Dynamic list read from the driver registry — auto-updates when drivers are added."""
|
||
return registry.choices()
|
||
|
||
|
||
# ── list ──────────────────────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/")
|
||
@login_required
|
||
def list_boards():
|
||
boards = Board.query.order_by(Board.name).all()
|
||
return render_template("boards/list.html", boards=boards)
|
||
|
||
|
||
# ── detail / quick controls ───────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>")
|
||
@login_required
|
||
def board_detail(board_id: int):
|
||
board = db.get_or_404(Board, board_id)
|
||
# Gateway boards have their own dedicated page
|
||
if board.board_type == "sonoff_ewelink":
|
||
return redirect(url_for("sonoff.gateway", board_id=board_id))
|
||
if board.board_type == "tuya_cloud":
|
||
return redirect(url_for("tuya.gateway", board_id=board_id))
|
||
# Refresh states from device
|
||
poll_board(current_app._get_current_object(), board_id)
|
||
board = db.session.get(Board, board_id)
|
||
return render_template("boards/detail.html", board=board)
|
||
|
||
|
||
# ── add ───────────────────────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/add", methods=["GET", "POST"])
|
||
@login_required
|
||
def add_board():
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
|
||
if request.method == "POST":
|
||
name = request.form.get("name", "").strip()
|
||
board_type = request.form.get("board_type", "olimex_esp32_c6_evb")
|
||
host = request.form.get("host", "").strip()
|
||
port = int(request.form.get("port", 80))
|
||
num_relays = int(request.form.get("num_relays", 4))
|
||
num_inputs = int(request.form.get("num_inputs", 4))
|
||
|
||
# Gateway boards don't need a real host address
|
||
is_gateway = board_type in ("sonoff_ewelink", "tuya_cloud")
|
||
if not name or (not host and not is_gateway):
|
||
flash("Name and host are required.", "danger")
|
||
return render_template("boards/add.html", board_types=_board_types(), drivers=registry.all())
|
||
|
||
if is_gateway:
|
||
if board_type == "tuya_cloud":
|
||
host = host or "openapi.tuyaeu.com"
|
||
else:
|
||
host = host or "ewelink.cloud"
|
||
num_relays = 0
|
||
num_inputs = 0
|
||
|
||
api_secret = request.form.get("api_secret", "").strip() or None
|
||
|
||
board = Board(
|
||
name=name,
|
||
board_type=board_type,
|
||
host=host,
|
||
port=port,
|
||
num_relays=num_relays,
|
||
num_inputs=num_inputs,
|
||
api_secret=api_secret,
|
||
)
|
||
db.session.add(board)
|
||
db.session.commit()
|
||
|
||
# Try to register webhook immediately
|
||
server_url = current_app.config.get("SERVER_BASE_URL", "http://localhost:5000")
|
||
register_webhook(board, server_url)
|
||
|
||
flash(f"Board '{name}' added successfully.", "success")
|
||
# Send gateway boards straight to their auth/settings page
|
||
if board_type == "sonoff_ewelink":
|
||
return redirect(url_for("sonoff.auth_settings", board_id=board.id))
|
||
if board_type == "tuya_cloud":
|
||
return redirect(url_for("tuya.auth_settings", board_id=board.id))
|
||
return redirect(url_for("boards.board_detail", board_id=board.id))
|
||
|
||
return render_template("boards/add.html", board_types=_board_types(), drivers=registry.all())
|
||
|
||
|
||
# ── edit ──────────────────────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/edit", methods=["GET", "POST"])
|
||
@login_required
|
||
def edit_board(board_id: int):
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
board = db.get_or_404(Board, board_id)
|
||
|
||
if request.method == "POST":
|
||
board.name = request.form.get("name", board.name).strip()
|
||
board.host = request.form.get("host", board.host).strip()
|
||
board.port = int(request.form.get("port", board.port))
|
||
board.board_type = request.form.get("board_type", board.board_type)
|
||
board.num_relays = int(request.form.get("num_relays", board.num_relays))
|
||
board.num_inputs = int(request.form.get("num_inputs", board.num_inputs))
|
||
|
||
# Update labels
|
||
labels = {}
|
||
for n in range(1, board.num_relays + 1):
|
||
lbl = request.form.get(f"relay_{n}_label", "").strip()
|
||
if lbl:
|
||
labels[f"relay_{n}"] = lbl
|
||
for n in range(1, board.num_inputs + 1):
|
||
lbl = request.form.get(f"input_{n}_label", "").strip()
|
||
if lbl:
|
||
labels[f"input_{n}"] = lbl
|
||
board.labels = labels
|
||
board.api_secret = request.form.get("api_secret", "").strip() or None
|
||
|
||
db.session.commit()
|
||
flash("Board updated.", "success")
|
||
return redirect(url_for("boards.board_detail", board_id=board.id))
|
||
|
||
return render_template("boards/edit.html", board=board, board_types=_board_types())
|
||
|
||
|
||
# ── delete ────────────────────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/delete", methods=["POST"])
|
||
@login_required
|
||
def delete_board(board_id: int):
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
board = db.get_or_404(Board, board_id)
|
||
db.session.delete(board)
|
||
db.session.commit()
|
||
flash(f"Board '{board.name}' deleted.", "warning")
|
||
return redirect(url_for("boards.list_boards"))
|
||
|
||
|
||
# ── quick relay toggle ───────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/relay/<int:relay_num>/toggle", methods=["POST"])
|
||
@login_required
|
||
def toggle_relay_view(board_id: int, relay_num: int):
|
||
from app import socketio
|
||
board = db.get_or_404(Board, board_id)
|
||
|
||
# Always flip the local cached state first (optimistic update)
|
||
states = board.relay_states
|
||
current = states.get(f"relay_{relay_num}", False)
|
||
new_state = not current
|
||
states[f"relay_{relay_num}"] = new_state
|
||
board.relay_states = states
|
||
db.session.commit()
|
||
|
||
# Best-effort: send the command to the physical board using set_relay
|
||
# (uses /relay/on or /relay/off — same endpoints as the detail page ON/OFF buttons)
|
||
hw_ok = set_relay(board, relay_num, new_state)
|
||
|
||
# Push live update to all clients
|
||
socketio.emit("board_update", {
|
||
"board_id": board_id,
|
||
"is_online": board.is_online,
|
||
"relay_states": board.relay_states,
|
||
"input_states": board.input_states,
|
||
})
|
||
|
||
label = board.get_relay_label(relay_num)
|
||
status_text = "ON" if new_state else "OFF"
|
||
hw_warning = not hw_ok # True when board was unreachable
|
||
|
||
# JSON response for AJAX callers (dashboard)
|
||
if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
|
||
return jsonify({
|
||
"relay": relay_num,
|
||
"state": new_state,
|
||
"label": label,
|
||
"hw_ok": not hw_warning,
|
||
})
|
||
|
||
# HTML response for form-submit callers (detail page)
|
||
if hw_warning:
|
||
flash(f"{label}: {status_text} (board unreachable — local state updated)", "warning")
|
||
else:
|
||
flash(f"{label}: {status_text}", "info")
|
||
return redirect(url_for("boards.board_detail", board_id=board_id))
|
||
|
||
|
||
# ── quick relay set ───────────────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/relay/<int:relay_num>/set", methods=["POST"])
|
||
@login_required
|
||
def set_relay_view(board_id: int, relay_num: int):
|
||
from app import socketio
|
||
board = db.get_or_404(Board, board_id)
|
||
state = request.form.get("state", "off") == "on"
|
||
|
||
# Always update local state
|
||
states = board.relay_states
|
||
states[f"relay_{relay_num}"] = state
|
||
board.relay_states = states
|
||
db.session.commit()
|
||
|
||
# Best-effort send to hardware
|
||
hw_ok = set_relay(board, relay_num, state)
|
||
|
||
socketio.emit("board_update", {
|
||
"board_id": board_id,
|
||
"is_online": board.is_online,
|
||
"relay_states": board.relay_states,
|
||
"input_states": board.input_states,
|
||
})
|
||
|
||
label = board.get_relay_label(relay_num)
|
||
status_text = "ON" if state else "OFF"
|
||
if not hw_ok:
|
||
flash(f"{label}: {status_text} (board unreachable — local state updated)", "warning")
|
||
else:
|
||
flash(f"{label}: {status_text}", "info")
|
||
return redirect(url_for("boards.board_detail", board_id=board_id))
|
||
|
||
|
||
# ── edit entity configuration ─────────────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/entities", methods=["GET", "POST"])
|
||
@login_required
|
||
def edit_entities(board_id: int):
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
board = db.get_or_404(Board, board_id)
|
||
|
||
if request.method == "POST":
|
||
entities = {}
|
||
for n in range(1, board.num_relays + 1):
|
||
entities[f"relay_{n}"] = {
|
||
"type": request.form.get(f"relay_{n}_type", "switch"),
|
||
"name": request.form.get(f"relay_{n}_name", "").strip()[:20],
|
||
"icon": request.form.get(f"relay_{n}_icon", "").strip(),
|
||
}
|
||
for n in range(1, board.num_inputs + 1):
|
||
entities[f"input_{n}"] = {
|
||
"type": request.form.get(f"input_{n}_type", "generic"),
|
||
"name": request.form.get(f"input_{n}_name", "").strip()[:20],
|
||
"icon": request.form.get(f"input_{n}_icon", "").strip(),
|
||
}
|
||
board.entities = entities
|
||
db.session.commit()
|
||
flash("Entity configuration saved.", "success")
|
||
return redirect(url_for("boards.board_detail", board_id=board_id))
|
||
|
||
from app.models.entity_types import RELAY_ENTITY_TYPES, INPUT_ENTITY_TYPES, ICON_PALETTE
|
||
return render_template(
|
||
"boards/edit_entities.html",
|
||
board=board,
|
||
relay_types=RELAY_ENTITY_TYPES,
|
||
input_types=INPUT_ENTITY_TYPES,
|
||
icon_palette=ICON_PALETTE,
|
||
)
|
||
|
||
|
||
# ── edit labels (legacy — redirects to edit_entities) ─────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/labels", methods=["GET", "POST"])
|
||
@login_required
|
||
def edit_labels(board_id: int):
|
||
return redirect(url_for("boards.edit_entities", board_id=board_id))
|
||
|
||
|
||
# ── NFC management (PN532 boards only) ────────────────────────────────────────
|
||
|
||
@boards_bp.route("/<int:board_id>/nfc")
|
||
@login_required
|
||
def nfc_management(board_id: int):
|
||
board = db.get_or_404(Board, board_id)
|
||
if board.board_type != _NFC_DRIVER_ID:
|
||
abort(404)
|
||
driver = registry.get(_NFC_DRIVER_ID)
|
||
nfc_status = driver.get_nfc_status(board) if driver else None
|
||
return render_template(
|
||
"boards/nfc.html",
|
||
board=board,
|
||
nfc=nfc_status or {},
|
||
)
|
||
|
||
|
||
@boards_bp.route("/<int:board_id>/nfc/status_json")
|
||
@login_required
|
||
def nfc_status_json(board_id: int):
|
||
board = db.get_or_404(Board, board_id)
|
||
if board.board_type != _NFC_DRIVER_ID:
|
||
abort(404)
|
||
driver = registry.get(_NFC_DRIVER_ID)
|
||
data = driver.get_nfc_status(board) if driver else None
|
||
if data is None:
|
||
return jsonify({"error": "Board unreachable"}), 502
|
||
return jsonify(data)
|
||
|
||
|
||
@boards_bp.route("/<int:board_id>/nfc/config", methods=["POST"])
|
||
@login_required
|
||
def nfc_config_save(board_id: int):
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
board = db.get_or_404(Board, board_id)
|
||
if board.board_type != _NFC_DRIVER_ID:
|
||
abort(404)
|
||
driver = registry.get(_NFC_DRIVER_ID)
|
||
if not driver:
|
||
flash("NFC driver not available.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
auth_uid = request.form.get("auth_uid", "").strip().upper()
|
||
relay_num = int(request.form.get("relay_num", 1))
|
||
pulse_ms = int(request.form.get("pulse_ms", 3000))
|
||
|
||
if relay_num < 1 or relay_num > 4:
|
||
flash("Relay number must be 1–4.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
if pulse_ms < 100 or pulse_ms > 60000:
|
||
flash("Absence timeout must be between 100 and 60 000 ms.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
ok = driver.set_nfc_config(board, auth_uid=auth_uid, relay_num=relay_num, pulse_ms=pulse_ms)
|
||
if ok:
|
||
uid_display = auth_uid if auth_uid else "(any card)"
|
||
flash(f"NFC config saved — authorized UID: {uid_display}, relay: {relay_num}, timeout: {pulse_ms} ms", "success")
|
||
else:
|
||
flash("Failed to push NFC config — board unreachable.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
|
||
@boards_bp.route("/<int:board_id>/nfc/enroll", methods=["POST"])
|
||
@login_required
|
||
def nfc_enroll(board_id: int):
|
||
"""Read the last-seen UID from the board and immediately authorize it."""
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
board = db.get_or_404(Board, board_id)
|
||
if board.board_type != _NFC_DRIVER_ID:
|
||
abort(404)
|
||
driver = registry.get(_NFC_DRIVER_ID)
|
||
if not driver:
|
||
flash("NFC driver not available.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
status = driver.get_nfc_status(board)
|
||
if not status:
|
||
flash("Board unreachable — could not read card UID.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
uid = (status.get("last_uid") or "").strip().upper()
|
||
if not uid:
|
||
flash("No card has been presented to the reader yet.", "warning")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|
||
|
||
relay_num = int(request.form.get("relay_num", status.get("relay_num", 1)))
|
||
pulse_ms = int(request.form.get("pulse_ms", status.get("pulse_ms", 3000)))
|
||
|
||
ok = driver.set_nfc_config(board, auth_uid=uid, relay_num=relay_num, pulse_ms=pulse_ms)
|
||
if ok:
|
||
flash(f"Card enrolled — UID: {uid}, relay: {relay_num}, timeout: {pulse_ms} ms", "success")
|
||
else:
|
||
flash("Card read OK but failed to push config to board.", "danger")
|
||
return redirect(url_for("boards.nfc_management", board_id=board_id))
|