Switch to using itterators

src/boot.py

@@ -6,4 +6,4 @@ s = Settings()
 
 name = s.get('name', 'led')
 password = s.get("ap_password", "")
-wifi.ap(name, password)
+# wifi.ap(name, password)

src/main.py

@@ -10,8 +10,9 @@ import time
 import wifi
 import json
 from p2p import p2p
+import espnow
+import network
 
-async def main():
 settings = Settings()
 
 patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"])
@@ -22,33 +23,33 @@ async def main():
 patterns.set_brightness(int(settings["brightness"]))
 patterns.set_delay(int(settings["delay"]))
 
-    async def tick():
+sta = network.WLAN(network.STA_IF)
-        while True:
+sta.active(True)
-            patterns.tick()
+sta.disconnect()   # Because ESP8266 auto-connects to last Access Point
-            await asyncio.sleep_ms(0)
 
-    async def system():
+e = espnow.ESPNow()
-        while True:
+e.active(True)
-            gc.collect()
-            for i in range(60):
-                wdt.feed()
-                await asyncio.sleep(1)
 
-    w = web(settings, patterns)
 print(settings)
 # start the server in a bacakground task
 print("Starting")
-    server = asyncio.create_task(w.start_server(host="0.0.0.0", port=80))
+
 wdt = machine.WDT(timeout=10000)
+
+while True:
+
     wdt.feed()
-
+    patterns.tick()
-    asyncio.create_task(tick())
+    host, msg = e.recv(0)
-    asyncio.create_task(p2p(settings, patterns))
+    if msg:
-    asyncio.create_task(system())
+        try:
-
+            data = json.loads(msg)
-
+        except:
-
+            print(f"Failed to load espnow data {msg}")
-    # cleanup before ending the application
+            continue
-    await server
+        print(data)
-
+        if "names" not in data or settings.get("name") in data.get("names", []):
-asyncio.run(main())
+            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))

src/patterns.py

@@ -2,211 +2,26 @@ from machine import Pin
 from neopixel import NeoPixel
 import utime
 import random
+from patterns_base import PatternsBase
 
-class Patterns:
+class Patterns(PatternsBase):
     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)
+        super().__init__(pin, num_leds, color1, color2, brightness, selected, delay)
-        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,
+            "off": self.off(),
-            "on" : self.on,
+            "on" : self.on(),
-            "color_wipe": self.color_wipe_step,
+            "color_wipe": self.color_wipe_step(),
-            "rainbow_cycle": self.rainbow_cycle_step,
+            "rainbow_cycle": self.rainbow_cycle_step(),
-            "theater_chase": self.theater_chase_step,
+            "theater_chase": self.theater_chase_step(),
-            "blink": self.blink_step,
+            "blink": self.blink_step(),
-            "color_transition": self.color_transition_step, # Added new pattern
+            "color_transition": self.color_transition_step(), # Added new pattern
-            "flicker": self.flicker_step,
+            "flicker": self.flicker_step(),
-            "scanner": self.scanner_step, # New: Single direction scanner
+            "scanner": self.scanner_step(), # New: Single direction scanner
-            "bidirectional_scanner": self.bidirectional_scanner_step, # New: Bidirectional scanner
+            "bidirectional_scanner": self.bidirectional_scanner_step(), # New: Bidirectional scanner
-            "external": None,
-            "pulse": self.pulse
         }
-        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
-
-        # New attributes for scanner patterns
-        self.scanner_direction = 1 # 1 for forward, -1 for backward
-        self.scanner_tail_length = 3 # Number of trailing pixels
-
-    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
-        # Reset scanner specific variables
-        self.scanner_direction = 1
-        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)
-        self.num_leds = num_leds
-        self.pattern_step = 0
-
-    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
-
-    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)
-
-    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
-
-    def set(self, i, color):
-        self.n[i] = color
-
-    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
-        self.n.write()
-
-    def off(self):
-        self.fill((0, 0, 0))
-
-    def on(self):
-        self.fill(self.apply_brightness(self.colors[0]))
 
     def color_wipe_step(self):
+        while True:
             color = self.apply_brightness(self.colors[0])
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay:
@@ -219,8 +34,10 @@ class Patterns:
                 else:
                     self.pattern_step = 0
                 self.last_update = current_time
+            yield
 
     def rainbow_cycle_step(self):
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay/5:
                 def wheel(pos):
@@ -239,8 +56,10 @@ class Patterns:
                 self.n.write()
                 self.pattern_step = (self.pattern_step + 1) % 256
                 self.last_update = current_time
+            yield
 
     def theater_chase_step(self):
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay:
                 for i in range(self.num_leds):
@@ -251,8 +70,10 @@ class Patterns:
                 self.n.write()
                 self.pattern_step = (self.pattern_step + 1) % 3
                 self.last_update = current_time
+            yield
 
     def blink_step(self):
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay:
                 if self.pattern_step % 2 == 0:
