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

Pi port: serial transport, addressed ESP-NOW bridge, port 80

Browse Source 
- Run app on Raspberry Pi: serial to ESP32 bridge at 912000 baud, /dev/ttyS0
- Remove ESP-NOW/MicroPython-only code from src (espnow, p2p, wifi, machine/Pin)
- Transport: always send 6-byte MAC + payload; optional to/destination_mac in API and WebSocket
- Settings and model DB use project paths (no root); fix sys.print_exception for CPython
- Preset/settings controllers use get_current_sender(); template paths for cwd=src
- Pipfile: run from src, PORT from env; scripts for port 80 (setcap) and test
- ESP32 bridge: receive 6-byte addr + payload, LRU peer management (20 max), handle ESP_ERR_ESPNOW_EXIST
- Add esp32/main.py, esp32/benchmark_peers.py, scripts/setup-port80.sh, scripts/test-port80.sh

Made-with: Cursor
This commit is contained in:
Jimmy
2026-03-15 17:16:07 +13:00
parent 0fdc11c0b0
commit ac9fca8d4b
19 changed files with 656 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/boot.py
View File

@@ -1,8 +1,6 @@
import settings
import util.wifi as wifi
# Boot script (ESP only; no-op on Pi)
import settings # noqa: F401
from settings import Settings
s = Settings()
name = s.get('name', 'led-controller')
wifi.ap(name, '')
# AP setup was here when running on ESP; Pi uses system networking.

32
src/controllers/preset.py
View File

@@ -2,7 +2,7 @@ from microdot import Microdot
from microdot.session import with_session
from models.preset import Preset
from models.profile import Profile
from models.espnow import ESPNow
from models.transport import get_current_sender
from util.espnow_message import build_message, build_preset_dict
import asyncio
import json
@@ -110,16 +110,13 @@ async def delete_preset(request, id, session):
@with_session
async def send_presets(request, session):
"""
Send one or more presets over ESPNow.
Send one or more presets to the LED driver (via serial transport).
Body JSON:
{"preset_ids": ["1", "2", ...]} or {"ids": ["1", "2", ...]}
The controller:
- looks up each preset in the Preset model
- converts them to API-compliant format
- splits into <= 240-byte ESPNow messages
- sends each message to all configured ESPNow peers.
The controller looks up each preset, converts to API format, chunks into
<= 240-byte messages, and sends them over the configured transport.
"""
try:
data = request.json or {}
@@ -132,6 +129,8 @@ async def send_presets(request, session):
save_flag = data.get('save', True)
save_flag = bool(save_flag)
default_id = data.get('default')
# Optional 12-char hex MAC to send to one device; omit for default (e.g. broadcast).
destination_mac = data.get('destination_mac') or data.get('to')
# Build API-compliant preset map keyed by preset ID, include name
current_profile_id = get_current_profile_id(session)
@@ -153,16 +152,17 @@ async def send_presets(request, session):
if default_id is not None and str(default_id) not in presets_by_name:
default_id = None
# Use shared ESPNow singleton
esp = ESPNow()
sender = get_current_sender()
if not sender:
return json.dumps({"error": "Transport not configured"}), 503, {'Content-Type': 'application/json'}
async def send_chunk(chunk_presets):
# Include save flag so the led-driver can persist when desired.
msg = build_message(presets=chunk_presets, save=save_flag, default=default_id)
await esp.send(msg)
await sender.send(msg, addr=destination_mac)
MAX_BYTES = 240
SEND_DELAY_MS = 100
send_delay_s = 0.1
entries = list(presets_by_name.items())
total_presets = len(entries)
messages_sent = 0
@@ -182,8 +182,8 @@ async def send_presets(request, session):
try:
await send_chunk(batch)
except Exception:
return json.dumps({"error": "ESP-NOW send failed"}), 503, {'Content-Type': 'application/json'}
await asyncio.sleep_ms(SEND_DELAY_MS)
return json.dumps({"error": "Send failed"}), 503, {'Content-Type': 'application/json'}
await asyncio.sleep(send_delay_s)
messages_sent += 1
batch = {name: preset_obj}
last_msg = build_message(presets=batch, save=save_flag, default=default_id)
@@ -192,12 +192,12 @@ async def send_presets(request, session):
try:
await send_chunk(batch)
except Exception:
return json.dumps({"error": "ESP-NOW send failed"}), 503, {'Content-Type': 'application/json'}
await asyncio.sleep_ms(SEND_DELAY_MS)
return json.dumps({"error": "Send failed"}), 503, {'Content-Type': 'application/json'}
await asyncio.sleep(send_delay_s)
messages_sent += 1
return json.dumps({
"message": "Presets sent via ESPNow",
"message": "Presets sent",
"presets_sent": total_presets,
"messages_sent": messages_sent
}), 200, {'Content-Type': 'application/json'}

