Compare commits

...

1 Commits

Author SHA1 Message Date
jimmy e83f0d607c Switch to async patterns 2025-08-28 22:55:10 +12:00
5 changed files with 149 additions and 312 deletions

View File

@ -17,16 +17,11 @@ async def main():
patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"]) patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"])
if settings["color_order"] == "rbg": color_order = (1, 5, 3) if settings["color_order"] == "rbg": color_order = (1, 5, 3)
else: color_order = (1, 3, 5) 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_brightness(int(settings["brightness"]))
patterns.set_delay(int(settings["delay"])) patterns.set_delay(int(settings["delay"]))
async def tick():
while True:
patterns.tick()
await asyncio.sleep_ms(1)
w = web(settings, patterns) w = web(settings, patterns)
print(settings) print(settings)
# start the server in a bacakground task # start the server in a bacakground task
@ -35,7 +30,6 @@ async def main():
wdt = machine.WDT(timeout=10000) wdt = machine.WDT(timeout=10000)
wdt.feed() wdt.feed()
asyncio.create_task(tick())
asyncio.create_task(p2p(settings, patterns)) asyncio.create_task(p2p(settings, patterns))
while True: while True:

View File

@ -16,5 +16,5 @@ async def p2p(settings, patterns):
if "step" in settings and isinstance(settings["step"], int): if "step" in settings and isinstance(settings["step"], int):
patterns.set_pattern_step(settings["step"]) patterns.set_pattern_step(settings["step"])
else: 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") print("should not print")

View File

