367 lines
16 KiB
Python
367 lines
16 KiB
Python
"""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 app import db
|
||
from app.models.board import Board
|
||
from app.models.device import Device
|
||
from app.models.tuya_device import TuyaDevice
|
||
from app.models.sonoff_device import SonoffDevice
|
||
from app.models.entity_types import (RELAY_ENTITY_TYPES, INPUT_ENTITY_TYPES,
|
||
ICON_PALETTE)
|
||
from app.services.board_service import set_relay
|
||
|
||
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
|
||
hardware_device_id = request.form.get("hardware_device_id") 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,
|
||
hardware_device_id = hardware_device_id,
|
||
)
|
||
device.app_id = Device.generate_app_id(name)
|
||
db.session.add(device)
|
||
db.session.commit()
|
||
flash(f"Device '{name}' created. You can now optionally bind it to hardware below.", "success")
|
||
return redirect(url_for("devices.edit_device", device_id=device.id))
|
||
|
||
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("/<int:device_id>/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
|
||
hardware_device_id = request.form.get("hardware_device_id") 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
|
||
device.hardware_device_id = hardware_device_id
|
||
|
||
# Regenerate app_id if name changed and app_id was based on the old name
|
||
custom_app_id = request.form.get("app_id", "").strip()
|
||
if custom_app_id:
|
||
# Accept a manually set app_id only if unique
|
||
existing = Device.query.filter(
|
||
Device.app_id == custom_app_id, Device.id != device.id
|
||
).first()
|
||
if not existing:
|
||
device.app_id = custom_app_id
|
||
elif device.app_id is None:
|
||
device.app_id = Device.generate_app_id(name, exclude_id=device.id)
|
||
|
||
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("/<int:device_id>/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("/<int:device_id>/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
|
||
|
||
# ── Sonoff eWeLink toggle ──────────────────────────────────────────────────
|
||
if device.entity_type == "sonoff":
|
||
from app.drivers.registry import registry
|
||
drv = registry.get("sonoff_ewelink")
|
||
if not drv:
|
||
return jsonify({"ok": False, "error": "Sonoff driver not available."}), 500
|
||
sd_id = device.entity_num // 100
|
||
channel = device.entity_num % 100
|
||
sd = db.session.get(SonoffDevice, sd_id)
|
||
if not sd:
|
||
return jsonify({"ok": False, "error": "Sonoff device not found."}), 404
|
||
if not sd.is_online:
|
||
return jsonify({"ok": False, "error": "Device is offline."})
|
||
current_state = sd.get_channel_state(channel)
|
||
ok = drv.set_device_channel(device.board, sd.device_id, channel, not current_state)
|
||
updated = db.session.get(Device, device_id)
|
||
error_msg = None if ok else "Cloud control failed. Device may be offline."
|
||
return jsonify({
|
||
"ok": ok,
|
||
"state": updated.current_state,
|
||
"state_label": updated.state_label,
|
||
"state_color": updated.state_color,
|
||
"error": error_msg,
|
||
})
|
||
|
||
# ── Tuya Cloud toggle ────────────────────────────────────────────────────
|
||
if device.entity_type == "tuya":
|
||
from app.drivers.registry import registry
|
||
drv = registry.get("tuya_cloud")
|
||
if not drv:
|
||
return jsonify({"ok": False, "error": "Tuya driver not available."}), 500
|
||
td_id = device.entity_num // 100
|
||
channel = (device.entity_num % 100) - 1 # 0-based index
|
||
td = db.session.get(TuyaDevice, td_id)
|
||
if not td:
|
||
return jsonify({"ok": False, "error": "Tuya device not found."}), 404
|
||
dps = td.switch_dps
|
||
if channel >= len(dps):
|
||
return jsonify({"ok": False, "error": "Invalid channel index."}), 400
|
||
dp_code = dps[channel]
|
||
ok = drv.toggle_dp(device.board, td.device_id, dp_code)
|
||
updated = db.session.get(Device, device_id)
|
||
return jsonify({
|
||
"ok": ok,
|
||
"state": updated.current_state,
|
||
"state_label": updated.state_label,
|
||
"state_color": updated.state_color,
|
||
})
|
||
|
||
# ── Generic relay toggle ─────────────────────────────────────────────────
|
||
# Optimistically write the new state to DB first (same pattern as boards.py
|
||
# toggle_relay_view) so the device card reflects the change immediately.
|
||
# poll_board() skips relay state updates for online boards by design, so
|
||
# without this the card would always show the stale pre-toggle state.
|
||
board = device.board
|
||
states = board.relay_states
|
||
current = states.get(f"relay_{device.entity_num}", False)
|
||
new_state = not current
|
||
states[f"relay_{device.entity_num}"] = new_state
|
||
board.relay_states = states
|
||
db.session.commit()
|
||
|
||
ok = set_relay(board, device.entity_num, new_state)
|
||
if not ok:
|
||
# Hardware command failed — roll back the optimistic state
|
||
states[f"relay_{device.entity_num}"] = current
|
||
board.relay_states = states
|
||
db.session.commit()
|
||
|
||
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/<int:board_id>/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)
|
||
|
||
# ── Tuya Cloud boards expose TuyaDevice rows, not relay/input channels ──
|
||
if board.board_type == "tuya_cloud":
|
||
from app.drivers.tuya_cloud.driver import KIND_ICON
|
||
tuya_devs = (
|
||
TuyaDevice.query.filter_by(board_id=board_id)
|
||
.order_by(TuyaDevice.name)
|
||
.all()
|
||
)
|
||
relays = []
|
||
for td in tuya_devs:
|
||
dps = td.switch_dps
|
||
for ch_idx, dp_code in enumerate(dps):
|
||
num = td.id * 100 + (ch_idx + 1)
|
||
state = bool(td.status.get(dp_code, False))
|
||
name = td.name if len(dps) == 1 else f"{td.name} – Ch {ch_idx + 1}"
|
||
relays.append({
|
||
"num": num,
|
||
"name": name,
|
||
"type": td.kind,
|
||
"icon": td.kind_icon_cls,
|
||
"state": state,
|
||
"state_label": "ON" if state else "OFF",
|
||
"on_color": "success",
|
||
"off_color": "secondary",
|
||
"entity_type": "tuya",
|
||
"device_id": td.device_id,
|
||
})
|
||
return jsonify({"relays": relays, "inputs": [], "board_type": "tuya_cloud"})
|
||
|
||
# ── Sonoff eWeLink boards expose SonoffDevice rows ─────────────────────────
|
||
if board.board_type == "sonoff_ewelink":
|
||
from app.models.sonoff_device import UIID_INFO
|
||
KIND_ICON_SONOFF = {
|
||
"switch": "bi-toggles",
|
||
"light": "bi-lightbulb-fill",
|
||
"fan": "bi-fan",
|
||
"sensor": "bi-thermometer-half",
|
||
}
|
||
sonoff_devs = (
|
||
SonoffDevice.query.filter_by(board_id=board_id)
|
||
.order_by(SonoffDevice.name)
|
||
.all()
|
||
)
|
||
relays = []
|
||
for sd in sonoff_devs:
|
||
uiid_meta = UIID_INFO.get(sd.uiid, {})
|
||
kind = uiid_meta.get("kind", "switch")
|
||
if kind not in ("switch", "light", "fan"):
|
||
continue # sensors / remotes are not controllable
|
||
icon = KIND_ICON_SONOFF.get(kind, "bi-toggles")
|
||
for ch in range(sd.num_channels):
|
||
num = sd.id * 100 + ch
|
||
state = sd.get_channel_state(ch)
|
||
name = sd.name if sd.num_channels == 1 else f"{sd.name} – Ch {ch + 1}"
|
||
relays.append({
|
||
"num": num,
|
||
"name": name,
|
||
"type": kind,
|
||
"icon": icon,
|
||
"state": state,
|
||
"state_label": "ON" if state else "OFF",
|
||
"on_color": "success",
|
||
"off_color": "secondary",
|
||
"entity_type": "sonoff",
|
||
"device_id": sd.device_id,
|
||
})
|
||
return jsonify({"relays": relays, "inputs": [], "board_type": "sonoff_ewelink"})
|
||
|
||
# ── Generic relay/input boards ──────────────────────────────────────────
|
||
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})
|