26
src/controllers/settings.py
View File

@@ -1,6 +1,5 @@
from microdot import Microdot, send_file
from settings import Settings
import util.wifi as wifi
import json
controller = Microdot()
@@ -15,19 +14,18 @@ async def get_settings(request):
@controller.get('/wifi/ap')
async def get_ap_config(request):
"""Get Access Point configuration."""
config = wifi.get_ap_config()
if config:
# Also get saved settings
config['saved_ssid'] = settings.get('wifi_ap_ssid')
config['saved_password'] = settings.get('wifi_ap_password')
config['saved_channel'] = settings.get('wifi_ap_channel')
return json.dumps(config), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Failed to get AP config"}), 500
"""Get saved AP configuration (Pi: no in-device AP)."""
config = {
'saved_ssid': settings.get('wifi_ap_ssid'),
'saved_password': settings.get('wifi_ap_password'),
'saved_channel': settings.get('wifi_ap_channel'),
'active': False,
}
return json.dumps(config), 200, {'Content-Type': 'application/json'}
@controller.post('/wifi/ap')
async def configure_ap(request):
"""Configure Access Point."""
"""Save AP configuration to settings (Pi: no in-device AP)."""
try:
data = request.json
ssid = data.get('ssid')
@@ -43,18 +41,14 @@ async def configure_ap(request):
if channel < 1 or channel > 11:
return json.dumps({"error": "Channel must be between 1 and 11"}), 400
# Save to settings
settings['wifi_ap_ssid'] = ssid
settings['wifi_ap_password'] = password
if channel is not None:
settings['wifi_ap_channel'] = channel
settings.save()
# Configure AP
wifi.ap(ssid, password, channel)
return json.dumps({
"message": "AP configured successfully",
"message": "AP settings saved",
"ssid": ssid,
"channel": channel
}), 200, {'Content-Type': 'application/json'}

58
src/main.py
View File

@@ -1,14 +1,11 @@
import asyncio
import gc
import json
import machine
from machine import Pin
import os
from microdot import Microdot, send_file
from microdot.websocket import with_websocket
from microdot.session import Session
from settings import Settings
import aioespnow
import controllers.preset as preset
import controllers.profile as profile
import controllers.group as group
@@ -18,7 +15,7 @@ import controllers.palette as palette
import controllers.scene as scene
import controllers.pattern as pattern
import controllers.settings as settings_controller
from models.espnow import ESPNow
from models.transport import get_sender, set_sender
async def main(port=80):
@@ -26,8 +23,9 @@ async def main(port=80):
print(settings)
print("Starting")
# Initialize ESPNow singleton (config + peers)
esp = ESPNow()
# Initialize transport (serial to ESP32 bridge)
sender = get_sender(settings)
set_sender(sender)
app = Microdot()
@@ -58,7 +56,7 @@ async def main(port=80):
app.mount(pattern.controller, '/patterns')
app.mount(settings_controller.controller, '/settings')
# Serve index.html at root
# Serve index.html at root (cwd is src/ when run via pipenv run run)
@app.route('/')
def index(request):
"""Serve the main web UI."""
@@ -91,19 +89,25 @@ async def main(port=80):
data = await ws.receive()
print(data)
if data:
# Debug: log incoming WebSocket data
try:
parsed = json.loads(data)
print("WS received JSON:", parsed)
except Exception:
print("WS received raw:", data)
# Forward raw JSON payload over ESPNow to configured peers
try:
await esp.send(data)
# Optional "to": 12-char hex MAC; rest is payload (sent with that address).
addr = parsed.pop("to", None)
payload = json.dumps(parsed) if parsed else data
await sender.send(payload, addr=addr)
except json.JSONDecodeError:
# Not JSON: send raw with default address
try:
await sender.send(data)
except Exception:
try:
await ws.send(json.dumps({"error": "Send failed"}))
except Exception:
pass
except Exception:
try:
await ws.send(json.dumps({"error": "ESP-NOW send failed"}))
await ws.send(json.dumps({"error": "Send failed"}))
except Exception:
pass
else:
@@ -113,25 +117,11 @@ async def main(port=80):
server = asyncio.create_task(app.start_server(host="0.0.0.0", port=port))
#wdt = machine.WDT(timeout=10000)
#wdt.feed()
# Initialize heartbeat LED (XIAO ESP32S3 built-in LED on GPIO 21)
led = Pin(15, Pin.OUT)
led_state = False
while True:
gc.collect()
for i in range(60):
#wdt.feed()
# Heartbeat: toggle LED every 500 ms
led.value(not led.value())
await asyncio.sleep_ms(500)
await asyncio.sleep(30)
# cleanup before ending the application
if __name__ == "__main__":
asyncio.run(main())
import os
port = int(os.environ.get("PORT", 80))
asyncio.run(main(port=port))

