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

feat(patterns): align manual and auto behaviour

Browse Source 
Unify manual/auto timing semantics for key patterns, add preset background support, and improve runtime observability while keeping the driver responsive under beat-triggered selects.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jimmy
2026-05-09 20:07:58 +12:00
parent 4879fcfe90
commit 170a0e05ab
15 changed files with 301 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/background_tasks.py
View File

@@ -26,7 +26,6 @@ async def udp_hello_loop_after_http_ready(sta_if, settings, wdt, runtime_state):
started_ms = utime.ticks_ms()
while True:
if runtime_state.hello:
print("UDP hello: broadcasting...")
try:
broadcast_hello_udp(
sta_if,

42
src/controller_messages.py
View File

@@ -12,8 +12,37 @@ except ImportError:
import os
def _log_rx(payload) -> None:
"""Serial log when led-controller sends a message into ``process_data``."""
try:
if isinstance(payload, (bytes, bytearray)):
n = len(payload)
if n == 0:
print("rx 0 B")
return
cap = 160
chunk = payload if n <= cap else payload[:cap]
try:
txt = bytes(chunk).decode("utf-8")
except Exception:
txt = str(chunk)
if n > cap:
txt = txt + "..."
print("rx", n, "B", txt)
else:
s = str(payload)
cap = 200
if len(s) <= cap:
print("rx", len(s), "C", s)
else:
print("rx", len(s), "C", s[:cap] + "...")
except Exception:
print("rx (logging failed)")
def process_data(payload, settings, presets, controller_ip=None):
"""Read one controller message; binary v1 envelope or JSON v1, then apply fields."""
_log_rx(payload)
data = None
if isinstance(payload, (bytes, bytearray)):
data = parse_binary_envelope(payload)
@@ -27,7 +56,6 @@ def process_data(payload, settings, presets, controller_ip=None):
data = json.loads(payload)
except (ValueError, TypeError):
return
print(payload)
if data.get("v", "") != "1":
return
if "b" in data:
@@ -71,8 +99,14 @@ def apply_presets(data, settings, presets):
)
except (TypeError, ValueError, KeyError):
continue
if "bg" in preset_data:
try:
bg_color = convert_and_reorder_colors([preset_data["bg"]], settings)
if bg_color:
preset_data["bg"] = bg_color[0]
except (TypeError, ValueError, KeyError):
pass
presets.edit(id, preset_data)
print(f"Edited preset {id}: {preset_data.get('name', '')}")
def apply_select(data, settings, presets):
@@ -99,7 +133,6 @@ def apply_clear_presets(data, presets):
if not should_clear:
return
presets.delete_all()
print("Cleared all presets.")
def apply_default(data, settings, presets):
@@ -244,8 +277,5 @@ def apply_patterns_ota(data, presets, controller_ip=None):
updated += 1
if updated > 0:
presets.reload_patterns()
print("patterns_ota: updated", updated, "pattern file(s)")
else:
print("patterns_ota: no valid files downloaded")
except Exception as e:
print("patterns_ota failed:", e)

23
src/hello.py
View File

