20 Commits

Author SHA1 Message Date
e83f0d607c Switch to async patterns 2025-08-28 22:55:10 +12:00
d2826a0f63 Swtich to names isntead of ids 2025-07-12 10:22:17 +12:00
87fc74bb51 Add flicker pattern 2025-07-12 10:21:43 +12:00
03f3f02da8 Remove wifi client 2025-06-19 19:13:13 +12:00
524db5e979 Move espnow to seperate file 2025-06-19 19:05:08 +12:00
279416cded Add set_pattern_step 2025-06-19 19:03:22 +12:00
fbd14f2e16 If no ids run set_settings 2025-06-12 21:32:24 +12:00
1989f6f5c9 Switch to list for colors 2025-06-12 21:29:31 +12:00
a19b1e86f2 Have to save when using espnow 2025-06-08 13:18:28 +12:00
c63e907204 espnow if id is 0 call set_settings 2025-06-04 21:07:15 +12:00
b7920e224f Add color order 2025-06-04 21:02:55 +12:00
42e92dafc8 Add led pin setting 2025-06-04 20:03:51 +12:00
0b6eb9724f Add device ID 2025-06-04 19:54:06 +12:00
55ef5c1580 Move json load out of set_settings 2025-06-04 19:22:09 +12:00
c15f9787a7 Sync after going through all the keys 2025-06-02 00:32:43 +12:00
3d0078f118 Update leds straight away after a sync 2025-06-02 00:18:25 +12:00
9e72dba035 Check all keys before returning 2025-06-02 00:16:58 +12:00
3d7dd754eb Wifi and ESPNOW don't work at the same time 2025-06-02 00:15:31 +12:00
2dd20fa51b Enable garabage collection 2025-05-28 21:19:56 +12:00
d33bd6b0e4 Enable watchdog timer 2025-05-28 21:17:23 +12:00
8 changed files with 279 additions and 251 deletions

View File

@@ -8,7 +8,8 @@ import utime
import machine
import time
import wifi
import json
from p2p import p2p
async def main():
settings = Settings()
@@ -16,57 +17,28 @@ async def main():
patterns = Patterns(settings["led_pin"], settings["num_leds"], selected=settings["pattern"])
if settings["color_order"] == "rbg": color_order = (1, 5, 3)
else: color_order = (1, 3, 5)
patterns.set_color1(tuple(int(settings["color1"][i:i+2], 16) for i in color_order))
patterns.set_color2(tuple(int(settings["color2"][i:i+2], 16) for i in color_order))
patterns.set_color(0,(tuple(int(settings["color1"][i:i+2], 16) for i in color_order)))
patterns.set_color(1,(tuple(int(settings["color2"][i:i+2], 16) for i in color_order)))
patterns.set_brightness(int(settings["brightness"]))
patterns.set_delay(int(settings["delay"]))
async def tick():
while True:
patterns.tick()
await asyncio.sleep_ms(1)
async def espnow():
e = aioespnow.AIOESPNow() # Returns AIOESPNow enhanced with async support
e.active(True)
async for mac, msg in e:
print(msg)
settings.set_settings(msg, patterns)
print("should not print")
async def wifi_connect():
for i in range(10):
config = wifi.connect(settings.get("wifi_ssid", ""),
settings.get("wifi_password", ""),
settings.get("wifi_ip", ""),
settings.get("wifi_gateway", "")
)
if config:
print(config)
break
await asyncio.sleep_ms(500)
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)
#wdt.feed()
wdt = machine.WDT(timeout=10000)
wdt.feed()
asyncio.create_task(wifi_connect())
asyncio.create_task(tick())
asyncio.create_task(espnow())
asyncio.create_task(p2p(settings, patterns))
while True:
#print(time.localtime())
# gc.collect()
for i in range(60):
#wdt.feed()
await asyncio.sleep_ms(500)
gc.collect()
for i in range(20):
wdt.feed()
await asyncio.sleep_ms(1000)
# cleanup before ending the application
await server

