466 lines
15 KiB
Python
466 lines
15 KiB
Python
from microdot import Microdot
|
|
from models.pattern import Pattern
|
|
from models.device import Device
|
|
from models.tcp_clients import send_json_line_to_ip
|
|
from util.driver_patterns import (
|
|
driver_patterns_dir,
|
|
is_firmware_builtin_pattern_module,
|
|
normalize_pattern_py_filename,
|
|
)
|
|
import json
|
|
import re
|
|
import sys
|
|
import os
|
|
|
|
controller = Microdot()
|
|
patterns = Pattern()
|
|
|
|
|
|
def _project_root():
|
|
"""Project root (parent of ``src/``). CWD is often ``src/`` when running ``main.py``."""
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
return os.path.abspath(os.path.join(here, "..", ".."))
|
|
|
|
|
|
def _safe_pattern_filename(name):
|
|
if not isinstance(name, str):
|
|
return False
|
|
if not name.endswith(".py"):
|
|
return False
|
|
if "/" in name or "\\" in name or ".." in name:
|
|
return False
|
|
return True
|
|
|
|
|
|
_PATTERN_KEY_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
|
|
|
|
|
|
def _normalize_pattern_key(raw):
|
|
"""Pattern id / module basename (no .py)."""
|
|
if not isinstance(raw, str):
|
|
return ""
|
|
s = raw.strip()
|
|
if s.lower().endswith(".py"):
|
|
s = s[:-3].strip()
|
|
return s
|
|
|
|
|
|
def _valid_pattern_key(key):
|
|
return bool(key and _PATTERN_KEY_RE.match(key))
|
|
|
|
def load_pattern_definitions():
|
|
"""Load pattern definitions from pattern.json file."""
|
|
try:
|
|
root = _project_root()
|
|
paths = [
|
|
os.path.join(root, "db", "pattern.json"),
|
|
os.path.join(root, "pattern.json"),
|
|
"db/pattern.json",
|
|
"pattern.json",
|
|
"/db/pattern.json",
|
|
]
|
|
for path in paths:
|
|
try:
|
|
with open(path, "r") as f:
|
|
return json.load(f)
|
|
except OSError:
|
|
continue
|
|
return {}
|
|
except Exception as e:
|
|
print(f"Error loading pattern.json: {e}")
|
|
return {}
|
|
|
|
|
|
def load_driver_pattern_names():
|
|
"""List available pattern module names from led-driver/src/patterns."""
|
|
try:
|
|
names = []
|
|
for filename in os.listdir(driver_patterns_dir()):
|
|
if not _safe_pattern_filename(filename) or filename == "__init__.py":
|
|
continue
|
|
names.append(filename[:-3])
|
|
names.sort()
|
|
return names
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def build_runtime_pattern_map():
|
|
"""
|
|
Runtime pattern map for UI menus.
|
|
Keep pattern DB metadata as primary, then add any local driver pattern files
|
|
missing from the DB so new OTA files still appear in menus.
|
|
"""
|
|
definitions = load_pattern_definitions()
|
|
available = load_driver_pattern_names()
|
|
result = {}
|
|
for name, meta in definitions.items():
|
|
result[name] = dict(meta) if isinstance(meta, dict) else {}
|
|
for name in available:
|
|
if name not in result:
|
|
result[name] = {}
|
|
return result
|
|
|
|
@controller.get('/definitions')
|
|
async def get_pattern_definitions(request):
|
|
"""Get definitions for patterns currently available on the driver."""
|
|
definitions = build_runtime_pattern_map()
|
|
return json.dumps(definitions), 200, {'Content-Type': 'application/json'}
|
|
|
|
|
|
@controller.get('/ota/manifest')
|
|
async def ota_manifest(request):
|
|
"""Manifest of driver pattern source files for OTA pulls."""
|
|
base_dir = driver_patterns_dir()
|
|
host = request.headers.get("Host", "")
|
|
if not host:
|
|
return json.dumps({"error": "Missing Host header"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
try:
|
|
names = sorted(os.listdir(base_dir))
|
|
except OSError as e:
|
|
return json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"}
|
|
|
|
files = []
|
|
for name in names:
|
|
if not _safe_pattern_filename(name) or name == "__init__.py":
|
|
continue
|
|
files.append({
|
|
"name": name,
|
|
"url": "http://%s/patterns/ota/file/%s" % (host, name),
|
|
})
|
|
|
|
return json.dumps({"files": files}), 200, {"Content-Type": "application/json"}
|
|
|
|
|
|
@controller.get('/ota/file/<name>')
|
|
async def ota_pattern_file(request, name):
|
|
"""Serve one driver pattern source file for OTA pulls."""
|
|
fname = normalize_pattern_py_filename(name)
|
|
if not fname or not _safe_pattern_filename(fname) or fname == "__init__.py":
|
|
return json.dumps({"error": "Invalid filename"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if is_firmware_builtin_pattern_module(fname):
|
|
return json.dumps(
|
|
{
|
|
"error": "on and off are built into the driver firmware; there is no module file to serve.",
|
|
}
|
|
), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
base = driver_patterns_dir()
|
|
path = os.path.join(base, fname)
|
|
try:
|
|
with open(path, "r") as f:
|
|
content = f.read()
|
|
except OSError:
|
|
return json.dumps(
|
|
{
|
|
"error": "Pattern file not found",
|
|
"path": path,
|
|
"hint": "Ensure led-driver is present or set LED_CONTROLLER_PATTERNS_DIR.",
|
|
}
|
|
), 404, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
return content, 200, {"Content-Type": "text/plain; charset=utf-8"}
|
|
|
|
|
|
@controller.post('/<name>/send')
|
|
async def send_pattern_to_device(request, name):
|
|
"""Tell Wi-Fi driver(s) to download one pattern source file over HTTP."""
|
|
if not isinstance(name, str):
|
|
return json.dumps({"error": "Invalid pattern name"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
filename = normalize_pattern_py_filename(name)
|
|
if not filename or not _safe_pattern_filename(filename) or filename == "__init__.py":
|
|
return json.dumps({"error": "Invalid pattern filename"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if is_firmware_builtin_pattern_module(filename):
|
|
return json.dumps(
|
|
{
|
|
"error": "on and off are built into the driver firmware; OTA send does not apply.",
|
|
}
|
|
), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
devices = Device()
|
|
body = request.json or {}
|
|
requested_device_id = str(body.get("device_id") or "").strip()
|
|
|
|
base = driver_patterns_dir()
|
|
path = os.path.join(base, filename)
|
|
if not os.path.exists(path):
|
|
return json.dumps(
|
|
{
|
|
"error": "Pattern file not found",
|
|
"path": path,
|
|
"hint": "Ensure led-driver is present or set LED_CONTROLLER_PATTERNS_DIR.",
|
|
}
|
|
), 404, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
file_url = "/patterns/ota/file/%s" % filename
|
|
|
|
msg = json.dumps(
|
|
{
|
|
"v": "1",
|
|
"manifest": {
|
|
"files": [
|
|
{
|
|
"name": filename,
|
|
"url": file_url,
|
|
}
|
|
]
|
|
},
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
target_ids = []
|
|
if requested_device_id:
|
|
dev = devices.read(requested_device_id)
|
|
if not dev:
|
|
return json.dumps({"error": "Device not found"}), 404, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if (dev.get("transport") or "").lower() != "wifi":
|
|
return json.dumps({"error": "Pattern send is only supported for Wi-Fi devices"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
target_ids = [requested_device_id]
|
|
else:
|
|
for did in devices.list():
|
|
dev = devices.read(did) or {}
|
|
if (dev.get("transport") or "").lower() == "wifi":
|
|
target_ids.append(str(did))
|
|
if not target_ids:
|
|
return json.dumps({"error": "No Wi-Fi devices found"}), 404, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
sent_ids = []
|
|
for did in target_ids:
|
|
dev = devices.read(did) or {}
|
|
ip = str(dev.get("address") or "").strip()
|
|
if not ip:
|
|
continue
|
|
ok = await send_json_line_to_ip(ip, msg)
|
|
if ok:
|
|
sent_ids.append(did)
|
|
|
|
if not sent_ids:
|
|
return json.dumps({"error": "No Wi-Fi drivers connected"}), 503, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
return json.dumps({"message": "Pattern sent", "pattern": filename, "device_ids": sent_ids, "sent_count": len(sent_ids)}), 200, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
|
|
@controller.post('/upload')
|
|
async def upload_pattern_file(request):
|
|
"""
|
|
Upload a pattern source file to led-controller local storage.
|
|
|
|
Body JSON:
|
|
{
|
|
"name": "sparkle.py" | "sparkle",
|
|
"code": "class Sparkle: ...",
|
|
"overwrite": true | false # optional, default true
|
|
}
|
|
"""
|
|
data = request.json or {}
|
|
raw_name = data.get("name") or data.get("filename")
|
|
code = data.get("code")
|
|
overwrite = data.get("overwrite", True)
|
|
overwrite = bool(overwrite)
|
|
|
|
if not isinstance(raw_name, str) or not raw_name.strip():
|
|
return json.dumps({"error": "name is required"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
filename = raw_name.strip()
|
|
if not filename.endswith(".py"):
|
|
filename += ".py"
|
|
if not _safe_pattern_filename(filename) or filename == "__init__.py":
|
|
return json.dumps({"error": "invalid pattern filename"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if is_firmware_builtin_pattern_module(filename):
|
|
return json.dumps(
|
|
{"error": "on and off are built into the driver firmware; use a different pattern name."}
|
|
), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
if not isinstance(code, str) or not code.strip():
|
|
return json.dumps({"error": "code is required"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
path = os.path.join(driver_patterns_dir(), filename)
|
|
exists = os.path.exists(path)
|
|
if exists and not overwrite:
|
|
return json.dumps({"error": "pattern file already exists", "name": filename}), 409, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
try:
|
|
with open(path, "w") as f:
|
|
f.write(code)
|
|
except OSError as e:
|
|
return json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"}
|
|
|
|
return json.dumps({
|
|
"message": "Pattern uploaded",
|
|
"name": filename,
|
|
"overwrote": bool(exists),
|
|
}), 201, {"Content-Type": "application/json"}
|
|
|
|
|
|
@controller.post('/driver')
|
|
async def create_driver_pattern(request):
|
|
"""
|
|
Create a driver pattern: save ``.py`` under led-driver/src/patterns and
|
|
metadata in db/pattern.json (Pattern model).
|
|
|
|
Body JSON:
|
|
name, code (required),
|
|
min_delay, max_delay, max_colors (optional numbers),
|
|
n1..n8 (optional string labels),
|
|
overwrite (optional, default true).
|
|
"""
|
|
data = request.json or {}
|
|
key = _normalize_pattern_key(data.get("name") or "")
|
|
if not _valid_pattern_key(key):
|
|
return json.dumps({
|
|
"error": "name must be a valid Python identifier (e.g. sparkle, my_pattern)",
|
|
}), 400, {"Content-Type": "application/json"}
|
|
if is_firmware_builtin_pattern_module(key):
|
|
return json.dumps(
|
|
{"error": "on and off are built into the driver firmware; use a different pattern name."}
|
|
), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
code = data.get("code")
|
|
if not isinstance(code, str) or not code.strip():
|
|
return json.dumps({"error": "code is required (upload a .py file or paste source)"}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
overwrite = bool(data.get("overwrite", True))
|
|
|
|
filename = key + ".py"
|
|
py_path = os.path.join(driver_patterns_dir(), filename)
|
|
if os.path.exists(py_path) and not overwrite:
|
|
return json.dumps({"error": "pattern file already exists", "name": filename}), 409, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
meta = {}
|
|
for fld in ("min_delay", "max_delay", "max_colors"):
|
|
if fld not in data:
|
|
continue
|
|
try:
|
|
meta[fld] = int(data[fld])
|
|
except (TypeError, ValueError):
|
|
return json.dumps({"error": "%s must be an integer" % fld}), 400, {
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
for i in range(1, 9):
|
|
nk = "n%d" % i
|
|
if nk not in data:
|
|
continue
|
|
lab = data[nk]
|
|
if lab is None:
|
|
continue
|
|
s = str(lab).strip()
|
|
if s:
|
|
meta[nk] = s
|
|
|
|
try:
|
|
with open(py_path, "w") as f:
|
|
f.write(code)
|
|
except OSError as e:
|
|
return json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"}
|
|
|
|
if patterns.read(key):
|
|
patterns.update(key, meta)
|
|
else:
|
|
patterns.create(key, meta)
|
|
|
|
return json.dumps({
|
|
"message": "Pattern created",
|
|
"name": key,
|
|
"file": filename,
|
|
"metadata": patterns.read(key),
|
|
}), 201, {"Content-Type": "application/json"}
|
|
|
|
|
|
@controller.get('')
|
|
async def list_patterns(request):
|
|
"""List patterns for UI (DB metadata + local driver additions)."""
|
|
return json.dumps(build_runtime_pattern_map()), 200, {'Content-Type': 'application/json'}
|
|
|
|
|
|
@controller.get('/<id>')
|
|
async def get_pattern(request, id):
|
|
"""Get a specific pattern by ID."""
|
|
pattern = patterns.read(id)
|
|
if pattern is not None:
|
|
return json.dumps(pattern), 200, {'Content-Type': 'application/json'}
|
|
return json.dumps({"error": "Pattern not found"}), 404
|
|
|
|
|
|
@controller.post('')
|
|
async def create_pattern(request):
|
|
"""Create a new pattern."""
|
|
try:
|
|
payload = request.json or {}
|
|
name = payload.get("name", "")
|
|
pattern_data = payload.get("data", {})
|
|
|
|
# IMPORTANT:
|
|
# `patterns.create()` stores `pattern_data` as the underlying dict value.
|
|
# If we then call `patterns.update(pattern_id, payload)` with the full
|
|
# request object, it may assign `payload["data"]` back onto that same
|
|
# dict object, creating a circular reference (json.dumps fails).
|
|
pattern_id = patterns.create(name, pattern_data)
|
|
|
|
# Only merge "extra" metadata fields (anything except name/data).
|
|
extra = dict(payload)
|
|
extra.pop("name", None)
|
|
extra.pop("data", None)
|
|
if extra:
|
|
patterns.update(pattern_id, extra)
|
|
return json.dumps(patterns.read(pattern_id)), 201, {'Content-Type': 'application/json'}
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)}), 400
|
|
|
|
|
|
@controller.put('/<id>')
|
|
async def update_pattern(request, id):
|
|
"""Update an existing pattern."""
|
|
try:
|
|
data = request.json
|
|
if patterns.update(id, data):
|
|
return json.dumps(patterns.read(id)), 200, {'Content-Type': 'application/json'}
|
|
return json.dumps({"error": "Pattern not found"}), 404
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)}), 400
|
|
|
|
|
|
@controller.delete('/<id>')
|
|
async def delete_pattern(request, id):
|
|
"""Delete a pattern."""
|
|
if patterns.delete(id):
|
|
return json.dumps({"message": "Pattern deleted successfully"}), 200
|
|
return json.dumps({"error": "Pattern not found"}), 404
|