Update backend models, controllers, and session

This commit is contained in:
2026-01-16 22:31:24 +13:00
parent d41faddfca
commit 9c43a0a22b
16 changed files with 916 additions and 64 deletions

50
dev.py
View File

@@ -6,28 +6,48 @@ import sys
print(sys.argv)
port = sys.argv[1]
# Extract port (first arg if it's not a command)
commands = ["src", "lib", "ls", "reset", "follow", "db"]
port = None
if len(sys.argv) > 1 and sys.argv[1] not in commands:
port = sys.argv[1]
cmd = sys.argv[1]
for cmd in sys.argv[1:]:
print(cmd)
match cmd:
case "src":
subprocess.call(["mpremote", "connect", port, "fs", "cp", "-r", ".", ":" ], cwd="src")
if port:
subprocess.call(["mpremote", "connect", port, "fs", "cp", "-r", ".", ":" ], cwd="src")
else:
print("Error: Port required for 'src' command")
case "lib":
subprocess.call(["mpremote", "connect", port, "fs", "cp", "-r", "lib", ":" ])
if port:
subprocess.call(["mpremote", "connect", port, "fs", "cp", "-r", "lib", ":" ])
else:
print("Error: Port required for 'lib' command")
case "ls":
subprocess.call(["mpremote", "connect", port, "fs", "ls", ":" ])
if port:
subprocess.call(["mpremote", "connect", port, "fs", "ls", ":" ])
else:
print("Error: Port required for 'ls' command")
case "reset":
with serial.Serial(port, baudrate=115200) as ser:
ser.write(b'\x03\x03\x04')
if port:
with serial.Serial(port, baudrate=115200) as ser:
ser.write(b'\x03\x03\x04')
else:
print("Error: Port required for 'reset' command")
case "follow":
with serial.Serial(port, baudrate=115200) as ser:
while True:
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)
if port:
with serial.Serial(port, baudrate=115200) as ser:
while True:
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)
else:
print("Error: Port required for 'follow' command")
case "db":
if port:
subprocess.call(["mpremote", "connect", port, "fs", "cp", "-r", "db", ":" ])
else:
print("Error: Port required for 'db' command")

159
lib/microdot/session.py Normal file
View File

