Compare commits
14 Commits
fe2717f51d
...
v2
| Author | SHA1 | Date | |
|---|---|---|---|
| af711bfb98 | |||
| 4d4b91e40d | |||
| b864f166cb | |||
| 05828833e0 | |||
| ce1db86fb6 | |||
| 43bd90f137 | |||
| d45846ec74 | |||
| 6ae6189519 | |||
| f53c57f036 | |||
| 47837c00e5 | |||
| 251b27133f | |||
| 10b6e5ee89 | |||
| 33aee7a3af | |||
| 501fd37def |
2
dev.py
2
dev.py
@@ -28,6 +28,8 @@ for cmd in sys.argv[1:]:
|
||||
if ser.in_waiting > 0: # Check if there is data in the buffer
|
||||
data = ser.readline().decode('utf-8').strip() # Read and decode the data
|
||||
print(data)
|
||||
case "clean":
|
||||
subprocess.call(["mpremote", "connect", port, "fs", "rm", ":/settings.json"])
|
||||
|
||||
|
||||
|
||||
|
||||
25
src/main.py
25
src/main.py
@@ -15,28 +15,37 @@ async def main():
|
||||
settings = Settings()
|
||||
|
||||
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.colors = [(8,0,0)]
|
||||
|
||||
# Initialize WDT only if debug is disabled
|
||||
wdt = None
|
||||
if not settings.get("debug", False):
|
||||
wdt = machine.WDT(timeout=10000)
|
||||
wdt.feed()
|
||||
print("Watchdog timer enabled")
|
||||
else:
|
||||
print("Debug mode: Watchdog timer disabled")
|
||||
|
||||
async def system():
|
||||
while True:
|
||||
gc.collect()
|
||||
for i in range(60):
|
||||
wdt.feed()
|
||||
await asyncio.sleep(1)
|
||||
if wdt is not None:
|
||||
for i in range(60):
|
||||
wdt.feed()
|
||||
await asyncio.sleep(1)
|
||||
else:
|
||||
# If WDT is disabled, just sleep
|
||||
await asyncio.sleep(60)
|
||||
|
||||
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()
|
||||
|
||||
asyncio.create_task(p2p(settings, patterns))
|
||||
asyncio.create_task(system())
|
||||
patterns.select(settings["pattern"])
|
||||
await patterns.run()
|
||||
|
||||
|
||||
|
||||
@@ -37,6 +37,7 @@ class Patterns(PatternsBase):
|
||||
"rainbow": self.rainbow,
|
||||
"pulse": self.pulse,
|
||||
"transition": self.transition,
|
||||
"chase": self.n_chase,
|
||||
"n_chase": self.n_chase,
|
||||
"circle": self.circle,
|
||||
}
|
||||
@@ -48,7 +49,8 @@ class Patterns(PatternsBase):
|
||||
state = True # True = on, False = off
|
||||
last_update = utime.ticks_ms()
|
||||
|
||||
while self.running:
|
||||
# Only continue running this pattern while it is the selected one
|
||||
while self.running and self.selected == "blink":
|
||||
current_time = utime.ticks_ms()
|
||||
if utime.ticks_diff(current_time, last_update) >= self.delay:
|
||||
if state:
|
||||
@@ -80,11 +82,12 @@ class Patterns(PatternsBase):
|
||||
return
|
||||
|
||||
# Auto is True: run continuously
|
||||
sleep_ms = max(1, int(self.delay))
|
||||
last_update = utime.ticks_ms()
|
||||
|
||||
while self.running:
|
||||
# Only continue running this pattern while it is the selected one
|
||||
while self.running and self.selected == "rainbow":
|
||||
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
|
||||
@@ -118,7 +121,8 @@ class Patterns(PatternsBase):
|
||||
min_write_time_ms = (self.num_leds * 30) // 1000 + 1 # Convert µs to ms, add 1ms overhead
|
||||
update_interval = max(10, min_write_time_ms + 4) # At least 10ms, add margin for safety
|
||||
|
||||
while self.running:
|
||||
# Only continue running this pattern while it is the selected one
|
||||
while self.running and self.selected == "pulse":
|
||||
cycle_start = utime.ticks_ms()
|
||||
|
||||
# Get the current color from the cycle
|
||||
@@ -167,7 +171,8 @@ class Patterns(PatternsBase):
|
||||
# Ensure the cycle takes exactly delay milliseconds before restarting
|
||||
if self.running:
|
||||
self.off()
|
||||
wait_until = utime.ticks_add(cycle_start, self.delay)
|
||||
delay_ms = int(self.delay) # Access delay directly
|
||||
wait_until = utime.ticks_add(cycle_start, delay_ms)
|
||||
while self.running and utime.ticks_diff(wait_until, utime.ticks_ms()) > 0:
|
||||
pass
|
||||
|
||||
@@ -189,7 +194,7 @@ class Patterns(PatternsBase):
|
||||
if len(self.colors) == 1:
|
||||
# Only one color, just stay that color
|
||||
last_update = utime.ticks_ms()
|
||||
while self.running:
|
||||
while self.running and self.selected == "transition":
|
||||
current_time = utime.ticks_ms()
|
||||
if utime.ticks_diff(current_time, last_update) >= 100:
|
||||
self.fill(self.apply_brightness(self.colors[0]))
|
||||
@@ -206,17 +211,18 @@ class Patterns(PatternsBase):
|
||||
self.stopped = True
|
||||
return
|
||||
|
||||
transition_duration = max(10, self.delay) # At least 10ms
|
||||
update_interval = max(10, transition_duration // 50) # Update every ~2% of transition
|
||||
|
||||
# Transition from color1 to color2
|
||||
color1 = self.colors[0]
|
||||
color2 = self.colors[1]
|
||||
|
||||
transition_start = utime.ticks_ms()
|
||||
last_update = transition_start
|
||||
|
||||
while self.running and utime.ticks_diff(utime.ticks_ms(), transition_start) < transition_duration:
|
||||
while self.running:
|
||||
# Access delay and colors directly for live updates
|
||||
transition_duration = max(10, int(self.delay)) # At least 10ms
|
||||
update_interval = max(10, transition_duration // 50) # Update every ~2% of transition
|
||||
color1 = self.colors[0] if len(self.colors) > 0 else (0, 0, 0)
|
||||
color2 = self.colors[1] if len(self.colors) > 1 else color1
|
||||
|
||||
if utime.ticks_diff(utime.ticks_ms(), transition_start) >= transition_duration:
|
||||
break
|
||||
now = utime.ticks_ms()
|
||||
if utime.ticks_diff(now, last_update) >= update_interval:
|
||||
# Calculate interpolation factor (0.0 to 1.0)
|
||||
@@ -239,10 +245,12 @@ class Patterns(PatternsBase):
|
||||
|
||||
# Auto is True: cycle through all colors continuously
|
||||
color_index = 0
|
||||
transition_duration = max(10, self.delay) # At least 10ms
|
||||
update_interval = max(10, transition_duration // 50) # Update every ~2% of transition
|
||||
|
||||
while self.running:
|
||||
# Auto is True: cycle through all colors continuously
|
||||
while self.running and self.selected == "transition":
|
||||
# Access colors directly for live updates
|
||||
if not self.colors:
|
||||
break
|
||||
# Get current and next color
|
||||
current_color = self.colors[color_index % len(self.colors)]
|
||||
next_color = self.colors[(color_index + 1) % len(self.colors)]
|
||||
@@ -251,7 +259,13 @@ class Patterns(PatternsBase):
|
||||
transition_start = utime.ticks_ms()
|
||||
last_update = transition_start
|
||||
|
||||
while self.running and utime.ticks_diff(utime.ticks_ms(), transition_start) < transition_duration:
|
||||
while self.running:
|
||||
# Access delay directly for live updates
|
||||
transition_duration = max(10, int(self.delay)) # At least 10ms
|
||||
update_interval = max(10, transition_duration // 50) # Update every ~2% of transition
|
||||
|
||||
if utime.ticks_diff(utime.ticks_ms(), transition_start) >= transition_duration:
|
||||
break
|
||||
now = utime.ticks_ms()
|
||||
if utime.ticks_diff(now, last_update) >= update_interval:
|
||||
# Calculate interpolation factor (0.0 to 1.0)
|
||||
@@ -280,28 +294,43 @@ class Patterns(PatternsBase):
|
||||
self.stopped = False
|
||||
self.running = True
|
||||
|
||||
if len(self.colors) < 2:
|
||||
# Need at least 2 colors
|
||||
if len(self.colors) < 1:
|
||||
# Need at least 1 color
|
||||
self.running = False
|
||||
self.stopped = True
|
||||
return
|
||||
|
||||
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
|
||||
segment_length = 0 # Will be calculated in loop
|
||||
position = 0 # Current position offset
|
||||
step_count = 0 # Track which step we're on
|
||||
|
||||
color0 = self.apply_brightness(self.colors[0])
|
||||
color1 = self.apply_brightness(self.colors[1])
|
||||
|
||||
transition_duration = max(10, self.delay)
|
||||
last_update = utime.ticks_ms()
|
||||
|
||||
while self.running:
|
||||
# Only continue running this pattern while it is the selected one
|
||||
# Note: this pattern can be selected as "n_chase" or "chase"
|
||||
while self.running and self.selected in ("n_chase", "chase"):
|
||||
# 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
|
||||
@@ -361,7 +390,8 @@ class Patterns(PatternsBase):
|
||||
|
||||
phase = "growing" # "growing", "shrinking", or "off"
|
||||
|
||||
while self.running:
|
||||
# Only continue running this pattern while it is the selected one
|
||||
while self.running and self.selected == "circle":
|
||||
current_time = utime.ticks_ms()
|
||||
|
||||
# Clear all LEDs
|
||||
|
||||
@@ -5,7 +5,6 @@ import random
|
||||
import _thread
|
||||
import asyncio
|
||||
import json
|
||||
from presets import Presets
|
||||
|
||||
# Short-key parameter mapping for convenience setters
|
||||
param_mapping = {
|
||||
@@ -27,7 +26,7 @@ param_mapping = {
|
||||
}
|
||||
|
||||
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="off", delay=100):
|
||||
self.n = NeoPixel(Pin(pin, Pin.OUT), num_leds)
|
||||
self.num_leds = num_leds
|
||||
self.pattern_step = 0
|
||||
@@ -55,7 +54,7 @@ class Patterns:
|
||||
self.scanner_tail_length = 3 # Number of trailing pixels
|
||||
self.running = False
|
||||
self.stopped = True
|
||||
self.presets = Presets()
|
||||
|
||||
self.n1 = 0
|
||||
self.n2 = 0
|
||||
self.n3 = 0
|
||||
@@ -64,28 +63,44 @@ class Patterns:
|
||||
self.n6 = 0
|
||||
|
||||
def select(self, pattern):
|
||||
|
||||
if pattern in self.patterns:
|
||||
self.selected = pattern
|
||||
return True
|
||||
# If pattern doesn't exist, default to "off"
|
||||
if "off" in self.patterns:
|
||||
self.selected = "off"
|
||||
return False
|
||||
|
||||
async def run(self):
|
||||
print(f"Stopping pattern")
|
||||
await self.stop()
|
||||
self.running = True
|
||||
# Ensure we wait a bit more to let the thread fully terminate
|
||||
# If selected pattern doesn't exist, default to "off"
|
||||
if self.selected not in self.patterns:
|
||||
print(f"Pattern {self.selected} not found, defaulting to 'off'")
|
||||
if "off" in self.patterns:
|
||||
self.selected = "off"
|
||||
else:
|
||||
print("No patterns available")
|
||||
self.running = False
|
||||
self.stopped = True
|
||||
return
|
||||
print(f"Starting pattern {self.selected}")
|
||||
if self.selected in self.patterns:
|
||||
_thread.start_new_thread(self.patterns[self.selected], ())
|
||||
else:
|
||||
print(f"Pattern {self.selected} not found")
|
||||
_thread.start_new_thread(self.patterns[self.selected], ())
|
||||
|
||||
async def stop(self):
|
||||
if not self.running:
|
||||
# Already stopped
|
||||
self.stopped = True
|
||||
return
|
||||
self.running = False
|
||||
start = utime.ticks_ms()
|
||||
while not self.stopped and utime.ticks_diff(utime.ticks_ms(), start) < 1000:
|
||||
await asyncio.sleep_ms(0)
|
||||
self.stopped = True
|
||||
timeout = 2000 # Increased timeout to 2 seconds
|
||||
while not self.stopped and utime.ticks_diff(utime.ticks_ms(), start) < timeout:
|
||||
await asyncio.sleep_ms(10) # Check every 10ms instead of 0ms
|
||||
if not self.stopped:
|
||||
# Timeout reached, force stop
|
||||
print("Warning: Pattern did not stop within timeout")
|
||||
self.stopped = True
|
||||
|
||||
def set_param(self, key, value):
|
||||
if key in param_mapping:
|
||||
|
||||
31
src/presets.py
Normal file
31
src/presets.py
Normal file
@@ -0,0 +1,31 @@
|
||||
import json
|
||||
import wifi
|
||||
import ubinascii
|
||||
import machine
|
||||
|
||||
class Presets(dict):
|
||||
FILE = "/presets.json"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.load() # Load settings from file during initialization
|
||||
|
||||
def save(self):
|
||||
try:
|
||||
j = json.dumps(self)
|
||||
with open(self.FILE, 'w') as file:
|
||||
file.write(j)
|
||||
print("Presets saved successfully.")
|
||||
except Exception as e:
|
||||
print(f"Error saving settings: {e}")
|
||||
|
||||
def load(self):
|
||||
try:
|
||||
with open(self.FILE, 'r') as file:
|
||||
self.update(json.load(file))
|
||||
print("Presets loaded successfully.")
|
||||
except Exception as e:
|
||||
print(f"Error loading presets")
|
||||
self.save()
|
||||
|
||||
|
||||
@@ -9,21 +9,19 @@ class Settings(dict):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.load() # Load settings from file during initialization
|
||||
if self["color_order"] == "rbg": self.color_order = (1, 5, 3)
|
||||
else: self.color_order = (1, 3, 5)
|
||||
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["color1"] = "#00ff00"
|
||||
self["color2"] = "#ff0000"
|
||||
self["delay"] = 100
|
||||
self["brightness"] = 10
|
||||
self["color_order"] = "rgb"
|
||||
self["name"] = f"led-{ubinascii.hexlify(wifi.get_mac()).decode()}"
|
||||
self["ap_password"] = ""
|
||||
self["id"] = 0
|
||||
self["debug"] = False
|
||||
|
||||
def save(self):
|
||||
try:
|
||||
@@ -84,8 +82,8 @@ class Settings(dict):
|
||||
self.save()
|
||||
machine.reset()
|
||||
elif key == "color_order":
|
||||
if value == "rbg": self.color_order = (1, 5, 3)
|
||||
else: self.color_order = (1, 3, 5)
|
||||
self["color_order"] = value
|
||||
self.color_order = self.get_color_order(value)
|
||||
pass
|
||||
elif key == "id":
|
||||
pass
|
||||
@@ -102,6 +100,18 @@ class Settings(dict):
|
||||
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()
|
||||
|
||||
@@ -107,3 +107,57 @@ input[type="range"]::-moz-range-thumb {
|
||||
margin-right: 10px;
|
||||
vertical-align: middle; /* Aligns them nicely if heights vary */
|
||||
}
|
||||
|
||||
#colors_palette {
|
||||
margin-bottom: 20px;
|
||||
}
|
||||
|
||||
#colors_container {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 10px;
|
||||
margin-bottom: 10px;
|
||||
}
|
||||
|
||||
.color-item {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 5px;
|
||||
}
|
||||
|
||||
.color-input {
|
||||
width: 60px !important;
|
||||
height: 40px;
|
||||
border: 2px solid #ddd;
|
||||
border-radius: 4px;
|
||||
cursor: pointer;
|
||||
}
|
||||
|
||||
.remove-color-btn {
|
||||
background-color: #f44336;
|
||||
color: white;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
width: 30px;
|
||||
height: 30px;
|
||||
cursor: pointer;
|
||||
font-size: 18px;
|
||||
line-height: 1;
|
||||
}
|
||||
|
||||
.remove-color-btn:hover {
|
||||
background-color: #da190b;
|
||||
}
|
||||
|
||||
#add_color_btn {
|
||||
background-color: #4caf50;
|
||||
color: white;
|
||||
border: none;
|
||||
padding: 8px 16px;
|
||||
border-radius: 4px;
|
||||
cursor: pointer;
|
||||
}
|
||||
|
||||
#add_color_btn:hover {
|
||||
background-color: #45a049;
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
let delayTimeout;
|
||||
let brightnessTimeout;
|
||||
let colorTimeout;
|
||||
let color2Timeout;
|
||||
let colorsTimeout;
|
||||
let ws; // Variable to hold the WebSocket connection
|
||||
let connectionStatusElement; // Variable to hold the connection status element
|
||||
|
||||
@@ -98,22 +97,60 @@ async function get(path) {
|
||||
}
|
||||
}
|
||||
|
||||
async function updateColor(event) {
|
||||
event.preventDefault();
|
||||
clearTimeout(colorTimeout);
|
||||
colorTimeout = setTimeout(function () {
|
||||
const color = document.getElementById("color").value;
|
||||
sendWebSocketData({ color1: color });
|
||||
function updateColors() {
|
||||
clearTimeout(colorsTimeout);
|
||||
colorsTimeout = setTimeout(function () {
|
||||
const colorInputs = document.querySelectorAll(".color-input");
|
||||
const colors = Array.from(colorInputs).map(input => input.value);
|
||||
sendWebSocketData({ colors: colors });
|
||||
}, 500);
|
||||
}
|
||||
|
||||
async function updateColor2(event) {
|
||||
event.preventDefault();
|
||||
clearTimeout(color2Timeout);
|
||||
color2Timeout = setTimeout(function () {
|
||||
const color = document.getElementById("color2").value;
|
||||
sendWebSocketData({ color2: color });
|
||||
}, 500);
|
||||
function addColorInput(color = "#ff0000") {
|
||||
const container = document.getElementById("colors_container");
|
||||
const colorDiv = document.createElement("div");
|
||||
colorDiv.className = "color-item";
|
||||
|
||||
const colorInput = document.createElement("input");
|
||||
colorInput.type = "color";
|
||||
colorInput.className = "color-input";
|
||||
colorInput.value = color;
|
||||
colorInput.addEventListener("input", updateColors);
|
||||
|
||||
const removeBtn = document.createElement("button");
|
||||
removeBtn.type = "button";
|
||||
removeBtn.textContent = "×";
|
||||
removeBtn.className = "remove-color-btn";
|
||||
removeBtn.addEventListener("click", function() {
|
||||
colorDiv.remove();
|
||||
updateColors();
|
||||
});
|
||||
|
||||
colorDiv.appendChild(colorInput);
|
||||
colorDiv.appendChild(removeBtn);
|
||||
container.appendChild(colorDiv);
|
||||
}
|
||||
|
||||
function initializeColors(initialColors = null) {
|
||||
const container = document.getElementById("colors_container");
|
||||
container.innerHTML = "";
|
||||
// Get initial colors from data attribute or use defaults
|
||||
if (initialColors === null) {
|
||||
const colorsData = document.getElementById("colors_container").dataset.colors;
|
||||
if (colorsData) {
|
||||
try {
|
||||
initialColors = JSON.parse(colorsData);
|
||||
} catch (e) {
|
||||
initialColors = ["#ff0000", "#00ff00"];
|
||||
}
|
||||
} else {
|
||||
initialColors = ["#ff0000", "#00ff00"];
|
||||
}
|
||||
}
|
||||
if (initialColors.length === 0) {
|
||||
initialColors = ["#ff0000"];
|
||||
}
|
||||
initialColors.forEach(color => addColorInput(color));
|
||||
}
|
||||
|
||||
async function updatePattern(pattern) {
|
||||
@@ -198,8 +235,11 @@ document.addEventListener("DOMContentLoaded", async function () {
|
||||
// Establish WebSocket connection on page load
|
||||
connectWebSocket();
|
||||
|
||||
document.getElementById("color").addEventListener("input", updateColor);
|
||||
document.getElementById("color2").addEventListener("input", updateColor2);
|
||||
// Initialize colors palette
|
||||
initializeColors();
|
||||
document.getElementById("add_color_btn").addEventListener("click", function() {
|
||||
addColorInput();
|
||||
});
|
||||
document.getElementById("delay").addEventListener("input", updateDelay);
|
||||
document
|
||||
.getElementById("brightness")
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
{% args settings, patterns, mac %}
|
||||
{% args settings, patterns, colors_json, mac %}
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
@@ -46,22 +46,13 @@
|
||||
step="1"
|
||||
/>
|
||||
</form>
|
||||
<form id="color_form" method="post" action="/color">
|
||||
<input
|
||||
type="color"
|
||||
id="color"
|
||||
name="color"
|
||||
value="{{settings['color1']}}"
|
||||
/>
|
||||
</form>
|
||||
<form id="color2_form" method="post" action="/color2">
|
||||
<input
|
||||
type="color"
|
||||
id="color2"
|
||||
name="color2"
|
||||
value="{{settings['color2']}}"
|
||||
/>
|
||||
</form>
|
||||
<div id="colors_palette">
|
||||
<label>Colors:</label>
|
||||
<div id="colors_container" data-colors='{{colors_json}}'>
|
||||
<!-- Color inputs will be added here dynamically -->
|
||||
</div>
|
||||
<button type="button" id="add_color_btn">+ Add Color</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Settings Menu for num_leds, Wi-Fi SSID, and Password -->
|
||||
|
||||
14
src/web.py
14
src/web.py
@@ -12,7 +12,19 @@ def web(settings, patterns):
|
||||
@app.route('/')
|
||||
async def index_hnadler(request):
|
||||
mac = wifi.get_mac().hex()
|
||||
return Template('index.html').render(settings=settings, patterns=patterns.patterns.keys())
|
||||
# Convert colors from RGB tuples to hex strings for display
|
||||
colors_hex = []
|
||||
for color in patterns.colors:
|
||||
# Convert (R, G, B) tuple to #RRGGBB hex string
|
||||
colors_hex.append(f"#{color[0]:02x}{color[1]:02x}{color[2]:02x}")
|
||||
# Convert to JSON string for data attribute
|
||||
colors_json = json.dumps(colors_hex)
|
||||
return Template('index.html').render(
|
||||
settings,
|
||||
patterns.patterns.keys(),
|
||||
colors_json,
|
||||
mac
|
||||
)
|
||||
|
||||
@app.route("/static/<path:path>")
|
||||
def static_handler(request, path):
|
||||
|
||||
63
test/circle.py
Normal file
63
test/circle.py
Normal file
@@ -0,0 +1,63 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Circle test: n1=50, n2=100, n3=200, n4=0 (Red)
|
||||
Runs forever
|
||||
Run with: mpremote run test/circle.py
|
||||
"""
|
||||
|
||||
import patterns
|
||||
import utime
|
||||
import _thread
|
||||
from settings import Settings
|
||||
from machine import WDT
|
||||
|
||||
print("Starting Circle Test: n1=50, n2=100, n3=200, n4=0 (Red)")
|
||||
print("Press Ctrl+C to stop")
|
||||
|
||||
# Load settings
|
||||
settings = Settings()
|
||||
|
||||
# Initialize patterns using settings
|
||||
p = patterns.Patterns(
|
||||
pin=settings["led_pin"],
|
||||
num_leds=settings["num_leds"],
|
||||
brightness=255,
|
||||
delay=2000
|
||||
)
|
||||
|
||||
# Configure test parameters
|
||||
p.n1 = 50 # Head moves 50 LEDs/second
|
||||
p.n2 = 100 # Max length 100 LEDs
|
||||
p.n3 = 200 # Tail moves 200 LEDs/second
|
||||
p.n4 = 0 # Min length 0 LEDs
|
||||
p.colors = [(255, 0, 0)] # Red
|
||||
|
||||
print(f"LED Pin: {settings['led_pin']}")
|
||||
print(f"LEDs: {settings['num_leds']}")
|
||||
print(f"Brightness: {p.brightness}")
|
||||
print(f"Parameters: n1={p.n1}, n2={p.n2}, n3={p.n3}, n4={p.n4}")
|
||||
print(f"Color: {p.colors[0]}")
|
||||
|
||||
# Initialize watchdog timer
|
||||
wdt = WDT(timeout=10000)
|
||||
wdt.feed()
|
||||
|
||||
# Start pattern
|
||||
p.select("circle")
|
||||
if p.selected in p.patterns:
|
||||
_thread.start_new_thread(p.patterns[p.selected], ())
|
||||
print("Pattern started. Running forever...")
|
||||
else:
|
||||
print(f"Pattern {p.selected} not found")
|
||||
|
||||
# Run forever
|
||||
try:
|
||||
while True:
|
||||
wdt.feed()
|
||||
utime.sleep_ms(100)
|
||||
except KeyboardInterrupt:
|
||||
print("\nStopping...")
|
||||
p.running = False
|
||||
p.off()
|
||||
print("LEDs turned off")
|
||||
|
||||
@@ -136,3 +136,9 @@ if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -108,3 +108,4 @@ async def test_patterns_save_load():
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_patterns_save_load())
|
||||
|
||||
|
||||
|
||||
@@ -109,3 +109,4 @@ def test_save_load():
|
||||
if __name__ == "__main__":
|
||||
test_save_load()
|
||||
|
||||
|
||||
|
||||
335
tool.py
Executable file
335
tool.py
Executable file
@@ -0,0 +1,335 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
LED Bar Configuration Tool
|
||||
A tkinter GUI for downloading, editing, and uploading settings.json to/from MicroPython devices via mpremote.
|
||||
"""
|
||||
|
||||
import tkinter as tk
|
||||
from tkinter import ttk, messagebox, filedialog
|
||||
import json
|
||||
import subprocess
|
||||
import os
|
||||
import tempfile
|
||||
import serial
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class LEDConfigTool:
|
||||
def __init__(self, root):
|
||||
self.root = root
|
||||
self.root.title("LED Bar Configuration Tool")
|
||||
self.root.geometry("600x700")
|
||||
|
||||
self.settings = {}
|
||||
self.temp_file = None
|
||||
|
||||
# Create main frame
|
||||
main_frame = ttk.Frame(root, padding="10")
|
||||
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
|
||||
|
||||
# Title
|
||||
title_label = ttk.Label(main_frame, text="LED Bar Configuration", font=("Arial", 16, "bold"))
|
||||
title_label.grid(row=0, column=0, columnspan=2, pady=(0, 20))
|
||||
|
||||
# Device connection section
|
||||
device_frame = ttk.LabelFrame(main_frame, text="Device Connection", padding="10")
|
||||
device_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(0, 10))
|
||||
|
||||
ttk.Label(device_frame, text="Device:").grid(row=0, column=0, sticky=tk.W, padx=(0, 5))
|
||||
self.device_entry = ttk.Entry(device_frame, width=30)
|
||||
self.device_entry.insert(0, "/dev/ttyACM0") # Default device
|
||||
self.device_entry.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=(0, 10))
|
||||
|
||||
ttk.Button(device_frame, text="Download Settings", command=self.download_settings).grid(row=0, column=2)
|
||||
|
||||
# Settings section
|
||||
settings_frame = ttk.LabelFrame(main_frame, text="Settings", padding="10")
|
||||
settings_frame.grid(row=2, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 10))
|
||||
|
||||
# Create scrollable frame for settings
|
||||
canvas = tk.Canvas(settings_frame, height=400)
|
||||
scrollbar = ttk.Scrollbar(settings_frame, orient="vertical", command=canvas.yview)
|
||||
scrollable_frame = ttk.Frame(canvas)
|
||||
|
||||
scrollable_frame.bind(
|
||||
"<Configure>",
|
||||
lambda e: canvas.configure(scrollregion=canvas.bbox("all"))
|
||||
)
|
||||
|
||||
canvas.create_window((0, 0), window=scrollable_frame, anchor="nw")
|
||||
canvas.configure(yscrollcommand=scrollbar.set)
|
||||
|
||||
# Settings fields
|
||||
self.setting_widgets = {}
|
||||
settings_config = [
|
||||
("led_pin", "LED Pin", "number"),
|
||||
("num_leds", "Number of LEDs", "number"),
|
||||
("color_order", "Color Order", "choice", ["rgb", "rbg", "grb", "gbr", "brg", "bgr"]),
|
||||
("name", "Device Name", "text"),
|
||||
("pattern", "Pattern", "text"),
|
||||
("delay", "Delay (ms)", "number"),
|
||||
("brightness", "Brightness", "number"),
|
||||
("n1", "N1", "number"),
|
||||
("n2", "N2", "number"),
|
||||
("n3", "N3", "number"),
|
||||
("n4", "N4", "number"),
|
||||
("n5", "N5", "number"),
|
||||
("n6", "N6", "number"),
|
||||
("ap_password", "AP Password", "text"),
|
||||
("id", "ID", "number"),
|
||||
("debug", "Debug Mode", "choice", ["True", "False"]),
|
||||
]
|
||||
|
||||
for idx, config in enumerate(settings_config):
|
||||
key = config[0]
|
||||
label_text = config[1]
|
||||
field_type = config[2]
|
||||
|
||||
ttk.Label(scrollable_frame, text=f"{label_text}:").grid(row=idx, column=0, sticky=tk.W, padx=(0, 10), pady=5)
|
||||
|
||||
if field_type == "number":
|
||||
widget = ttk.Entry(scrollable_frame, width=20)
|
||||
elif field_type == "choice":
|
||||
widget = ttk.Combobox(scrollable_frame, width=17, values=config[3], state="readonly")
|
||||
elif field_type == "color":
|
||||
widget = ttk.Entry(scrollable_frame, width=20)
|
||||
else: # text
|
||||
widget = ttk.Entry(scrollable_frame, width=20)
|
||||
|
||||
widget.grid(row=idx, column=1, sticky=(tk.W, tk.E), pady=5)
|
||||
self.setting_widgets[key] = widget
|
||||
|
||||
canvas.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
|
||||
scrollbar.grid(row=0, column=1, sticky=(tk.N, tk.S))
|
||||
settings_frame.grid_rowconfigure(0, weight=1)
|
||||
settings_frame.grid_columnconfigure(0, weight=1)
|
||||
|
||||
# Buttons section
|
||||
button_frame = ttk.Frame(main_frame)
|
||||
button_frame.grid(row=3, column=0, columnspan=2, pady=(10, 0))
|
||||
|
||||
ttk.Button(button_frame, text="Load from File", command=self.load_from_file).grid(row=0, column=0, padx=5)
|
||||
ttk.Button(button_frame, text="Save to File", command=self.save_to_file).grid(row=0, column=1, padx=5)
|
||||
ttk.Button(button_frame, text="Upload Settings", command=self.upload_settings).grid(row=0, column=2, padx=5)
|
||||
|
||||
# Status bar
|
||||
self.status_label = ttk.Label(main_frame, text="Ready", relief=tk.SUNKEN, anchor=tk.W)
|
||||
self.status_label.grid(row=4, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=(10, 0))
|
||||
|
||||
# Configure grid weights
|
||||
root.columnconfigure(0, weight=1)
|
||||
root.rowconfigure(0, weight=1)
|
||||
main_frame.columnconfigure(0, weight=1)
|
||||
main_frame.rowconfigure(2, weight=1)
|
||||
device_frame.columnconfigure(1, weight=1)
|
||||
|
||||
def update_status(self, message):
|
||||
"""Update the status bar message."""
|
||||
self.status_label.config(text=message)
|
||||
self.root.update_idletasks()
|
||||
|
||||
def download_settings(self):
|
||||
"""Download settings.json from the device using mpremote."""
|
||||
device = self.device_entry.get().strip()
|
||||
if not device:
|
||||
messagebox.showerror("Error", "Please specify a device")
|
||||
return
|
||||
|
||||
self.update_status("Downloading settings...")
|
||||
|
||||
try:
|
||||
# Create temporary file
|
||||
self.temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
|
||||
temp_path = self.temp_file.name
|
||||
self.temp_file.close()
|
||||
|
||||
# Download file using mpremote
|
||||
cmd = ["mpremote", "connect", device, "cp", ":/settings.json", temp_path]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise Exception(f"mpremote error: {result.stderr}")
|
||||
|
||||
# Load the downloaded file
|
||||
with open(temp_path, 'r') as f:
|
||||
self.settings = json.load(f)
|
||||
|
||||
# Update UI with loaded settings
|
||||
self.update_ui_from_settings()
|
||||
|
||||
self.update_status(f"Settings downloaded successfully from {device}")
|
||||
messagebox.showinfo("Success", "Settings downloaded successfully!")
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
self.update_status("Error: Connection timeout")
|
||||
messagebox.showerror("Error", "Connection timeout. Check device connection.")
|
||||
except FileNotFoundError:
|
||||
self.update_status("Error: mpremote not found")
|
||||
messagebox.showerror("Error", "mpremote not found. Please install it:\npip install mpremote")
|
||||
except Exception as e:
|
||||
self.update_status(f"Error: {str(e)}")
|
||||
messagebox.showerror("Error", f"Failed to download settings:\n{str(e)}")
|
||||
finally:
|
||||
# Clean up temp file
|
||||
if self.temp_file and os.path.exists(temp_path):
|
||||
try:
|
||||
os.unlink(temp_path)
|
||||
except:
|
||||
pass
|
||||
|
||||
def upload_settings(self):
|
||||
"""Upload settings.json to the device using mpremote."""
|
||||
device = self.device_entry.get().strip()
|
||||
if not device:
|
||||
messagebox.showerror("Error", "Please specify a device")
|
||||
return
|
||||
|
||||
if not self.settings:
|
||||
messagebox.showerror("Error", "No settings to upload. Please download or load settings first.")
|
||||
return
|
||||
|
||||
self.update_status("Uploading settings...")
|
||||
|
||||
try:
|
||||
# Get current settings from UI
|
||||
self.update_settings_from_ui()
|
||||
|
||||
# Create temporary file with current settings
|
||||
temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
|
||||
temp_path = temp_file.name
|
||||
json.dump(self.settings, temp_file, indent=2)
|
||||
temp_file.close()
|
||||
|
||||
# Upload file using mpremote
|
||||
cmd = ["mpremote", "connect", device, "cp", temp_path, ":/settings.json"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise Exception(f"mpremote error: {result.stderr}")
|
||||
|
||||
# Reset the device
|
||||
self.update_status("Resetting device...")
|
||||
try:
|
||||
with serial.Serial(device, baudrate=115200) as ser:
|
||||
ser.write(b'\x03\x03\x04')
|
||||
except Exception as e:
|
||||
# If serial reset fails, try mpremote method as fallback
|
||||
reset_cmd = ["mpremote", "connect", device, "exec", "import machine; machine.reset()"]
|
||||
subprocess.run(reset_cmd, capture_output=True, text=True, timeout=5)
|
||||
|
||||
self.update_status(f"Settings uploaded and device reset on {device}")
|
||||
messagebox.showinfo("Success", "Settings uploaded successfully and device reset!")
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
self.update_status("Error: Connection timeout")
|
||||
messagebox.showerror("Error", "Connection timeout. Check device connection.")
|
||||
except FileNotFoundError:
|
||||
self.update_status("Error: mpremote not found")
|
||||
messagebox.showerror("Error", "mpremote not found. Please install it:\npip install mpremote")
|
||||
except Exception as e:
|
||||
self.update_status(f"Error: {str(e)}")
|
||||
messagebox.showerror("Error", f"Failed to upload settings:\n{str(e)}")
|
||||
finally:
|
||||
# Clean up temp file
|
||||
if os.path.exists(temp_path):
|
||||
try:
|
||||
os.unlink(temp_path)
|
||||
except:
|
||||
pass
|
||||
|
||||
def load_from_file(self):
|
||||
"""Load settings from a local JSON file."""
|
||||
file_path = filedialog.askopenfilename(
|
||||
title="Load Settings",
|
||||
filetypes=[("JSON files", "*.json"), ("All files", "*.*")]
|
||||
)
|
||||
|
||||
if not file_path:
|
||||
return
|
||||
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
self.settings = json.load(f)
|
||||
|
||||
self.update_ui_from_settings()
|
||||
self.update_status(f"Settings loaded from {os.path.basename(file_path)}")
|
||||
messagebox.showinfo("Success", "Settings loaded successfully!")
|
||||
|
||||
except Exception as e:
|
||||
self.update_status(f"Error: {str(e)}")
|
||||
messagebox.showerror("Error", f"Failed to load settings:\n{str(e)}")
|
||||
|
||||
def save_to_file(self):
|
||||
"""Save current settings to a local JSON file."""
|
||||
if not self.settings:
|
||||
messagebox.showerror("Error", "No settings to save. Please download or load settings first.")
|
||||
return
|
||||
|
||||
file_path = filedialog.asksaveasfilename(
|
||||
title="Save Settings",
|
||||
defaultextension=".json",
|
||||
filetypes=[("JSON files", "*.json"), ("All files", "*.*")]
|
||||
)
|
||||
|
||||
if not file_path:
|
||||
return
|
||||
|
||||
try:
|
||||
# Get current settings from UI
|
||||
self.update_settings_from_ui()
|
||||
|
||||
with open(file_path, 'w') as f:
|
||||
json.dump(self.settings, f, indent=2)
|
||||
|
||||
self.update_status(f"Settings saved to {os.path.basename(file_path)}")
|
||||
messagebox.showinfo("Success", "Settings saved successfully!")
|
||||
|
||||
except Exception as e:
|
||||
self.update_status(f"Error: {str(e)}")
|
||||
messagebox.showerror("Error", f"Failed to save settings:\n{str(e)}")
|
||||
|
||||
def update_ui_from_settings(self):
|
||||
"""Update UI widgets with current settings values."""
|
||||
for key, widget in self.setting_widgets.items():
|
||||
if key in self.settings:
|
||||
value = self.settings[key]
|
||||
if isinstance(widget, ttk.Combobox):
|
||||
# For debug, convert boolean to string
|
||||
if key == "debug":
|
||||
widget.set(str(value))
|
||||
else:
|
||||
widget.set(str(value))
|
||||
else:
|
||||
widget.delete(0, tk.END)
|
||||
widget.insert(0, str(value))
|
||||
|
||||
def update_settings_from_ui(self):
|
||||
"""Update settings dictionary from UI widget values."""
|
||||
for key, widget in self.setting_widgets.items():
|
||||
value = widget.get().strip()
|
||||
if value:
|
||||
# Try to convert to appropriate type
|
||||
if key in ["led_pin", "num_leds", "delay", "brightness", "id", "n1", "n2", "n3", "n4", "n5", "n6"]:
|
||||
try:
|
||||
self.settings[key] = int(value)
|
||||
except ValueError:
|
||||
pass # Keep as string if conversion fails
|
||||
elif key == "debug":
|
||||
# Convert string "True"/"False" to boolean
|
||||
self.settings[key] = value == "True"
|
||||
else:
|
||||
self.settings[key] = value
|
||||
elif key in self.settings:
|
||||
# Keep existing value if widget is empty
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
root = tk.Tk()
|
||||
app = LEDConfigTool(root)
|
||||
root.mainloop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user