"""Devices module routes. Devices are user-defined friendly aliases for board relay/input entities. E.g. Board "Garage" / Relay 4 → "Outdoor Light 1 (Courtyard)". """ import json from flask import (Blueprint, render_template, redirect, url_for, flash, request, abort, jsonify) from flask_login import login_required, current_user from flask import current_app from app import db from app.models.board import Board from app.models.device import Device from app.models.entity_types import (RELAY_ENTITY_TYPES, INPUT_ENTITY_TYPES, ICON_PALETTE) from app.services.board_service import set_relay, poll_board devices_bp = Blueprint("devices", __name__) # ── helpers ─────────────────────────────────────────────────────────────────── def _class_choices(entity_type: str) -> list[tuple[str, str]]: """Return (value, label) pairs for the device_class selector.""" if entity_type == "relay": return [(k, v["label"]) for k, v in RELAY_ENTITY_TYPES.items()] return [(k, v["label"]) for k, v in INPUT_ENTITY_TYPES.items()] # ── list ────────────────────────────────────────────────────────────────────── @devices_bp.route("/") @login_required def list_devices(): devices = Device.query.order_by(Device.area, Device.name).all() return render_template("devices/list.html", devices=devices) # ── add ─────────────────────────────────────────────────────────────────────── @devices_bp.route("/add", methods=["GET", "POST"]) @login_required def add_device(): if not current_user.is_admin(): abort(403) boards = Board.query.order_by(Board.name).all() if request.method == "POST": name = request.form.get("name", "").strip() if not name: flash("Name is required.", "danger") return render_template( "devices/edit.html", device=None, boards=boards, relay_types=RELAY_ENTITY_TYPES, input_types=INPUT_ENTITY_TYPES, icon_palette=ICON_PALETTE, ) board_id = request.form.get("board_id") or None entity_type = request.form.get("entity_type") or None entity_num = request.form.get("entity_num") or None device = Device( name = name, description = request.form.get("description", "").strip() or None, area = request.form.get("area", "").strip() or None, device_class= request.form.get("device_class", "switch"), icon = request.form.get("icon", "").strip() or None, board_id = int(board_id) if board_id else None, entity_type = entity_type, entity_num = int(entity_num) if entity_num else None, ) db.session.add(device) db.session.commit() flash(f"Device '{name}' created successfully.", "success") return redirect(url_for("devices.list_devices")) return render_template( "devices/edit.html", device=None, boards=boards, relay_types=RELAY_ENTITY_TYPES, input_types=INPUT_ENTITY_TYPES, icon_palette=ICON_PALETTE, ) # ── edit ────────────────────────────────────────────────────────────────────── @devices_bp.route("//edit", methods=["GET", "POST"]) @login_required def edit_device(device_id: int): if not current_user.is_admin(): abort(403) device = db.get_or_404(Device, device_id) boards = Board.query.order_by(Board.name).all() if request.method == "POST": name = request.form.get("name", "").strip() if not name: flash("Name is required.", "danger") return render_template( "devices/edit.html", device=device, boards=boards, relay_types=RELAY_ENTITY_TYPES, input_types=INPUT_ENTITY_TYPES, icon_palette=ICON_PALETTE, ) board_id = request.form.get("board_id") or None entity_type = request.form.get("entity_type") or None entity_num = request.form.get("entity_num") or None device.name = name device.description = request.form.get("description", "").strip() or None device.area = request.form.get("area", "").strip() or None device.device_class = request.form.get("device_class", device.device_class) device.icon = request.form.get("icon", "").strip() or None device.board_id = int(board_id) if board_id else None device.entity_type = entity_type device.entity_num = int(entity_num) if entity_num else None db.session.commit() flash(f"Device '{name}' updated.", "success") return redirect(url_for("devices.list_devices")) return render_template( "devices/edit.html", device=device, boards=boards, relay_types=RELAY_ENTITY_TYPES, input_types=INPUT_ENTITY_TYPES, icon_palette=ICON_PALETTE, ) # ── delete ──────────────────────────────────────────────────────────────────── @devices_bp.route("//delete", methods=["POST"]) @login_required def delete_device(device_id: int): if not current_user.is_admin(): abort(403) device = db.get_or_404(Device, device_id) name = device.name db.session.delete(device) db.session.commit() flash(f"Device '{name}' deleted.", "warning") return redirect(url_for("devices.list_devices")) # ── toggle relay (AJAX) ─────────────────────────────────────────────────────── @devices_bp.route("//toggle", methods=["POST"]) @login_required def toggle_device(device_id: int): device = db.get_or_404(Device, device_id) if not device.is_controllable: return jsonify({"ok": False, "error": "Device is not controllable."}), 400 if not device.board: return jsonify({"ok": False, "error": "No board linked."}), 400 # Determine new state (toggle) current = device.board.relay_states.get(f"relay_{device.entity_num}", False) new_state = not current app = current_app._get_current_object() ok = set_relay(device.board, device.entity_num, new_state) # Re-read updated relay state poll_board(app, device.board_id) updated_device = db.session.get(Device, device_id) return jsonify({ "ok": ok, "state": updated_device.current_state, "state_label": updated_device.state_label, "state_color": updated_device.state_color, }) # ── API: board entity info (for dynamic form population) ───────────────────── @devices_bp.route("/api/boards//entities") @login_required def board_entities(board_id: int): """Return relay and input entity info for the given board as JSON.""" board = db.get_or_404(Board, board_id) relays = [] for n in range(1, board.num_relays + 1): e = board.get_relay_entity(n) state = board.relay_states.get(f"relay_{n}", False) relays.append({ "num": n, "name": e["name"], "type": e["type"], "icon": e["icon"], "state": state, "state_label": e["on_label"] if state else e["off_label"], "on_color": e["on_color"], "off_color": e["off_color"], }) inputs = [] for n in range(1, board.num_inputs + 1): e = board.get_input_entity(n) raw = board.input_states.get(f"input_{n}", True) active = not raw inputs.append({ "num": n, "name": e["name"], "type": e["type"], "icon": e["icon"], "active": active, "state_label": e["active_label"] if active else e["idle_label"], "active_color": e["active_color"], "idle_color": e["idle_color"], }) return jsonify({"relays": relays, "inputs": inputs})