Move to src. Add midi and sound

This commit is contained in:
2025-09-07 21:15:42 +12:00
parent b77b29415c
commit 9fc58a827b
9 changed files with 568 additions and 7 deletions

68
src/color_utils.py Normal file
View File

@@ -0,0 +1,68 @@
def adjust_brightness(rgb_color, brightness):
r, g, b = rgb_color
# Convert 0-255 brightness to a scale of 0-1
scale_factor = brightness / 255.0
adjusted_r = int(r * scale_factor)
adjusted_g = int(g * scale_factor)
adjusted_b = int(b * scale_factor)
# Ensure values are within 0-255
adjusted_r = max(0, min(255, adjusted_r))
adjusted_g = max(0, min(255, adjusted_g))
adjusted_b = max(0, min(255, adjusted_b))
return (adjusted_r, adjusted_g, adjusted_b)
def hex_to_rgb(hex_color: str) -> tuple[int, int, int]:
"""Converts a hex color string (e.g., "#RRGGBB") to an RGB tuple."""
hex_color = hex_color.lstrip('#')
return int(hex_color[0:2], 16), int(hex_color[2:4], 16), int(hex_color[4:6], 16)
def rgb_to_hex(r: int, g: int, b: int) -> str:
"""Converts an RGB tuple to a hex color string (e.g., "#RRGGBB")."""
return f"#{r:02x}{g:02x}{b:02x}"
def get_contrast_text_color(background_hex_color: str) -> str:
"""
Determines whether black or white text is more readable on a given background color.
Uses the WCAG 2.0 contrast recommendations (Luminosity calculation).
"""
r, g, b = hex_to_rgb(background_hex_color)
# Convert RGB to sRGB (0-1 range)
# The linear RGB values are normalized by dividing by 255
r_linear = r / 255.0
g_linear = g / 255.0
b_linear = b / 255.0
# Apply the sRGB to linear conversion for gamma correction
# This is a simplified approximation for readability, a more accurate one involves if/else for values <= 0.03928
# For a general "light vs dark" determination, this simplified approach is often sufficient.
# The formula used here is often simplified as (R*0.299 + G*0.587 + B*0.114) for quick luminance.
# A more precise relative luminance (L) calculation:
def srgb_to_linear(c):
if c <= 0.03928:
return c / 12.92
else:
return ((c + 0.055) / 1.055) ** 2.4
L = (0.2126 * srgb_to_linear(r_linear) +
0.7152 * srgb_to_linear(g_linear) +
0.0722 * srgb_to_linear(b_linear))
# For general UI elements, a luminance threshold around 0.179 (sqrt(0.032)) is often used
# or simply checking if (R*0.299 + G*0.587 + B*0.114) is > 186 for light background / dark text
# A common rule of thumb for perceived brightness (closer to the one used in many UIs):
# (R*299 + G*587 + B*114) / 1000
# Let's use a simpler luminance check based on the first example's intention:
# If the perceived brightness is above a certain threshold, use black text. Otherwise, use white.
# A simpler luminance check often used for text contrast:
luminance = (0.299 * r + 0.587 * g + 0.114 * b) / 255
if luminance > 0.5: # Adjust this threshold as needed for perceived contrast
return "black"
else:
return "white"

745
src/main.py Normal file
View File