20
src/p2p.py Normal file
View File

@@ -0,0 +1,20 @@
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
print(data)
if "names" not in data or settings.get("name") in data.get("names", []):
if "step" in settings and isinstance(settings["step"], int):
patterns.set_pattern_step(settings["step"])
else:
await settings.set_settings(data.get("settings", {}), patterns, data.get("save", False))
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
import utime
import random
@@ -7,38 +8,24 @@ class Patterns:
def __init__(self, pin, num_leds, color1=(0,0,0), color2=(0,0,0), brightness=127, selected="rainbow_cycle", delay=100):
self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
self.num_leds = num_leds
self.pattern_step = 0
self.last_update = utime.ticks_ms()
self.delay = delay
self.brightness = brightness
self.patterns = {
"off": self.off,
"on" : self.on,
"color_wipe": self.color_wipe_step,
"rainbow_cycle": self.rainbow_cycle_step,
"theater_chase": self.theater_chase_step,
"blink": self.blink_step,
"random_color_wipe": self.random_color_wipe_step,
"random_rainbow_cycle": self.random_rainbow_cycle_step,
"random_theater_chase": self.random_theater_chase_step,
"random_blink": self.random_blink_step,
"color_transition": self.color_transition_step,
"external": None
"blink": self.blink,
"rainbow": self.rainbow,
"theater chase": self.theater_chase,
"flicker": self.flicker # Added flicker pattern
}
self.selected = selected
self.color1 = color1
self.color2 = color2
self.transition_duration = delay * 10 # Default transition duration is 10 times the delay
self.transition_step = 0
self.current_color = self.color1
def sync(self):
self.pattern_step=0
self.last_update = utime.ticks_ms()
def tick(self):
if self.patterns[self.selected]:
self.patterns[self.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.pattern_step = 0
def update_num_leds(self, pin, num_leds):
self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
@@ -47,215 +34,189 @@ class Patterns:
def set_delay(self, delay):
self.delay = delay
# Update transition duration when delay changes for color_transition pattern
if self.selected == "color_transition":
self.transition_duration = self.delay * 10 # Or some other multiplier
def set_brightness(self, brightness):
self.brightness = brightness
def set_color1(self, color):
self.color1 = color
if self.selected == "color_transition":
self.transition_step = 0
self.current_color = self.color1
def set_colors(self, colors):
self.colors = colors
def set_color2(self, color):
self.color2 = color
if self.selected == "color_transition":
self.transition_step = 0
self.current_color = self.color1
def apply_brightness(self, color):
return tuple(int(c * self.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":
self.transition_step = 0
self.current_color = self.color1
self.transition_duration = self.delay * 10 # Initialize transition duration
def set_color(self, num, color):
# Changed: More robust index check
if 0 <= num < len(self.colors):
self.colors[num] = color
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)
async def select(self, pattern, reset = True):
if pattern not in self.patterns:
return False
self.selected = pattern
if self.task is not None:
self.task.cancel()
if reset: self.pattern_step = 0
print(pattern)
self.task = asyncio.create_task(self.patterns[pattern]())
return True
def set(self, i, color):
self.n[i] = color
def write(self):
self.n.write()
def fill(self, color=None):
fill_color = color if color is not None else self.color1
for i in range(self.num_leds):
self.n[i] = fill_color
def fill(self, color):
self.n.fill(color)
self.n.write()
def off(self):
async def off(self):
self.fill((0, 0, 0))
def on(self):
self.fill(self.apply_brightness(self.color1))
async def on(self):
self.fill(self.apply_brightness(self.colors[0]))
def sync(self):
self.pattern_step = 0
def color_wipe_step(self):
color = self.apply_brightness(self.color1)
current_time = utime.ticks_ms()
if utime.ticks_diff(current_time, self.last_update) >= self.delay:
if self.pattern_step < self.num_leds:
async def rainbow(self):
def wheel(pos):
if pos < 85:
return (pos * 3, 255 - pos * 3, 0)
elif pos < 170:
pos -= 85
return (255 - pos * 3, 0, pos * 3)
else:
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):
self.n[i] = (0, 0, 0)
self.n[self.pattern_step] = self.apply_brightness(color)
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 += 1
else:
self.pattern_step = 0
self.last_update = current_time
self.pattern_step = (self.pattern_step + 1) % 256
last_update += self.delay
await asyncio.sleep(0)
def rainbow_cycle_step(self):
current_time = utime.ticks_ms()
if utime.ticks_diff(current_time, self.last_update) >= self.delay/5:
def wheel(pos):
if pos < 85:
return (pos * 3, 255 - pos * 3, 0)
elif pos < 170:
pos -= 85
return (255 - pos * 3, 0, pos * 3)
else:
pos -= 170
return (0, pos * 3, 255 - pos * 3)
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
def theater_chase_step(self):
current_time = utime.ticks_ms()
if utime.ticks_diff(current_time, self.last_update) >= self.delay:
for i in range(self.num_leds):
if (i + self.pattern_step) % 3 == 0:
self.n[i] = self.apply_brightness(self.color1)
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.color1))
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:
async def theater_chase(self):
last_update = utime.ticks_ms()
while True:
if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
for i in range(self.num_leds):
self.n[i] = (0, 0, 0)
self.n[self.pattern_step] = self.apply_brightness(color)
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 += 1
else:
self.pattern_step = 0
self.last_update = current_time
self.pattern_step = (self.pattern_step + 1) % 3
last_update += self.delay
await asyncio.sleep(0)
def random_rainbow_cycle_step(self):
current_time = utime.ticks_ms()
if utime.ticks_diff(current_time, self.last_update) >= self.delay:
def wheel(pos):
if pos < 85:
return (pos * 3, 255 - pos * 3, 0)
elif pos < 170:
pos -= 85
return (255 - pos * 3, 0, pos * 3)
async def blink(self):
last_update = utime.ticks_ms()
self.pattern_step = 0
while True:
if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
if self.pattern_step:
self.off()
self.pattern_step = 0
else:
pos -= 170
return (0, pos * 3, 255 - pos * 3)
self.on()
self.pattern_step = 1
last_update += self.delay
await asyncio.sleep(0)
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
async def flicker(self):
last_update = utime.ticks_ms()
while True:
if utime.ticks_diff(utime.ticks_ms(), last_update) >= self.delay:
# Calculate a single flicker amount for all LEDs
flicker_amount = random.randint(int(-self.brightness // 1.5), int(self.brightness // 1.5))
flicker_brightness = max(0, min(255, self.brightness + flicker_amount))
self.fill(self.apply_brightness(self.colors[0], brightness_override=flicker_brightness))
last_update += self.delay
await asyncio.sleep(0)
def 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
async def color_transition(self):
if len(self.colors) < 2:
# If there's only one color or no colors, just display that color (or off)
self.fill(self.apply_brightness(self.colors[0]))
return
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
last_transition_start_time = utime.ticks_ms()
current_color_index = 0
transition_duration_ms = self.delay # Use self.delay as the transition time
def interpolate_color(self, color_a, color_b, factor):
"""Interpolates between two colors."""
return tuple(int(a + (b - a) * factor) for a, b in zip(color_a, color_b))
while True:
color_from = self.colors[current_color_index]
color_to = self.colors[(current_color_index + 1) % len(self.colors)]
def color_transition_step(self):
current_time = utime.ticks_ms()
# Use delay for how often to update the transition, not for the duration
if utime.ticks_diff(current_time, self.last_update) >= 1: # Update frequently for smooth transition
self.transition_step += utime.ticks_diff(current_time, self.last_update)
self.last_update = current_time
start_time = utime.ticks_ms()
elapsed_time = 0
if self.transition_step >= self.transition_duration:
# Transition complete, swap colors and restart
self.color1, self.color2 = self.color2, self.color1
self.transition_step = 0
while elapsed_time < transition_duration_ms:
# Calculate the interpolation factor (0.0 to 1.0)
# Maximize to avoid division by zero if delay is 0, though a meaningful delay is expected
t = min(1.0, elapsed_time / max(1, transition_duration_ms))
# Calculate the interpolation factor (0 to 1)
factor = self.transition_step / self.transition_duration
# 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)
)
# Get the interpolated color and apply brightness
interpolated_color = self.interpolate_color(self.color1, self.color2, factor)
self.current_color = self.apply_brightness(interpolated_color)
self.fill(self.apply_brightness(interpolated_color))
await asyncio.sleep(0) # Update smoothly
elapsed_time = utime.ticks_diff(utime.ticks_ms(), start_time)
# Fill the LEDs with the current interpolated color
self.fill(self.current_color)
# Ensure the final color is set precisely after interpolation loop
self.fill(self.apply_brightness(color_to))
current_color_index = (current_color_index + 1) % len(self.colors)
await asyncio.sleep(0) # Yield control
async def main():
w = WDT(timeout = 10000)
p = Patterns(num_leds=10, pin=10, color1=(16,16,0))
# p.set_delay(100)
# await p.select("blink")
# await asyncio.sleep(2)
# p.set_delay(10)
# await p.select("rainbow")
# await asyncio.sleep(2)
# p.set_delay(100)
# await p.select("theater chase")
# await asyncio.sleep(2)
# p.set_colors([(255, 100, 0)]) # Set a base color for flicker (e.g., orange for a candle effect)
# p.set_brightness(200) # Set a brighter base for flicker to allow for dimming
# p.set_delay(100) # Faster updates for a more convincing flicker
# await p.select("flicker")
# await asyncio.sleep(2)
w.feed()
# Test the new color transition pattern
print("Starting color transition...")
p.set_colors([(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0)]) # Red, Green, Blue, Yellow
p.set_delay(1000) # 1 second transition between colors
p.set_brightness(150)
await p.select("color transition")
await asyncio.sleep(10) # Let it run for 10 seconds
if __name__ == "__main__":
p = Patterns(4, 180)
p.set_color1((255,0,0))
p.set_color2((0,255,0))
#p.set_delay(10)
try:
while True:
for key in p.patterns:
print(key)
p.select(key)
for _ in range(2000):
p.tick()
utime.sleep_ms(1)
except KeyboardInterrupt:
p.fill((0, 0, 0))
asyncio.run(main())

View File

@@ -23,10 +23,7 @@ class Settings(dict):
self["color_order"] = "rgb"
self["name"] = f"led-{ubinascii.hexlify(wifi.get_mac()).decode()}"
self["ap_password"] = ""
self["wifi_ssid"] = ""
self["wifi_password"] = ""
self["wifi_ip"] = ""
self["wifi_gateway"] = ""
self["id"] = 0
def save(self):
try:
@@ -48,21 +45,24 @@ class Settings(dict):
self.set_defaults()
self.save()
def set_settings(self, raw_json, patterns):
patterns.sync()
async def set_settings(self, data, patterns, save):
try:
data = json.loads(raw_json)
print(data)
for key, value in data.items():
print(key, value)
if key == "color1":
patterns.set_color1(tuple(int(value[i:i+2], 16) for i in self.color_order)) # Convert hex to RGB
if key == "colors":
buff = []
for color in value:
buff.append(tuple(int(color[i:i+2], 16) for i in self.color_order))
patterns.set_colors(buff)
elif key == "color1":
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(4, value)
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"])
@@ -77,11 +77,19 @@ class Settings(dict):
elif key == "color_order":
if value == "rbg": self.color_order = (1, 5, 3)
else: self.color_order = (1, 3, 5)
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)
patterns.sync()
if save:
self.save()
return "OK", 200
return "OK", 200
except (KeyError, ValueError):
return "Bad request", 400