@@ -0,0 +1,159 @@
import jwt
from microdot.microdot import invoke_handler
from microdot.helpers import wraps
class SessionDict(dict):
"""A session dictionary.
The session dictionary is a standard Python dictionary that has been
extended with convenience ``save()`` and ``delete()`` methods.
"""
def __init__(self, request, session_dict):
super().__init__(session_dict)
self.request = request
def save(self):
"""Update the session cookie."""
self.request.app._session.update(self.request, self)
def delete(self):
"""Delete the session cookie."""
self.request.app._session.delete(self.request)
class Session:
"""Session handling
:param app: The application instance.
:param secret_key: The secret key, as a string or bytes object.
:param cookie_options: A dictionary with cookie options to pass as
arguments to :meth:`Response.set_cookie()
<microdot.Response.set_cookie>`.
"""
secret_key = None
def __init__(self, app=None, secret_key=None, cookie_options=None):
self.secret_key = secret_key
self.cookie_options = cookie_options or {}
if app is not None:
self.initialize(app)
def initialize(self, app, secret_key=None, cookie_options=None):
if secret_key is not None:
self.secret_key = secret_key
if cookie_options is not None:
self.cookie_options = cookie_options
if 'path' not in self.cookie_options:
self.cookie_options['path'] = '/'
if 'http_only' not in self.cookie_options:
self.cookie_options['http_only'] = True
app._session = self
def get(self, request):
"""Retrieve the user session.
:param request: The client request.
The return value is a session dictionary with the data stored in the
user's session, or ``{}`` if the session data is not available or
invalid.
"""
if not self.secret_key:
raise ValueError('The session secret key is not configured')
if hasattr(request.g, '_session'):
return request.g._session
session = request.cookies.get('session')
if session is None:
request.g._session = SessionDict(request, {})
return request.g._session
request.g._session = SessionDict(request, self.decode(session))
return request.g._session
def update(self, request, session):
"""Update the user session.
:param request: The client request.
:param session: A dictionary with the update session data for the user.
Applications would normally not call this method directly, instead they
would use the :meth:`SessionDict.save` method on the session
dictionary, which calls this method. For example::
@app.route('/')
@with_session
def index(request, session):
session['foo'] = 'bar'
session.save()
return 'Hello, World!'
Calling this method adds a cookie with the updated session to the
request currently being processed.
"""
if not self.secret_key:
raise ValueError('The session secret key is not configured')
encoded_session = self.encode(session)
@request.after_request
def _update_session(request, response):
response.set_cookie('session', encoded_session,
**self.cookie_options)
return response
def delete(self, request):
"""Remove the user session.
:param request: The client request.
Applications would normally not call this method directly, instead they
would use the :meth:`SessionDict.delete` method on the session
dictionary, which calls this method. For example::
@app.route('/')
@with_session
def index(request, session):
session.delete()
return 'Hello, World!'
Calling this method adds a cookie removal header to the request
currently being processed.
"""
@request.after_request
def _delete_session(request, response):
response.delete_cookie('session', **self.cookie_options)
return response
def encode(self, payload, secret_key=None):
return jwt.encode(payload, secret_key or self.secret_key,
algorithm='HS256')
def decode(self, session, secret_key=None):
try:
payload = jwt.decode(session, secret_key or self.secret_key,
algorithms=['HS256'])
except jwt.exceptions.PyJWTError: # pragma: no cover
return {}
return payload
def with_session(f):
"""Decorator that passes the user session to the route handler.
The session dictionary is passed to the decorated function as an argument
after the request object. Example::
@app.route('/')
@with_session
def index(request, session):
return 'Hello, World!'
Note that the decorator does not save the session. To update the session,
call the :func:`session.save() <microdot.session.SessionDict.save>` method.
"""
@wraps(f)
async def wrapper(request, *args, **kwargs):
return await invoke_handler(
f, request, request.app._session.get(request), *args, **kwargs)
return wrapper

View File

@@ -1,8 +1,8 @@
import settings
import wifi
import util.wifi as wifi
from settings import Settings
s = Settings()
name = s.get('name', 'led')
name = s.get('name', 'led-controller')
wifi.ap(name, '')

View File

@@ -0,0 +1 @@
# Controllers package

View File

@@ -0,0 +1,55 @@
from microdot import Microdot
from models.pattern import Pattern
import json
controller = Microdot()
patterns = Pattern()
@controller.get('')
async def list_patterns(request):
"""List all patterns."""
return json.dumps(patterns), 200, {'Content-Type': 'application/json'}
@controller.get('/<id>')
async def get_pattern(request, id):
"""Get a specific pattern by ID."""
pattern = patterns.read(id)
if pattern is not None:
return json.dumps(pattern), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Pattern not found"}), 404
@controller.post('')
async def create_pattern(request):
"""Create a new pattern."""
try:
data = request.json or {}
name = data.get("name", "")
pattern_id = patterns.create(name, data.get("data", {}))
if data:
patterns.update(pattern_id, data)
return json.dumps(patterns.read(pattern_id)), 201, {'Content-Type': 'application/json'}
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.put('/<id>')
async def update_pattern(request, id):
"""Update an existing pattern."""
try:
data = request.json
if patterns.update(id, data):
return json.dumps(patterns.read(id)), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Pattern not found"}), 404
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.delete('/<id>')
async def delete_pattern(request, id):
"""Delete a pattern."""
if patterns.delete(id):
return json.dumps({"message": "Pattern deleted successfully"}), 200
return json.dumps({"error": "Pattern not found"}), 404

View File