@@ -0,0 +1,745 @@
import asyncio
import tkinter as tk
from tkinter import ttk, messagebox # Import messagebox for confirmations
import json
from async_tkinter_loop import async_handler, async_mainloop
from networking import WebSocketClient
import color_utils
from settings import Settings
import time
# Dark theme colors (unchanged)
bg_color = "#2e2e2e"
fg_color = "white"
trough_color_red = "#4a0000"
trough_color_green = "#004a00"
trough_color_blue = "#00004a"
trough_color_brightness = "#4a4a4a"
trough_color_delay = "#4a4a4a"
active_bg_color = "#4a4a4a"
highlight_pattern_color = "#6a5acd"
# New color for active color in palette
active_palette_color_border = "#FFD700" # Gold color
class App:
def __init__(self) -> None:
self.settings = Settings()
self.root = tk.Tk()
self.root.attributes("-fullscreen", True)
self.root.configure(bg=bg_color)
# Debouncing variables (remain the same)
self.last_rgb_update_time = 0
self.rgb_update_interval_ms = 100
self.pending_rgb_update_id = None
self.last_brightness_update_time = 0
self.brightness_update_interval_ms = 100
self.pending_brightness_update_id = None
self.last_delay_update_time = 0
self.delay_update_interval_ms = 100
self.pending_delay_update_id = None
# --- WebSocketClient ---
self.websocket_client = WebSocketClient("ws://192.168.4.1:80/ws")
self.root.after(100, async_handler(self.websocket_client.connect))
# Configure ttk style (unchanged)
style = ttk.Style()
style.theme_use("alt")
style.configure(".", background=bg_color, foreground=fg_color, font=("Arial", 14))
style.configure("TNotebook", background=bg_color, borderwidth=0)
style.configure(
"TNotebook.Tab", background=bg_color, foreground=fg_color, font=("Arial", 30), padding=[10, 5]
)
style.map("TNotebook.Tab", background=[("selected", active_bg_color)], foreground=[("selected", fg_color)])
style.configure("TFrame", background=bg_color)
# Create Notebook for tabs (unchanged)
self.notebook = ttk.Notebook(self.root)
self.notebook.pack(expand=1, fill="both")
self.tabs = {}
self.create_tabs()
self.notebook.bind("<<NotebookTabChanged>>", self.on_tab_change)
# Add Reload Config Button (unchanged)
reload_button = tk.Button(
self.root,
text="Reload Config",
command=self.reload_config,
bg=active_bg_color,
fg=fg_color,
font=("Arial", 20),
padx=20,
pady=10,
relief=tk.FLAT,
)
reload_button.pack(side=tk.BOTTOM, pady=20)
self.root.protocol("WM_DELETE_WINDOW", self.on_closing)
async_mainloop(self.root)
def on_closing(self):
print("Closing application...")
asyncio.create_task(self.websocket_client.close())
self.root.destroy()
def create_tabs(self):
for tab_name in list(self.tabs.keys()):
self.notebook.forget(self.tabs[tab_name])
del self.tabs[tab_name]
for key, value in self.settings["lights"].items():
tab = ttk.Frame(self.notebook)
self.notebook.add(tab, text=key)
self.create_light_control_widgets(tab, key, value["names"], value["settings"])
self.tabs[key] = tab
def create_light_control_widgets(self, tab, tab_name, ids, initial_settings):
slider_length = 800
slider_width = 50
# Extract initial color, brightness, and delay
initial_colors = initial_settings.get("colors", ["#000000"])
initial_hex_color = initial_colors[0] if initial_colors else "#000000"
initial_brightness = initial_settings.get("brightness", 127)
initial_delay = initial_settings.get("delay", 0)
initial_pattern = initial_settings.get("pattern", "on")
initial_r, initial_g, initial_b = color_utils.hex_to_rgb(initial_hex_color)
# Main frame to hold everything within the tab
main_tab_frame = tk.Frame(tab, bg=bg_color)
main_tab_frame.pack(expand=True, fill="both", padx=10, pady=10)
# Left panel for sliders
slider_panel_frame = tk.Frame(main_tab_frame, bg=bg_color)
slider_panel_frame.pack(side=tk.LEFT, padx=10, pady=10)
# Common slider configuration
slider_config = {
"from_": 255,
"to": 0,
"orient": tk.VERTICAL,
"length": slider_length,
"width": slider_width,
"bg": bg_color,
"fg": fg_color,
"highlightbackground": bg_color,
"activebackground": active_bg_color,
"resolution": 1,
"sliderlength": 70,
}
# Red Slider
red_slider = tk.Scale(slider_panel_frame, label="Red", troughcolor=trough_color_red, **slider_config)
red_slider.set(initial_r)
red_slider.pack(side=tk.LEFT, padx=10)
red_slider.bind("<B1-Motion>", lambda _: self.schedule_update_rgb(tab))
red_slider.bind("<ButtonRelease-1>", lambda _: self.schedule_update_rgb(tab, force_send=True))
# Green Slider
green_slider = tk.Scale(slider_panel_frame, label="Green", troughcolor=trough_color_green, **slider_config)
green_slider.set(initial_g)
green_slider.pack(side=tk.LEFT, padx=10)
green_slider.bind("<B1-Motion>", lambda _: self.schedule_update_rgb(tab))
green_slider.bind("<ButtonRelease-1>", lambda _: self.schedule_update_rgb(tab, force_send=True))
# Blue Slider
blue_slider = tk.Scale(slider_panel_frame, label="Blue", troughcolor=trough_color_blue, **slider_config)
blue_slider.set(initial_b)
blue_slider.pack(side=tk.LEFT, padx=10)
blue_slider.bind("<B1-Motion>", lambda _: self.schedule_update_rgb(tab))
blue_slider.bind("<ButtonRelease-1>", lambda _: self.schedule_update_rgb(tab, force_send=True))
# Brightness Slider
brightness_slider = tk.Scale(
slider_panel_frame, label="Brightness", troughcolor=trough_color_brightness, **slider_config
)
brightness_slider.set(initial_brightness)
brightness_slider.pack(side=tk.LEFT, padx=10)
brightness_slider.bind("<B1-Motion>", lambda _: self.schedule_update_brightness(tab))
brightness_slider.bind("<ButtonRelease-1>", lambda _: self.schedule_update_brightness(tab, force_send=True))
# Delay Slider
delay_slider_config = slider_config.copy()
delay_slider_config.update(
{
"from_": 1000,
"to": 0,
"resolution": 10,
"label": "Delay (ms)",
"troughcolor": trough_color_delay,
}
)
delay_slider = tk.Scale(slider_panel_frame, **delay_slider_config)
delay_slider.set(initial_delay)
delay_slider.pack(side=tk.LEFT, padx=10)
delay_slider.bind("<B1-Motion>", lambda _: self.schedule_update_delay(tab))
delay_slider.bind("<ButtonRelease-1>", lambda _: self.schedule_update_delay(tab, force_send=True))
# Store references to widgets for this tab
tab.widgets = {
"red_slider": red_slider,
"green_slider": green_slider,
"blue_slider": blue_slider,
"brightness_slider": brightness_slider,
"delay_slider": delay_slider,
"selected_color_index": 0, # Default to the first color
}
tab.colors_in_palette = initial_colors.copy() # Store the list of hex colors for this tab
tab.color_swatch_frames = [] # To hold references to the color swatches
# Right panel for IDs, Patterns, and NEW Color Palette
right_panel_frame = tk.Frame(main_tab_frame, bg=bg_color)
right_panel_frame.pack(side=tk.LEFT, padx=20, pady=10, anchor="n", expand=True, fill="both")
# IDs section - MODIFIED TO BE SIDE-BY-SIDE
ids_frame = tk.Frame(right_panel_frame, bg=bg_color)
ids_frame.pack(pady=10, fill=tk.X)
tk.Label(ids_frame, text="Associated Names:", font=("Arial", 20), bg=bg_color, fg=fg_color).pack(pady=10)
# New inner frame for the IDs to be displayed horizontally
ids_inner_frame = tk.Frame(ids_frame, bg=bg_color)
ids_inner_frame.pack(fill=tk.X, expand=True) # Pack this frame to fill available width
for light_id in ids:
tk.Label(ids_inner_frame, text=str(light_id), font=("Arial", 18), bg=bg_color, fg=fg_color).pack(
side=tk.LEFT, padx=5, pady=2
) # Pack labels horizontally
# --- New Frame to hold Patterns and Color Palette side-by-side ---
patterns_and_palette_frame = tk.Frame(right_panel_frame, bg=bg_color)
patterns_and_palette_frame.pack(pady=20, fill=tk.BOTH, expand=True)
# Patterns section
patterns_frame = tk.Frame(patterns_and_palette_frame, bg=bg_color, bd=2, relief=tk.GROOVE)
patterns_frame.pack(side=tk.LEFT, padx=10, pady=5, fill=tk.BOTH, expand=True) # Pack to the left
tk.Label(patterns_frame, text="Patterns:", font=("Arial", 20), bg=bg_color, fg=fg_color).pack(pady=10)
tab.pattern_buttons = {}
patterns = self.settings.get("patterns", [])
for pattern_name in patterns:
button = tk.Button(
patterns_frame,
text=pattern_name,
command=lambda p=pattern_name: self.send_pattern(tab_name, p),
bg=active_bg_color,
fg=fg_color,
font=("Arial", 18),
padx=15,
pady=5,
relief=tk.FLAT,
)
button.pack(pady=5, fill=tk.X)
tab.pattern_buttons[pattern_name] = button
self.highlight_pattern_button(tab, initial_pattern)
# --- Color Palette Editor Section ---
color_palette_editor_frame = tk.Frame(patterns_and_palette_frame, bg=bg_color, bd=2, relief=tk.GROOVE)
color_palette_editor_frame.pack(side=tk.LEFT, padx=10, pady=5, fill=tk.BOTH, expand=True) # Pack to the left
tab.color_palette_editor_frame = color_palette_editor_frame # Store reference for update_ui_for_pattern
tk.Label(color_palette_editor_frame, text="Color Palette:", font=("Arial", 20), bg=bg_color, fg=fg_color).pack(
pady=10
)
# Frame to hold color swatches (will be dynamic)
tab.color_swatches_container = tk.Frame(color_palette_editor_frame, bg=bg_color)
tab.color_swatches_container.pack(pady=5, fill=tk.BOTH, expand=True)
# Buttons for Add/Remove Color
palette_buttons_frame = tk.Frame(color_palette_editor_frame, bg=bg_color)
palette_buttons_frame.pack(pady=10, fill=tk.X)
add_color_button = tk.Button(
palette_buttons_frame,
text="Add Color",
command=lambda t=tab: self.add_color_to_palette(t),
bg=active_bg_color,
fg=fg_color,
font=("Arial", 16),
padx=10,
pady=5,
relief=tk.FLAT,
)
add_color_button.pack(side=tk.LEFT, expand=True, padx=5)
remove_color_button = tk.Button(
palette_buttons_frame,
text="Remove Selected",
command=lambda t=tab: self.remove_selected_color_from_palette(t),
bg=active_bg_color,
fg=fg_color,
font=("Arial", 16),
padx=10,
pady=5,
relief=tk.FLAT,
)
remove_color_button.pack(side=tk.RIGHT, expand=True, padx=5)
# Initial population of the color palette
self.refresh_color_palette_display(tab)
# The initial call to update_ui_for_pattern now only sets slider values and highlights
self.update_ui_for_pattern(tab, initial_pattern)
def refresh_color_palette_display(self, tab):
"""Clears and repopulates the color swatches in the palette display."""
# Clear existing swatches
for frame in tab.color_swatch_frames:
frame.destroy()
tab.color_swatch_frames.clear()
for i, hex_color in enumerate(tab.colors_in_palette):
swatch_frame = tk.Frame(
tab.color_swatches_container, bg=hex_color, width=100, height=50, bd=2, relief=tk.SOLID
)
swatch_frame.pack(pady=3, padx=5, fill=tk.X)
# Bind click to select this color for editing
swatch_frame.bind("<Button-1>", lambda event, idx=i, t=tab: self.select_color_in_palette(t, idx))
# Add a label inside to make it clickable too
swatch_label = tk.Label(
swatch_frame,
text=f"Color {i+1}",
bg=hex_color,
fg=color_utils.get_contrast_text_color(hex_color),
font=("Arial", 14),
width=5,
height=3,
)
swatch_label.pack(expand=True, fill=tk.BOTH)
swatch_label.bind("<Button-1>", lambda event, idx=i, t=tab: self.select_color_in_palette(t, idx))
tab.color_swatch_frames.append(swatch_frame)
# Re-highlight the currently selected color
self._highlight_selected_color_swatch(tab)
def _highlight_selected_color_swatch(self, tab):
"""Applies/removes highlight border to the selected color swatch."""
current_index = tab.widgets["selected_color_index"]
for i, swatch_frame in enumerate(tab.color_swatch_frames):
if i == current_index:
swatch_frame.config(highlightbackground=active_palette_color_border, highlightthickness=3)
else:
swatch_frame.config(highlightbackground=swatch_frame.cget("bg"), highlightthickness=0) # Reset to no highlight
def select_color_in_palette(self, tab, index: int):
"""Selects a color in the palette, updates sliders, and highlights swatch.
This now also triggers an RGB update to the device."""
if not (0 <= index < len(tab.colors_in_palette)):
return
tab.widgets["selected_color_index"] = index
self._highlight_selected_color_swatch(tab)
# Update RGB sliders with the selected color
hex_color = tab.colors_in_palette[index]
r, g, b = color_utils.hex_to_rgb(hex_color)
tab.widgets["red_slider"].set(r)
tab.widgets["green_slider"].set(g)
tab.widgets["blue_slider"].set(b)
print(f"Selected color index {index}: {hex_color}")
# Immediately send the update, as changing the selected color implies
# a desire to change the light's current color, regardless of pattern.
# This will also save the settings.
self.schedule_update_rgb(tab, force_send=True)
def add_color_to_palette(self, tab):
"""Adds a new black color to the palette and selects it, with a limit of 10 colors."""
MAX_COLORS = 8 # Define the maximum number of colors allowed
if len(tab.colors_in_palette) >= MAX_COLORS:
messagebox.showwarning("Color Limit Reached", f"You can add a maximum of {MAX_COLORS} colors to the palette.")
return
# Simplified: just add black. If unique colors were required globally,
# more complex logic would be needed here.
tab.colors_in_palette.append("#000000") # Add black as default
self.refresh_color_palette_display(tab)
# Select the newly added color
self.select_color_in_palette(tab, len(tab.colors_in_palette) - 1)
self.save_current_tab_settings() # Save changes to settings.json
def remove_selected_color_from_palette(self, tab):
"""Removes the currently selected color from the palette."""
current_index = tab.widgets["selected_color_index"]
if len(tab.colors_in_palette) <= 1:
messagebox.showwarning("Cannot Remove", "There must be at least one color in the palette.")
return
if messagebox.askyesno("Confirm Delete", f"Are you sure you want to remove Color {current_index + 1}?"):
del tab.colors_in_palette[current_index]
# Adjust selected index if the removed color was the last one
if current_index >= len(tab.colors_in_palette):
tab.widgets["selected_color_index"] = len(tab.colors_in_palette) - 1
if tab.widgets["selected_color_index"] < 0: # Should not happen with 1-color check
tab.widgets["selected_color_index"] = 0
self.refresh_color_palette_display(tab)
# Update sliders with the new selected color (if any)
if tab.colors_in_palette:
self.select_color_in_palette(tab, tab.widgets["selected_color_index"])
else: # If palette became empty (shouldn't happen with 1-color check)
tab.widgets["red_slider"].set(0)
tab.widgets["green_slider"].set(0)
tab.widgets["blue_slider"].set(0)
self.save_current_tab_settings() # Save changes to settings.json
def update_ui_for_pattern(self, tab, current_pattern: str):
"""
Manages the state of the UI elements based on the selected pattern.
The Color Palette Editor is always visible. RGB sliders update
based on the currently selected color in the palette, or the first
color if the palette is empty or not in transition mode and a new tab/pattern is selected.
"""
# The color_palette_editor_frame is always packed, so no visibility control needed here.
# When the pattern changes, we need to ensure the RGB sliders reflect
# the appropriate color based on the context.
if tab.colors_in_palette:
# If in 'transition' mode, set sliders to the currently selected color in the palette.
if current_pattern == "transition":
self.select_color_in_palette(tab, tab.widgets["selected_color_index"])
else:
# If not in 'transition' mode, but a color is selected, update sliders to that.
# Or, if this is a fresh load/tab change, ensure it's the first color.
# This ensures the sliders consistently show the color that will be sent
# for 'on'/'blink' based on the palette's first entry.
hex_color = tab.colors_in_palette[tab.widgets["selected_color_index"]]
r, g, b = color_utils.hex_to_rgb(hex_color)
tab.widgets["red_slider"].set(r)
tab.widgets["green_slider"].set(g)
tab.widgets["blue_slider"].set(b)
self._highlight_selected_color_swatch(tab) # Re-highlight even if index didn't change
else:
# Handle empty palette scenario (shouldn't happen with default ["#000000"])
tab.widgets["red_slider"].set(0)
tab.widgets["green_slider"].set(0)
tab.widgets["blue_slider"].set(0)
tab.widgets["selected_color_index"] = 0 # Ensure index is valid
self._highlight_selected_color_swatch(tab)
# Brightness and Delay sliders are always visible.
def highlight_pattern_button(self, tab_widget, active_pattern_name):
if hasattr(tab_widget, "pattern_buttons"):
for pattern_name, button in tab_widget.pattern_buttons.items():
if pattern_name == active_pattern_name:
button.config(bg=highlight_pattern_color)
else:
button.config(bg=active_bg_color)
def on_tab_change(self, event):
selected_tab_name = self.notebook.tab(self.notebook.select(), "text")
current_tab_widget = self.notebook.nametowidget(self.notebook.select())
initial_settings = self.settings["lights"][selected_tab_name]["settings"]
# Ensure current_tab_widget has the necessary attributes
if not hasattr(current_tab_widget, "colors_in_palette"):
# This tab might not have been fully initialized yet, or recreated
# In a full reload, create_tabs ensures it is.
return
# Update the local colors_in_palette list for the tab
current_tab_widget.colors_in_palette = initial_settings.get("colors", ["#000000"]).copy()
current_tab_widget.widgets["selected_color_index"] = 0 # Default to first color
# Refresh the color palette display and select the first color
self.refresh_color_palette_display(current_tab_widget)
if current_tab_widget.colors_in_palette:
self.select_color_in_palette(current_tab_widget, 0)
else: # If palette became empty (shouldn't happen with default ["#000000"])
current_tab_widget.widgets["red_slider"].set(0)
current_tab_widget.widgets["green_slider"].set(0)
current_tab_widget.widgets["blue_slider"].set(0)
# Update brightness and delay sliders
current_tab_widget.widgets["brightness_slider"].set(initial_settings.get("brightness", 127))
current_tab_widget.widgets["delay_slider"].set(initial_settings.get("delay", 0))
# Highlight the active pattern button
initial_pattern = initial_settings.get("pattern", "on")
self.highlight_pattern_button(current_tab_widget, initial_pattern)
# Update UI visibility based on the current pattern
self.update_ui_for_pattern(current_tab_widget, initial_pattern)
def reload_config(self):
print("Reloading configuration...")
self.settings = Settings()
self.create_tabs()
# After recreating, ensure the currently selected tab's sliders are updated
# Trigger on_tab_change manually for the currently selected tab
self.on_tab_change(None)
# --- Debouncing functions (no change to core logic, just how they call update_rgb) ---
def schedule_update_rgb(self, tab, force_send=False):
current_time = time.time() * 1000
if force_send:
if self.pending_rgb_update_id:
self.root.after_cancel(self.pending_rgb_update_id)
self.pending_rgb_update_id = None
self.update_rgb(tab)
self.last_rgb_update_time = current_time
elif current_time - self.last_rgb_update_time >= self.rgb_update_interval_ms:
if self.pending_rgb_update_id:
self.root.after_cancel(self.pending_rgb_update_id)
self.pending_rgb_update_id = None
self.update_rgb(tab)
self.last_rgb_update_time = current_time
else:
if self.pending_rgb_update_id:
self.root.after_cancel(self.pending_rgb_update_id)
time_to_wait = int(self.rgb_update_interval_ms - (current_time - self.last_rgb_update_time))
self.pending_rgb_update_id = self.root.after(time_to_wait, lambda: self.update_rgb(tab))
def schedule_update_brightness(self, tab, force_send=False):
current_time = time.time() * 1000
if force_send:
if self.pending_brightness_update_id:
self.root.after_cancel(self.pending_brightness_update_id)
self.pending_brightness_update_id = None
self.update_brightness(tab)
self.last_brightness_update_time = current_time
elif current_time - self.last_brightness_update_time >= self.brightness_update_interval_ms:
if self.pending_brightness_update_id:
self.root.after_cancel(self.pending_brightness_update_id)
self.pending_brightness_update_id = None
self.update_brightness(tab)
self.last_brightness_update_time = current_time
else:
if self.pending_brightness_update_id:
self.root.after_cancel(self.pending_brightness_update_id)
time_to_wait = int(self.brightness_update_interval_ms - (current_time - self.last_brightness_update_time))
self.pending_brightness_update_id = self.root.after(time_to_wait, lambda: self.update_brightness(tab))
def schedule_update_delay(self, tab, force_send=False):
current_time = time.time() * 1000
if force_send:
if self.pending_delay_update_id:
self.root.after_cancel(self.pending_delay_update_id)
self.pending_delay_update_id = None
self.update_delay(tab)
self.last_delay_update_time = current_time
elif current_time - self.last_delay_update_time >= self.delay_update_interval_ms:
if self.pending_delay_update_id:
self.root.after_cancel(self.pending_delay_update_id)
self.pending_delay_update_id = None
self.update_delay(tab)
self.last_delay_update_time = current_time
else:
if self.pending_delay_update_id:
self.root.after_cancel(self.pending_delay_update_id)
time_to_wait = int(self.delay_update_interval_ms - (current_time - self.last_delay_update_time))
self.pending_delay_update_id = self.root.after(time_to_wait, lambda: self.update_delay(tab))
# --- Asynchronous Update Functions ---
@async_handler
async def update_rgb(self, tab):
"""Update the currently selected color in the palette and send to the server."""
try:
red_slider = tab.widgets["red_slider"]
green_slider = tab.widgets["green_slider"]
blue_slider = tab.widgets["blue_slider"]
r = red_slider.get()
g = green_slider.get()
b = blue_slider.get()
hex_color = f"#{r:02x}{g:02x}{b:02x}"
print(f"Updating selected color to: {hex_color}")
selected_color_index = tab.widgets["selected_color_index"]
if 0 <= selected_color_index < len(tab.colors_in_palette):
tab.colors_in_palette[selected_color_index] = hex_color
self.refresh_color_palette_display(tab) # Update swatch immediately
selected_server = self.notebook.tab(self.notebook.select(), "text")
names = self.settings["lights"][selected_server]["names"]
# Determine which colors to send based on the current pattern.
current_pattern = self.settings["lights"][selected_server]["settings"].get("pattern", "on")
colors_to_send = []
if current_pattern == "transition":
colors_to_send = tab.colors_in_palette.copy()
elif current_pattern in ["on", "blink", "theater_chase", "flicker"]: # Add other patterns that use a single color
if tab.colors_in_palette:
# For non-transition patterns, the device typically uses only the first color.
# However, if a user picks a color from the palette, we want THAT color to be the one
# sent and active. So, the selected color from the palette *becomes* the first color
# in the list we send to the device for these modes.
# This ensures the light matches the selected palette color.
colors_to_send = [hex_color] # Send the color currently set by the sliders
else:
colors_to_send = ["#000000"] # Default if palette is empty
else: # For other patterns like "off", "rainbow" where colors might not be primary
# We still want to send the *current* palette state for saving,
# but the device firmware might ignore it for these patterns.
colors_to_send = tab.colors_in_palette.copy()
payload = {
"save": True, # Always save this change to config
"names": names,
"settings": {
"colors": colors_to_send, # This now dynamically changes based on pattern
"brightness": tab.widgets["brightness_slider"].get(),
"delay": tab.widgets["delay_slider"].get(),
"pattern": current_pattern, # Always send the current pattern
},
}
# Update the settings object with the new color list (and potentially other synced values)
self.settings["lights"][selected_server]["settings"]["colors"] = tab.colors_in_palette.copy()
self.settings["lights"][selected_server]["settings"]["brightness"] = tab.widgets["brightness_slider"].get()
self.settings["lights"][selected_server]["settings"]["delay"] = tab.widgets["delay_slider"].get()
self.settings.save()
await self.websocket_client.send_data(payload)
print(f"Sent RGB payload: {payload}")
except Exception as e:
print(f"Error updating RGB: {e}")
@async_handler
async def update_brightness(self, tab):
try:
brightness_slider = tab.widgets["brightness_slider"]
brightness = brightness_slider.get()
print(f"Brightness: {brightness}")
selected_server = self.notebook.tab(self.notebook.select(), "text")
names = self.settings["lights"][selected_server]["names"]
payload = {
"save": True,
"names": names,
"settings": {
"brightness": brightness,
},
}
# Update the settings object with the new brightness
self.settings["lights"][selected_server]["settings"]["brightness"] = brightness
self.settings.save()
await self.websocket_client.send_data(payload)
print(f"Sent brightness payload: {payload}")
except Exception as e:
print(f"Error updating brightness: {e}")
@async_handler
async def update_delay(self, tab):
try:
delay_slider = tab.widgets["delay_slider"]
delay = delay_slider.get()
print(f"Delay: {delay}")
selected_server = self.notebook.tab(self.notebook.select(), "text")
names = self.settings["lights"][selected_server]["names"]
payload = {
"save": True,
"names": names,
"settings": {
"delay": delay,
},
}
# Update the settings object with the new delay
self.settings["lights"][selected_server]["settings"]["delay"] = delay
self.settings.save()
await self.websocket_client.send_data(payload)
print(f"Sent delay payload: {payload}")
except Exception as e:
print(f"Error updating delay: {e}")
@async_handler
async def send_pattern(self, tab_name: str, pattern_name: str):
try:
names = self.settings["lights"][tab_name]["names"]
# Get the actual tab widget to access its `colors_in_palette` and other attributes
current_tab_widget = None
for key, tab_widget in self.tabs.items():
if key == tab_name:
current_tab_widget = tab_widget
break
if not current_tab_widget:
print(f"Error: Could not find tab widget for {tab_name}")
return
current_settings_for_tab = self.settings["lights"][tab_name]["settings"]
payload_settings = {
"pattern": pattern_name,
"brightness": current_settings_for_tab.get("brightness", 127),
"delay": current_settings_for_tab.get("delay", 0),
}
# Determine colors to send based on the *newly selected* pattern
if pattern_name == "transition":
payload_settings["colors"] = current_tab_widget.colors_in_palette.copy()
elif pattern_name in ["on", "blink"]: # Add other patterns that use a single color here
# When switching TO 'on' or 'blink', ensure the color sent is the one
# currently displayed on the sliders (which reflects the selected palette color).
r = current_tab_widget.widgets["red_slider"].get()
g = current_tab_widget.widgets["green_slider"].get()
b = current_tab_widget.widgets["blue_slider"].get()
hex_color_from_sliders = f"#{r:02x}{g:02x}{b:02x}"
payload_settings["colors"] = [hex_color_from_sliders]
else:
# For other patterns, send the full palette, device might ignore or use default
payload_settings["colors"] = current_tab_widget.colors_in_palette.copy()
payload = {
"save": True,
"names": names,
"settings": payload_settings,
}
# Update the settings object with the new pattern and current colors/brightness/delay
self.settings["lights"][tab_name]["settings"]["pattern"] = pattern_name
# Always save the full current palette state in settings.
self.settings["lights"][tab_name]["settings"]["colors"] = current_tab_widget.colors_in_palette.copy()
self.settings.save()
self.highlight_pattern_button(current_tab_widget, pattern_name)
self.update_ui_for_pattern(current_tab_widget, pattern_name) # Update UI based on new pattern
await self.websocket_client.send_data(payload)
print(f"Sent pattern payload: {payload}")
except Exception as e:
print(f"Error sending pattern: {e}")
def save_current_tab_settings(self):
"""Saves the current state of the active tab's settings (colors, brightness, delay, pattern) to config."""
selected_server = self.notebook.tab(self.notebook.select(), "text")
current_tab_widget = self.notebook.nametowidget(self.notebook.select())
if not hasattr(current_tab_widget, "colors_in_palette"):
return # Tab not fully initialized yet
# Update settings for the current tab in the self.settings object
self.settings["lights"][selected_server]["settings"]["colors"] = current_tab_widget.colors_in_palette
self.settings["lights"][selected_server]["settings"]["brightness"] = current_tab_widget.widgets["brightness_slider"].get()
self.settings["lights"][selected_server]["settings"]["delay"] = current_tab_widget.widgets["delay_slider"].get()
# The pattern is updated in send_pattern already, but ensure consistency
# For simplicity, we assume send_pattern is the primary way to change pattern.
self.settings.save()
print(f"Saved settings for {selected_server}")
if __name__ == "__main__":
app = App()