@@ -92,7 +92,6 @@ def broadcast_hello_udp(
"""
ip, mask, _gw, _dns = sta.ifconfig()
msg = pack_hello_line(sta, device_name)
print("hello:", msg)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
@@ -121,11 +120,9 @@ def broadcast_hello_udp(
for dest_ip, dest_port in targets:
if wdt is not None:
wdt.feed()
label = "%s:%s" % (dest_ip, dest_port)
target = (dest_ip, dest_port)
try:
sock.sendto(msg, target)
print("sent hello ->", target)
except OSError as e:
print("sendto failed:", e)
continue
@@ -134,20 +131,12 @@ def broadcast_hello_udp(
if wdt is not None:
wdt.feed()
try:
data, addr = sock.recvfrom(2048)
print("reply from", addr, ":", data)
_data, addr = sock.recvfrom(2048)
remote_ip = addr[0]
if data != msg:
print("(warning: reply payload differs from hello; still using source IP.)")
discovered = remote_ip
print("Discovered controller at", remote_ip)
break
except OSError as e:
print("recv (no reply):", e, "via", label)
if dest_ip == "255.255.255.255":
print(
"(hint: many APs drop Wi-Fi client broadcast; try wired server or AP without client isolation.)"
)
except OSError:
pass
sock.close()
return discovered
@@ -171,18 +160,12 @@ def discover_controller_udp(device_name="", wdt=None):
print("hello: STA has no IP address.")
raise SystemExit(1)
print("STA IP:", ip, "mask:", mask)
discovered = broadcast_hello_udp(
sta,
device_name,
wait_reply=True,
wdt=wdt,
)
if discovered:
print("discover done; controller =", repr(discovered))
else:
print("discover done; controller not found")
return discovered

18
src/http_routes.py
View File

@@ -23,7 +23,6 @@ def register_routes(app, settings, presets, runtime_state):
@app.route("/ws")
@with_websocket
async def ws_handler(request, ws):
print("WS client connected")
runtime_state.ws_connected()
controller_ip = None
try:
@@ -34,15 +33,11 @@ def register_routes(app, settings, presets, runtime_state):
controller_ip = client_addr
except Exception:
controller_ip = None
print("WS controller_ip:", controller_ip)
try:
while True:
data = await ws.receive()
if not data:
print("WS client disconnected (closed)")
break
print("WS recv bytes:", len(data) if isinstance(data, (bytes, bytearray)) else len(str(data)))
print(data)
process_data(data, settings, presets, controller_ip=controller_ip)
except WebSocketError as e:
print("WS client disconnected:", e)
@@ -50,12 +45,6 @@ def register_routes(app, settings, presets, runtime_state):
print("WS client dropped (OSError):", e)
finally:
runtime_state.ws_disconnected()
print(
"WS client disconnected: hello=",
runtime_state.hello,
"ws_client_count=",
runtime_state.ws_client_count,
)
@app.post("/patterns/upload")
async def upload_pattern(request):
@@ -63,19 +52,15 @@ def register_routes(app, settings, presets, runtime_state):
raw_name = request.args.get("name")
reload_raw = request.args.get("reload", "1")
reload_patterns = str(reload_raw).strip().lower() not in ("0", "false", "no", "off")
print("patterns/upload request:", {"name": raw_name, "reload": reload_patterns})
if not isinstance(raw_name, str) or not raw_name.strip():
return json.dumps({"error": "name is required"}), 400, {"Content-Type": "application/json"}
body = request.body
if not isinstance(body, (bytes, bytearray)) or not body:
print("patterns/upload rejected: empty body")
return json.dumps({"error": "code is required"}), 400, {"Content-Type": "application/json"}
print("patterns/upload body_bytes:", len(body))
try:
code = body.decode("utf-8")
except UnicodeError:
print("patterns/upload rejected: body not utf-8")
return json.dumps({"error": "body must be utf-8 text"}), 400, {"Content-Type": "application/json"}
if not code.strip():
return json.dumps({"error": "code is required"}), 400, {"Content-Type": "application/json"}
@@ -93,16 +78,13 @@ def register_routes(app, settings, presets, runtime_state):
path = "patterns/" + name
try:
print("patterns/upload writing:", path)
with open(path, "w") as f:
f.write(code)
if reload_patterns:
print("patterns/upload reloading patterns")
presets.reload_patterns()
except OSError as e:
print("patterns/upload failed:", e)
return json.dumps({"error": str(e)}), 500, {"Content-Type": "application/json"}
print("patterns/upload success:", {"name": name, "reloaded": reload_patterns})
return json.dumps(
{

42
src/main.py
View File

@@ -15,30 +15,27 @@ try:
except ImportError:
import os
wdt = machine.WDT(timeout=10000)
wdt.feed()
machine.freq(160000000)
settings = Settings()
print(settings)
wdt = machine.WDT(timeout=10000)
wdt.feed()
gc.collect()
print("mem before presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
presets = Presets(settings["led_pin"], settings["num_leds"])
presets.load(settings)
presets.b = settings.get("brightness", 255)
presets.debug = bool(settings.get("debug", False))
gc.collect()
print("mem after presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
default_preset = settings.get("default", "")
if default_preset and default_preset in presets.presets:
if presets.select(default_preset):
print(f"Selected startup preset: {default_preset}")
else:
if not presets.select(default_preset):
print("Startup preset failed (invalid pattern?):", default_preset)
# On ESP32-C3, soft reboots can leave Wi-Fi driver state allocated.
@@ -54,11 +51,22 @@ sta_if.active(True)
sta_if.config(pm=network.WLAN.PM_NONE)
sta_if.connect(settings["ssid"], settings["password"])
while not sta_if.isconnected():
print("Connecting")
print("Waiting for network connection...")
utime.sleep(1)
wdt.feed()
print(sta_if.ifconfig())
def _print_network_ips(controller_ip=None):
"""Always log STA address and led-controller (WS client) address when known."""
try:
led_ip = sta_if.ifconfig()[0]
except Exception:
led_ip = "?"
ctrl = controller_ip if controller_ip else "(not connected)"
print("led-driver IP:", led_ip, " led-controller IP:", ctrl)
_print_network_ips()
app = Microdot()
@@ -76,7 +84,6 @@ def _safe_pattern_filename(name):
@app.route("/ws")
@with_websocket
async def ws_handler(request, ws):
print("WS client connected")
controller_ip = None
try:
client_addr = getattr(request, "client_addr", None)
@@ -86,15 +93,12 @@ async def ws_handler(request, ws):
controller_ip = client_addr
except Exception:
controller_ip = None
print("WS controller_ip:", controller_ip)
_print_network_ips(controller_ip)
try:
while True:
data = await ws.receive()
if not data:
print("WS client disconnected (closed)")
break
print("WS recv bytes:", len(data) if isinstance(data, (bytes, bytearray)) else len(str(data)))
print(data)
process_data(data, settings, presets, controller_ip=controller_ip)
except WebSocketError as e:
print("WS client disconnected:", e)
@@ -108,7 +112,6 @@ async def upload_pattern(request):
raw_name = request.args.get("name")
reload_raw = request.args.get("reload", "1")
reload_patterns = str(reload_raw).strip().lower() not in ("0", "false", "no", "off")
print("patterns/upload request:", {"name": raw_name, "reload": reload_patterns})
if not isinstance(raw_name, str) or not raw_name.strip():
return json.dumps({"error": "name is required"}), 400, {
@@ -116,15 +119,12 @@ async def upload_pattern(request):
}
body = request.body
if not isinstance(body, (bytes, bytearray)) or not body:
print("patterns/upload rejected: empty body")
return json.dumps({"error": "code is required"}), 400, {
"Content-Type": "application/json"
}
print("patterns/upload body_bytes:", len(body))
try:
code = body.decode("utf-8")
except UnicodeError:
print("patterns/upload rejected: body not utf-8")
return json.dumps({"error": "body must be utf-8 text"}), 400, {
"Content-Type": "application/json"
}
@@ -148,18 +148,15 @@ async def upload_pattern(request):
path = "patterns/" + name
try:
print("patterns/upload writing:", path)
with open(path, "w") as f:
f.write(code)
if reload_patterns:
print("patterns/upload reloading patterns")
presets.reload_patterns()
except OSError as e:
print("patterns/upload failed:", e)
return json.dumps({"error": str(e)}), 500, {
"Content-Type": "application/json"
}
print("patterns/upload success:", {"name": name, "reloaded": reload_patterns})
return json.dumps({
"message": "pattern uploaded",
@@ -179,7 +176,6 @@ async def presets_loop():
async def _udp_hello_after_http_ready():
"""Hello must run after the HTTP server binds, or discovery clients time out on /ws."""
await asyncio.sleep(1)
print("UDP hello: broadcasting…")
try:
broadcast_hello_udp(
sta_if,

8
src/patterns/chase.py
View File

@@ -36,7 +36,7 @@ class Chase:
segment_length = n1 + n2
# Calculate position from step_count
step_count = self.driver.step
step_count = int(self.driver.step) % 2
# Position alternates: step 0 adds n3, step 1 adds n4, step 2 adds n3, etc.
if step_count % 2 == 0:
# Even steps: (step_count//2) pairs of (n3+n4) plus one extra n3
@@ -70,9 +70,10 @@ class Chase:
self.driver.n[i] = color1
self.driver.n.write()
print("[chase] step", step_count)
# Increment step for next beat
self.driver.step = step_count + 1
self.driver.step = (step_count + 1) % 2
# Allow tick() to advance the generator once
yield
@@ -115,9 +116,10 @@ class Chase:
self.driver.n[i] = color1
self.driver.n.write()
print("[chase] step", step_count)
# Increment step
step_count += 1
step_count = (step_count + 1) % 2
self.driver.step = step_count
last_update = utime.ticks_add(last_update, transition_duration)
transition_duration = max(10, int(preset.d))

14
src/patterns/pulse.py
View File

@@ -14,6 +14,12 @@ class Pulse:
self.driver.fill(self.driver.apply_brightness(bg_base, preset.b))
color_index = self.driver.step % max(1, len(colors))
if not preset.a:
# Manual / beat trigger: each select restarts this generator and resets
# cycle_start below. Advancing step here makes each beat the next colour
# without requiring a full wall-clock cycle between beats.
nclr = max(1, len(colors))
self.driver.step = (color_index + 1) % nclr
cycle_start = utime.ticks_ms()
# State machine based pulse using a single generator loop
@@ -52,13 +58,13 @@ class Pulse:
# Delay phase: LEDs off between pulses
self.driver.fill(bg_color)
else:
# End of cycle, move to next color and restart timing
# End of cycle: auto advances colour and loops; manual already
# advanced step at run start for the next beat.
if not preset.a:
break
color_index = (color_index + 1) % max(1, len(colors))
self.driver.step = color_index
cycle_start = now
if not preset.a:
break
# Skip drawing this tick, start next cycle
yield
continue

112
src/patterns/radiate.py
View File

@@ -1,11 +1,13 @@
import utime
_RADIATE_DBG_INTERVAL_MS = 1000
# When ``driver.debug`` is True (``settings["debug"]``), log at most this often (ms).
_RADIATE_DBG_INTERVAL_MS = 800
class Radiate:
def __init__(self, driver):
self.driver = driver
self._color_step = 0
def run(self, preset):
"""Radiate from nodes every n1 LEDs, retriggering every delay (d).
@@ -16,7 +18,6 @@ class Radiate:
- d: retrigger interval in ms
"""
colors = preset.c if preset.c else [(255, 255, 255)]
base_on = colors[0]
base_off = preset.background_or(colors)
spacing = max(1, int(preset.n1))
@@ -24,7 +25,7 @@ class Radiate:
return_ms = max(1, int(preset.n3))
max_dist = spacing // 2
lit_color = self.driver.apply_brightness(base_on, preset.b)
lit_color = self.driver.apply_brightness(colors[self._color_step % max(1, len(colors))], preset.b)
off_color = self.driver.apply_brightness(base_off, preset.b)
now = utime.ticks_ms()
@@ -34,18 +35,70 @@ class Radiate:
dbg_banner = False
if not preset.a:
# Single-shot mode exits after one rendered frame. Seed the pulse
# slightly in the past so this frame is visible before returning.
active_pulses = [utime.ticks_add(utime.ticks_ms(), -1)]
# Manual mode: one-shot pulse using the same ms-based timing as auto.
cycle_start = utime.ticks_ms()
while True:
dbg = bool(getattr(self.driver, "debug", False))
spacing = max(1, int(preset.n1))
outward_ms = max(1, int(preset.n2))
return_ms = max(1, int(preset.n3))
max_dist = spacing // 2
on_color = colors[self._color_step % max(1, len(colors))]
lit_color = self.driver.apply_brightness(on_color, preset.b)
off_color = self.driver.apply_brightness(base_off, preset.b)
pulse_lifetime = outward_ms + return_ms
now = utime.ticks_ms()
age = utime.ticks_diff(now, cycle_start)
if age < 1:
age = 1
if age <= outward_ms:
front = (age * max_dist + outward_ms - 1) // outward_ms
elif age <= outward_ms + return_ms:
back_age = age - outward_ms
remaining = return_ms - back_age
front = (remaining * max_dist + return_ms - 1) // return_ms
else:
front = 0
lit_count = 0
for i in range(self.driver.num_leds):
offset = (i + (spacing // 2)) % spacing
dist = min(offset, spacing - offset)
lit = dist <= front
self.driver.n[i] = lit_color if lit else off_color
if lit:
lit_count += 1
self.driver.n.write()
if dbg:
if not dbg_banner:
dbg_banner = True
print(
"[radiate] debug on n1=%s n2=%s n3=%s d=%s auto=%s num_leds=%d"
% (preset.n1, preset.n2, preset.n3, preset.d, preset.a, self.driver.num_leds)
)
print(
"[radiate] manual frame age=%d/%d front=%d lit=%d"
% (age, pulse_lifetime, front, lit_count)
)
yield
if age >= pulse_lifetime:
self._color_step += 1
return
while True:
now = utime.ticks_ms()
dbg = bool(getattr(self.driver, "debug", False))
delay_ms = max(1, int(preset.d))
spacing = max(1, int(preset.n1))
outward_ms = max(1, int(preset.n2))
return_ms = max(1, int(preset.n3))
pulse_lifetime = outward_ms + return_ms
max_dist = spacing // 2
lit_color = self.driver.apply_brightness(base_on, preset.b)
on_color = colors[self._color_step % max(1, len(colors))]
lit_color = self.driver.apply_brightness(on_color, preset.b)
off_color = self.driver.apply_brightness(base_off, preset.b)
if preset.a and utime.ticks_diff(now, last_trigger) >= delay_ms:
@@ -53,33 +106,26 @@ class Radiate:
# prevents overlap from keeping color[0] continuously visible.
active_pulses = [now]
last_trigger = utime.ticks_add(last_trigger, delay_ms)
if bool(getattr(self.driver, "debug", False)):
print(
"[radiate] trigger spacing=%d out=%d in=%d delay=%d"
% (spacing, outward_ms, return_ms, delay_ms)
)
self._color_step += 1
# Drop pulses once their out-and-back lifetime ends.
pulse_lifetime = outward_ms + return_ms
kept = []
for start in active_pulses:
age = utime.ticks_diff(now, start)
if age < pulse_lifetime:
kept.append(start)
active_pulses = kept
debug_front = -1
lit_count = 0
lit_count = 0
for i in range(self.driver.num_leds):
# Nearest node distance for a repeating node grid every `spacing` LEDs.
offset = i % spacing
offset = (i + (spacing // 2)) % spacing
dist = min(offset, spacing - offset)
lit = False
for start in active_pulses:
age = utime.ticks_diff(now, start)
# Do not render on the exact trigger tick; this avoids
# node LEDs appearing "stuck on" between cycles.
# Auto: skip the exact trigger tick (age==0) so nodes are not stuck on.
if age <= 0:
continue
if age <= outward_ms:
@@ -95,8 +141,6 @@ class Radiate:
if dist <= front:
lit = True
if front > debug_front:
debug_front = front
break
self.driver.n[i] = lit_color if lit else off_color
@@ -105,33 +149,21 @@ class Radiate:
self.driver.n.write()
if bool(getattr(self.driver, "debug", False)):
if dbg:
if not dbg_banner:
dbg_banner = True
print(
"[radiate] debug on: spacing=%s out=%s in=%s d=%s num=%d"
% (
preset.n1,
preset.n2,
preset.n3,
preset.d,
self.driver.num_leds,
)
"[radiate] debug on n1=%s n2=%s n3=%s d=%s auto=%s num_leds=%d"
% (preset.n1, preset.n2, preset.n3, preset.d, preset.a, self.driver.num_leds)
)
pulse_age = -1
if active_pulses:
pulse_age = utime.ticks_diff(now, active_pulses[0])
if utime.ticks_diff(now, last_dbg) >= _RADIATE_DBG_INTERVAL_MS:
pulse_age = -1
if active_pulses:
pulse_age = utime.ticks_diff(now, active_pulses[0])
print(
"[radiate] age=%d front=%d max=%d active=%d lit=%d"
% (pulse_age, debug_front, max_dist, len(active_pulses), lit_count)
"[radiate] pulses=%d first_age=%d lit=%d lifetime=%d"
% (len(active_pulses), pulse_age, lit_count, pulse_lifetime)
)
if lit_count == 0:
print("[radiate] fully off")
last_dbg = now
if not preset.a:
yield
return
yield

68
src/patterns/twinkle.py
View File

@@ -2,9 +2,6 @@ import random
import utime
# Default cool palette (icy blues, violet, mint) when preset has no colours.
# When `driver.debug` is True, print stats every N twinkle ticks (serial can be slow).
_TWINKLE_DBG_INTERVAL = 40
_DEFAULT_COOL = (
(120, 200, 255),
(80, 140, 255),
@@ -93,32 +90,6 @@ class Twinkle:
on = [random.randint(0, 255) < dens for _ in range(num)]
colour_i = [random.randint(0, len(palette) - 1) for _ in range(num)]
last_update = utime.ticks_ms()
dbg_tick = 0
dbg_banner = False
def on_run_min_max(bits):
"""Smallest and largest contiguous run of True in bits (0,0 if all off)."""
best_min = num + 1
best_max = 0
cur = 0
for j in range(num):
if bits[j]:
cur += 1
else:
if cur:
if cur < best_min:
best_min = cur
if cur > best_max:
best_max = cur
cur = 0
if cur:
if cur < best_min:
best_min = cur
if cur > best_max:
best_max = cur
if best_min == num + 1:
return 0, 0
return best_min, best_max
if not preset.a:
for i in range(num):
@@ -137,15 +108,12 @@ class Twinkle:
if utime.ticks_diff(now, last_update) >= delay_ms:
rate = activity_rate()
dens = density255()
dbg = bool(getattr(self.driver, "debug", False))
dbg_tick += 1
# Snapshot for decisions; apply all darks then all lights so
# overlaps in the same tick favour lit runs (lights win).
prev_on = on[:]
prev_ci = colour_i[:]
next_on = list(prev_on)
next_ci = list(prev_ci)
dbg_ops = {"L": 0, "D": 0}
light_i = []
dark_i = []
@@ -160,7 +128,6 @@ class Twinkle:
dark_i.append(i)
def light_adjacent(start):
dbg_ops["L"] += 1
k = random_cluster_len()
b = cluster_base_index(start, k)
for dj in range(k):
@@ -169,7 +136,6 @@ class Twinkle:
next_ci[idx] = random.randint(0, len(palette) - 1)
def dark_adjacent(start):
dbg_ops["D"] += 1
k = random_cluster_len()
b = cluster_base_index(start, k)
for dj in range(k):
@@ -191,38 +157,4 @@ class Twinkle:
on = next_on
colour_i = next_ci
last_update = utime.ticks_add(last_update, delay_ms)
if dbg:
lo, hi = cluster_len_bounds()
if not dbg_banner:
dbg_banner = True
print(
"[twinkle] debug on: n1=%s n2=%s n3=%s n4=%s d=%s -> lo=%d hi=%d num=%d"
% (
preset.n1,
preset.n2,
preset.n3,
preset.n4,
preset.d,
lo,
hi,
num,
)
)
rmin, rmax = on_run_min_max(on)
bad = lo > 0 and rmin > 0 and rmin < lo and num >= lo
if bad or (dbg_tick % _TWINKLE_DBG_INTERVAL == 0):
print(
"[twinkle] tick=%d rate=%d dens=%d L=%d D=%d on_runs min=%d max=%d%s"
% (
dbg_tick,
rate,
dens,
dbg_ops["L"],
dbg_ops["D"],
rmin,
rmax,
" **run<lo**" if bad else "",
)
)
yield

26
src/preset.py
View File

@@ -12,6 +12,7 @@ class Preset:
self.n4 = 0
self.n5 = 0
self.n6 = 0
self.bg = (0, 0, 0)
# Override defaults with provided data
self.edit(data)
@@ -25,9 +26,10 @@ class Preset:
"delay": "d",
"brightness": "b",
"auto": "a",
"background": "bg",
}
int_fields = {"d", "b", "n1", "n2", "n3", "n4", "n5", "n6"}
allowed_fields = {"p", "c", "d", "b", "a", "n1", "n2", "n3", "n4", "n5", "n6"}
allowed_fields = {"p", "c", "d", "b", "a", "bg", "n1", "n2", "n3", "n4", "n5", "n6"}
for key, value in data.items():
key = aliases.get(key, key)
if key not in allowed_fields:
@@ -56,6 +58,21 @@ class Preset:
elif key == "c":
if isinstance(value, (list, tuple)):
self.c = value
elif key == "bg":
if isinstance(value, str) and value.startswith("#") and len(value) == 7:
try:
self.bg = (
int(value[1:3], 16),
int(value[3:5], 16),
int(value[5:7], 16),
)
except (TypeError, ValueError):
continue
elif isinstance(value, (list, tuple)) and len(value) == 3:
try:
self.bg = tuple(max(0, min(255, int(x))) for x in value)
except (TypeError, ValueError):
continue
else:
setattr(self, key, value)
return True
@@ -101,6 +118,12 @@ class Preset:
self.a = value
def background_or(self, colors=None, default=(0, 0, 0)):
bg = getattr(self, "bg", None)
if isinstance(bg, (list, tuple)) and len(bg) == 3:
try:
return tuple(max(0, min(255, int(x))) for x in bg)
except (TypeError, ValueError):
return default
return default
def to_dict(self):
@@ -110,6 +133,7 @@ class Preset:
"b": self.b,
"c": self.c,
"a": self.a,
"bg": self.bg,
"n1": self.n1,
"n2": self.n2,
"n3": self.n3,

24
src/presets.py
View File

@@ -106,10 +106,6 @@ class Presets:
preset_data[color_key], order
)
self.presets[name] = Preset(preset_data)
if self.presets:
print("Loaded presets:")
#for name in sorted(self.presets.keys()):
# print(f" {name}: {self.presets[name].to_dict()}")
return True
def edit(self, name, data):
@@ -117,6 +113,15 @@ class Presets:
if name in self.presets:
# Update existing preset
self.presets[name].edit(data)
# Editing the live preset (e.g. toggling auto/manual) must reset runtime
# state; re-select alone keeps step because preset name is unchanged.
if self.selected == name:
self.step = 0
self.generator = None
self.fill((0, 0, 0))
# Re-start pattern so manual/auto and other edits apply without a
# separate select message (controller usually sends both).
self.select(name)
else:
if len(self.presets) >= MAX_PRESETS and name not in ("on", "off"):
print("Preset limit reached:", MAX_PRESETS)
@@ -159,6 +164,16 @@ class Presets:
if preset_name in self.presets:
preset = self.presets[preset_name]
if preset.p in self.patterns:
# 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 (
preset_name == self.selected
and not preset.a
and preset.p == "chase"
and self.generator is not None
):
while self.generator is not None:
self.tick()
# Set step value if explicitly provided
if step is not None:
self.step = step
@@ -166,6 +181,7 @@ class Presets:
self.step = 0
self.generator = self.patterns[preset.p](preset)
self.selected = preset_name # Store the preset name, not the object
self.tick()
return True
print("select failed: pattern not found for preset", preset_name, "pattern=", preset.p)
return False

2
src/settings.py
View File

@@ -39,7 +39,6 @@ class Settings(dict):
j = json.dumps(self)
with open(self.SETTINGS_FILE, 'w') as file:
file.write(j)
print("Settings saved successfully.")
except Exception as e:
print(f"Error saving settings: {e}")
@@ -48,7 +47,6 @@ class Settings(dict):
with open(self.SETTINGS_FILE, 'r') as file:
loaded_settings = json.load(file)
self.update(loaded_settings)
print("Settings loaded successfully.")
except Exception as e:
print(f"Error loading settings")
self.set_defaults()

14
src/startup.py
View File

@@ -11,26 +11,21 @@ def initialize_runtime():
machine.freq(160000000)
settings = Settings()
print(settings)
wdt = machine.WDT(timeout=10000)
wdt.feed()
gc.collect()
print("mem before presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
presets = Presets(settings["led_pin"], settings["num_leds"])
presets.load(settings)
presets.b = settings.get("brightness", 255)
presets.debug = bool(settings.get("debug", False))
gc.collect()
print("mem after presets:", {"free": gc.mem_free(), "alloc": gc.mem_alloc()})
default_preset = settings.get("default", "")
if default_preset and default_preset in presets.presets:
if presets.select(default_preset):
print("Selected startup preset:", default_preset)
else:
if not presets.select(default_preset):
print("Startup preset failed (invalid pattern?):", default_preset)
# On ESP32-C3, soft reboots can leave Wi-Fi driver state allocated.
@@ -49,5 +44,10 @@ def initialize_runtime():
utime.sleep(1)
wdt.feed()
print(sta_if.ifconfig())
try:
led_ip = sta_if.ifconfig()[0]
except Exception:
led_ip = "?"
print("led-driver IP:", led_ip, " led-controller IP:", "(not connected)")
return settings, presets, wdt, sta_if