1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
#!/usr/bin/env python3
import array
import socket
import threading
import time
from tkinter import *
from tkinter.ttk import *
CHANNEL_COUNT = 64
class Main(Frame):
def _update_display(self):
with self.lock:
for i, s in zip(self.channels, self.sliders):
s.config(state=NORMAL)
s.set(i)
s.config(state=DISABLED)
self.master.after(16, self._update_display)
def update(self, b: bytes):
with self.lock:
self.channels = array.array('B', b)
def __init__(self, root):
super().__init__(root)
self.sliders = []
self.rowconfigure(0, weight=1)
for i in range(CHANNEL_COUNT):
self.columnconfigure(i, weight=1)
self.sliders.append(Scale(self, from_=255, to=0, orient=VERTICAL, state=DISABLED))
self.sliders[-1].grid(row=0, column=i, sticky=N+E+S+W)
Label(self, text=str(i+1)).grid(row=1, column=i, sticky=N+E+S+W)
self.channels = array.array('B', (0 for i in range(CHANNEL_COUNT)))
self.lock = threading.RLock()
self.master.after(0, self._update_display)
def handle_conn(conn, m):
while True:
a = conn.recv(1024)
if not a:
break
m.update(a)
def socket_main(m):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 6969))
s.listen()
print("Listening")
while True:
conn, addr = s.accept()
threading.Thread(target=handle_conn, args=(conn, m,)).start()
if __name__ == "__main__":
root = Tk()
root.rowconfigure(0, weight=1)
root.columnconfigure(0, weight=1)
root.wm_title("Lighting Output")
main = Main(root)
main.grid(row=0, column=0, sticky=N+E+S+W)
root.after(0, threading.Thread(target=socket_main, args=(main,)).start)
main.mainloop()
|