View File

@@ -98,3 +98,12 @@ input[type="range"]::-moz-range-thumb {
#connection-status.closed {
background-color: red;
}
#color_order_form label,
#color_order_form input[type="radio"] {
/* Ensures they behave as inline elements */
display: inline-block;
/* Adds some space between them for readability */
margin-right: 10px;
vertical-align: middle; /* Aligns them nicely if heights vary */
}

View File

@@ -150,6 +150,30 @@ async function updateName(event) {
sendWebSocketData({ name: name });
}
async function updateID(event) {
event.preventDefault();
const id = document.getElementById("id").value;
sendWebSocketData({ id: parseInt(id) });
}
async function updateLedPin(event) {
event.preventDefault();
const ledpin = document.getElementById("led_pin").value;
sendWebSocketData({ led_pin: parseInt(ledpin) });
}
function handleRadioChange(event) {
event.preventDefault();
console.log("Selected color order:", event.target.value);
// Add your specific logic here
if (event.target.value === "rgb") {
console.log("RGB order selected!");
} else if (event.target.value === "rbg") {
console.log("RBG order selected!");
}
sendWebSocketData({ color_order: event.target.value });
}
function createPatternButtons(patterns) {
const container = document.getElementById("pattern_buttons");
container.innerHTML = ""; // Clear previous buttons
@@ -184,11 +208,17 @@ document.addEventListener("DOMContentLoaded", async function () {
.getElementById("num_leds_form")
.addEventListener("submit", updateNumLeds);
document.getElementById("name_form").addEventListener("submit", updateName);
document.getElementById("id_form").addEventListener("submit", updateID);
document
.getElementById("led_pin_form")
.addEventListener("submit", updateLedPin);
document.getElementById("delay").addEventListener("touchend", updateDelay);
document
.getElementById("brightness")
.addEventListener("touchend", updateBrightness);
document.getElementById("rgb").addEventListener("change", handleRadioChange);
document.getElementById("rbg").addEventListener("change", handleRadioChange);
document.querySelectorAll(".pattern_button").forEach((button) => {
console.log(button.value);
button.addEventListener("click", async (event) => {

View File

@@ -79,6 +79,16 @@
/>
<input type="submit" value="Update Name" />
</form>
<form id="id_form" method="post" action="/id">
<label for="id">ID:</label>
<input
type="text"
id="id"
name="id"
value="{{settings['id']}}"
/>
<input type="submit" value="Update ID" />
</form>
<!-- Separate form for submitting num_leds -->
<form id="num_leds_form" method="post" action="/num_leds">
<label for="num_leds">Number of LEDs:</label>
@@ -90,6 +100,23 @@
/>
<input type="submit" value="Update Number of LEDs" />
</form>
<form id="led_pin_form" method="post" action="/led_pin">
<label for="num_leds">Led pin:</label>
<input
type="text"
id="led_pin"
name="led_pin"
value="{{settings['led_pin']}}"
/>
<input type="submit" value="Update Led Pin" />
</form>
<form id="color_order_form">
<label for="rgb">RGB:</label>
<input type="radio" id="rgb" name="color_order" value="rgb" {{'checked' if settings["color_order"]=="rgb" else ''}} />
<label for="rbg">RBG</label>
<input type="radio" id="rbg" name="color_order" value="rbg" {{'checked' if settings["color_order"]=="rbg" else ''}}/>
</form>
<p>Mac address: {{mac}}</p>
</div>
<div id="connection-status"></div>

View File

@@ -33,8 +33,9 @@ def web(settings, patterns):
while True:
data = await ws.receive()
if data:
# Process the received data
_, status_code = settings.set_settings(data, patterns)
_, status_code = await settings.set_settings(json.loads(data), patterns, True)
#await ws.send(status_code)
else:
break