From 170a0e05ab592f8e183213ae9095f17c4252c924 Mon Sep 17 00:00:00 2001 From: Jimmy Date: Sat, 9 May 2026 20:07:58 +1200 Subject: [PATCH] feat(patterns): align manual and auto behaviour 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 --- src/background_tasks.py | 1 - src/controller_messages.py | 42 ++++++++++++-- src/hello.py | 23 +------- src/http_routes.py | 18 ------ src/main.py | 42 +++++++------- src/patterns/chase.py | 8 ++- src/patterns/pulse.py | 14 +++-- src/patterns/radiate.py | 112 ++++++++++++++++++++++++------------- src/patterns/twinkle.py | 68 ---------------------- src/preset.py | 26 ++++++++- src/presets.py | 24 ++++++-- src/settings.py | 2 - src/startup.py | 14 ++--- tests/patterns/radiate.py | 52 +++++++++++++++++ tests/patterns/twinkle.py | 52 +++++++++++++++++ 15 files changed, 301 insertions(+), 197 deletions(-) create mode 100644 tests/patterns/radiate.py create mode 100644 tests/patterns/twinkle.py diff --git a/src/background_tasks.py b/src/background_tasks.py index 32638e6..64e7397 100644 --- a/src/background_tasks.py +++ b/src/background_tasks.py @@ -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, diff --git a/src/controller_messages.py b/src/controller_messages.py index d4bd98f..3f5fd04 100644 --- a/src/controller_messages.py +++ b/src/controller_messages.py @@ -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) diff --git a/src/hello.py b/src/hello.py index 155cec2..2673d11 100644 --- a/src/hello.py +++ b/src/hello.py @@ -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 diff --git a/src/http_routes.py b/src/http_routes.py index aa5177a..913b398 100644 --- a/src/http_routes.py +++ b/src/http_routes.py @@ -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( { diff --git a/src/main.py b/src/main.py index 1f7bebb..ba2f02a 100644 --- a/src/main.py +++ b/src/main.py @@ -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, diff --git a/src/patterns/chase.py b/src/patterns/chase.py index a532848..7b07946 100644 --- a/src/patterns/chase.py +++ b/src/patterns/chase.py @@ -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)) diff --git a/src/patterns/pulse.py b/src/patterns/pulse.py index 474a5df..1a74ce5 100644 --- a/src/patterns/pulse.py +++ b/src/patterns/pulse.py @@ -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 diff --git a/src/patterns/radiate.py b/src/patterns/radiate.py index 33b57ec..ce682cd 100644 --- a/src/patterns/radiate.py +++ b/src/patterns/radiate.py @@ -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 diff --git a/src/patterns/twinkle.py b/src/patterns/twinkle.py index a5825f7..bc56455 100644 --- a/src/patterns/twinkle.py +++ b/src/patterns/twinkle.py @@ -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= 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 diff --git a/src/settings.py b/src/settings.py index 31b4929..8724d9c 100644 --- a/src/settings.py +++ b/src/settings.py @@ -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() diff --git a/src/startup.py b/src/startup.py index bdb8345..62243f4 100644 --- a/src/startup.py +++ b/src/startup.py @@ -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 diff --git a/tests/patterns/radiate.py b/tests/patterns/radiate.py new file mode 100644 index 0000000..a2bc61e --- /dev/null +++ b/tests/patterns/radiate.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +import utime +from machine import WDT +from settings import Settings +from presets import Presets + + +def main(): + print("[test] radiate: start") + s = Settings() + p = Presets(pin=s.get("led_pin", 10), num_leds=s.get("num_leds", 30)) + p.debug = True + wdt = WDT(timeout=10000) + + print("[test] radiate: auto phase begin") + p.edit("test_pattern", {"p": "radiate", "b": 64, "a": True, "d": 3000, "c": [(255, 0, 0), (0, 0, 255)]}) + if not p.select("test_pattern"): + raise Exception("radiate select failed in auto phase") + auto_start = utime.ticks_ms() + while utime.ticks_diff(utime.ticks_ms(), auto_start) < 2500: + wdt.feed() + p.run_step() + utime.sleep_ms(20) + remaining_ms = utime.ticks_diff(p.next_tick_ms, utime.ticks_ms()) + if p.next_tick_ms == 0 or remaining_ms <= 0: + raise Exception("radiate delay scheduling invalid") + print("[test] radiate: auto phase end") + + print("[test] radiate: manual phase begin") + p.edit("test_pattern", {"p": "radiate", "b": 64, "a": False, "d": 3000, "c": [(255, 0, 0), (0, 0, 255)]}) + if not p.select("test_pattern", step=0): + raise Exception("radiate select failed in manual phase") + for _ in range(6): + current_step = int(p.step) + if not p.select("test_pattern", step=current_step): + raise Exception("radiate external select failed") + p.run_step() + wdt.feed() + if int(p.step) == current_step: + raise Exception("radiate external step did not advance") + if p.generator is not None: + raise Exception("radiate manual mode rescheduled generator") + hold_start = utime.ticks_ms() + while utime.ticks_diff(utime.ticks_ms(), hold_start) < 700: + wdt.feed() + utime.sleep_ms(20) + print("[test] radiate: manual phase end") + print("[test] radiate: pass") + + +if __name__ == "__main__": + main() diff --git a/tests/patterns/twinkle.py b/tests/patterns/twinkle.py new file mode 100644 index 0000000..7c13983 --- /dev/null +++ b/tests/patterns/twinkle.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +import utime +from machine import WDT +from settings import Settings +from presets import Presets + + +def main(): + print("[test] twinkle: start") + s = Settings() + p = Presets(pin=s.get("led_pin", 10), num_leds=s.get("num_leds", 30)) + p.debug = True + wdt = WDT(timeout=10000) + + print("[test] twinkle: auto phase begin") + p.edit("test_pattern", {"p": "twinkle", "b": 64, "a": True, "d": 3000, "c": [(255, 0, 0), (0, 0, 255)]}) + if not p.select("test_pattern"): + raise Exception("twinkle select failed in auto phase") + auto_start = utime.ticks_ms() + while utime.ticks_diff(utime.ticks_ms(), auto_start) < 2500: + wdt.feed() + p.run_step() + utime.sleep_ms(20) + remaining_ms = utime.ticks_diff(p.next_tick_ms, utime.ticks_ms()) + if p.next_tick_ms == 0 or remaining_ms <= 0: + raise Exception("twinkle delay scheduling invalid") + print("[test] twinkle: auto phase end") + + print("[test] twinkle: manual phase begin") + p.edit("test_pattern", {"p": "twinkle", "b": 64, "a": False, "d": 3000, "c": [(255, 0, 0), (0, 0, 255)]}) + if not p.select("test_pattern", step=0): + raise Exception("twinkle select failed in manual phase") + for _ in range(6): + current_step = int(p.step) + if not p.select("test_pattern", step=current_step): + raise Exception("twinkle external select failed") + p.run_step() + wdt.feed() + if int(p.step) == current_step: + raise Exception("twinkle external step did not advance") + if p.generator is not None: + raise Exception("twinkle manual mode rescheduled generator") + hold_start = utime.ticks_ms() + while utime.ticks_diff(utime.ticks_ms(), hold_start) < 700: + wdt.feed() + utime.sleep_ms(20) + print("[test] twinkle: manual phase end") + print("[test] twinkle: pass") + + +if __name__ == "__main__": + main()