technicalkiwi/led-controller
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Remove obsolete scripts and root config files

Browse Source 
Drop clear-debug-log, install, run_web, send_empty_json, esp32 helpers,
and root msg.json/settings.json in favor of current layout.

Made-with: Cursor
This commit is contained in:
Jimmy
2026-03-21 19:47:29 +13:00
parent b56af23cbf
commit aa3546e9ac
8 changed files with 0 additions and 422 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
clear-debug-log.sh
View File

@@ -1,2 +0,0 @@
#!/usr/bin/env sh
rm -f /home/pi/led-controller/.cursor/debug.log

112
esp32/benchmark_peers.py
View File

@@ -1,112 +0,0 @@
# Benchmark: LRU eviction vs add-then-remove-after-use on ESP32.
# Run on device: mpremote run esp32/benchmark_peers.py
# (add/del_peer are timed; send() may fail if no peer is listening - timing still valid)
import espnow
import network
import time
BROADCAST = b"\xff\xff\xff\xff\xff\xff"
MAX_PEERS = 20
ITERATIONS = 50
PAYLOAD = b"x" * 32 # small payload
network.WLAN(network.STA_IF).active(True)
esp = espnow.ESPNow()
esp.active(True)
esp.add_peer(BROADCAST)
# Build 19 dummy MACs so we have 20 peers total (broadcast + 19).
def mac(i):
return bytes([0, 0, 0, 0, 0, i])
peers_list = [mac(i) for i in range(1, 20)]
for p in peers_list:
esp.add_peer(p)
# One "new" MAC we'll add/remove.
new_mac = bytes([0, 0, 0, 0, 0, 99])
def bench_lru():
"""LRU: ensure_peer (evict oldest + add new), send, update last_used."""
last_used = {BROADCAST: time.ticks_ms()}
for p in peers_list:
last_used[p] = time.ticks_ms()
# Pre-remove one so we have 19; ensure_peer(new) will add 20th.
esp.del_peer(peers_list[-1])
last_used.pop(peers_list[-1], None)
# Now 19 peers. Each iteration: ensure_peer(new) -> add_peer(new), send, update.
# Next iter: ensure_peer(new) -> already there, just send. So we need to force
# eviction each time: use a different "new" each time so we always evict+add.
t0 = time.ticks_us()
for i in range(ITERATIONS):
addr = bytes([0, 0, 0, 0, 0, 50 + (i % 30)]) # 30 different "new" MACs
peers = esp.get_peers()
peer_macs = [p[0] for p in peers]
if addr not in peer_macs:
if len(peer_macs) >= MAX_PEERS:
oldest_mac = None
oldest_ts = time.ticks_ms()
for m in peer_macs:
if m == BROADCAST:
continue
ts = last_used.get(m, 0)
if ts <= oldest_ts:
oldest_ts = ts
oldest_mac = m
if oldest_mac is not None:
esp.del_peer(oldest_mac)
last_used.pop(oldest_mac, None)
esp.add_peer(addr)
esp.send(addr, PAYLOAD)
last_used[addr] = time.ticks_ms()
t1 = time.ticks_us()
return time.ticks_diff(t1, t0)
def bench_add_then_remove():
"""Add peer, send, del_peer (remove after use). At 20 we must del one first."""
# Start full: 20 peers. To add new we del any one, add new, send, del new.
victim = peers_list[0]
t0 = time.ticks_us()
for i in range(ITERATIONS):
esp.del_peer(victim) # make room
esp.add_peer(new_mac)
esp.send(new_mac, PAYLOAD)
esp.del_peer(new_mac)
esp.add_peer(victim) # put victim back so we're at 20 again
t1 = time.ticks_us()
return time.ticks_diff(t1, t0)
def bench_send_existing():
"""Baseline: send to existing peer only (no add/del)."""
t0 = time.ticks_us()
for _ in range(ITERATIONS):
esp.send(peers_list[0], PAYLOAD)
t1 = time.ticks_us()
return time.ticks_diff(t1, t0)
print("ESP-NOW peer benchmark ({} iterations)".format(ITERATIONS))
print()
# Baseline: send to existing peer
try:
us = bench_send_existing()
print("Send to existing peer only: {:>8} us total {:>7.1f} us/iter".format(us, us / ITERATIONS))
except Exception as e:
print("Send existing failed:", e)
print()
# LRU: evict oldest then add new, send
try:
us = bench_lru()
print("LRU (evict oldest + add + send): {:>8} us total {:>7.1f} us/iter".format(us, us / ITERATIONS))
except Exception as e:
print("LRU failed:", e)
print()
# Add then remove after use
try:
us = bench_add_then_remove()
print("Add then remove after use: {:>8} us total {:>7.1f} us/iter".format(us, us / ITERATIONS))
except Exception as e:
print("Add-then-remove failed:", e)
print()
print("Done.")