69
src/models/espnow.py
View File

@@ -1,69 +0,0 @@
import network
import aioespnow
class ESPNow:
"""
Singleton ESPNow helper:
- Manages a single AIOESPNow instance
- Adds a single broadcast-like peer
- Exposes async send(data) to send to that peer.
"""
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if getattr(self, "_initialized", False):
return
# ESP-NOW requires a WiFi interface to be active (STA or AP). Activate STA
# so ESP-NOW has an interface to use; we don't need to connect to an AP.
try:
sta = network.WLAN(network.STA_IF)
sta.active(True)
except Exception as e:
print("ESPNow: STA active failed:", e)
self._esp = aioespnow.AIOESPNow()
self._esp.active(True)
try:
self._esp.add_peer(b"\xff\xff\xff\xff\xff\xff")
except Exception:
# Ignore add_peer failures (e.g. duplicate)
pass
self._initialized = True
async def send(self, data):
"""
Async send to the broadcast peer.
- data: bytes or str (JSON)
"""
if isinstance(data, str):
payload = data.encode()
else:
payload = data
# Debug: show what we're sending and its size
try:
preview = payload.decode('utf-8')
except Exception:
preview = str(payload)
if len(preview) > 200:
preview = preview[:200] + "...(truncated)"
print("ESPNow.send len=", len(payload), "payload=", preview)
try:
await self._esp.asend(b"\xff\xff\xff\xff\xff\xff", payload)
except Exception as e:
print("ESPNow.send error:", e)
raise

27
src/models/model.py
View File

@@ -1,5 +1,15 @@
import json
import os
import traceback
# DB directory: project root / db (writable without root)
def _db_dir():
try:
# src/models/model.py -> project root
base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
return os.path.join(base, "db")
except Exception:
return "db"
class Model(dict):
def __new__(cls, *args, **kwargs):
@@ -13,13 +23,13 @@ class Model(dict):
if hasattr(self, '_initialized'):
return
# Create /db directory if it doesn't exist (MicroPython compatible)
db_dir = _db_dir()
try:
os.mkdir("/db")
os.makedirs(db_dir, exist_ok=True)
except OSError:
pass # Directory already exists, which is fine
pass
self.class_name = self.__class__.__name__
self.file = f"/db/{self.class_name.lower()}.json"
self.file = os.path.join(db_dir, f"{self.class_name.lower()}.json")
super().__init__()
self.load() # Load settings from file during initialization
@@ -37,11 +47,11 @@ class Model(dict):
def save(self):
try:
# Ensure directory exists
db_dir = os.path.dirname(self.file)
try:
os.mkdir("/db")
os.makedirs(db_dir, exist_ok=True)
except OSError:
pass # Directory already exists
pass
j = json.dumps(self)
with open(self.file, 'w') as file:
file.write(j)
@@ -54,8 +64,7 @@ class Model(dict):
print(f"{self.class_name} saved successfully to {self.file}")
except Exception as e:
print(f"Error saving {self.class_name} to {self.file}: {e}")
import sys
sys.print_exception(e)
traceback.print_exception(type(e), e, e.__traceback__)
def load(self):
try:

12
src/models/serial.py Normal file
View File

@@ -0,0 +1,12 @@
class Serial:
def __init__(self, port, baudrate):
self.port = port
self.baudrate = baudrate
self.uart = UART(1, baudrate, tx=Pin(21), rx=Pin(6))
def send(self, data):
self.uart.write(data)
def receive(self):
return self.uart.read()

66
src/models/transport.py Normal file
View File