@@ -1,4 +1,5 @@
from microdot import Microdot
from microdot.session import with_session
from models.profile import Profile
import json
@@ -10,6 +11,21 @@ async def list_profiles(request):
"""List all profiles."""
return json.dumps(profiles), 200, {'Content-Type': 'application/json'}
@controller.get('/current')
@with_session
async def get_current_profile(request, session):
"""Get the current profile ID from session (or fallback)."""
profile_list = profiles.list()
current_id = session.get('current_profile')
if not current_id and profile_list:
current_id = profile_list[0]
session['current_profile'] = str(current_id)
session.save()
if current_id:
profile = profiles.read(current_id)
return json.dumps({"id": current_id, "profile": profile}), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "No profile available"}), 404
@controller.get('/<id>')
async def get_profile(request, id):
"""Get a specific profile by ID."""
@@ -18,6 +34,16 @@ async def get_profile(request, id):
return json.dumps(profile), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Profile not found"}), 404
@controller.post('/<id>/apply')
@with_session
async def apply_profile(request, session, id):
"""Apply a profile by saving it to session."""
if not profiles.read(id):
return json.dumps({"error": "Profile not found"}), 404
session['current_profile'] = str(id)
session.save()
return json.dumps({"message": "Profile applied", "id": str(id)}), 200, {'Content-Type': 'application/json'}
@controller.post('')
async def create_profile(request):
"""Create a new profile."""
@@ -31,6 +57,26 @@ async def create_profile(request):
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.put('/current')
@with_session
async def update_current_profile(request, session):
"""Update the current profile using session (or fallback)."""
try:
data = request.json or {}
profile_list = profiles.list()
current_id = session.get('current_profile')
if not current_id and profile_list:
current_id = profile_list[0]
session['current_profile'] = str(current_id)
session.save()
if not current_id:
return json.dumps({"error": "No profile available"}), 404
if profiles.update(current_id, data):
return json.dumps(profiles.read(current_id)), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Profile not found"}), 404
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.put('/<id>')
async def update_profile(request, id):
"""Update an existing profile."""

49
src/controllers/scene.py Normal file
View File

@@ -0,0 +1,49 @@
from microdot import Microdot
from models.scene import Scene
import json
controller = Microdot()
scenes = Scene()
@controller.get('')
async def list_scenes(request):
"""List all scenes."""
return json.dumps(scenes), 200, {'Content-Type': 'application/json'}
@controller.get('/<id>')
async def get_scene(request, id):
"""Get a specific scene by ID."""
scene = scenes.read(id)
if scene:
return json.dumps(scene), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Scene not found"}), 404
@controller.post('')
async def create_scene(request):
"""Create a new scene."""
try:
data = request.json
scene_id = scenes.create()
if scenes.update(scene_id, data):
return json.dumps(scenes.read(scene_id)), 201, {'Content-Type': 'application/json'}
return json.dumps({"error": "Failed to create scene"}), 400
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.put('/<id>')
async def update_scene(request, id):
"""Update an existing scene."""
try:
data = request.json
if scenes.update(id, data):
return json.dumps(scenes.read(id)), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Scene not found"}), 404
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.delete('/<id>')
async def delete_scene(request, id):
"""Delete a scene."""
if scenes.delete(id):
return json.dumps({"message": "Scene deleted successfully"}), 200
return json.dumps({"error": "Scene not found"}), 404

View File

