broadlinkmanager/app/routers/devices.py
Rama master · réplica de lectura del repositorio original.
import json
import re
import subprocess
from os import path
_HOST_RE = re.compile(r"^[a-zA-Z0-9._-]{1,253}$")
import broadlink
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse
from loguru import logger
from app.config import args, discovery_ip_address_list, get_devices_file_path
router = APIRouter(tags=["Devices"])
_DEVICE_NAMES: dict[int, str] = {
0x0000: "SP1 (Broadlink)", 0x2717: "NEO (Ankuoo)", 0x2719: "SP2-compatible (Honeywell)",
0x271A: "SP2-compatible (Honeywell)", 0x2720: "SPmini (Broadlink)", 0x2728: "SP2-compatible (URANT)",
0x273E: "SPmini (Broadlink)", 0x7530: "SP2 (Broadlink OEM)", 0x7539: "SP2-IL (Broadlink OEM)",
0x753E: "SPmini3 (Broadlink)", 0x7540: "MP2 (Broadlink)", 0x7544: "SP2-CL (Broadlink)",
0x7546: "SP2-UK/BR/IN (Broadlink OEM)", 0x7547: "SC1 (Broadlink)", 0x7918: "SP2 (Broadlink OEM)",
0x7919: "SP2-compatible (Honeywell)", 0x791A: "SP2-compatible (Honeywell)",
0x7D0D: "SPmini3 (Broadlink OEM)", 0x2711: "SP2 (Broadlink)", 0x2716: "NEOPRO (Ankuoo)",
0x271D: "Ego (Efergy)", 0x2736: "SPmini+ (Broadlink)", 0x2733: "SP3 (Broadlink)",
0x7D00: "SP3-EU (Broadlink OEM)", 0x9479: "SP3S-US (Broadlink)", 0x947A: "SP3S-EU (Broadlink)",
0x7568: "SP4L-CN (Broadlink)", 0x756C: "SP4M (Broadlink)", 0x756F: "MCB1 (Broadlink)",
0x7579: "SP4L-EU (Broadlink)", 0x757B: "SP4L-AU (Broadlink)", 0x7583: "SPmini3 (Broadlink)",
0x7587: "SP4L-UK (Broadlink)", 0x7D11: "SPmini3 (Broadlink)", 0xA56A: "MCB1 (Broadlink)",
0xA56B: "SCB1E (Broadlink)", 0xA56C: "SP4L-EU (Broadlink)", 0xA589: "SP4L-UK (Broadlink)",
0xA5D3: "SP4L-EU (Broadlink)", 0x5115: "SCB1E (Broadlink)", 0x51E2: "AHC/U-01 (BGElectrical)",
0x6111: "MCB1 (Broadlink)", 0x6113: "SCB1E (Broadlink)", 0x618B: "SP4L-EU (Broadlink)",
0x6489: "SP4L-AU (Broadlink)", 0x648B: "SP4M-US (Broadlink)", 0x6494: "SCB2 (Broadlink)",
0x2737: "RMmini3 (Broadlink)", 0x278F: "RMmini (Broadlink)", 0x27C2: "RMmini3 (Broadlink)",
0x27C7: "RMmini3 (Broadlink)", 0x27CC: "RMmini3 (Broadlink)", 0x27CD: "RMmini3 (Broadlink)",
0x27D0: "RMmini3 (Broadlink)", 0x27D1: "RMmini3 (Broadlink)", 0x27D3: "RMmini3 (Broadlink)",
0x27DC: "RMmini3 (Broadlink)", 0x27DE: "RMmini3 (Broadlink)", 0x2712: "RMpro/pro+ (Broadlink)",
0x272A: "RMpro (Broadlink)", 0x273D: "RMpro (Broadlink)", 0x277C: "RMhome (Broadlink)",
0x2783: "RMhome (Broadlink)", 0x2787: "RMpro (Broadlink)", 0x278B: "RMplus (Broadlink)",
0x2797: "RMpro+ (Broadlink)", 0x279D: "RMpro+ (Broadlink)", 0x27A1: "RMplus (Broadlink)",
0x27A6: "RMplus (Broadlink)", 0x27A9: "RMpro+ (Broadlink)", 0x27C3: "RMpro+ (Broadlink)",
0x5F36: "RMmini3 (Broadlink)", 0x6507: "RMmini3 (Broadlink)", 0x6508: "RMmini3 (Broadlink)",
0x51DA: "RM4mini (Broadlink)", 0x520D: "RM4C mini (Broadlink)", 0x6070: "RM4Cmini (Broadlink)",
0x610E: "RM4mini (Broadlink)", 0x610F: "RM4Cmini (Broadlink)", 0x62BC: "RM4mini (Broadlink)",
0x62BE: "RM4Cmini (Broadlink)", 0x6364: "RM4S (Broadlink)", 0x648D: "RM4mini (Broadlink)",
0x6539: "RM4Cmini (Broadlink)", 0x653A: "RM4mini (Broadlink)", 0x6026: "RM4pro (Broadlink)",
0x6184: "RM4Cpro (Broadlink)", 0x61A2: "RM4pro (Broadlink)", 0x649B: "RM4pro (Broadlink)",
0x653C: "RM4pro (Broadlink)", 0x2714: "e-Sensor (Broadlink)", 0x5043: "SB800TD (Broadlink OEM)",
0x504E: "LB1 (Broadlink)", 0x606E: "SB500TD (Broadlink OEM)", 0x60C7: "LB1 (Broadlink)",
0x60C8: "LB1 (Broadlink)", 0x6112: "LB1 (Broadlink)", 0xA4F4: "LB27R1 (Broadlink)",
0x2722: "S2KIT (Broadlink)", 0xA59C: "S3 (Broadlink)", 0xA64D: "S3 (Broadlink)",
0x4EAD: "HY02/HY03 (Hysen)", 0x4E4D: "DT360E-45/20 (Dooya)", 0x51E3: "BG800/BG900 (BGElectrical)",
0xA569: "SP4L-UK (Broadlink)", 0x520C: "RM4 mini (Broadlink)", 0x5212: "RM4 TV mate (Broadlink)",
0x5211: "RM4C mate (Broadlink)", 0x521C: "RM4 mini (Broadlink)", 0x5216: "RM4 mini (Broadlink)",
0x520B: "RM4 pro (Broadlink)", 0x5218: "RM4C pro (Broadlink)", 0x5209: "RM4 TV mate (Broadlink)",
0x5213: "RM4 pro (Broadlink)", 0x644B: "LB1 (Broadlink)", 0x644C: "LB27 R1 (Broadlink)",
0x644E: "LB26 R1 (Broadlink)", 0xA5F7: "LB27 R1 (Broadlink)", 0x648C: "SP4L-US (Broadlink)",
}
def _get_device_name(devtype: int) -> str:
return _DEVICE_NAMES.get(devtype, "Not Supported")
def _process_device(device) -> dict | None:
try:
if not device.auth():
logger.warning(f"Auth failed for {device.host[0]}")
return None
mac = "".join(format(x, "02x") for x in device.mac)
name = _get_device_name(device.devtype)
logger.info(f"Found: {name} at {device.host[0]} ({mac})")
return {"name": name, "type": format(hex(device.devtype)), "ip": device.host[0], "mac": mac}
except Exception as e:
logger.error(f"Error processing device: {e}")
return None
@router.get("/autodiscover", tags=["Devices"])
def autodiscover(freshscan: str = "1"):
devices_path = get_devices_file_path()
if path.exists(devices_path) and freshscan != "1":
try:
with open(devices_path, "r") as f:
return JSONResponse(json.load(f))
except Exception as e:
logger.error(f"Failed to load devices file: {e}")
logger.info("Scanning for devices...")
found: list[dict] = []
for iface in discovery_ip_address_list:
try:
devices = broadlink.discover(timeout=5, local_ip_address=iface, discover_ip_address=args.dst_ip)
for d in devices:
info = _process_device(d)
if info:
found.append(info)
except OSError as e:
logger.error(f"Discovery failed on {iface}: {e}")
logger.info(f"Found {len(found)} device(s)")
return JSONResponse(found)
@router.get("/device/ping", tags=["Devices"])
def ping_device(host: str = ""):
if not host:
return JSONResponse({"status": "Host parameter is required", "success": False})
if not _HOST_RE.match(host):
logger.warning(f"Rejected invalid host value: {host!r}")
return JSONResponse({"status": "invalid host", "success": False})
try:
# "--" terminates option parsing so a host that starts with "-" is never
# interpreted as a ping flag (defence-in-depth; regex above already rejects it)
result = subprocess.run(["ping", "-c", "1", "-W", "3", "--", host], capture_output=True, timeout=5)
status = "online" if result.returncode == 0 else "offline"
return JSONResponse({"status": status, "success": True})
except subprocess.TimeoutExpired:
return JSONResponse({"status": "timeout", "success": True})
except Exception as e:
logger.error(f"Ping error for {host}: {e}")
return JSONResponse({"status": "error", "success": False})
_DEVICE_KEYS = {"name", "type", "ip", "mac"}
def _validate_device_list(data: object) -> list[dict]:
"""Validate that data is a list of device dicts with expected string fields."""
if not isinstance(data, list):
raise ValueError("Expected a list of devices")
result = []
for item in data:
if not isinstance(item, dict):
raise ValueError("Each device must be an object")
if not _DEVICE_KEYS.issubset(item.keys()):
raise ValueError(f"Device missing required keys: {_DEVICE_KEYS - item.keys()}")
result.append({k: str(item[k]) for k in _DEVICE_KEYS})
return result
@router.post("/devices/save", include_in_schema=False)
async def save_devices(request: Request):
try:
data = await request.json()
devices = _validate_device_list(data)
with open(get_devices_file_path(), "w") as f:
f.write(json.dumps(devices))
return JSONResponse({"success": 1})
except (ValueError, TypeError) as e:
logger.warning(f"Save devices rejected invalid payload: {e}")
return JSONResponse({"success": 0, "message": str(e)})
except Exception as e:
logger.error(f"Save devices failed: {e}")
return JSONResponse({"success": 0})
@router.get("/devices/load", include_in_schema=False)
def load_devices():
try:
with open(get_devices_file_path(), "r") as f:
return JSONResponse(json.load(f))
except Exception as e:
logger.error(f"Load devices failed: {e}")
return JSONResponse({"success": 0})