@@ -0,0 +1,66 @@
import asyncio
import json
# Default: broadcast (6 bytes). Pi always sends 6-byte address + payload to ESP32.
BROADCAST_MAC = bytes.fromhex("ffffffffffff")
def _encode_payload(data):
if isinstance(data, str):
return data.encode()
if isinstance(data, dict):
return json.dumps(data).encode()
return data
def _parse_mac(addr):
"""Convert 12-char hex string or 6-byte bytes to 6-byte MAC."""
if addr is None or addr == b"":
return BROADCAST_MAC
if isinstance(addr, bytes) and len(addr) == 6:
return addr
if isinstance(addr, str) and len(addr) == 12:
return bytes.fromhex(addr)
return BROADCAST_MAC
async def _to_thread(func, *args):
to_thread = getattr(asyncio, "to_thread", None)
if to_thread:
return await to_thread(func, *args)
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, func, *args)
class SerialSender:
def __init__(self, port, baudrate, default_addr=None):
import serial
self._serial = serial.Serial(port, baudrate=baudrate, timeout=1)
self._default_addr = _parse_mac(default_addr)
async def send(self, data, addr=None):
mac = _parse_mac(addr) if addr is not None else self._default_addr
payload = _encode_payload(data)
await _to_thread(self._serial.write, mac + payload)
return True
_current_sender = None
def set_sender(sender):
global _current_sender
_current_sender = sender
def get_current_sender():
return _current_sender
def get_sender(settings):
port = settings.get("serial_port", "/dev/ttyS0")
baudrate = settings.get("serial_baudrate", 912000)
default_addr = settings.get("serial_destination_mac", "ffffffffffff")
return SerialSender(port, baudrate, default_addr=default_addr)

39
src/p2p.py
View File

@@ -1,39 +0,0 @@
import network
import aioespnow
import asyncio
import json
from time import sleep
class P2P:
def __init__(self):
network.WLAN(network.STA_IF).active(True)
self.broadcast = bytes.fromhex("ffffffffffff")
self.e = aioespnow.AIOESPNow()
self.e.active(True)
try:
self.e.add_peer(self.broadcast)
except:
pass
async def send(self, data):
# Convert data to bytes if it's a string or dict
if isinstance(data, str):
payload = data.encode()
elif isinstance(data, dict):
payload = json.dumps(data).encode()
else:
payload = data # Assume it's already bytes
# Use asend for async sending - returns boolean indicating success
result = await self.e.asend(self.broadcast, payload)
return result
async def main():
p = P2P()
await p.send(json.dumps({"dj": {"p": "on", "colors": ["#ff0000"]}}))
if __name__ == "__main__":
asyncio.run(main())

14
src/settings.py
View File

@@ -2,11 +2,23 @@ import json
import os
import binascii
def _settings_path():
"""Path to settings.json in project root (writable without root)."""
try:
base = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
return os.path.join(base, "settings.json")
except Exception:
return "settings.json"
class Settings(dict):
SETTINGS_FILE = "/settings.json"
SETTINGS_FILE = None # Set in __init__ from _settings_path()
def __init__(self):
super().__init__()
if Settings.SETTINGS_FILE is None:
Settings.SETTINGS_FILE = _settings_path()
self.load() # Load settings from file during initialization
def generate_secret_key(self):

13
src/util/espnow_message.py
View File

@@ -1,7 +1,8 @@
"""
ESPNow message builder utility for LED driver communication.
Message builder for LED driver API communication.
This module provides utilities to build ESPNow messages according to the API specification.
Builds JSON messages according to the LED driver API specification
for sending presets and select commands over the transport (e.g. serial).
"""
import json
@@ -9,14 +10,14 @@ import json
def build_message(presets=None, select=None, save=False, default=None):
"""
Build an ESPNow message according to the API specification.
Build an API message (presets and/or select) as a JSON string.
Args:
presets: Dictionary mapping preset names to preset objects, or None
select: Dictionary mapping device names to select lists, or None
Returns:
JSON string ready to send via ESPNow
JSON string ready to send over the transport
Example:
message = build_message(

42
src/util/wifi.py
View File

@@ -1,42 +0,0 @@
import network
def ap(ssid, password, channel=None):
ap_if = network.WLAN(network.AP_IF)
ap_mac = ap_if.config('mac')
print(ssid)
ap_if.active(True)
if channel is not None:
ap_if.config(essid=ssid, password=password, channel=channel)
else:
ap_if.config(essid=ssid, password=password)
ap_if.active(False)
ap_if.active(True)
print(ap_if.ifconfig())
def get_mac():
ap_if = network.WLAN(network.AP_IF)
return ap_if.config('mac')
def get_ap_config():
"""Get current AP configuration."""
try:
ap_if = network.WLAN(network.AP_IF)
if ap_if.active():
config = ap_if.ifconfig()
return {
'ssid': ap_if.config('essid'),
'channel': ap_if.config('channel'),
'ip': config[0] if config else None,
'active': True
}
return {
'ssid': None,
'channel': None,
'ip': None,
'active': False
}
except Exception as e:
print(f"Error getting AP config: {e}")
return None