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

Scope presets to active profiles and support cloning.

Browse Source 
This keeps data isolated per profile while letting users duplicate setups quickly.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jimmy
2026-02-08 13:51:02 +13:00
parent 00514f0525
commit 6c6ed22dbe
8 changed files with 406 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/controllers/preset.py
View File

@@ -1,31 +1,67 @@
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 util.espnow_message import build_message, build_preset_dict
import json
controller = Microdot()
presets = Preset()
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
@controller.get('')
async def list_presets(request):
"""List all presets."""
return json.dumps(presets), 200, {'Content-Type': 'application/json'}
@with_session
async def list_presets(request, session):
"""List presets for the current profile."""
current_profile_id = get_current_profile_id(session)
if not current_profile_id:
return json.dumps({}), 200, {'Content-Type': 'application/json'}
scoped = {
pid: pdata for pid, pdata in presets.items()
if isinstance(pdata, dict) and str(pdata.get("profile_id")) == str(current_profile_id)
}
return json.dumps(scoped), 200, {'Content-Type': 'application/json'}
@controller.get('/<id>')
async def get_preset(request, id):
"""Get a specific preset by ID."""
@with_session
async def get_preset(request, id, session):
"""Get a specific preset by ID (current profile only)."""
preset = presets.read(id)
if preset:
current_profile_id = get_current_profile_id(session)
if preset and str(preset.get("profile_id")) == str(current_profile_id):
return json.dumps(preset), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Preset not found"}), 404
@controller.post('')
async def create_preset(request):
"""Create a new preset."""
@with_session
async def create_preset(request, session):
"""Create a new preset for the current profile."""
try:
data = request.json
preset_id = presets.create()
try:
data = request.json or {}
except Exception:
return json.dumps({"error": "Invalid JSON"}), 400, {'Content-Type': 'application/json'}
current_profile_id = get_current_profile_id(session)
if not current_profile_id:
return json.dumps({"error": "No profile available"}), 404
preset_id = presets.create(current_profile_id)
if not isinstance(data, dict):
data = {}
data = dict(data)
data["profile_id"] = str(current_profile_id)
if presets.update(preset_id, data):
preset_data = presets.read(preset_id)
return json.dumps({preset_id: preset_data}), 201, {'Content-Type': 'application/json'}
@@ -34,10 +70,22 @@ async def create_preset(request):
return json.dumps({"error": str(e)}), 400
@controller.put('/<id>')
async def update_preset(request, id):
"""Update an existing preset."""
@with_session
async def update_preset(request, id, session):
"""Update an existing preset (current profile only)."""
try:
data = request.json
preset = presets.read(id)
current_profile_id = get_current_profile_id(session)
if not preset or str(preset.get("profile_id")) != str(current_profile_id):
return json.dumps({"error": "Preset not found"}), 404
try:
data = request.json or {}
except Exception:
return json.dumps({"error": "Invalid JSON"}), 400, {'Content-Type': 'application/json'}
if not isinstance(data, dict):
data = {}
data = dict(data)
data["profile_id"] = str(current_profile_id)
if presets.update(id, data):
return json.dumps(presets.read(id)), 200, {'Content-Type': 'application/json'}
return json.dumps({"error": "Preset not found"}), 404
@@ -45,15 +93,21 @@ async def update_preset(request, id):
return json.dumps({"error": str(e)}), 400
@controller.delete('/<id>')
async def delete_preset(request, id):
"""Delete a preset."""
@with_session
async def delete_preset(request, id, session):
"""Delete a preset (current profile only)."""
preset = presets.read(id)
current_profile_id = get_current_profile_id(session)
if not preset or str(preset.get("profile_id")) != str(current_profile_id):
return json.dumps({"error": "Preset not found"}), 404
if presets.delete(id):
return json.dumps({"message": "Preset deleted successfully"}), 200
return json.dumps({"error": "Preset not found"}), 404
@controller.post('/send')
async def send_presets(request):
@with_session
async def send_presets(request, session):
"""
Send one or more presets over ESPNow.
@@ -74,13 +128,18 @@ async def send_presets(request):
preset_ids = data.get('preset_ids') or data.get('ids')
if not isinstance(preset_ids, list) or not preset_ids:
return json.dumps({"error": "preset_ids must be a non-empty list"}), 400, {'Content-Type': 'application/json'}
save_flag = data.get('save', True)
save_flag = bool(save_flag)
# Build API-compliant preset map keyed by preset ID (not name)
current_profile_id = get_current_profile_id(session)
presets_by_name = {}
for pid in preset_ids:
preset_data = presets.read(str(pid))
if not preset_data:
continue
if str(preset_data.get("profile_id")) != str(current_profile_id):
continue
preset_id_key = str(pid)
presets_by_name[preset_id_key] = build_preset_dict(preset_data)
@@ -91,7 +150,8 @@ async def send_presets(request):
esp = ESPNow()
async def send_chunk(chunk_presets):
msg = build_message(presets=chunk_presets)
# Include save flag so the led-driver can persist when desired.
msg = build_message(presets=chunk_presets, save=save_flag)
await esp.send(msg)
MAX_BYTES = 240
@@ -104,7 +164,7 @@ async def send_presets(request):
for name, preset_obj in entries:
test_batch = dict(batch)
test_batch[name] = preset_obj
test_msg = build_message(presets=test_batch)
test_msg = build_message(presets=test_batch, save=save_flag)
size = len(test_msg)
if size <= MAX_BYTES or not batch:
@@ -114,7 +174,7 @@ async def send_presets(request):
await send_chunk(batch)
messages_sent += 1
batch = {name: preset_obj}
last_msg = build_message(presets=batch)
last_msg = build_message(presets=batch, save=save_flag)
if batch:
await send_chunk(batch)