technicalkiwi/led-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(espnow): groups filter and v1 select list on driver

Browse Source 
Apply group membership on RX, accept select as [preset_id, step?],
and fix identify/off plus presets layout for manual beat stepping.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jimmy
2026-05-24 01:44:21 +12:00
parent 1fdb2c9441
commit a97f6c7c2c
7 changed files with 402 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
src/controller_messages.py
View File

@@ -2,7 +2,11 @@
import json
import socket
import network
import ubinascii
import device_groups as dg
from v1_wire import expand_v1
from binary_envelope import parse_binary_envelope
from utils import convert_and_reorder_colors
@@ -58,8 +62,18 @@ def process_data(payload, settings, presets, controller_ip=None, save=False):
return
if data.get("v", "") != "1":
return
data = expand_v1(data)
if save:
data["save"] = True
set_groups = bool(data.get("set_groups"))
groups = data.get("groups")
if set_groups and isinstance(groups, list):
dg.groups_replace(groups)
print("groups set", dg.list_groups())
elif isinstance(groups, list) and groups:
if not any(dg.in_group(str(g)) for g in groups):
print("ignored: not in groups", groups)
return
if "device_config" in data:
apply_device_config(data, settings, presets)
if "b" in data:
@@ -68,7 +82,7 @@ def process_data(payload, settings, presets, controller_ip=None, save=False):
apply_presets(data, settings, presets)
if "clear_presets" in data:
apply_clear_presets(data, presets)
if "select" in data:
elif "select" in data or "s" in data:
apply_select(data, settings, presets)
if "default" in data:
apply_default(data, settings, presets)
@@ -172,7 +186,18 @@ def apply_brightness(data, settings, presets):
pass
_pending_select = None
def _run_select(presets, settings, preset_name, step=None):
if presets.select(preset_name, step=step):
record_last_preset(settings, preset_name)
return True
return False
def apply_presets(data, settings, presets):
global _pending_select
presets_map = data["presets"]
for id, preset_data in presets_map.items():
if not preset_data:
@@ -183,8 +208,8 @@ def apply_presets(data, settings, presets):
preset_data[color_key] = convert_and_reorder_colors(
preset_data[color_key], settings
)
except (TypeError, ValueError, KeyError):
continue
except (TypeError, ValueError, KeyError) as err:
print("preset color convert failed:", id, err)
if "bg" in preset_data:
try:
bg_color = convert_and_reorder_colors([preset_data["bg"]], settings)
@@ -193,18 +218,72 @@ def apply_presets(data, settings, presets):
except (TypeError, ValueError, KeyError):
pass
presets.edit(id, preset_data)
# Same message often carries select; apply now while presets are loaded.
if "select" in data:
apply_select(data, settings, presets)
elif _pending_select is not None:
preset_name, step = _pending_select
_pending_select = None
if preset_name in presets.presets or preset_name in ("on", "off"):
_run_select(presets, settings, preset_name, step)
def _select_list_for_this_device(select_val, settings):
"""Resolve select to ``[preset_id, step?]`` (wire list or legacy name map)."""
if isinstance(select_val, list) and select_val:
return select_val
if isinstance(select_val, str) and str(select_val).strip():
return [str(select_val).strip()]
if not isinstance(select_val, dict) or not select_val:
return None
if "preset" in select_val:
preset_name = select_val.get("preset")
if preset_name is None:
return None
out = [str(preset_name)]
if "step" in select_val:
out.append(select_val["step"])
return out
device_name = str(settings.get("name") or "").strip()
select_list = select_val.get(device_name)
if select_list:
return select_list
try:
sta = network.WLAN(network.STA_IF)
mac_hex = ubinascii.hexlify(sta.config("mac")).decode().lower()
except Exception:
mac_hex = ""
if mac_hex:
for key in select_val:
k = str(key).lower().replace(":", "").replace("-", "")
if mac_hex in k:
return select_val[key]
if len(select_val) == 1:
return next(iter(select_val.values()))
return None
def apply_select(data, settings, presets):
select_map = data["select"]
device_name = settings["name"]
select_list = select_map.get(device_name, [])
global _pending_select
select_val = data.get("select")
if select_val is None:
select_val = data.get("s")
select_list = _select_list_for_this_device(select_val, settings)
if not select_list:
print("select ignored:", repr(select_val))
return
preset_name = str(select_list[0]).strip()
if not preset_name:
return
preset_name = select_list[0]
step = select_list[1] if len(select_list) > 1 else None
if presets.select(preset_name, step=step):
record_last_preset(settings, preset_name)
if preset_name not in presets.presets and preset_name not in ("on", "off"):
_pending_select = (preset_name, step)
print("select deferred (preset not loaded yet):", preset_name)
return
if _run_select(presets, settings, preset_name, step):
_pending_select = None
else:
print("select failed:", preset_name)
def apply_clear_presets(data, presets):