105
src/midi.py Normal file
View File

@@ -0,0 +1,105 @@
import mido
import asyncio
import time
import networking # <--- This will now correctly import your module
async def midi_to_websocket_listener(midi_port_index: int, websocket_uri: str):
"""
Listens to a specific MIDI port and sends data to a WebSocket server
when Note 32 (and 33) is pressed.
"""
delay = 100 # Default delay value
# 1. Get MIDI port name
port_names = mido.get_input_names()
if not port_names:
print("No MIDI input ports found. Please connect your device.")
return
if not (0 <= midi_port_index < len(port_names)):
print(f"Error: MIDI port index {midi_port_index} out of range. Available ports: {port_names}")
print("Available ports:")
for i, name in enumerate(port_names):
print(f" {i}: {name}")
return
midi_port_name = port_names[midi_port_index]
print(f"Selected MIDI input port: {midi_port_name}")
# 2. Initialize WebSocket client (using your actual networking.py)
ws_client = networking.WebSocketClient(websocket_uri)
try:
# 3. Connect WebSocket
await ws_client.connect()
print(f"WebSocket client connected to {ws_client.uri}")
# 4. Open MIDI port and start listening loop
with mido.open_input(midi_port_name) as port:
print(f"MIDI port '{midi_port_name}' opened. Press Ctrl+C to stop.")
while True:
msg = port.receive(block=False) # Non-blocking read
if msg:
match msg.type:
case 'note_on':
print(f" Note ON: Note={msg.note}, Velocity={msg.velocity}, Channel={msg.channel}")
match msg.note:
case 32:
await ws_client.send_data({
"names": ["1"],
"settings": {
"pattern": "pulse",
"delay": delay,
"colors": ["#00ff00"],
"brightness": 100,
"num_leds": 200,
}
})
case 33:
await ws_client.send_data({
"names": ["2"],
"settings": {
"pattern": "chase",
"speed": 10,
"color": "#00FFFF",
}
})
case 'control_change':
match msg.control:
case 36:
delay = msg.value * 4
print(f"Delay set to {delay} ms")
await asyncio.sleep(0.001) # Important: Yield control to asyncio event loop
except mido.PortsError as e:
print(f"Error opening MIDI port '{midi_port_name}': {e}")
except asyncio.CancelledError:
print(f"MIDI listener cancelled.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
finally:
# 5. Disconnect WebSocket and clean up
# This assumes your WebSocketClient has a ._connected attribute or similar way to check state.
# If your client's disconnect method is safe to call even if not connected, you can simplify.
await ws_client.close()
print("MIDI listener stopped and cleaned up.")
async def main():
# --- Configuration ---
MIDI_PORT_INDEX = 1 # <--- IMPORTANT: Change this to the correct index for your device
WEBSOCKET_SERVER_URI = "ws://192.168.4.1:80/ws"
# --- End Configuration ---
try:
await midi_to_websocket_listener(MIDI_PORT_INDEX, WEBSOCKET_SERVER_URI)
except KeyboardInterrupt:
print("\nProgram interrupted by user.")
finally:
print("Main program finished.")
if __name__ == "__main__":
asyncio.run(main())

53
src/networking.py Normal file
View File

@@ -0,0 +1,53 @@
import asyncio
import websockets
import json
class WebSocketClient:
def __init__(self, uri):
self.uri = uri
self.websocket = None
self.is_connected = False
async def connect(self):
"""Establishes the WebSocket connection."""
if self.is_connected and self.websocket:
print("Already connected.")
return
try:
print(f"Connecting to {self.uri}...")
self.websocket = await websockets.connect(self.uri)
self.is_connected = True
print("WebSocket connected.")
except (ConnectionError, websockets.exceptions.ConnectionClosedOK) as e:
print(f"Error connecting: {e}")
self.is_connected = False
self.websocket = None
async def send_data(self, data):
print(data)
"""Sends data over the open WebSocket connection."""
if not self.is_connected or not self.websocket:
print("WebSocket not connected. Attempting to reconnect...")
await self.connect()
if not self.is_connected:
print("Failed to reconnect. Cannot send data.")
return
try:
await self.websocket.send(json.dumps(data))
print(f"Sent: {data}")
except (ConnectionError, websockets.exceptions.ConnectionClosed) as e:
print(f"Error sending data: {e}")
self.is_connected = False
self.websocket = None # Reset connection on error
await self.connect() # Attempt to reconnect
async def close(self):
"""Closes the WebSocket connection."""
if self.websocket and self.is_connected:
await self.websocket.close()
self.is_connected = False
self.websocket = None
print("WebSocket closed.")

27
src/settings.py Normal file
View File

@@ -0,0 +1,27 @@
import json
class Settings(dict):
SETTINGS_FILE = "settings.json"
def __init__(self):
super().__init__()
self.load() # Load settings from file during initialization
def save(self):
try:
j = json.dumps(self, indent=4)
with open(self.SETTINGS_FILE, 'w') as file:
file.write(j)
print("Settings saved successfully.")
except Exception as e:
print(f"Error saving settings: {e}")
def load(self):
try:
with open(self.SETTINGS_FILE, 'r') as file:
loaded_settings = json.load(file)
self.update(loaded_settings)
print("Settings loaded successfully.")
except Exception as e:
print(f"Error loading settings {e}")
self.save()

124
src/sound.py Normal file
View File

@@ -0,0 +1,124 @@
#!/usr/bin/python
import pyaudio
import aubio
import numpy as np
from time import sleep
import websocket # pip install websocket-client
import json
seconds = 10 # how long this script should run (if not using infinite loop)
bufferSize = 512
windowSizeMultiple = 2 # or 4 for higher accuracy, but more computational cost
audioInputDeviceIndex = 7 # use 'arecord -l' to check available audio devices
audioInputChannels = 1
pa = pyaudio.PyAudio()
print("Available audio input devices:")
info = pa.get_host_api_info_by_index(0)
num_devices = info.get('deviceCount')
found_device = False
for i in range(0, num_devices):
device_info = pa.get_device_info_by_host_api_device_index(0, i)
if (device_info.get('maxInputChannels')) > 0:
print(f" Input Device id {i} - {device_info.get('name')}")
if i == audioInputDeviceIndex:
found_device = True
if not found_device:
print(f"Warning: Audio input device index {audioInputDeviceIndex} not found or has no input channels.")
# Consider exiting or picking a default if necessary
try:
audioInputDevice = pa.get_device_info_by_index(audioInputDeviceIndex)
audioInputSampleRate = int(audioInputDevice['defaultSampleRate'])
except Exception as e:
print(f"Error getting audio device info for index {audioInputDeviceIndex}: {e}")
pa.terminate()
exit()
# create the aubio tempo detection:
hopSize = bufferSize
winSize = hopSize * windowSizeMultiple
tempoDetection = aubio.tempo(method='default', buf_size=winSize, hop_size=hopSize, samplerate=audioInputSampleRate)
# --- WebSocket Setup ---
websocket_url = "ws://192.168.4.1:80/ws"
ws = None
try:
ws = websocket.create_connection(websocket_url)
print(f"Successfully connected to WebSocket at {websocket_url}")
except Exception as e:
print(f"Failed to connect to WebSocket: {e}. Data will not be sent over WebSocket.")
# --- End WebSocket Setup ---
# this function gets called by the input stream, as soon as enough samples are collected from the audio input:
def readAudioFrames(in_data, frame_count, time_info, status):
global ws # Allow modification of the global ws variable
signal = np.frombuffer(in_data, dtype=np.float32)
beat = tempoDetection(signal)
if beat:
bpm = tempoDetection.get_bpm()
print(f"beat! (running with {bpm:.2f} bpm)") # Use f-string for cleaner formatting, removed extra bells
data_to_send = {
"names": ["1"],
"settings": {
"pattern": "pulse",
"delay": 10,
"colors": ["#00ff00"],
"brightness": 10,
"num_leds": 200,
},
}
if ws: # Only send if the websocket connection is established
try:
ws.send(json.dumps(data_to_send))
# print("Sent data over WebSocket") # Optional: for debugging
except websocket.WebSocketConnectionClosedException:
print("WebSocket connection closed, attempting to reconnect...")
ws = None # Mark as closed, connection will need to be re-established if desired
except Exception as e:
print(f"Error sending over WebSocket: {e}")
return (in_data, pyaudio.paContinue)
# create and start the input stream
try:
inputStream = pa.open(format=pyaudio.paFloat32,
input=True,
channels=audioInputChannels,
input_device_index=audioInputDeviceIndex,
frames_per_buffer=bufferSize,
rate=audioInputSampleRate,
stream_callback=readAudioFrames)
inputStream.start_stream()
print("\nAudio stream started. Detecting beats. Press Ctrl+C to stop.")
# Loop to keep the script running, allowing graceful shutdown
while inputStream.is_active():
sleep(0.1) # Small delay to prevent busy-waiting
except KeyboardInterrupt:
print("\nKeyboardInterrupt: Stopping script gracefully.")
except Exception as e:
print(f"An error occurred with the audio stream: {e}")
finally:
# Ensure streams and resources are closed
if 'inputStream' in locals() and inputStream.is_active():
inputStream.stop_stream()
if 'inputStream' in locals() and not inputStream.is_stopped():
inputStream.close()
pa.terminate()
if ws:
print("Closing WebSocket connection.")
ws.close()
print("Script finished.")