@@ -1,15 +1,253 @@
from microdot import Microdot
from microdot import Microdot, send_file
from microdot.session import with_session
from models.tab import Tab
from models.profile import Profile
import json
import os
import time
controller = Microdot()
tabs = Tab()
profiles = Profile()
def get_current_profile_id(session=None):
"""Get the current active profile ID from session or fallback to first."""
profile_list = profiles.list()
session_profile = None
if session is not None:
session_profile = session.get('current_profile')
if session_profile and session_profile in profile_list:
return session_profile
if profile_list:
return profile_list[0]
return None
def get_profile_tab_order(profile_id):
"""Get the tab order for a profile."""
if not profile_id:
return []
profile = profiles.read(profile_id)
if profile:
# Support both "tab_order" (old) and "tabs" (new) format
return profile.get("tabs", profile.get("tab_order", []))
return []
def get_current_tab_id(request, session=None):
"""Get the current tab ID from session."""
if session:
current_tab = session.get('current_tab')
if current_tab:
return current_tab
# Fallback to first tab in current profile if no session
profile_id = get_current_profile_id(session)
if profile_id:
profile = profiles.read(profile_id)
if profile:
# Support both "tabs" (new) and "tab_order" (old) format
tabs_list = profile.get("tabs", profile.get("tab_order", []))
if tabs_list:
return tabs_list[0]
return None
@controller.get('')
async def list_tabs(request):
"""List all tabs."""
return json.dumps(tabs), 200, {'Content-Type': 'application/json'}
# HTML Fragment endpoints for htmx - must be before /<id> route
@controller.get('/list-fragment')
@with_session
async def tabs_list_fragment(request, session):
"""Return HTML fragment for the tabs list."""
profile_id = get_current_profile_id(session)
# #region agent log
try:
os.makedirs('/home/pi/led-controller/.cursor', exist_ok=True)
with open('/home/pi/led-controller/.cursor/debug.log', 'a') as _log:
_log.write(json.dumps({
"sessionId": "debug-session",
"runId": "tabs-pre-fix",
"hypothesisId": "H1",
"location": "src/controllers/tab.py:tabs_list_fragment",
"message": "tabs list fragment",
"data": {
"profile_id": profile_id,
"profile_count": len(profiles.list())
},
"timestamp": int(time.time() * 1000)
}) + "\n")
except Exception:
pass
# #endregion
if not profile_id:
return '<div class="tabs-list">No profile selected</div>', 200, {'Content-Type': 'text/html'}
tab_order = get_profile_tab_order(profile_id)
current_tab_id = get_current_tab_id(request, session)
html = '<div class="tabs-list">'
for tab_id in tab_order:
tab_data = tabs.read(tab_id)
if tab_data:
active_class = 'active' if str(tab_id) == str(current_tab_id) else ''
tab_name = tab_data.get('name', 'Tab ' + str(tab_id))
html += (
'<button class="tab-button ' + active_class + '" '
'hx-get="/tabs/' + str(tab_id) + '/content-fragment" '
'hx-target="#tab-content" '
'hx-swap="innerHTML" '
'hx-push-url="true" '
'hx-trigger="click" '
'onclick="document.querySelectorAll(\'.tab-button\').forEach(b => b.classList.remove(\'active\')); this.classList.add(\'active\');">'
+ tab_name +
'</button>'
)
html += '</div>'
return html, 200, {'Content-Type': 'text/html'}
@controller.get('/create-form-fragment')
async def create_tab_form_fragment(request):
"""Return the create tab form HTML fragment."""
html = '''
<h2>Add New Tab</h2>
<form hx-post="/tabs"
hx-target="#tabs-list"
hx-swap="innerHTML"
hx-headers='{"Accept": "text/html"}'
hx-on::after-request="if(event.detail.successful) { document.getElementById('add-tab-modal').classList.remove('active'); document.body.dispatchEvent(new Event('tabs-updated')); }">
<label>Tab Name:</label>
<input type="text" name="name" placeholder="Enter tab name" required>
<label>Device IDs (comma-separated):</label>
<input type="text" name="ids" placeholder="1,2,3" value="1">
<div class="modal-actions">
<button type="submit" class="btn btn-primary">Add</button>
<button type="button" class="btn btn-secondary" onclick="document.getElementById('add-tab-modal').classList.remove('active')">Cancel</button>
</div>
</form>
'''
return html, 200, {'Content-Type': 'text/html'}
@controller.get('/current')
@with_session
async def get_current_tab(request, session):
"""Get the current tab from session."""
current_tab_id = get_current_tab_id(request, session)
if not current_tab_id:
accept_header = request.headers.get('Accept', '')
wants_html = 'text/html' in accept_header
if wants_html:
return '<div class="error">No current tab set</div>', 404, {'Content-Type': 'text/html'}
return json.dumps({"error": "No current tab set"}), 404
return await tab_content_fragment.__wrapped__(request, session, current_tab_id)
@controller.get('/<id>/content-fragment')
@with_session
async def tab_content_fragment(request, session, id):
"""Return HTML fragment for tab content."""
# Handle 'current' as a special case
if id == 'current':
return await get_current_tab(request, session)
tab = tabs.read(id)
if not tab:
return '<div>Tab not found</div>', 404, {'Content-Type': 'text/html'}
# Set this tab as the current tab in session
session['current_tab'] = str(id)
session.save()
# If this is a direct page load (not HTMX), return full UI so CSS loads.
if not request.headers.get('HX-Request'):
return send_file('templates/index.html')
tab_name = tab.get('name', 'Tab ' + str(id))
device_ids = ', '.join(tab.get('names', []))
html = (
'<div class="left-panel">'
'<div class="left-panel-header">'
'<div class="ids-display">'
'<label>IDs: </label>'
'<span id="current-ids">' + device_ids + '</span>'
'</div>'
'<button id="toggle-left-panel" class="btn btn-small left-panel-toggle" title="Collapse/expand controls">◀</button>'
'</div>'
'<div class="left-panel-body">'
'<div class="color-palette-section">'
'<h3>Color Palette</h3>'
'<div id="color-palette" class="color-palette" data-tab-id="' + str(id) + '">'
'<!-- Colors will be loaded here -->'
'</div>'
'<div class="palette-actions">'
'<input type="color" id="tab-color-input" value="#ffffff">'
'<button class="btn btn-small" id="tab-color-add-btn">Add Color</button>'
'<button class="btn btn-small" id="tab-color-add-from-palette-btn">Add from Palette</button>'
'</div>'
'</div>'
'<div class="controls-section">'
'<div class="control-group">'
'<label for="brightness-slider">Brightness:</label>'
'<input type="range" id="brightness-slider" min="0" max="255" value="127" class="slider">'
'<span id="brightness-value" class="slider-value">127</span>'
'</div>'
'<div class="control-group">'
'<label for="delay-slider">Delay:</label>'
'<input type="range" id="delay-slider" min="0" max="1000" value="0" class="slider">'
'<span id="delay-value" class="slider-value">100 ms</span>'
'</div>'
'</div>'
'<div class="n-params-section">'
'<h3>N Parameters</h3>'
'<div class="n-params-grid">'
'<div class="n-param-group">'
'<label for="n1-input" id="n1-label">n1:</label>'
'<input type="number" id="n1-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n2-input" id="n2-label">n2:</label>'
'<input type="number" id="n2-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n3-input" id="n3-label">n3:</label>'
'<input type="number" id="n3-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n4-input" id="n4-label">n4:</label>'
'<input type="number" id="n4-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n5-input" id="n5-label">n5:</label>'
'<input type="number" id="n5-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n6-input" id="n6-label">n6:</label>'
'<input type="number" id="n6-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n7-input" id="n7-label">n7:</label>'
'<input type="number" id="n7-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'<div class="n-param-group">'
'<label for="n8-input" id="n8-label">n8:</label>'
'<input type="number" id="n8-input" min="0" max="255" value="10" class="n-input">'
'</div>'
'</div>'
'</div>'
'</div>'
'</div>'
'<div class="right-panel">'
'<div class="presets-section">'
'<h3>Presets</h3>'
'<div id="presets-list-tab" class="presets-list">'
'<!-- Presets will be loaded here -->'
'</div>'
'</div>'
'</div>'
)
return html, 200, {'Content-Type': 'text/html'}
@controller.get('/<id>')
async def get_tab(request, id):
"""Get a specific tab by ID."""
@@ -18,21 +256,6 @@ async def get_tab(request, id):
return json.dumps(tab), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Tab not found"}), 404
@controller.post('')
async def create_tab(request):
"""Create a new tab."""
try:
data = request.json or {}
name = data.get("name", "")
names = data.get("names", None)
preset_ids = data.get("presets", None)
tab_id = tabs.create(name, names, preset_ids)
if data:
tabs.update(tab_id, data)
return json.dumps(tabs.read(tab_id)), 201, {'Content-Type': 'application/json'}
except Exception as e:
return json.dumps({"error": str(e)}), 400
@controller.put('/<id>')
async def update_tab(request, id):
"""Update an existing tab."""
@@ -45,8 +268,147 @@ async def update_tab(request, id):
return json.dumps({"error": str(e)}), 400
@controller.delete('/<id>')
async def delete_tab(request, id):
@with_session
async def delete_tab(request, id, session):
"""Delete a tab."""
# Check if this is an htmx request (wants HTML fragment)
accept_header = request.headers.get('Accept', '')
wants_html = 'text/html' in accept_header
# Handle 'current' tab ID
if id == 'current':
current_tab_id = get_current_tab_id(request, session)
if current_tab_id:
id = current_tab_id
else:
if wants_html:
return '<div class="error">No current tab to delete</div>', 404, {'Content-Type': 'text/html'}
return json.dumps({"error": "No current tab to delete"}), 404
if tabs.delete(id):
return json.dumps({"message": "Tab deleted successfully"}), 200
# Remove from profile's tabs
profile_id = get_current_profile_id(session)
if profile_id:
profile = profiles.read(profile_id)
if profile:
# Support both "tabs" (new) and "tab_order" (old) format
tabs_list = profile.get('tabs', profile.get('tab_order', []))
if id in tabs_list:
tabs_list.remove(id)
profile['tabs'] = tabs_list
# Remove old tab_order if it exists
if 'tab_order' in profile:
del profile['tab_order']
profiles.update(profile_id, profile)
# Clear session if the deleted tab was the current tab
current_tab_id = get_current_tab_id(request, session)
if current_tab_id == id:
if 'current_tab' in session:
session.pop('current_tab', None)
session.save()
if wants_html:
return await tabs_list_fragment.__wrapped__(request, session)
else:
return json.dumps({"message": "Tab deleted successfully"}), 200, {'Content-Type': 'application/json'}
if wants_html:
return '<div class="error">Tab not found</div>', 404, {'Content-Type': 'text/html'}
return json.dumps({"error": "Tab not found"}), 404
@controller.post('')
@with_session
async def create_tab(request, session):
"""Create a new tab."""
# Check if this is an htmx request (wants HTML fragment)
accept_header = request.headers.get('Accept', '')
wants_html = 'text/html' in accept_header
# #region agent log
try:
os.makedirs('/home/pi/led-controller/.cursor', exist_ok=True)
with open('/home/pi/led-controller/.cursor/debug.log', 'a') as _log:
_log.write(json.dumps({
"sessionId": "debug-session",
"runId": "tabs-pre-fix",
"hypothesisId": "H3",
"location": "src/controllers/tab.py:create_tab_htmx",
"message": "create tab with session",
"data": {
"wants_html": wants_html,
"has_form": bool(request.form),
"accept": accept_header
},
"timestamp": int(time.time() * 1000)
}) + "\n")
except Exception:
pass
# #endregion
try:
# Handle form data (htmx) or JSON
if request.form:
name = request.form.get('name', '').strip()
ids_str = request.form.get('ids', '1').strip()
names = [id.strip() for id in ids_str.split(',') if id.strip()]
preset_ids = None
else:
data = request.json or {}
name = data.get("name", "")
names = data.get("names", None)
preset_ids = data.get("presets", None)
if not name:
if wants_html:
return '<div class="error">Tab name cannot be empty</div>', 400, {'Content-Type': 'text/html'}
return json.dumps({"error": "Tab name cannot be empty"}), 400
tab_id = tabs.create(name, names, preset_ids)
# Add to current profile's tabs
profile_id = get_current_profile_id(session)
if profile_id:
profile = profiles.read(profile_id)
if profile:
# Support both "tabs" (new) and "tab_order" (old) format
tabs_list = profile.get('tabs', profile.get('tab_order', []))
if tab_id not in tabs_list:
tabs_list.append(tab_id)
profile['tabs'] = tabs_list
# Remove old tab_order if it exists
if 'tab_order' in profile:
del profile['tab_order']
profiles.update(profile_id, profile)
# #region agent log
try:
os.makedirs('/home/pi/led-controller/.cursor', exist_ok=True)
with open('/home/pi/led-controller/.cursor/debug.log', 'a') as _log:
_log.write(json.dumps({
"sessionId": "debug-session",
"runId": "tabs-pre-fix",
"hypothesisId": "H4",
"location": "src/controllers/tab.py:create_tab_htmx",
"message": "tab created and profile updated",
"data": {
"tab_id": tab_id,
"profile_id": profile_id,
"profile_tabs": tabs_list if profile_id and profile else None
},
"timestamp": int(time.time() * 1000)
}) + "\n")
except Exception:
pass
# #endregion
if wants_html:
# Return HTML fragment for tabs list
return await tabs_list_fragment.__wrapped__(request, session)
else:
# Return JSON response
return json.dumps(tabs.read(tab_id)), 201, {'Content-Type': 'application/json'}
except Exception as e:
import sys
sys.print_exception(e)
if wants_html:
return f'<div class="error">Error: {str(e)}</div>', 400, {'Content-Type': 'text/html'}
return json.dumps({"error": str(e)}), 400

