Convert app to Flask web application with color pickers

- Created Flask backend with REST API endpoints
- Built HTML/CSS/JavaScript frontend
- Replaced RGB sliders with color pickers for each palette color
- Reorganized layout: color palette on left, patterns on right
- Added persistence for color changes
- Integrated WebSocket client for lighting controller communication
- Added tab management, profile support, and pattern selection
This commit is contained in:
2026-01-04 15:59:19 +13:00
parent c8ae113355
commit 5aa500a7fb
8 changed files with 4853 additions and 0 deletions

448
src/flask_app.py Normal file
View File

@@ -0,0 +1,448 @@
"""
Flask web application for the lighting controller.
Provides REST API and serves the web UI.
"""
import asyncio
import json
import os
import math
from flask import Flask, render_template, request, jsonify
from flask_cors import CORS
from networking import WebSocketClient
import color_utils
from settings import Settings
app = Flask(__name__,
template_folder='../templates',
static_folder='../static')
CORS(app)
# Global settings and WebSocket client
settings = Settings()
websocket_client = None
websocket_uri = "ws://192.168.4.1:80/ws"
def delay_to_slider(delay_ms, min_delay=10, max_delay=10000):
"""Convert delay in ms to slider position (0-1000) using logarithmic scale."""
if delay_ms <= min_delay:
return 0
if delay_ms >= max_delay:
return 1000
if min_delay == max_delay:
return 0
return 1000 * math.log(delay_ms / min_delay) / math.log(max_delay / min_delay)
def slider_to_delay(slider_value, min_delay=10, max_delay=10000):
"""Convert slider position (0-1000) to delay in ms using logarithmic scale."""
if slider_value <= 0:
return min_delay
if slider_value >= 1000:
return max_delay
if min_delay == max_delay:
return min_delay
return int(min_delay * ((max_delay / min_delay) ** (slider_value / 1000)))
def get_pattern_settings(tab_name, pattern_name):
"""Get pattern-specific settings."""
light_settings = settings["lights"][tab_name]["settings"]
if "patterns" not in light_settings:
light_settings["patterns"] = {}
if pattern_name not in light_settings["patterns"]:
light_settings["patterns"][pattern_name] = {}
pattern_settings = light_settings["patterns"][pattern_name]
return {
"colors": pattern_settings.get("colors", ["#000000"]),
"delay": pattern_settings.get("delay", 100),
"n1": pattern_settings.get("n1", 10),
"n2": pattern_settings.get("n2", 10),
"n3": pattern_settings.get("n3", 10),
"n4": pattern_settings.get("n4", 10),
}
def save_pattern_settings(tab_name, pattern_name, colors=None, delay=None, n_params=None):
"""Save pattern-specific settings."""
light_settings = settings["lights"][tab_name]["settings"]
if "patterns" not in light_settings:
light_settings["patterns"] = {}
if pattern_name not in light_settings["patterns"]:
light_settings["patterns"][pattern_name] = {}
pattern_settings = light_settings["patterns"][pattern_name]
if colors is not None:
pattern_settings["colors"] = colors
if delay is not None:
pattern_settings["delay"] = delay
if n_params is not None:
for i in range(1, 5):
if f"n{i}" in n_params:
pattern_settings[f"n{i}"] = n_params[f"n{i}"]
async def send_to_lighting_controller(payload):
"""Send data to the lighting controller via WebSocket."""
global websocket_client
if websocket_client is None:
websocket_client = WebSocketClient(websocket_uri)
await websocket_client.connect()
if not websocket_client.is_connected:
await websocket_client.connect()
if websocket_client.is_connected:
await websocket_client.send_data(payload)
def run_async(coro):
"""Run async function in sync context."""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
finally:
loop.close()
@app.route('/')
def index():
"""Serve the main web UI."""
return render_template('index.html')
@app.route('/api/state', methods=['GET'])
def get_state():
"""Get the current state of all lights."""
return jsonify({
"lights": settings.get("lights", {}),
"patterns": settings.get("patterns", {}),
"tab_order": settings.get("tab_order", [])
})
@app.route('/api/pattern', methods=['POST'])
def set_pattern():
"""Set the pattern for a light group."""
data = request.json
tab_name = data.get("tab_name")
pattern_name = data.get("pattern")
if not tab_name or not pattern_name:
return jsonify({"error": "Missing tab_name or pattern"}), 400
if tab_name not in settings.get("lights", {}):
return jsonify({"error": f"Tab '{tab_name}' not found"}), 404
# Save current pattern's settings before switching
old_pattern = settings["lights"][tab_name]["settings"].get("pattern", "on")
# Get current delay (would need to be passed from frontend)
current_delay = data.get("delay", 100)
current_n_params = {
f"n{i}": data.get(f"n{i}", 10) for i in range(1, 5)
}
save_pattern_settings(
tab_name,
old_pattern,
colors=data.get("colors", ["#000000"]),
delay=current_delay,
n_params=current_n_params
)
# Load new pattern's settings
new_pattern_settings = get_pattern_settings(tab_name, pattern_name)
# Update settings
settings["lights"][tab_name]["settings"]["pattern"] = pattern_name
# Prepare payload for lighting controller
names = settings["lights"][tab_name]["names"]
payload = {
"save": True,
"names": names,
"settings": {
"pattern": pattern_name,
"brightness": settings["lights"][tab_name]["settings"].get("brightness", 127),
"delay": new_pattern_settings["delay"],
"colors": new_pattern_settings["colors"],
**{f"n{i}": new_pattern_settings[f"n{i}"] for i in range(1, 5)}
}
}
# Send to lighting controller
run_async(send_to_lighting_controller(payload))
settings.save()
return jsonify({
"success": True,
"pattern": pattern_name,
"settings": new_pattern_settings
})
@app.route('/api/parameters', methods=['POST'])
def set_parameters():
"""Update parameters (RGB, brightness, delay, n params) for a light group."""
data = request.json
tab_name = data.get("tab_name")
if not tab_name:
return jsonify({"error": "Missing tab_name"}), 400
if tab_name not in settings.get("lights", {}):
return jsonify({"error": f"Tab '{tab_name}' not found"}), 404
current_pattern = settings["lights"][tab_name]["settings"].get("pattern", "on")
pattern_config = settings.get("patterns", {}).get(current_pattern, {})
min_delay = pattern_config.get("min_delay", 10)
max_delay = pattern_config.get("max_delay", 10000)
# Build settings payload
payload_settings = {}
# Handle RGB colors
if "red" in data or "green" in data or "blue" in data:
r = data.get("red", 0)
g = data.get("green", 0)
b = data.get("blue", 0)
hex_color = f"#{r:02x}{g:02x}{b:02x}"
# Update color in palette
pattern_settings = get_pattern_settings(tab_name, current_pattern)
colors = pattern_settings["colors"].copy()
selected_index = data.get("color_index", 0)
if 0 <= selected_index < len(colors):
colors[selected_index] = hex_color
else:
# If index is out of range, append the color
colors.append(hex_color)
# Save pattern settings to persist the color change
save_pattern_settings(tab_name, current_pattern, colors=colors)
payload_settings["colors"] = colors
# Handle brightness
if "brightness" in data:
brightness = int(data["brightness"])
settings["lights"][tab_name]["settings"]["brightness"] = brightness
payload_settings["brightness"] = brightness
# Handle delay
if "delay_slider" in data:
slider_value = int(data["delay_slider"])
delay = slider_to_delay(slider_value, min_delay, max_delay)
save_pattern_settings(tab_name, current_pattern, delay=delay)
payload_settings["delay"] = delay
# Handle n parameters
n_params = {}
for i in range(1, 5):
if f"n{i}" in data:
n_params[f"n{i}"] = int(data[f"n{i}"])
if n_params:
save_pattern_settings(tab_name, current_pattern, n_params=n_params)
payload_settings.update(n_params)
# Send to lighting controller
if payload_settings:
names = settings["lights"][tab_name]["names"]
payload = {
"save": True,
"names": names,
"settings": payload_settings
}
run_async(send_to_lighting_controller(payload))
settings.save()
return jsonify({"success": True})
@app.route('/api/tabs', methods=['GET'])
def get_tabs():
"""Get list of tabs."""
return jsonify({
"tabs": settings.get("tab_order", []),
"lights": settings.get("lights", {})
})
@app.route('/api/tabs', methods=['POST'])
def create_tab():
"""Create a new tab."""
data = request.json
tab_name = data.get("name")
ids = data.get("ids", ["1"])
if not tab_name:
return jsonify({"error": "Missing name"}), 400
if tab_name in settings.get("lights", {}):
return jsonify({"error": f"Tab '{tab_name}' already exists"}), 400
settings.setdefault("lights", {})[tab_name] = {
"names": ids if isinstance(ids, list) else [ids],
"settings": {
"pattern": "on",
"brightness": 127,
"colors": ["#000000"],
"delay": 100,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"patterns": {}
}
}
if "tab_order" not in settings:
settings["tab_order"] = []
settings["tab_order"].append(tab_name)
settings.save()
return jsonify({"success": True, "tab_name": tab_name})
@app.route('/api/tabs/<tab_name>', methods=['PUT'])
def update_tab(tab_name):
"""Update a tab."""
data = request.json
new_name = data.get("name", tab_name)
ids = data.get("ids")
if tab_name not in settings.get("lights", {}):
return jsonify({"error": f"Tab '{tab_name}' not found"}), 404
if new_name != tab_name:
if new_name in settings.get("lights", {}):
return jsonify({"error": f"Tab '{new_name}' already exists"}), 400
# Rename tab
settings["lights"][new_name] = settings["lights"][tab_name]
del settings["lights"][tab_name]
# Update tab order
if "tab_order" in settings and tab_name in settings["tab_order"]:
index = settings["tab_order"].index(tab_name)
settings["tab_order"][index] = new_name
tab_name = new_name
if ids is not None:
settings["lights"][tab_name]["names"] = ids if isinstance(ids, list) else [ids]
settings.save()
return jsonify({"success": True, "tab_name": tab_name})
@app.route('/api/tabs/<tab_name>', methods=['DELETE'])
def delete_tab(tab_name):
"""Delete a tab."""
if tab_name not in settings.get("lights", {}):
return jsonify({"error": f"Tab '{tab_name}' not found"}), 404
del settings["lights"][tab_name]
if "tab_order" in settings and tab_name in settings["tab_order"]:
settings["tab_order"].remove(tab_name)
settings.save()
return jsonify({"success": True})
@app.route('/api/profiles', methods=['GET'])
def get_profiles():
"""Get list of profiles."""
profiles_dir = "profiles"
profiles = []
if os.path.exists(profiles_dir):
for filename in os.listdir(profiles_dir):
if filename.endswith('.json'):
profiles.append(filename[:-5])
profiles.sort()
return jsonify({
"profiles": profiles,
"current_profile": settings.get("current_profile", "")
})
@app.route('/api/profiles', methods=['POST'])
def create_profile():
"""Create a new profile."""
data = request.json
profile_name = data.get("name")
if not profile_name:
return jsonify({"error": "Missing name"}), 400
profiles_dir = "profiles"
os.makedirs(profiles_dir, exist_ok=True)
profile_path = os.path.join(profiles_dir, f"{profile_name}.json")
if os.path.exists(profile_path):
return jsonify({"error": f"Profile '{profile_name}' already exists"}), 400
empty_profile = {
"lights": {},
"tab_password": "",
"tab_order": []
}
with open(profile_path, 'w') as file:
json.dump(empty_profile, file, indent=4)
return jsonify({"success": True, "profile_name": profile_name})
@app.route('/api/profiles/<profile_name>', methods=['POST'])
def load_profile(profile_name):
"""Load a profile."""
profile_path = os.path.join("profiles", f"{profile_name}.json")
if not os.path.exists(profile_path):
return jsonify({"error": f"Profile '{profile_name}' not found"}), 404
with open(profile_path, 'r') as file:
profile_data = json.load(file)
# Update settings with profile data
profile_data.pop("current_profile", None)
patterns_backup = settings.get("patterns", {})
tab_password_backup = settings.get("tab_password", "")
settings.update(profile_data)
settings["patterns"] = patterns_backup
settings["current_profile"] = profile_name
settings_to_save = {
"tab_password": tab_password_backup,
"current_profile": profile_name,
"patterns": patterns_backup
}
with open("settings.json", 'w') as f:
json.dump(settings_to_save, f, indent=4)
return jsonify({"success": True})
def init_websocket():
"""Initialize WebSocket connection in background."""
global websocket_client
if websocket_client is None:
websocket_client = WebSocketClient(websocket_uri)
run_async(websocket_client.connect())
if __name__ == '__main__':
# Initialize WebSocket connection
init_websocket()
app.run(host='0.0.0.0', port=5000, debug=True)

1054
src/main_textual.py Normal file

File diff suppressed because it is too large Load Diff

2135
src/main_tkinter.py.bak Normal file

File diff suppressed because it is too large Load Diff