-
Notifications
You must be signed in to change notification settings - Fork 0
/
predict.py
178 lines (160 loc) · 5.28 KB
/
predict.py
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' # Ignore tensorflow warnings
import argparse
import serial
import time
import threading
import numpy as np
from tensorflow import keras
from scipy.ndimage import uniform_filter1d
def read_serial(model, HAND):
global event
gesture = "4-4" # defaults to crash
prev_gesture = gesture
# Keep track of sample number (to ensure a certain no. of samples between each beat)
sample_num = 0
prev_sample_num = sample_num
timestep = 30
window_size = 10 # prediction occurs every <window_size> samples
data = {
"index": [],
"middle": [],
"ring": [],
"pinky": []
}
TRANSITIONS = [
"0-0","0-1","0-2","0-3","0-4",
"1-0","1-1","1-2","1-3","1-4",
"2-0","2-1","2-2","2-3","2-4",
"3-0","3-1","3-2","3-3","3-4",
"4-0","4-1","4-2","4-3","4-4"
]
BEAT = False #
BEAT_ACTIVATED = False #
BEAT_INTERVAL = 60 # number of samples between each beat (to prevent double-triggering)
LOW = 100
HIGH = 1000
while event.is_set():
line = serial.readline() # read a byte
if not line:
continue
try:
string = line.decode() # convert the byte string to a unicode string
# print(string, end="")
if prev_gesture != gesture:
# print(gesture)
prev_gesture = gesture
# Parse data into 5 values for flex sensors + accX
lst = [float(k) for k in string.split(',')[:-1]]
if len(lst) != 5:
continue
sample_num += 1
data["index"].append(lst[0])
data["middle"].append(lst[1])
data["ring"].append(lst[2])
data["pinky"].append(lst[3])
accX = lst[4]
# If next beat is ready to be triggered
if not BEAT:
# Right hand (accX goes up then down)
if HAND == "right":
# Activate beat to be sounded
if accX > HIGH and not BEAT_ACTIVATED:
BEAT_ACTIVATED = True
prev_sample_num = sample_num
# Actually sound beat
if BEAT_ACTIVATED and accX < LOW:
BEAT = True
BEAT_ACTIVATED = False # reset to False to allow next activation of beat
ind = int(gesture[2]) # gesture index
t = threading.Thread(target=play_sound, args=[ind])
t.start()
# Left hand (accX goes down then up)
else:
if accX < -HIGH and not BEAT_ACTIVATED:
BEAT_ACTIVATED = True
prev_sample_num = sample_num
if BEAT_ACTIVATED and accX > -LOW:
BEAT = True
BEAT_ACTIVATED = False
ind = int(gesture[2])
t = threading.Thread(target=play_sound, args=[ind])
t.start()
# Ensure that there're 60 samples between each beat
# Prevents double triggering
else:
if sample_num - prev_sample_num > BEAT_INTERVAL:
BEAT = False
# Parse data and pre-process for input to model
if len(data["index"]) == timestep:
x = np.c_[
np.array(data["index"]).reshape(-1,1),
np.array(data["middle"]).reshape(-1,1),
np.array(data["ring"]).reshape(-1,1),
np.array(data["pinky"]).reshape(-1,1)
]
x = uniform_filter1d(x, axis=0, size=10)
for k in range(4):
data_min, data_max = np.min(x), np.max(x)
x[:,k] = (x[:,k] - data_min) / (data_max - data_min)
x = np.ravel(x, order="F").reshape(1,-1)
for k,v in data.items():
data[k] = v[window_size:] # prediction occurs every 10 samples
# Get model prediction
y_pred = model.predict(x, verbose=0)[0]
if max(y_pred) > 0.90:
transition = np.argmax(y_pred)
predicted_gesture = TRANSITIONS[transition]
# print(predicted_gesture, end='\r')
if not ((predicted_gesture[0] == predicted_gesture[2]) and
(predicted_gesture[0] != prev_gesture[2])):
gesture = predicted_gesture
except Exception as e:
print("exception in main loop: {!s}".format(e), end='\r')
# Play the sound according to the gesture
beat_count = 1
def play_sound(ind):
global PLAY_SOUND, SOUNDS, beat_count
if PLAY_SOUND == "play":
SOUNDS[ind].play()
else:
global keyboard
keyboard.press(SOUNDS[ind])
time.sleep(0.05)
keyboard.release(SOUNDS[ind])
print("BEAT %d: %s " % (beat_count, GESTURES[ind]), end="\r")
beat_count += 1
return
# ======== Main code ========
parser = argparse.ArgumentParser()
parser.add_argument('-d', '--dev', default="ttyUSB0", type=str, help="Device ID under /dev/ (e.g. cu.usbserial-XXXXXX")
parser.add_argument('-n', '--hand', default="r", type=str, help="Hand: 'r' for right, 'l' for left. Default is right.")
parser.add_argument('-s', '--sound', default="p", type=str, help="Method for playing sound: 'k' for keyboard press, or 'p' for playing sample. Default is playing.")
args = parser.parse_args()
SERIAL_PATH = "/dev/" + str(args.dev)
BAUD_RATE = 115200
HAND = "left" if args.hand == "l" else "right"
MODEL_PATH = "models/mlp_right" if HAND == "right" else "models/mlp_left"
GESTURES = ["Kick","Hihat","Snare","Tom","Crash"]
PLAY_SOUND = "play" if args.sound == "p" else "keyboard"
if PLAY_SOUND == "play":
import pyglet
SOUNDS = [pyglet.resource.media("samples/" + file + ".wav", streaming=False)
for file in ["kick","hihat","snare","tom","crash"]
]
else:
from pynput.keyboard Controller
keyboard = Controller()
SOUNDS = ["a", "t", "s", "h", "o"]
event = threading.Event()
model = keras.models.load_model(MODEL_PATH)
serial = serial.Serial(SERIAL_PATH, BAUD_RATE, timeout=1)
event.set()
try:
print("Ready.")
read_serial(model, HAND)
except KeyboardInterrupt:
event.clear()
if PLAY_SOUND == "play":
pyglet.app.exit()
serial.close()