View File

@@ -4,19 +4,21 @@ import gc
import machine
from microdot import Microdot, send_file
from microdot.websocket import with_websocket
from microdot.session import Session
import aioespnow
import network
from controllers.preset import preset
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
async def main():
async def main(port=80):
settings = Settings()
print("Starting")
@@ -29,13 +31,37 @@ async def main():
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('/presets', preset.controller)
app.mount('/profiles', profile.controller)
app.mount('/groups', group.controller)
app.mount('/sequences', sequence.controller)
app.mount('/tabs', tab.controller)
app.mount('/palettes', palette.controller)
# Verify controllers are Microdot instances before mounting
controllers_to_mount = [
('/presets', preset, 'preset'),
('/profiles', profile, 'profile'),
('/groups', group, 'group'),
('/sequences', sequence, 'sequence'),
('/tabs', tab, 'tab'),
('/palettes', palette, 'palette'),
('/scenes', scene, 'scene'),
]
# 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')
# Serve index.html at root
@app.route('/')
def index(request):
"""Serve the main web UI."""
return send_file('templates/index.html')
# Static file route
@app.route("/static/<path:path>")
@@ -59,7 +85,7 @@ async def main():
server = asyncio.create_task(app.start_server(host="0.0.0.0", port=80))
server = asyncio.create_task(app.start_server(host="0.0.0.0", port=port))
wdt = machine.WDT(timeout=10000)
wdt.feed()
@@ -71,4 +97,5 @@ async def main():
await asyncio.sleep_ms(500)
# cleanup before ending the application
asyncio.run(main())
if __name__ == "__main__":
asyncio.run(main())