63
esp32/main.py
View File

@@ -1,63 +0,0 @@
# Serial-to-ESP-NOW bridge: receives from Pi on UART, forwards to ESP-NOW peers.
# Wire format: first 6 bytes = destination MAC, rest = payload. Address is always 6 bytes.
from machine import Pin, UART
import espnow
import network
import time
UART_BAUD = 912000
BROADCAST = b"\xff\xff\xff\xff\xff\xff"
MAX_PEERS = 20
network.WLAN(network.STA_IF).active(True)
esp = espnow.ESPNow()
esp.active(True)
esp.add_peer(BROADCAST)
uart = UART(1, UART_BAUD, tx=Pin(21), rx=Pin(6))
# Track last send time per peer for LRU eviction (remove oldest when at limit).
last_used = {BROADCAST: time.ticks_ms()}
# ESP_ERR_ESPNOW_EXIST: peer already registered (ignore when adding).
ESP_ERR_ESPNOW_EXIST = -12395
def ensure_peer(addr):
"""Ensure addr is in the peer list. When at 20 peers, remove the oldest-used (LRU)."""
peers = esp.get_peers()
peer_macs = [p[0] for p in peers]
if addr in peer_macs:
return
if len(peer_macs) >= MAX_PEERS:
# Remove the peer we used least recently (oldest).
oldest_mac = None
oldest_ts = time.ticks_ms()
for mac in peer_macs:
if mac == BROADCAST:
continue
ts = last_used.get(mac, 0)
if ts <= oldest_ts:
oldest_ts = ts
oldest_mac = mac
if oldest_mac is not None:
esp.del_peer(oldest_mac)
last_used.pop(oldest_mac, None)
try:
esp.add_peer(addr)
except OSError as e:
if e.args[0] != ESP_ERR_ESPNOW_EXIST:
raise
while True:
if uart.any():
data = uart.read()
if not data or len(data) < 6:
continue
addr = data[:6]
payload = data[6:]
ensure_peer(addr)
esp.send(addr, payload)
last_used[addr] = time.ticks_ms()

4
install.sh
View File

@@ -1,4 +0,0 @@
#!/usr/bin/env bash
# Install script - runs pipenv install
pipenv install "$@"

23
msg.json
View File

@@ -1,23 +0,0 @@
{
"g":{
"df": {
"pt": "on",
"cl": ["#ff0000"],
"br": 200,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"dl": 100
},
"dj": {
"pt": "blink",
"cl": ["#00ff00"],
"dl": 500
}
},
"sv": true,
"st": 0
}

173
run_web.py
View File

