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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
import json
from flask import Flask, send_from_directory
from flask_sockets import Sockets
from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler
import os
import os.path
import pickle
try:
import unicornhathd as unicorn
print("unicorn hat hd detected")
except ImportError:
from unicorn_hat_sim import unicornhathd as unicorn
# Actions that can be sent
NO_OP = 'NO_OP'
SET_PIXEL = 'SET_PIXEL'
CLEAR = 'CLEAR'
SAVE = 'SAVE'
LOAD = 'LOAD'
# GLobal references to app & clients
all_clients = set()
app = Flask(__name__, static_folder='build/')
sockets = Sockets(app)
def send_state():
global all_clients
state = {
"saves": os.listdir(SAVES_DIR),
"pixels": unicorn.get_pixels()
}
stateJson = json.dumps(state)
for client in all_clients:
client.send(stateJson)
def execute_command(command):
cmd_type = command['type']
if cmd_type == NO_OP: # Used just to get state without doing an action
pass
elif cmd_type == SET_PIXEL:
x = int(command['x'])
y = int(command['y'])
r = int(command['r'])
g = int(command['g'])
b = int(command['b'])
unicorn.set_pixel(x, y, r, g, b)
unicorn.show()
elif cmd_type == CLEAR:
unicorn.clear()
unicorn.show()
elif cmd_type == SAVE:
saveName = command["saveName"]
save(saveName)
elif cmd_type == LOAD:
saveName = command["saveName"]
load(saveName)
# Constants
SAVES_DIR = 'saves/'
def save(saveName):
with open(os.path.join(SAVES_DIR, saveName), "wb") as f:
pickle.dump(unicorn.get_pixels(), f)
def load(saveName):
with open(os.path.join(SAVES_DIR, saveName), "rb") as f:
pixels = pickle.load(f)
for x, row in enumerate(pixels):
for y, pixel in enumerate(row):
unicorn.set_pixel(x, y, *pixel)
unicorn.show()
@sockets.route('/ws')
def do_websocket(websocket):
global all_clients
all_clients.add(websocket)
try:
while not websocket.closed:
command = websocket.receive()
if not command:
break
cmd = json.loads(command)
execute_command(cmd)
send_state()
finally:
all_clients.remove(websocket)
@app.route('/<path:path>')
def send_static(path):
return send_from_directory('build/', path)
def main():
try:
os.mkdir(SAVES_DIR)
except:
pass
try:
print("Serving on port 3001")
server = pywsgi.WSGIServer(('', 3001), app, handler_class=WebSocketHandler)
server.serve_forever()
finally:
unicorn.off()
if __name__=="__main__":
main()
|