1
src/models/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Models package

View File

@@ -1,18 +1,32 @@
import json
import wifi
import ubinascii
import machine
import os
class Model(dict):
def __new__(cls, *args, **kwargs):
# Singleton pattern: return existing instance if it exists
if not hasattr(cls, '_instance'):
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
self.file = self.__class__.__name__ + ".json"
# Only initialize once (check if already initialized)
if hasattr(self, '_initialized'):
return
# Create /db directory if it doesn't exist (MicroPython compatible)
try:
os.mkdir("/db")
except OSError:
pass # Directory already exists, which is fine
self.class_name = self.__class__.__name__
self.file = f"/db/{self.class_name.lower()}.json"
super().__init__()
self.load() # Load settings from file during initialization
self._initialized = True
def set_defaults(self):
self = {}
self.clear()
def get_next_id(self):
"""Get the next available ID for creating a new record."""
@@ -23,20 +37,27 @@ class Model(dict):
def save(self):
try:
# Ensure directory exists
try:
os.mkdir("/db")
except OSError:
pass # Directory already exists
j = json.dumps(self)
with open(self.file, 'w') as file:
file.write(j)
print("Settings saved successfully.")
print(f"{self.class_name} saved successfully to {self.file}")
except Exception as e:
print(f"Error saving settings: {e}")
print(f"Error saving {self.class_name} to {self.file}: {e}")
import sys
sys.print_exception(e)
def load(self):
try:
with open(self.file, 'r') as file:
loaded_settings = json.load(file)
self.update(loaded_settings)
print("Settings loaded successfully.")
print(f"{self.class_name} loaded successfully.")
except Exception as e:
print(f"Error loading settings")
print(f"Error loading {self.class_name}")
self.set_defaults()
self.save()