@@ -261,8 +82,10 @@ class Patterns:
                     self.fill((0, 0, 0))
                 self.pattern_step = (self.pattern_step + 1) % 2
                 self.last_update = current_time
+            yield
 
     def color_transition_step(self):
+        while True:
             current_time = utime.ticks_ms()
 
             # Check for hold duration first
@@ -270,7 +93,7 @@ class Patterns:
                 # 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
+                yield
 
             # If hold duration is over, proceed with transition
             if utime.ticks_diff(current_time, self.last_update) >= self.delay:
@@ -278,7 +101,7 @@ class Patterns:
                 if num_colors < 2:
                     # Should not happen if select handles it, but as a safeguard
                     self.select("on")
-                return
+                    yield
 
                 from_color = self.colors[self.current_color_idx]
                 to_color_idx = (self.current_color_idx + 1) % num_colors
@@ -309,8 +132,10 @@ class Patterns:
                     self.hold_start_time = current_time # Start hold phase for the new color
 
                 self.last_update = current_time
+            yield
 
     def flicker_step(self):
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay/5:
                 base_color = self.colors[0]
@@ -322,11 +147,13 @@ class Patterns:
                 flicker_color = self.apply_brightness(base_color, brightness_override=flicker_brightness)
                 self.fill(flicker_color)
                 self.last_update = current_time
+            yield
 
     def scanner_step(self):
         """
         Mimics a 'Knight Rider' style scanner, moving in one direction.
         """
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay:
                 self.fill((0, 0, 0)) # Clear all LEDs
@@ -356,11 +183,13 @@ class Patterns:
                     self.pattern_step = 0 # Reset to start
 
                 self.last_update = current_time
+            yield
 
     def bidirectional_scanner_step(self):
         """
         Mimics a 'Knight Rider' style scanner, moving back and forth.
         """
+        while True:
             current_time = utime.ticks_ms()
             if utime.ticks_diff(current_time, self.last_update) >= self.delay/100:
                 self.fill((0, 0, 0)) # Clear all LEDs
@@ -395,23 +224,4 @@ class Patterns:
                     self.pattern_step = 0 # Start moving forward from the first LED
 
                 self.last_update = current_time
-
+            yield
-    def pulse(self):
-        if self.pattern_step == 0:
-            self.fill(self.apply_brightness(self.colors[0]))
-            self.pattern_step = 1
-            self.last_update = utime.ticks_ms()
-        if utime.ticks_diff(utime.ticks_ms(), self.last_update) > self.delay:
-            self.fill((0, 0, 0))
-
-
-if __name__ == "__main__":
-    import time
-    from machine import WDT
-    wdt = WDT(timeout=2000)  # Enable watchdog with a 2 second timeout
-    p = Patterns(pin=10, num_leds=200, color1=(255,0,0), color2=(0,0,255), brightness=127, selected="bidirectional_scanner", delay=50)
-    p.select("pulse")
-    for i in range(1000):
-        p.tick()
-        wdt.feed()
-        time.sleep_ms(1)

src/patterns_base.py

@@ -2,16 +2,16 @@ from machine import Pin
 from neopixel import NeoPixel
 import utime
 
-
+class PatternsBase:
-class PatternBase:
     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 = {}
+        self. run = True
         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
@@ -29,6 +29,8 @@ class PatternBase:
         # New attributes for scanner patterns
         self.scanner_direction = 1 # 1 for forward, -1 for backward
         self.scanner_tail_length = 3 # Number of trailing pixels
+        self.patterns = {}
+        self.run = True
 
     def sync(self):
         self.pattern_step=0
@@ -45,10 +47,6 @@ class PatternBase:
     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)
         self.num_leds = num_leds
@@ -60,7 +58,6 @@ class PatternBase:
         self.transition_duration = self.delay * 50
         self.hold_duration = self.delay * 10
 
-
     def set_brightness(self, brightness):
         self.brightness = brightness
 
@@ -157,8 +154,9 @@ class PatternBase:
         return tuple(int(c * effective_brightness / 255) for c in color)
 
     def select(self, pattern):
-        if pattern in self.patterns:
+        if pattern in self.patterns and hasattr(self.patterns[pattern], "__next__"):
             self.selected = pattern
+            self.run = True
             self.sync() # Reset pattern state when selecting a new pattern
             if pattern == "color_transition":
                 if len(self.colors) < 2:
@@ -175,6 +173,13 @@ class PatternBase:
             return True
         return False
 
+    def tick(self):
+        if self.run:
+            try:
+                next(self.patterns[self.selected])
+            except StopIteration:
+                self.run = False
+
     def set(self, i, color):
         self.n[i] = color
 
@@ -188,7 +193,13 @@ class PatternBase:
         self.n.write()
 
     def off(self):
+        while True:
             self.fill((0, 0, 0))
+            self.run = False
+            yield
 
     def on(self):
+        while True:
             self.fill(self.apply_brightness(self.colors[0]))
+            self.run = False
+            yield