2
src/espnow_transport.py
View File

@@ -28,7 +28,7 @@ def init_espnow(settings):
global _esp
ch = 6
try:
ch = int(settings.get("wifi_channel", 6))
ch = int(settings.get("wifi_channel", 1))
except (TypeError, ValueError):
pass
ch = max(1, min(11, ch))

11
src/main.py
View File

@@ -17,6 +17,7 @@ wdt.feed()
machine.freq(160000000)
settings = Settings()
print(settings)
gc.collect()
presets = Presets(settings["led_pin"], settings["num_leds"])
@@ -35,11 +36,10 @@ hello = json.dumps({
"name": settings.get("name", "led"),
"type": "led",
})
try:
esp.send(BROADCAST_MAC, hello)
print("espnow hello", len(hello), "B")
except Exception as e:
print("espnow hello failed:", e)
print(hello)
esp.send(BROADCAST_MAC, hello)
print("espnow hello", len(hello), "B")
def _on_espnow_message(msg):
@@ -62,6 +62,7 @@ async def main():
print(host, len(msg), "B")
try:
_on_espnow_message(msg)
print(msg)
except Exception as e:
print("espnow rx error:", e)
await asyncio.sleep(0)

41
src/presets.py
View File

@@ -4,6 +4,7 @@ from preset import Preset
from utils import convert_and_reorder_colors
import json
import sys
import utime
try:
import uos as os
except ImportError:
@@ -31,6 +32,7 @@ class Presets:
self.patterns = {
"off": self.off,
"on": self.on,
"blink": self.blink,
}
self.patterns.update(self._load_dynamic_patterns())
@@ -193,6 +195,12 @@ class Presets:
if preset_name in self.presets:
preset = self.presets[preset_name]
if preset.p in self.patterns:
if preset.p == "off":
self.generator = None
self.step = 0
self.fill((0, 0, 0))
self.selected = preset_name
return True
# Manual single-shot patterns: if this select arrives before the main loop has
# tick()'d the previous frame, completing it first keeps step in sync with beats.
if (
@@ -206,7 +214,7 @@ class Presets:
# Set step value if explicitly provided
if step is not None:
self.step = step
elif preset.p == "off" or self.selected != preset_name:
elif self.selected != preset_name:
self.step = 0
self.generator = self.patterns[preset.p](preset)
self.selected = preset_name # Store the preset name, not the object
@@ -256,4 +264,33 @@ class Presets:
def on(self, preset):
colors = preset.c
color = colors[0] if colors else (255, 255, 255)
self.fill(self.apply_brightness(color, preset.b))
lit = self.apply_brightness(color, preset.b)
while True:
self.fill(lit)
yield
def blink(self, preset):
"""Built-in blink (used by controller identify); no patterns/ deploy required."""
colors = preset.c if preset.c else [(255, 255, 255)]
bg_color = self.apply_brightness(preset.background_or(colors), preset.b)
color_index = 0
state = True
last_update = utime.ticks_ms()
while True:
now = utime.ticks_ms()
delay_ms = max(1, int(preset.d))
if utime.ticks_diff(now, last_update) >= delay_ms:
if state:
base = colors[color_index % len(colors)]
self.fill(self.apply_brightness(base, preset.b))
color_index += 1
else:
self.fill(bg_color)
state = not state
last_update = utime.ticks_add(last_update, delay_ms)
yield
def run_tick(presets):
"""Advance one animation frame (standalone tests / mpremote demos)."""
presets.tick()

36
src/v1_wire.py Normal file
View File

@@ -0,0 +1,36 @@
"""Expand short v1 wire keys to long names (MicroPython)."""
K_PRESETS = "p"
K_SELECT = "s"
K_GROUPS = "g"
K_SET_GROUPS = "sg"
K_SAVE = "sv"
K_DEFAULT = "df"
K_DEVICE_CONFIG = "dc"
K_CLEAR_PRESETS = "cp"
K_MANIFEST = "mf"
_SHORT_TO_LONG = {
K_PRESETS: "presets",
K_SELECT: "select",
K_GROUPS: "groups",
K_SET_GROUPS: "set_groups",
K_SAVE: "save",
K_DEFAULT: "default",
K_DEVICE_CONFIG: "device_config",
K_CLEAR_PRESETS: "clear_presets",
K_MANIFEST: "manifest",
}
def expand_v1(data):
if not isinstance(data, dict):
return data
out = dict(data)
for short_key, long_key in _SHORT_TO_LONG.items():
if short_key in data and long_key not in out:
out[long_key] = data[short_key]
if short_key in out:
del out[short_key]
return out