38
src/models/pattern.py Normal file
View File

@@ -0,0 +1,38 @@
from models.model import Model
class Pattern(Model):
def __init__(self):
super().__init__()
def create(self, name="", data=None):
pattern_name = str(name).strip()
if not pattern_name:
pattern_name = self.get_next_id()
self[pattern_name] = data if isinstance(data, dict) else {}
self.save()
return pattern_name
def read(self, id):
id_str = str(id)
return self.get(id_str, None)
def update(self, id, data):
id_str = str(id)
if id_str not in self:
return False
if isinstance(data, dict):
self[id_str].update(data)
self.save()
return True
def delete(self, id):
id_str = str(id)
if id_str not in self:
return False
self.pop(id_str)
self.save()
return True
def list(self):
return list(self.keys())

View File

@@ -18,6 +18,8 @@ class Preset(Model):
"n4": 0,
"n5": 0,
"n6": 0,
"n7": 0,
"n8": 0,
}
self.save()
return next_id

View File

@@ -4,13 +4,18 @@ class Profile(Model):
def __init__(self):
super().__init__()
def create(self, name=""):
def create(self, name="", profile_type="tabs"):
"""
Create a new profile.
profile_type: "tabs" or "scenes" (ignoring scenes for now)
"""
next_id = self.get_next_id()
self[next_id] = {
"name": name,
"tabs": {},
"palette": [],
"tab_order": []
"type": profile_type, # "tabs" or "scenes"
"tabs": [], # Array of tab IDs
"scenes": [], # Array of scene IDs (for future use)
"palette": []
}
self.save()
return next_id

