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

Change to websocket

This commit is contained in:
Jimmy 2025-07-11 21:33:50 +12:00
parent f302be85c0
commit 65774837c7
2 changed files with 77 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
main.py
View File

@ -1,20 +1,24 @@
import asyncio
import tkinter as tk
from tkinter import ttk
import json
from async_tkinter_loop import async_handler, async_mainloop
from networking import send_to_server
import color_utils
bg_color="#2e2e2e"
bg_color = "#2e2e2e"
class App:
def __init__(self) -> None:
self.root = tk.Tk()
self.root.attributes('-fullscreen', True)
self.root.configure(bg=bg_color) # Set background color
# List of servers (IP, Port)
self.servers = [("192.168.0.208", 80),("192.168.0.201", 80), ("192.168.0.202", 80), ("192.168.0.203", 80), ("192.168.0.206", 80)]
self.lightgroups = {
"light1": {"ids": [0], "settings": {"color": "#00ff00"}},
"light2": {"ids": [0], "settings": {"color": "#ff0000"}}
}
# Create Notebook for tabs
self.notebook = ttk.Notebook(self.root)
@ -22,12 +26,11 @@ class App:
# Create a tab for each server
self.tabs = {}
for server in self.servers:
for key, value in self.lightgroups.items():
tab = ttk.Frame(self.notebook)
self.notebook.add(tab, text=f"{server[0]}:{server[1]}")
self.notebook.add(tab, text=key)
self.create_sliders(tab)
self.tabs[server] = tab
#self.root.configure(bg=bg_color)
self.tabs[key] = tab
async_mainloop(self.root)
@ -40,25 +43,25 @@ class App:
red_slider = tk.Scale(tab, from_=255, to=0, orient=tk.VERTICAL, length=slider_length, width=slider_width, label="Red")
red_slider.set(0)
red_slider.pack(side=tk.LEFT, padx=10)
red_slider.bind("<ButtonRelease-1>", lambda event, srv=tab: self.update_colour(event, srv))
red_slider.bind("<ButtonRelease-1>", lambda _: self.update_colour(tab))
# Green Slider
green_slider = tk.Scale(tab, from_=255, to=0, orient=tk.VERTICAL, length=slider_length, width=slider_width, label="Green")
green_slider.set(0)
green_slider.pack(side=tk.LEFT, padx=10)
green_slider.bind("<ButtonRelease-1>", lambda event, srv=tab: self.update_colour(event, srv))
green_slider.bind("<ButtonRelease-1>", lambda _: self.update_colour(tab))
# Blue Slider
blue_slider = tk.Scale(tab, from_=255, to=0, orient=tk.VERTICAL, length=slider_length, width=slider_width, label="Blue")
blue_slider.set(0)
blue_slider.pack(side=tk.LEFT, padx=10)
blue_slider.bind("<ButtonRelease-1>", lambda event, srv=tab: self.update_colour(event, srv))
blue_slider.bind("<ButtonRelease-1>", lambda _: self.update_colour(tab))
# Brightness Slider
brightness_slider = tk.Scale(tab, from_=255, to=0, orient=tk.VERTICAL, length=slider_length, width=slider_width, label="Brightness")
brightness_slider.set(127)
brightness_slider.pack(side=tk.LEFT, padx=10)
brightness_slider.bind("<ButtonRelease-1>", lambda event, srv=tab: self.update_colour(event, srv))
brightness_slider.bind("<ButtonRelease-1>", lambda _: self.update_colour(tab))
tab.widgets = {
'red_slider': red_slider,
@ -68,23 +71,49 @@ class App:
}
@async_handler
async def update_colour(self, event, tab):
async def update_colour(self, tab):
"""Update color based on the slider values and send to the selected server."""
red_slider = tab.widgets['red_slider']
green_slider = tab.widgets['green_slider']
blue_slider = tab.widgets['blue_slider']
brightness_slider = tab.widgets['brightness_slider']
try:
# Retrieve slider values
red_slider = tab.widgets['red_slider']
green_slider = tab.widgets['green_slider']
blue_slider = tab.widgets['blue_slider']
brightness_slider = tab.widgets['brightness_slider']
r = red_slider.get()
g = green_slider.get()
b = blue_slider.get()
brightness = brightness_slider.get()
r = red_slider.get()
g = green_slider.get()
b = blue_slider.get()
brightness = brightness_slider.get()
color = color_utils.adjust_brightness((r, b, g), brightness)
print(color, brightness)
# Adjust brightness
color = color_utils.adjust_brightness((r, g, b), brightness)
print(f"Adjusted color: {color}, Brightness: {brightness}")
# Convert RGB to hex color
hex_color = f"#{color[0]:02x}{color[1]:02x}{color[2]:02x}"
# Get the selected server
selected_server = self.notebook.tab(self.notebook.select(), "text")
ids = self.lightgroups[selected_server]["ids"]
# Construct WebSocket payload
payload = {
"save": True,
"ids": [0],
"settings": {
"colors": [hex_color], # Use the dynamically calculated hex color
"brightness": brightness, # Use the brightness slider value
"pattern": "on"
}
}
# Send the payload to the server
await send_to_server(payload)
print(f"Sent payload: {payload}")
except Exception as e:
print(f"Error updating color: {e}")
# Optionally, display the error in the GUI
selected_server = self.notebook.tab(self.notebook.select(), "text").split(":")
await send_to_server(bytes(color), (selected_server[0], int(selected_server[1])))
if __name__ == "__main__":
app = App()

72
networking.py
View File

@ -1,53 +1,29 @@
import socket
import asyncio
import httpx
from websockets import connect
import websockets
import json
async def send_color_data(color_data, servers):
"""Send color data to all configured servers."""
tasks = [send_to_server(color_data, server) for server in servers]
await asyncio.gather(*tasks)
async def send_to_server(color_data, server):
"""Connect to a server, send color data, and close the connection."""
server_ip, server_port = server
async def send_to_server(data):
"""
Send WebSocket data to the server.
"""
try:
reader, writer = await asyncio.open_connection(server_ip, server_port)
writer.write(color_data)
await writer.drain()
writer.close()
await writer.wait_closed()
print(f"Sent data to {server_ip}:{server_port}")
except (socket.error, ConnectionError) as e:
print(f"Error sending to {server_ip}:{server_port}: {e}")
# Connect to the WebSocket server
async with websockets.connect("ws://192.168.4.1:80/ws") as websocket:
# Serialize data to JSON and send it
await websocket.send(json.dumps(data))
except (ConnectionError, websockets.exceptions.ConnectionClosed) as e:
print(f"Error sending to {e}")
async def post(server, endpoint, data=""):
async with httpx.AsyncClient() as client:
r = await client.post(f"http://{server}/{endpoint}", data=data)
print(server, r)
async def main():
b = bytearray(180)
try:
ws1 = await connect("ws://10.1.1.200/external")
await asyncio.gather(post("10.1.1.200", "pattern", "external"))
while True:
print("sync")
for i in range(len(b)):
b[i] = 128
await asyncio.gather(ws1.send(b))
await asyncio.sleep(5)
finally:
await ws1.close()
await asyncio.gather(post("10.1.1.200", "pattern", "off"))
if __name__ == "__main__":
asyncio.run(main())
# Define the data to be sent
data = {
"settings": {
"color": "#00ff00"
}
}
# Server details
server = ("192.168.4.1", 80) # Example WebSocket server port
# Run the asynchronous function using asyncio.run
asyncio.run(send_to_server(data, server))