Switch to async patterns

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_color(0,(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(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:

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")

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,
+            "blink": self.blink,
-            "rainbow_cycle": self.rainbow_cycle_step,
+            "rainbow": self.rainbow,
-            "theater_chase": self.theater_chase_step,
+            "theater chase": self.theater_chase,
-            "blink": self.blink_step,
+            "flicker": self.flicker # Added flicker pattern
-            "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
         }
         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.task = None
-        self.transition_duration = delay * 50 # Default transition duration
+        self.pattern_step = 0
-        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]()
 
     def update_num_leds(self, pin, num_leds):
         self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
@@ -65,99 +34,27 @@ 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()
 
     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
 
@@ -165,24 +62,16 @@ class Patterns:
             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):
+    async def select(self, pattern, reset = True):
-        if pattern in self.patterns:
+        if pattern not 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
+        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,35 +79,20 @@ class Patterns:
     def write(self):
         self.n.write()
 
-    def fill(self, color=None):
+    def fill(self, color):
-        fill_color = color if color is not None else self.colors[0]
+        self.n.fill(color)
-        for i in range(self.num_leds):
-            self.n[i] = 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):
+    def sync(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
-            else:
         self.pattern_step = 0
-            self.last_update = current_time
 
-    def rainbow_cycle_step(self):
+    async def rainbow(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)
@@ -228,17 +102,22 @@ class Patterns:
             else:
                 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
-            self.last_update = current_time
+                last_update += self.delay
+                await asyncio.sleep(0)
 
-    def theater_chase_step(self):
+
-        current_time = utime.ticks_ms()
+    async def theater_chase(self):
-        if utime.ticks_diff(current_time, self.last_update) >= self.delay:
+        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])
@@ -246,134 +125,98 @@ class Patterns:
                         self.n[i] = (0, 0, 0)
                 self.n.write()
                 self.pattern_step = (self.pattern_step + 1) % 3
-            self.last_update = current_time
+                last_update += self.delay
+            await asyncio.sleep(0)
 
-    def blink_step(self):
+    async def blink(self):
-        current_time = utime.ticks_ms()
+        last_update = utime.ticks_ms()
-        if utime.ticks_diff(current_time, self.last_update) >= self.delay:
+        self.pattern_step = 0
-            if self.pattern_step % 2 == 0:
+        while True:
-                self.fill(self.apply_brightness(self.colors[0]))
+            if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
-            else:
+                if self.pattern_step:
-                self.fill((0, 0, 0))
+                    self.off()
-            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
+                    self.on()
-                    return (0, pos * 3, 255 - pos * 3)
+                    self.pattern_step = 1
+                last_update += self.delay
+                await asyncio.sleep(0)
 
-            random_offset = random.randint(0, 255)
+    async def flicker(self):
-            for i in range(self.num_leds):
+            last_update = utime.ticks_ms()
-                rc_index = (i * 256 // self.num_leds) + self.pattern_step + random_offset
+            while True:
-                self.n[i] = self.apply_brightness(wheel(rc_index & 255))
+                if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
-            self.n.write()
+                    # Calculate a single flicker amount for all LEDs
-            self.pattern_step = (self.pattern_step + 1) % 256
+                    flicker_amount = random.randint(int(-self.brightness // 1.5), int(self.brightness // 1.5))
-            self.last_update = current_time
+                    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 random_theater_chase_step(self):
+    async def color_transition(self):
-        current_time = utime.ticks_ms()
+            if len(self.colors) < 2:
-        if utime.ticks_diff(current_time, self.last_update) >= self.delay:
+                # If there's only one color or no colors, just display that color (or off)
-            color = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
+                self.fill(self.apply_brightness(self.colors[0]))
-            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
+            last_transition_start_time = utime.ticks_ms()
-        if utime.ticks_diff(current_time, self.last_update) >= self.delay:
+            current_color_index = 0
-            num_colors = len(self.colors)
+            transition_duration_ms = self.delay # Use self.delay as the transition time
-            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]
+            while True:
-            to_color_idx = (self.current_color_idx + 1) % num_colors
+                color_from = self.colors[current_color_index]
-            to_color = self.colors[to_color_idx]
+                color_to = self.colors[(current_color_index + 1) % len(self.colors)]
 
-            # Calculate interpolation factor (0.0 to 1.0)
+                start_time = utime.ticks_ms()
-            # transition_step goes from 0 to transition_duration - 1
+                elapsed_time = 0
-            if self.transition_duration > 0:
+
-                interp_factor = self.transition_step / self.transition_duration
+                while elapsed_time < transition_duration_ms:
-            else:
+                    # Calculate the interpolation factor (0.0 to 1.0)
-                interp_factor = 1.0 # Immediately transition if duration is zero
+                    # 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))
 
                     # Interpolate each color component
-            r = int(from_color[0] + (to_color[0] - from_color[0]) * interp_factor)
+                    interpolated_color = (
-            g = int(from_color[1] + (to_color[1] - from_color[1]) * interp_factor)
+                        int(color_from[0] + (color_to[0] - color_from[0]) * t),
-            b = int(from_color[2] + (to_color[2] - from_color[2]) * interp_factor)
+                        int(color_from[1] + (color_to[1] - color_from[1]) * t),
+                        int(color_from[2] + (color_to[2] - color_from[2]) * t)
+                    )
 
-            self.current_color = (r, g, b)
+                    self.fill(self.apply_brightness(interpolated_color))
-            self.fill(self.apply_brightness(self.current_color))
+                    await asyncio.sleep(0) # Update smoothly
+                    elapsed_time = utime.ticks_diff(utime.ticks_ms(), start_time)
 
-            self.transition_step += self.delay # Advance the transition step by the delay
+                # Ensure the final color is set precisely after interpolation loop
+                self.fill(self.apply_brightness(color_to))
 
-            if self.transition_step >= self.transition_duration:
+                current_color_index = (current_color_index + 1) % len(self.colors)
-                # Transition complete, move to the next color and reset for hold phase
+                await asyncio.sleep(0) # Yield control
-                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.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
 
-    def flicker_step(self):
+if __name__ == "__main__":
-        current_time = utime.ticks_ms()
+    asyncio.run(main())
-        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))
-
-            flicker_color = self.apply_brightness(base_color, brightness_override=flicker_brightness)
-            self.fill(flicker_color)
-            self.last_update = current_time

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"])

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