38
src/models/scene.py Normal file
View File

@@ -0,0 +1,38 @@
from models.model import Model
class Scene(Model):
def __init__(self):
super().__init__()
def create(self, name="", sequences=None, groups=None):
next_id = self.get_next_id()
self[next_id] = {
"name": name,
"sequences": sequences if sequences else [],
"groups": groups if groups else []
}
self.save()
return next_id
def read(self, id):
id_str = str(id)
return self.get(id_str, None)
def update(self, id, data):
id_str = str(id)
if id_str not in self:
return False
self[id_str].update(data)
self.save()
return True
def delete(self, id):
id_str = str(id)
if id_str not in self:
return False
self.pop(id_str)
self.save()
return True
def list(self):
return list(self.keys())

View File

@@ -1,7 +1,6 @@
import json
import wifi
import ubinascii
import machine
import os
import binascii
class Settings(dict):
SETTINGS_FILE = "/settings.json"
@@ -10,8 +9,30 @@ class Settings(dict):
super().__init__()
self.load() # Load settings from file during initialization
def generate_secret_key(self):
"""Generate a random secret key for session signing."""
try:
# Try to use os.urandom for secure random bytes
random_bytes = os.urandom(32)
return binascii.hexlify(random_bytes).decode('utf-8')
except (AttributeError, NotImplementedError):
# Fallback for MicroPython or systems without os.urandom
try:
import secrets
return secrets.token_hex(32)
except ImportError:
# Last resort: use a combination of time and random
import time
import random
random.seed(time.time())
return binascii.hexlify(bytes([random.randint(0, 255) for _ in range(32)])).decode('utf-8')
def set_defaults(self):
self = {}
"""Set default settings if they don't exist."""
if 'session_secret_key' not in self:
self['session_secret_key'] = self.generate_secret_key()
# Save immediately when generating a new key
self.save()
def save(self):
try:
@@ -23,12 +44,19 @@ class Settings(dict):
print(f"Error saving settings: {e}")
def load(self):
loaded_from_file = False
try:
with open(self.SETTINGS_FILE, 'r') as file:
loaded_settings = json.load(file)
self.update(loaded_settings)
loaded_from_file = True
print("Settings loaded successfully.")
except Exception as e:
print(f"Error loading settings")
self.clear()
finally:
# Ensure defaults are set even if file exists but is missing keys
self.set_defaults()
self.save()
# Only save if file didn't exist or was invalid
if not loaded_from_file:
self.save()