Initial commit

src/main.py

@@ -0,0 +1,34 @@
+from settings import Settings
+from machine import WDT
+from espnow import ESPNow
+import network
+from patterns import Patterns
+import json
+
+settings = Settings()
+print(settings)
+
+patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"])
+patterns.colors = [(8,0,0)]
+patterns.select("rainbow")
+
+wdt = WDT(timeout=10000)
+wdt.feed()
+
+sta_if = network.WLAN(network.STA_IF)
+sta_if.active(True)
+sta_if.disconnect()
+sta_if.config(channel=1)
+e = ESPNow()
+e.active(True)
+
+
+while True:
+    wdt.feed()
+    patterns.tick()
+    if e.any():
+        host, msg = e.recv()
+        data = json.loads(msg)
+        if settings.get("name") in data.get("names", []):
+            settings.set_settings(data.get("settings", {}), patterns, data.get("save", False))
+

src/p2p.py

@@ -0,0 +1,16 @@
+import asyncio
+import aioespnow
+import json
+
+async def p2p(settings, patterns):
+    e = aioespnow.AIOESPNow()  # Returns AIOESPNow enhanced with async support
+    e.active(True)
+    async for mac, msg in e:
+        try:
+            data = json.loads(msg)
+        except:
+            print(f"Failed to load espnow data {msg}")
+            continue
+ 
+        if "names" not in data or settings.get("name") in data.get("names", []):       
+            await settings.set_settings(data.get("settings", {}), patterns, data.get("save", False))

src/patterns.py