@@ -1,173 +0,0 @@
#!/usr/bin/env python3
"""
Local development web server - imports and runs main.py with port 5000
"""
import sys
import os
import asyncio
# Add src and lib to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'lib'))
# Import the main module
from src import main as main_module
# Override the port in the main function
async def run_local():
"""Run main with port 5000 for local development."""
from settings import Settings
import gc
# Mock MicroPython modules for local development
class MockMachine:
class WDT:
def __init__(self, timeout):
pass
def feed(self):
pass
import sys as sys_module
sys_module.modules['machine'] = MockMachine()
class MockESPNow:
def __init__(self):
self.active_value = False
self.peers = []
def active(self, value):
self.active_value = value
print(f"[MOCK] ESPNow active: {value}")
def add_peer(self, peer):
self.peers.append(peer)
print(f"[MOCK] Added peer: {peer.hex() if hasattr(peer, 'hex') else peer}")
async def asend(self, peer, data):
print(f"[MOCK] Would send to {peer.hex() if hasattr(peer, 'hex') else peer}: {data}")
class MockAIOESPNow:
def __init__(self):
pass
def active(self, value):
return MockESPNow()
def add_peer(self, peer):
pass
class MockNetwork:
class WLAN:
def __init__(self, interface):
self.interface = interface
def active(self, value):
print(f"[MOCK] WLAN({self.interface}) active: {value}")
STA_IF = 0
# Replace MicroPython modules with mocks
sys_module.modules['aioespnow'] = type('module', (), {'AIOESPNow': MockESPNow})()
sys_module.modules['network'] = MockNetwork()
# Mock gc if needed
if not hasattr(gc, 'collect'):
class MockGC:
def collect(self):
pass
gc = MockGC()
settings = Settings()
print("Starting LED Controller Web Server (Local Development)")
print("=" * 60)
# Mock network
import network
network.WLAN(network.STA_IF).active(True)
# Mock ESPNow
import aioespnow
e = aioespnow.AIOESPNow()
e.active(True)
e.add_peer(b"\xbb\xbb\xbb\xbb\xbb\xbb")
from microdot import Microdot, send_file
from microdot.websocket import with_websocket
from microdot.session import Session
import controllers.preset as preset
import controllers.profile as profile
import controllers.group as group
import controllers.sequence as sequence
import controllers.tab as tab
import controllers.palette as palette
import controllers.scene as scene
import controllers.pattern as pattern
import controllers.settings as settings_controller
app = Microdot()
# Initialize sessions with a secret key from settings
secret_key = settings.get('session_secret_key', 'led-controller-secret-key-change-in-production')
Session(app, secret_key=secret_key)
# Mount model controllers as subroutes
app.mount(preset.controller, '/presets')
app.mount(profile.controller, '/profiles')
app.mount(group.controller, '/groups')
app.mount(sequence.controller, '/sequences')
app.mount(tab.controller, '/tabs')
app.mount(palette.controller, '/palettes')
app.mount(scene.controller, '/scenes')
app.mount(pattern.controller, '/patterns')
app.mount(settings_controller.controller, '/settings')
# Serve index.html at root
@app.route('/')
def index(request):
"""Serve the main web UI."""
return send_file('src/templates/index.html')
# Serve settings page
@app.route('/settings')
def settings_page(request):
"""Serve the settings page."""
return send_file('src/templates/settings.html')
# Favicon: avoid 404 in browser console (no file needed)
@app.route('/favicon.ico')
def favicon(request):
return '', 204
# Static file route
@app.route("/static/<path:path>")
def static_handler(request, path):
"""Serve static files."""
if '..' in path:
return 'Not found', 404
return send_file('src/static/' + path)
@app.route('/ws')
@with_websocket
async def ws(request, ws):
while True:
data = await ws.receive()
if data:
await e.asend(b"\xbb\xbb\xbb\xbb\xbb\xbb", data)
print(data)
else:
break
# Use port 5000 for local development
port = 5000
print(f"Starting server on http://0.0.0.0:{port}")
print(f"Open http://localhost:{port} in your browser")
print("=" * 60)
try:
await app.start_server(host="0.0.0.0", port=port, debug=True)
except KeyboardInterrupt:
print("\nShutting down server...")
if __name__ == '__main__':
# Change to project root
os.chdir(os.path.dirname(os.path.abspath(__file__)))
# Override settings path for local development
import settings as settings_module
settings_module.Settings.SETTINGS_FILE = os.path.join(os.getcwd(), 'settings.json')
asyncio.run(run_local())

44
send_empty_json.py
View File

@@ -1,44 +0,0 @@
#!/usr/bin/env python3
import socket
import struct
import base64
import hashlib
# Connect to the WebSocket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('192.168.4.1', 80))
# Send HTTP WebSocket upgrade request
key = base64.b64encode(b'test-nonce').decode('utf-8')
request = f'''GET /ws HTTP/1.1\r
Host: 192.168.4.1\r
Upgrade: websocket\r
Connection: Upgrade\r
Sec-WebSocket-Key: {key}\r
Sec-WebSocket-Version: 13\r
\r
'''
s.send(request.encode())
# Read upgrade response
response = s.recv(4096)
print(response.decode())
# Send WebSocket TEXT frame with empty JSON '{}'
payload = b'{}'
mask = b'\x12\x34\x56\x78'
payload_masked = bytes(p ^ mask[i % 4] for i, p in enumerate(payload))
frame = struct.pack('BB', 0x81, 0x80 | len(payload))
frame += mask
frame += payload_masked
s.send(frame)
print("Sent empty JSON to WebSocket")
s.close()

1
settings.json
View File

@@ -1 +0,0 @@
{"session_secret_key": "06a2a6ac631cc8db0658373b37f7fe922a8a3ed3a27229e1adb5448cb368f2b7"}