@ -1,4 +1,5 @@
from machine import Pin import asyncio
from machine import Pin, WDT
from neopixel import NeoPixel from neopixel import NeoPixel
import utime import utime
import random 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): 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.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
self.num_leds = num_leds self.num_leds = num_leds
self.pattern_step = 0
self.last_update = utime.ticks_ms()
self.delay = delay self.delay = delay
self.brightness = brightness self.brightness = brightness
self.patterns = { self.patterns = {
"off": self.off, "off": self.off,
"on" : self.on, "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 self.selected = selected
# Ensure colors list always starts with at least two for robust transition handling # 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 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 if not self.colors: # Ensure at least one color exists
self.colors = [(0, 0, 0)] 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): def update_num_leds(self, pin, num_leds):
self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds) self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
@ -65,124 +34,44 @@ class Patterns:
def set_delay(self, delay): def set_delay(self, delay):
self.delay = 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): def set_brightness(self, brightness):
self.brightness = 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): def set_colors(self, colors):
if colors and len(colors) >= 2: self.colors = colors
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): def set_color(self, num, color):
# Changed: More robust index check # Changed: More robust index check
if 0 <= num < len(self.colors): if 0 <= num < len(self.colors):
self.colors[num] = color 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 return True
elif num == len(self.colors): # Allow setting a new color at the end elif num == len(self.colors): # Allow setting a new color at the end
self.colors.append(color) self.colors.append(color)
return True return True
return False 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): def del_color(self, num):
# Changed: More robust index check and using del for lists # Changed: More robust index check and using del for lists
if 0 <= num < len(self.colors): if 0 <= num < len(self.colors):
del self.colors[num] 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 True
return False return False
def apply_brightness(self, color, brightness_override=None): def apply_brightness(self, color, brightness_override=None):
effective_brightness = brightness_override if brightness_override is not None else self.brightness effective_brightness = brightness_override if brightness_override is not None else self.brightness
return tuple(int(c * effective_brightness / 255) for c in color) 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 return False
self.sync() # Reset pattern state when selecting a new pattern self.selected = pattern
if pattern == "color_transition": if self.task is not None:
if len(self.colors) < 2: self.task.cancel()
print("Warning: 'color_transition' requires at least two colors. Switching to 'on'.") if reset: self.pattern_step = 0
self.selected = "on" # Fallback if not enough colors print(pattern)
self.sync() # Re-sync for the new pattern self.task = asyncio.create_task(self.patterns[pattern]())
else: return True
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): def set(self, i, color):
self.n[i] = color self.n[i] = color
@ -190,190 +79,144 @@ class Patterns:
def write(self): def write(self):
self.n.write() 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() self.n.write()
def off(self): async def off(self):
self.fill((0, 0, 0)) self.fill((0, 0, 0))
def on(self): async def on(self):
self.fill(self.apply_brightness(self.colors[0])) self.fill(self.apply_brightness(self.colors[0]))
def color_wipe_step(self): def sync(self):
color = self.apply_brightness(self.colors[0]) self.pattern_step = 0
current_time = utime.ticks_ms()
if utime.ticks_diff(current_time, self.last_update) >= self.delay: async def rainbow(self):
if self.pattern_step < self.num_leds: def wheel(pos):
for i in range(self.num_leds): if pos < 85:
self.n[i] = (0, 0, 0) return (pos * 3, 255 - pos * 3, 0)
self.n[self.pattern_step] = self.apply_brightness(color) elif pos < 170:
self.n.write() pos -= 85
self.pattern_step += 1 return (255 - pos * 3, 0, pos * 3)
else: else:
self.pattern_step = 0 pos -= 170
self.last_update = current_time 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() async def theater_chase(self):
if utime.ticks_diff(current_time, self.last_update) >= self.delay/5: last_update = utime.ticks_ms()
def wheel(pos): while True:
if pos < 85: if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
return (pos * 3, 255 - pos * 3, 0) for i in range(self.num_leds):
elif pos < 170: if (i + self.pattern_step) % 3 == 0:
pos -= 85 self.n[i] = self.apply_brightness(self.colors[0])
return (255 - pos * 3, 0, pos * 3) 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: else:
pos -= 170 self.on()
return (0, pos * 3, 255 - pos * 3) self.pattern_step = 1
last_update += self.delay
await asyncio.sleep(0)
for i in range(self.num_leds): async def flicker(self):
rc_index = (i * 256 // self.num_leds) + self.pattern_step last_update = utime.ticks_ms()
self.n[i] = self.apply_brightness(wheel(rc_index & 255)) while True:
self.n.write() if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
self.pattern_step = (self.pattern_step + 1) % 256 # Calculate a single flicker amount for all LEDs
self.last_update = current_time 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): 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)
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:
self.fill(self.apply_brightness(self.colors[0])) 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 return
from_color = self.colors[self.current_color_idx] last_transition_start_time = utime.ticks_ms()
to_color_idx = (self.current_color_idx + 1) % num_colors current_color_index = 0
to_color = self.colors[to_color_idx] transition_duration_ms = self.delay # Use self.delay as the transition time
# Calculate interpolation factor (0.0 to 1.0) while True:
# transition_step goes from 0 to transition_duration - 1 color_from = self.colors[current_color_index]
if self.transition_duration > 0: color_to = self.colors[(current_color_index + 1) % len(self.colors)]
interp_factor = self.transition_step / self.transition_duration
else:
interp_factor = 1.0 # Immediately transition if duration is zero
# Interpolate each color component start_time = utime.ticks_ms()
r = int(from_color[0] + (to_color[0] - from_color[0]) * interp_factor) elapsed_time = 0
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)
self.current_color = (r, g, b) while elapsed_time < transition_duration_ms:
self.fill(self.apply_brightness(self.current_color)) # 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: self.fill(self.apply_brightness(interpolated_color))
# Transition complete, move to the next color and reset for hold phase await asyncio.sleep(0) # Update smoothly
self.current_color_idx = to_color_idx elapsed_time = utime.ticks_diff(utime.ticks_ms(), start_time)
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 # Ensure the final color is set precisely after interpolation loop
self.fill(self.apply_brightness(color_to))
def flicker_step(self): current_color_index = (current_color_index + 1) % len(self.colors)
current_time = utime.ticks_ms() await asyncio.sleep(0) # Yield control
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) async def main():
self.fill(flicker_color) w = WDT(timeout = 10000)
self.last_update = current_time 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())

View File

@ -45,7 +45,7 @@ class Settings(dict):
self.set_defaults() self.set_defaults()
self.save() self.save()
def set_settings(self, data, patterns, save): async def set_settings(self, data, patterns, save):
try: try:
print(data) print(data)
for key, value in data.items(): 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)) buff.append(tuple(int(color[i:i+2], 16) for i in self.color_order))
patterns.set_colors(buff) patterns.set_colors(buff)
elif key == "color1": 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": 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": elif key == "num_leds":
patterns.update_num_leds(self["led_pin"], value) patterns.update_num_leds(self["led_pin"], value)
elif key == "pattern": elif key == "pattern":
if not patterns.select(value): if not await patterns.select(value):
return "Pattern doesn't exist", 400 return "Pattern doesn't exist", 400
elif key == "delay": elif key == "delay":
delay = int(data["delay"]) delay = int(data["delay"])

View File

@ -35,7 +35,7 @@ def web(settings, patterns):
if data: if data:
# Process the received 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) #await ws.send(status_code)
else: else:
break break