From e83f0d607c9429b58696d00e8202f3558e3bac01 Mon Sep 17 00:00:00 2001 From: jimmy Date: Thu, 28 Aug 2025 22:55:10 +1200 Subject: [PATCH] Switch to async patterns --- src/main.py | 10 +- src/p2p.py | 2 +- src/patterns.py | 439 ++++++++++++++++-------------------------------- src/settings.py | 8 +- src/web.py | 2 +- 5 files changed, 149 insertions(+), 312 deletions(-) diff --git a/src/main.py b/src/main.py index db405b0..c17dae4 100644 --- a/src/main.py +++ b/src/main.py @@ -17,16 +17,11 @@ async def main(): patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"]) if settings["color_order"] == "rbg": color_order = (1, 5, 3) else: color_order = (1, 3, 5) - patterns.set_color1(tuple(int(settings["color1"][i:i+2], 16) for i in color_order)) - patterns.set_color2(tuple(int(settings["color2"][i:i+2], 16) for i in color_order)) + patterns.set_color(0,(tuple(int(settings["color1"][i:i+2], 16) for i in color_order))) + patterns.set_color(1,(tuple(int(settings["color2"][i:i+2], 16) for i in color_order))) patterns.set_brightness(int(settings["brightness"])) patterns.set_delay(int(settings["delay"])) - async def tick(): - while True: - patterns.tick() - await asyncio.sleep_ms(1) - w = web(settings, patterns) print(settings) # start the server in a bacakground task @@ -35,7 +30,6 @@ async def main(): wdt = machine.WDT(timeout=10000) wdt.feed() - asyncio.create_task(tick()) asyncio.create_task(p2p(settings, patterns)) while True: diff --git a/src/p2p.py b/src/p2p.py index d73636c..2e953ab 100644 --- a/src/p2p.py +++ b/src/p2p.py @@ -16,5 +16,5 @@ async def p2p(settings, patterns): if "step" in settings and isinstance(settings["step"], int): patterns.set_pattern_step(settings["step"]) else: - settings.set_settings(data.get("settings", {}), patterns, data.get("save", False)) + await settings.set_settings(data.get("settings", {}), patterns, data.get("save", False)) print("should not print") diff --git a/src/patterns.py b/src/patterns.py index c42b56d..83fb35e 100644 --- a/src/patterns.py +++ b/src/patterns.py @@ -1,4 +1,5 @@ -from machine import Pin +import asyncio +from machine import Pin, WDT from neopixel import NeoPixel import utime import random @@ -7,56 +8,24 @@ class Patterns: def __init__(self, pin, num_leds, color1=(0,0,0), color2=(0,0,0), brightness=127, selected="rainbow_cycle", delay=100): self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds) self.num_leds = num_leds - self.pattern_step = 0 - self.last_update = utime.ticks_ms() self.delay = delay self.brightness = brightness self.patterns = { "off": self.off, "on" : self.on, - "color_wipe": self.color_wipe_step, - "rainbow_cycle": self.rainbow_cycle_step, - "theater_chase": self.theater_chase_step, - "blink": self.blink_step, - "random_color_wipe": self.random_color_wipe_step, - "random_rainbow_cycle": self.random_rainbow_cycle_step, - "random_theater_chase": self.random_theater_chase_step, - "random_blink": self.random_blink_step, - "color_transition": self.color_transition_step, # Added new pattern - "flicker": self.flicker_step, - "external": None + "blink": self.blink, + "rainbow": self.rainbow, + "theater chase": self.theater_chase, + "flicker": self.flicker # Added flicker pattern + } self.selected = selected # Ensure colors list always starts with at least two for robust transition handling self.colors = [color1, color2] if color1 != color2 else [color1, (255, 255, 255)] # Fallback if initial colors are same if not self.colors: # Ensure at least one color exists self.colors = [(0, 0, 0)] - - self.transition_duration = delay * 50 # Default transition duration - self.hold_duration = delay * 10 # Default hold duration at each color - self.transition_step = 0 # Current step in the transition - self.current_color_idx = 0 # Index of the color currently being held/transitioned from - self.current_color = self.colors[self.current_color_idx] # The actual blended color - - self.hold_start_time = utime.ticks_ms() # Time when the current color hold started - - - def sync(self): - self.pattern_step=0 - self.last_update = utime.ticks_ms() - self.delay - if self.selected == "color_transition": - self.transition_step = 0 - self.current_color_idx = 0 - self.current_color = self.colors[self.current_color_idx] - self.hold_start_time = utime.ticks_ms() # Reset hold time - self.tick() - - def set_pattern_step(self, step): - self.pattern_step = step - - def tick(self): - if self.patterns[self.selected]: - self.patterns[self.selected]() + self.task = None + self.pattern_step = 0 def update_num_leds(self, pin, num_leds): self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds) @@ -65,124 +34,44 @@ class Patterns: def set_delay(self, delay): self.delay = delay - # Update transition duration and hold duration when delay changes - self.transition_duration = self.delay * 50 - self.hold_duration = self.delay * 10 - def set_brightness(self, brightness): self.brightness = brightness - def set_color1(self, color): - if len(self.colors) > 0: - self.colors[0] = color - if self.selected == "color_transition": - # If the first color is changed, potentially reset transition - # to start from this new color if we were about to transition from it - if self.current_color_idx == 0: - self.transition_step = 0 - self.current_color = self.colors[0] - self.hold_start_time = utime.ticks_ms() - else: - self.colors.append(color) - - - def set_color2(self, color): - if len(self.colors) > 1: - self.colors[1] = color - elif len(self.colors) == 1: - self.colors.append(color) - else: # List is empty - self.colors.append((0,0,0)) # Dummy color - self.colors.append(color) - - def set_colors(self, colors): - if colors and len(colors) >= 2: - self.colors = colors - if self.selected == "color_transition": - self.sync() # Reset transition if new color list is provided - elif colors and len(colors) == 1: - self.colors = [colors[0], (255,255,255)] # Add a default second color - if self.selected == "color_transition": - print("Warning: 'color_transition' requires at least two colors. Adding a default second color.") - self.sync() - else: - print("Error: set_colors requires a list of at least one color.") - self.colors = [(0,0,0), (255,255,255)] # Fallback - if self.selected == "color_transition": - self.sync() + self.colors = colors def set_color(self, num, color): # Changed: More robust index check if 0 <= num < len(self.colors): self.colors[num] = color - # If the changed color is part of the current or next transition, - # restart the transition for smoother updates - if self.selected == "color_transition": - current_from_idx = self.current_color_idx - current_to_idx = (self.current_color_idx + 1) % len(self.colors) - if num == current_from_idx or num == current_to_idx: - # If we change a color involved in the current transition, - # it's best to restart the transition state for smoothness. - self.transition_step = 0 - self.current_color_idx = current_from_idx # Stay at the current starting color - self.current_color = self.colors[self.current_color_idx] - self.hold_start_time = utime.ticks_ms() # Reset hold return True elif num == len(self.colors): # Allow setting a new color at the end self.colors.append(color) return True return False - def add_color(self, color): - self.colors.append(color) - if self.selected == "color_transition" and len(self.colors) == 2: - # If we just added the second color needed for transition - self.sync() - - def del_color(self, num): # Changed: More robust index check and using del for lists if 0 <= num < len(self.colors): del self.colors[num] - # If the color being deleted was part of the current transition, - # re-evaluate the current_color_idx - if self.selected == "color_transition": - if len(self.colors) < 2: # Need at least two colors for transition - print("Warning: Not enough colors for 'color_transition'. Switching to 'on'.") - self.select("on") # Or some other default - else: - # Adjust index if it's out of bounds after deletion or was the one transitioning from - self.current_color_idx %= len(self.colors) - self.transition_step = 0 - self.current_color = self.colors[self.current_color_idx] - self.hold_start_time = utime.ticks_ms() return True return False def apply_brightness(self, color, brightness_override=None): - effective_brightness = brightness_override if brightness_override is not None else self.brightness - return tuple(int(c * effective_brightness / 255) for c in color) + effective_brightness = brightness_override if brightness_override is not None else self.brightness + return tuple(int(c * effective_brightness / 255) for c in color) - def select(self, pattern): - if pattern in self.patterns: - self.selected = pattern - self.sync() # Reset pattern state when selecting a new pattern - if pattern == "color_transition": - if len(self.colors) < 2: - print("Warning: 'color_transition' requires at least two colors. Switching to 'on'.") - self.selected = "on" # Fallback if not enough colors - self.sync() # Re-sync for the new pattern - else: - self.transition_step = 0 - self.current_color_idx = 0 # Start from the first color in the list - self.current_color = self.colors[self.current_color_idx] - self.hold_start_time = utime.ticks_ms() # Reset hold timer - self.transition_duration = self.delay * 50 # Initialize transition duration - self.hold_duration = self.delay * 10 # Initialize hold duration - return True - return False + async def select(self, pattern, reset = True): + if pattern not in self.patterns: + return False + self.selected = pattern + if self.task is not None: + self.task.cancel() + if reset: self.pattern_step = 0 + print(pattern) + self.task = asyncio.create_task(self.patterns[pattern]()) + return True def set(self, i, color): self.n[i] = color @@ -190,190 +79,144 @@ class Patterns: def write(self): self.n.write() - def fill(self, color=None): - fill_color = color if color is not None else self.colors[0] - for i in range(self.num_leds): - self.n[i] = fill_color + def fill(self, color): + self.n.fill(color) self.n.write() - def off(self): + async def off(self): self.fill((0, 0, 0)) - def on(self): + async def on(self): self.fill(self.apply_brightness(self.colors[0])) - def color_wipe_step(self): - color = self.apply_brightness(self.colors[0]) - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - if self.pattern_step < self.num_leds: - for i in range(self.num_leds): - self.n[i] = (0, 0, 0) - self.n[self.pattern_step] = self.apply_brightness(color) - self.n.write() - self.pattern_step += 1 + def sync(self): + self.pattern_step = 0 + + async def rainbow(self): + def wheel(pos): + if pos < 85: + return (pos * 3, 255 - pos * 3, 0) + elif pos < 170: + pos -= 85 + return (255 - pos * 3, 0, pos * 3) else: - self.pattern_step = 0 - self.last_update = current_time + pos -= 170 + return (0, pos * 3, 255 - pos * 3) + last_update = utime.ticks_ms() + while True: + if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay: + for i in range(self.num_leds): + rc_index = (i * 256 // self.num_leds) + self.pattern_step + self.n[i] = self.apply_brightness(wheel(rc_index & 255)) + self.n.write() + self.pattern_step = (self.pattern_step + 1) % 256 + last_update += self.delay + await asyncio.sleep(0) - def rainbow_cycle_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay/5: - def wheel(pos): - if pos < 85: - return (pos * 3, 255 - pos * 3, 0) - elif pos < 170: - pos -= 85 - return (255 - pos * 3, 0, pos * 3) + + async def theater_chase(self): + last_update = utime.ticks_ms() + while True: + if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay: + for i in range(self.num_leds): + if (i + self.pattern_step) % 3 == 0: + self.n[i] = self.apply_brightness(self.colors[0]) + else: + self.n[i] = (0, 0, 0) + self.n.write() + self.pattern_step = (self.pattern_step + 1) % 3 + last_update += self.delay + await asyncio.sleep(0) + + async def blink(self): + last_update = utime.ticks_ms() + self.pattern_step = 0 + while True: + if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay: + if self.pattern_step: + self.off() + self.pattern_step = 0 else: - pos -= 170 - return (0, pos * 3, 255 - pos * 3) + self.on() + self.pattern_step = 1 + last_update += self.delay + await asyncio.sleep(0) - for i in range(self.num_leds): - rc_index = (i * 256 // self.num_leds) + self.pattern_step - self.n[i] = self.apply_brightness(wheel(rc_index & 255)) - self.n.write() - self.pattern_step = (self.pattern_step + 1) % 256 - self.last_update = current_time + async def flicker(self): + last_update = utime.ticks_ms() + while True: + if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay: + # Calculate a single flicker amount for all LEDs + flicker_amount = random.randint(int(-self.brightness // 1.5), int(self.brightness // 1.5)) + flicker_brightness = max(0, min(255, self.brightness + flicker_amount)) + self.fill(self.apply_brightness(self.colors[0], brightness_override=flicker_brightness)) + last_update += self.delay + await asyncio.sleep(0) - def theater_chase_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - for i in range(self.num_leds): - if (i + self.pattern_step) % 3 == 0: - self.n[i] = self.apply_brightness(self.colors[0]) - else: - self.n[i] = (0, 0, 0) - self.n.write() - self.pattern_step = (self.pattern_step + 1) % 3 - self.last_update = current_time - - def blink_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - if self.pattern_step % 2 == 0: + async def color_transition(self): + if len(self.colors) < 2: + # If there's only one color or no colors, just display that color (or off) self.fill(self.apply_brightness(self.colors[0])) - else: - self.fill((0, 0, 0)) - self.pattern_step = (self.pattern_step + 1) % 2 - self.last_update = current_time - - def random_color_wipe_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) - if self.pattern_step < self.num_leds: - for i in range(self.num_leds): - self.n[i] = (0, 0, 0) - self.n[self.pattern_step] = self.apply_brightness(color) - self.n.write() - self.pattern_step += 1 - else: - self.pattern_step = 0 - self.last_update = current_time - - def random_rainbow_cycle_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: # Kept original delay for now - def wheel(pos): - if pos < 85: - return (pos * 3, 255 - pos * 3, 0) - elif pos < 170: - pos -= 85 - return (255 - pos * 3, 0, pos * 3) - else: - pos -= 170 - return (0, pos * 3, 255 - pos * 3) - - random_offset = random.randint(0, 255) - for i in range(self.num_leds): - rc_index = (i * 256 // self.num_leds) + self.pattern_step + random_offset - self.n[i] = self.apply_brightness(wheel(rc_index & 255)) - self.n.write() - self.pattern_step = (self.pattern_step + 1) % 256 - self.last_update = current_time - - def random_theater_chase_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) - for i in range(self.num_leds): - if (i + self.pattern_step) % 3 == 0: - self.n[i] = self.apply_brightness(color) - else: - self.n[i] = (0, 0, 0) - self.n.write() - self.pattern_step = (self.pattern_step + 1) % 3 - self.last_update = current_time - - def random_blink_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay*10: - color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) - if self.pattern_step % 2 == 0: - self.fill(self.apply_brightness(color)) - else: - self.fill((0, 0, 0)) - self.pattern_step = (self.pattern_step + 1) % 2 - self.last_update = current_time - - def color_transition_step(self): - current_time = utime.ticks_ms() - - # Check for hold duration first - if utime.ticks_diff(current_time, self.hold_start_time) < self.hold_duration: - # Still in hold phase, just display the current solid color - self.fill(self.apply_brightness(self.current_color)) - self.last_update = current_time # Keep updating last_update to avoid skipping frames - return - - # If hold duration is over, proceed with transition - if utime.ticks_diff(current_time, self.last_update) >= self.delay: - num_colors = len(self.colors) - if num_colors < 2: - # Should not happen if select handles it, but as a safeguard - self.select("on") return - from_color = self.colors[self.current_color_idx] - to_color_idx = (self.current_color_idx + 1) % num_colors - to_color = self.colors[to_color_idx] + last_transition_start_time = utime.ticks_ms() + current_color_index = 0 + transition_duration_ms = self.delay # Use self.delay as the transition time - # Calculate interpolation factor (0.0 to 1.0) - # transition_step goes from 0 to transition_duration - 1 - if self.transition_duration > 0: - interp_factor = self.transition_step / self.transition_duration - else: - interp_factor = 1.0 # Immediately transition if duration is zero + while True: + color_from = self.colors[current_color_index] + color_to = self.colors[(current_color_index + 1) % len(self.colors)] - # Interpolate each color component - r = int(from_color[0] + (to_color[0] - from_color[0]) * interp_factor) - g = int(from_color[1] + (to_color[1] - from_color[1]) * interp_factor) - b = int(from_color[2] + (to_color[2] - from_color[2]) * interp_factor) + start_time = utime.ticks_ms() + elapsed_time = 0 - self.current_color = (r, g, b) - self.fill(self.apply_brightness(self.current_color)) + while elapsed_time < transition_duration_ms: + # Calculate the interpolation factor (0.0 to 1.0) + # Maximize to avoid division by zero if delay is 0, though a meaningful delay is expected + t = min(1.0, elapsed_time / max(1, transition_duration_ms)) - self.transition_step += self.delay # Advance the transition step by the delay + # Interpolate each color component + interpolated_color = ( + int(color_from[0] + (color_to[0] - color_from[0]) * t), + int(color_from[1] + (color_to[1] - color_from[1]) * t), + int(color_from[2] + (color_to[2] - color_from[2]) * t) + ) - if self.transition_step >= self.transition_duration: - # Transition complete, move to the next color and reset for hold phase - self.current_color_idx = to_color_idx - self.current_color = self.colors[self.current_color_idx] # Ensure current_color is the exact target color - self.transition_step = 0 # Reset transition progress - self.hold_start_time = current_time # Start hold phase for the new color + self.fill(self.apply_brightness(interpolated_color)) + await asyncio.sleep(0) # Update smoothly + elapsed_time = utime.ticks_diff(utime.ticks_ms(), start_time) - self.last_update = current_time + # Ensure the final color is set precisely after interpolation loop + self.fill(self.apply_brightness(color_to)) - def flicker_step(self): - current_time = utime.ticks_ms() - if utime.ticks_diff(current_time, self.last_update) >= self.delay/5: - base_color = self.colors[0] - # Increase the range for flicker_brightness_offset - # Changed from self.brightness // 4 to self.brightness // 2 (or even self.brightness for max intensity) - flicker_brightness_offset = random.randint(-int(self.brightness // 1.5), int(self.brightness // 1.5)) - flicker_brightness = max(0, min(255, self.brightness + flicker_brightness_offset)) + current_color_index = (current_color_index + 1) % len(self.colors) + await asyncio.sleep(0) # Yield control - flicker_color = self.apply_brightness(base_color, brightness_override=flicker_brightness) - self.fill(flicker_color) - self.last_update = current_time +async def main(): + w = WDT(timeout = 10000) + p = Patterns(num_leds=10, pin=10, color1=(16,16,0)) + # p.set_delay(100) + # await p.select("blink") + # await asyncio.sleep(2) + # p.set_delay(10) + # await p.select("rainbow") + # await asyncio.sleep(2) + # p.set_delay(100) + # await p.select("theater chase") + # await asyncio.sleep(2) + # p.set_colors([(255, 100, 0)]) # Set a base color for flicker (e.g., orange for a candle effect) + # p.set_brightness(200) # Set a brighter base for flicker to allow for dimming + # p.set_delay(100) # Faster updates for a more convincing flicker + # await p.select("flicker") + # await asyncio.sleep(2) + w.feed() +# Test the new color transition pattern + print("Starting color transition...") + p.set_colors([(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0)]) # Red, Green, Blue, Yellow + p.set_delay(1000) # 1 second transition between colors + p.set_brightness(150) + await p.select("color transition") + await asyncio.sleep(10) # Let it run for 10 seconds + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/settings.py b/src/settings.py index ea51d03..5481ed8 100644 --- a/src/settings.py +++ b/src/settings.py @@ -45,7 +45,7 @@ class Settings(dict): self.set_defaults() self.save() - def set_settings(self, data, patterns, save): + async def set_settings(self, data, patterns, save): try: print(data) for key, value in data.items(): @@ -56,13 +56,13 @@ class Settings(dict): buff.append(tuple(int(color[i:i+2], 16) for i in self.color_order)) patterns.set_colors(buff) elif key == "color1": - patterns.set_color1(tuple(int(value[i:i+2], 16) for i in self.color_order)) # Convert hex to RGB + patterns.set_color(0,(tuple(int(value[i:i+2], 16) for i in self.color_order))) # Convert hex to RGB elif key == "color2": - patterns.set_color2(tuple(int(value[i:i+2], 16) for i in self.color_order)) # Convert hex to RGB + patterns.set_color(1,(tuple(int(value[i:i+2], 16) for i in self.color_order))) # Convert hex to RGB elif key == "num_leds": patterns.update_num_leds(self["led_pin"], value) elif key == "pattern": - if not patterns.select(value): + if not await patterns.select(value): return "Pattern doesn't exist", 400 elif key == "delay": delay = int(data["delay"]) diff --git a/src/web.py b/src/web.py index a66c4f0..9c55d41 100644 --- a/src/web.py +++ b/src/web.py @@ -35,7 +35,7 @@ def web(settings, patterns): if data: # Process the received data - _, status_code = settings.set_settings(json.loads(data), patterns, True) + _, status_code = await settings.set_settings(json.loads(data), patterns, True) #await ws.send(status_code) else: break