@@ -0,0 +1,322 @@
+import utime
+from patterns_base import Patterns as PatternsBase
+
+class Patterns(PatternsBase):
+    def __init__(self, pin, num_leds, color1=(0,0,0), color2=(0,0,0), brightness=127, selected="off", delay=100):
+        super().__init__(pin, num_leds, color1, color2, brightness, selected, delay)
+        self.auto = True
+        self.step = 0
+        self.patterns = {
+            "off": self.off,
+            "on" : self.on,
+            "blink": self.blink,
+            "rainbow": self.rainbow,
+            "pulse": self.pulse,
+            "transition": self.transition,
+            "chase": self.chase,
+            "circle": self.circle,
+        }
+
+
+    def blink(self):
+        state = True  # True = on, False = off
+        last_update = utime.ticks_ms()
+
+        while True:
+            current_time = utime.ticks_ms()
+            if utime.ticks_diff(current_time, last_update) >= self.delay:
+                if state:
+                    self.fill(self.apply_brightness(self.colors[0]))
+                else:
+                    self.fill((0, 0, 0))
+                state = not state
+                last_update = current_time
+            # Yield once per tick so other logic can run
+            yield
+
+
+    def rainbow(self):
+        step = self.step % 256
+        step_amount = max(1, int(self.n1))  # n1 controls step increment
+
+        # If auto is False, run a single step and then stop
+        if not self.auto:
+            for i in range(self.num_leds):
+                rc_index = (i * 256 // self.num_leds) + step
+                self.n[i] = self.apply_brightness(self.wheel(rc_index & 255))
+            self.n.write()
+            # Increment step by n1 for next manual call
+            self.step = (step + step_amount) % 256
+            # Allow tick() to advance the generator once
+            yield
+            return
+        
+        last_update = utime.ticks_ms()
+
+        while True:
+            current_time = utime.ticks_ms()
+            sleep_ms = max(1, int(self.delay))  # Access delay directly
+            if utime.ticks_diff(current_time, last_update) >= sleep_ms:
+                for i in range(self.num_leds):
+                    rc_index = (i * 256 // self.num_leds) + step
+                    self.n[i] = self.apply_brightness(self.wheel(rc_index & 255))
+                self.n.write()
+                step = (step + step_amount) % 256
+                self.step = step
+                last_update = current_time
+            # Yield once per tick so other logic can run
+            yield
+
+   
+    def pulse(self):
+        self.off()
+
+        # Ensure we have at least one color
+        if not self.colors:
+            self.colors = [(255, 255, 255)]
+
+        color_index = 0
+        cycle_start = utime.ticks_ms()
+
+        # State machine based pulse using a single generator loop
+        while True:
+            # Read current timing parameters each cycle so they can be changed live
+            attack_ms = max(0, int(self.n1))  # Attack time in ms
+            hold_ms = max(0, int(self.n2))    # Hold time in ms
+            decay_ms = max(0, int(self.n3))   # Decay time in ms
+            delay_ms = max(0, int(self.delay))
+
+            total_ms = attack_ms + hold_ms + decay_ms + delay_ms
+            if total_ms <= 0:
+                total_ms = 1
+
+            now = utime.ticks_ms()
+            elapsed = utime.ticks_diff(now, cycle_start)
+
+            base_color = self.colors[color_index % len(self.colors)]
+
+            if elapsed < attack_ms and attack_ms > 0:
+                # Attack: fade 0 -> 1
+                factor = elapsed / attack_ms
+                color = tuple(int(c * factor) for c in base_color)
+                self.fill(self.apply_brightness(color))
+            elif elapsed < attack_ms + hold_ms:
+                # Hold: full brightness
+                self.fill(self.apply_brightness(base_color))
+            elif elapsed < attack_ms + hold_ms + decay_ms and decay_ms > 0:
+                # Decay: fade 1 -> 0
+                dec_elapsed = elapsed - attack_ms - hold_ms
+                factor = max(0.0, 1.0 - (dec_elapsed / decay_ms))
+                color = tuple(int(c * factor) for c in base_color)
+                self.fill(self.apply_brightness(color))
+            elif elapsed < total_ms:
+                # Delay phase: LEDs off between pulses
+                self.fill((0, 0, 0))
+            else:
+                # End of cycle, move to next color and restart timing
+                color_index += 1
+                cycle_start = now
+                if not self.auto:
+                    break
+                # Skip drawing this tick, start next cycle
+                yield
+                continue
+
+            # Yield once per tick
+            yield
+
+    def transition(self):
+        """Transition between colors, blending over `delay` ms."""
+        if not self.colors:
+            self.off()
+            yield
+            return
+
+        # Only one color: just keep it on
+        if len(self.colors) == 1:
+            while True:
+                self.fill(self.apply_brightness(self.colors[0]))
+                yield
+            return
+
+        color_index = 0
+        start_time = utime.ticks_ms()
+
+        while True:
+            if not self.colors:
+                break
+
+            # Get current and next color based on live list
+            c1 = self.colors[color_index % len(self.colors)]
+            c2 = self.colors[(color_index + 1) % len(self.colors)]
+
+            duration = max(10, int(self.delay))  # At least 10ms
+            now = utime.ticks_ms()
+            elapsed = utime.ticks_diff(now, start_time)
+
+            if elapsed >= duration:
+                # End of this transition step
+                if not self.auto and color_index >= 0:
+                    # One-shot: transition from first to second color only
+                    self.fill(self.apply_brightness(c2))
+                    break
+                # Auto: move to next pair
+                color_index = (color_index + 1) % len(self.colors)
+                start_time = now
+                yield
+                continue
+
+            # Interpolate between c1 and c2
+            factor = elapsed / duration
+            interpolated = tuple(
+                int(c1[i] + (c2[i] - c1[i]) * factor) for i in range(3)
+            )
+            self.fill(self.apply_brightness(interpolated))
+
+            yield
+
+    def chase(self):
+        """Chase pattern: n1 LEDs of color0, n2 LEDs of color1, repeating.
+        Moves by n3 on even steps, n4 on odd steps (n3/n4 can be positive or negative)"""
+        if len(self.colors) < 1:
+            # Need at least 1 color
+            return
+        
+        segment_length = 0  # Will be calculated in loop
+        position = 0  # Current position offset
+        step_count = 0  # Track which step we're on
+        
+        last_update = utime.ticks_ms()
+
+        while True:
+            # Access colors, delay, and n values directly for live updates
+            if not self.colors:
+                break
+            # If only one color provided, use it for both colors
+            if len(self.colors) < 2:
+                color0 = self.colors[0]
+                color1 = self.colors[0]
+            else:
+                color0 = self.colors[0]
+                color1 = self.colors[1]
+            
+            color0 = self.apply_brightness(color0)
+            color1 = self.apply_brightness(color1)
+            
+            n1 = max(1, int(self.n1))  # LEDs of color 0
+            n2 = max(1, int(self.n2))  # LEDs of color 1
+            n3 = int(self.n3)  # Step movement on odd steps (can be negative)
+            n4 = int(self.n4)  # Step movement on even steps (can be negative)
+            
+            segment_length = n1 + n2
+            transition_duration = max(10, int(self.delay))
+            
+            current_time = utime.ticks_ms()
+            if utime.ticks_diff(current_time, last_update) >= transition_duration:
+                # Clear all LEDs
+                self.n.fill((0, 0, 0))
+                
+                # Draw repeating pattern starting at position
+                for i in range(self.num_leds):
+                    # Calculate position in the repeating segment
+                    relative_pos = (i - position) % segment_length
+                    if relative_pos < 0:
+                        relative_pos = (relative_pos + segment_length) % segment_length
+                    
+                    # Determine which color based on position in segment
+                    if relative_pos < n1:
+                        self.n[i] = color0
+                    else:
+                        self.n[i] = color1
+                
+                self.n.write()
+                
+                # Move position by n3 or n4 on alternate steps
+                if step_count % 2 == 0:
+                    position = position + n3
+                else:
+                    position = position + n4
+                
+                # Wrap position to keep it reasonable
+                max_pos = self.num_leds + segment_length
+                position = position % max_pos
+                if position < 0:
+                    position += max_pos
+                
+                step_count += 1
+                last_update = current_time
+
+            # Yield once per tick so other logic can run
+            yield
+
+    def circle(self):
+        """Circle loading pattern - grows to n2, then tail moves forward at n3 until min length n4"""
+        head = 0
+        tail = 0
+        
+        # Calculate timing
+        head_rate = max(1, int(self.n1))  # n1 = head moves per second
+        tail_rate = max(1, int(self.n3))  # n3 = tail moves per second
+        max_length = max(1, int(self.n2))  # n2 = max length
+        min_length = max(0, int(self.n4))  # n4 = min length
+        
+        head_delay = 1000 // head_rate  # ms between head movements
+        tail_delay = 1000 // tail_rate  # ms between tail movements
+        
+        last_head_move = utime.ticks_ms()
+        last_tail_move = utime.ticks_ms()
+        
+        phase = "growing"  # "growing", "shrinking", or "off"
+        
+        while True:
+            current_time = utime.ticks_ms()
+            
+            # Clear all LEDs
+            self.n.fill((0, 0, 0))
+            
+            # Calculate segment length
+            segment_length = (head - tail) % self.num_leds
+            if segment_length == 0 and head != tail:
+                segment_length = self.num_leds
+            
+            # Draw segment from tail to head
+            color = self.apply_brightness(self.colors[0])
+            for i in range(segment_length + 1):
+                led_pos = (tail + i) % self.num_leds
+                self.n[led_pos] = color
+            
+            # Move head continuously at n1 LEDs per second
+            if utime.ticks_diff(current_time, last_head_move) >= head_delay:
+                head = (head + 1) % self.num_leds
+                last_head_move = current_time
+            
+            # Tail behavior based on phase
+            if phase == "growing":
+                # Growing phase: tail stays at 0 until max length reached
+                if segment_length >= max_length:
+                    phase = "shrinking"
+            elif phase == "shrinking":
+                # Shrinking phase: move tail forward at n3 LEDs per second
+                if utime.ticks_diff(current_time, last_tail_move) >= tail_delay:
+                    tail = (tail + 1) % self.num_leds
+                    last_tail_move = current_time
+                    
+                    # Check if we've reached min length
+                    current_length = (head - tail) % self.num_leds
+                    if current_length == 0 and head != tail:
+                        current_length = self.num_leds
+                    
+                    # For min_length = 0, we need at least 1 LED (the head)
+                    if min_length == 0 and current_length <= 1:
+                        phase = "off"  # All LEDs off for 1 step
+                    elif min_length > 0 and current_length <= min_length:
+                        phase = "growing"  # Cycle repeats
+            else:  # phase == "off"
+                # Off phase: all LEDs off for 1 step, then restart
+                tail = head  # Reset tail to head position to start fresh
+                phase = "growing"
+            
+            self.n.write()
+
+            # Yield once per tick so other logic can run
+            yield

src/patterns_base.py

@@ -0,0 +1,145 @@
+from machine import Pin
+from neopixel import NeoPixel
+import utime
+
+
+
+
+# Short-key parameter mapping for convenience setters
+param_mapping = {
+    "pt": "selected",
+    "pa": "selected",
+    "cl": "colors",
+    "br": "brightness",
+    "dl": "delay",
+    "nl": "num_leds",
+    "co": "color_order",
+    "lp": "led_pin",
+    "n1": "n1",
+    "n2": "n2",
+    "n3": "n3",
+    "n4": "n4",
+    "n5": "n5",
+    "n6": "n6",
+    "auto": "auto",
+}
+
+class Patterns:
+    def __init__(self, pin, num_leds, color1=(0,0,0), color2=(0,0,0), brightness=127, selected="off", 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.auto = False
+        self.patterns = {}
+        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
+
+        self.n1 = 0
+        self.n2 = 0
+        self.n3 = 0
+        self.n4 = 0
+        self.n5 = 0
+        self.n6 = 0       
+
+        self.generator = None
+        self.select(self.selected)       
+
+
+    def tick(self):
+        if self.generator is None:
+            return
+        try:
+            next(self.generator)
+        except StopIteration:
+            self.generator = None
+
+    def select(self, pattern):
+        if pattern in self.patterns:
+            self.selected = pattern
+            self.generator = self.patterns[pattern]()
+            print(f"Selected pattern: {pattern}")
+            return True
+        # If pattern doesn't exist, default to "off"
+        return False
+
+    def set_param(self, key, value):
+        if key in param_mapping:
+            setattr(self, param_mapping[key], value)
+            return True
+        print(f"Invalid parameter: {key}")
+        return False
+
+    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_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
+           return True
+        elif num == len(self.colors): # Allow setting a new color at the end
+            self.colors.append(color)
+            return True
+        return False
+
+
+    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]
+           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 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 wheel(self, 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)
+
+    

src/settings.py

@@ -0,0 +1,131 @@
+import json
+import ubinascii
+import machine
+import network
+
+class Settings(dict):
+    SETTINGS_FILE = "/settings.json"
+
+    def __init__(self):
+        super().__init__()
+        self.load()  # Load settings from file during initialization
+        self.color_order = self.get_color_order(self["color_order"])
+
+    def set_defaults(self):
+        self["led_pin"] = 10
+        self["num_leds"] = 50
+        self["pattern"] = "on"
+        self["delay"] = 100
+        self["brightness"] = 10
+        self["color_order"] = "rgb"
+        self["name"] = f"led-{ubinascii.hexlify(network.WLAN(network.AP_IF).config('mac')).decode()}"
+        self["ap_password"] = ""
+        self["id"] = 0
+        self["debug"] = False
+
+    def save(self):
+        try:
+            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}")
+
+    def load(self):
+        try:
+            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()
+            self.save()
+
+    def set_settings(self, data, patterns, save):
+        try:
+            print(f"Setting settings: {data}")
+            for key, value in data.items():
+                print(key, value)
+                if key == "colors":
+                    buff = []
+                    for color in value:
+                        buff.append(tuple(int(color[i:i+2], 16) for i in self.color_order))
+                    patterns.colors = buff
+                elif key == "num_leds":
+                    patterns.update_num_leds(self["led_pin"], value)
+                elif key ==  "pattern":
+                    if not patterns.select(value):
+                        return "Pattern doesn't exist", 400
+                elif key == "delay":
+                    delay = int(data["delay"])
+                    patterns.delay = delay
+                elif key == "brightness":
+                    brightness = int(data["brightness"])
+                    patterns.brightness = brightness
+                elif key == "n1":
+                    patterns.n1 = value
+                elif key == "n2":
+                    patterns.n2 = value
+                elif key == "n3":
+                    patterns.n3 = value
+                elif key == "n4":
+                    patterns.n4 = value
+                elif key == "n5":
+                    patterns.n5 = value
+                elif key == "n6":
+                    patterns.n6 = value
+                elif key == "name":
+                    self[key] = value
+                    self.save()
+                    machine.reset()
+                elif key == "color_order":
+                    self["color_order"] = value
+                    self.color_order = self.get_color_order(value)
+                    pass
+                elif key == "id":
+                    pass
+                elif key == "led_pin":
+                    patterns.update_num_leds(value, self["num_leds"])
+                else:
+                    return "Invalid key", 400
+                self[key] = value
+            #print(self)
+            if save:
+                self.save()
+            print(self)
+            return "OK", 200
+        except (KeyError, ValueError):
+            return "Bad request", 400
+
+    def get_color_order(self, color_order):
+        """Convert color order string to tuple of hex string indices."""
+        color_orders = {
+            "rgb": (1, 3, 5),
+            "rbg": (1, 5, 3),
+            "grb": (3, 1, 5),
+            "gbr": (3, 5, 1),
+            "brg": (5, 1, 3),
+            "bgr": (5, 3, 1)
+        }
+        return color_orders.get(color_order.lower(), (1, 3, 5))  # Default to RGB
+
+# Example usage
+def main():
+    settings = Settings()
+    print(f"Number of LEDs: {settings['num_leds']}")
+    settings['num_leds'] = 100
+    print(f"Updated number of LEDs: {settings['num_leds']}")
+    settings.save()
+
+    # Create a new Settings object to test loading
+    new_settings = Settings()
+    print(f"Loaded number of LEDs: {new_settings['num_leds']}")
+    print(settings)
+
+
+
+# Run the example
+if __name__ == "__main__":
+    main()