Adding the Kivy package as gui with streaming data, their native LinePlot works much smoother than matplotlib embedding in tkinter

Author: matthijscox

symbionic/_model.py

@@ -0,0 +1,67 @@
+import random
+import numpy as np
+import symbionic
+
+
+def prepare_data_for_keras(input_data):
+    # only select last 195 samples
+    output_data = input_data[-195:, :]
+    # preparation for 1D Neural Network
+    # TODO: it's too slow for fast updating at 0.1 second intervals...
+    output_data = np.apply_along_axis(lambda x: symbionic.calc_envelope(x,smooth=51), 0, output_data)
+    output_data = output_data.reshape((1,output_data.shape[0]*output_data.shape[1]))
+    return output_data
+
+
+def absmax_at_end(signal):
+    return np.max(abs(signal[-50:]))
+
+
+def take_max_abs(input_data):
+    features =  np.apply_along_axis(absmax_at_end, 0, input_data)
+    features = features.reshape(1, -1)
+    return features
+
+
+class FakeModel:
+    def __init__(self, n_gestures=6):
+        self.number_of_gestures = n_gestures
+
+    def predict(self, input_data):
+        p = self.number_of_gestures*[0]
+        p[random.randrange(self.number_of_gestures)] = 1
+        return p
+
+
+class GestureModel:
+    def __init__(self,model=None, data_prepare=None):
+        self.model = model
+        self.data_prepare = data_prepare
+        self.collapse = False
+
+    def predict(self,input_data):
+        prediction = 0
+        if self.data_prepare is not None:
+            input_data = self.data_prepare(input_data)
+        if self.model is not None:
+            prediction = self.model.predict(input_data)
+            if self.collapse:
+                prediction = np.argmax(prediction)
+        return prediction
+
+
+class PredictionBuffer:
+    def __init__(self, n_gestures=6):
+        self.number_of_gestures = n_gestures
+        self.buffer = [[] for n in range(self.number_of_gestures)]
+        self.stored_predictions = 1
+
+    def append(self,predictions):
+        for gesture, gesture_probability in zip(range(self.number_of_gestures), predictions):
+            self.buffer[gesture].append(gesture_probability)
+        self.stored_predictions += 1
+
+    def get_predictions(self,number_of_predictions=30):
+        predictions = self.buffer
+        predictions = [p[-number_of_predictions:] for p in predictions]
+        return np.array(predictions).transpose()

symbionic/gui/__init__.py

@@ -0,0 +1,148 @@
+#https://kivy.org/doc/stable/installation/installation-windows.html
+from kivy.lang import Builder
+from kivy.app import App
+from kivy.uix.boxlayout import BoxLayout
+from kivy.clock import Clock
+import symbionic
+from symbionic.gui.graph import Graph, LinePlot
+from symbionic._model import FakeModel, GestureModel, PredictionBuffer, take_max_abs
+from kivy.core.window import Window
+from sklearn.externals import joblib
+
+
+class DataIterator:
+    def __init__(self, data):
+        self.index = 0
+        self.step = 1
+        self.size = 1
+        self.data = data
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self.index >= len(self.data):
+            #raise StopIteration
+            self.index = 0 # start over
+        start_index = self.index
+        end_index = start_index + self.step + self.size - 1
+        self.index = start_index + self.step
+        return self.data[start_index:end_index, :]
+
+    def goto(self, index):
+        self.index = index
+
+
+class FileDataStub:
+    def __init__(self):
+        emg_data = symbionic.EmgData()
+        emg_data.load(symbionic.example_data_directory() + "raw3.bin")
+        emg_array = emg_data.data['g1'].drop('time', axis=1).values
+        emg_iterator = DataIterator(emg_array)
+        emg_iterator.step = 80
+        emg_iterator.size = 500
+        self.emg_iterator = emg_iterator
+
+    def __next__(self):
+        return next(self.emg_iterator)
+
+
+class RandomForestModel:
+    def __init__(self):
+        # using a pre-trained random forest stored via joblib
+        self.fitted_model = joblib.load('maxabs_forest.joblib')
+
+    def predict(self,data):
+        # need to convert to a 1D array for single sample data
+        predictions = self.fitted_model.predict(data)
+        return predictions.ravel()
+
+
+class Logic(BoxLayout):
+    def __init__(self):
+        super().__init__()
+        self.number_of_channels = 8
+        self.number_of_packages = 15
+        self.number_of_gestures = 7
+        self.update_interval = 0.1
+        self.graphs = []
+        self.plots = []
+        self.receiver = symbionic.GFDataReceiverSocket(stub=True)
+        self.fake_data = FileDataStub()
+        self.rawDataBox = []
+        self.gestureBox = []
+        self.gestureModel = GestureModel(FakeModel(self.number_of_gestures), data_prepare=take_max_abs)
+        #self.gestureModel.model = RandomForestModel()
+        self.predictions = PredictionBuffer(self.number_of_gestures)
+
+    def init_graphs(self):
+        self.rawDataBox = GraphBox(self.ids.raw_data_box,self.number_of_channels)
+        self.rawDataBox.init_graphs()
+        self.gestureBox = GraphBox(self.ids.gesture_box, self.number_of_gestures)
+        self.gestureBox.ylim = (-0.1,1.1)
+        self.gestureBox.init_graphs()
+
+    def start(self):
+        self.receiver.start()
+        Clock.schedule_interval(self.update_gui, self.update_interval)
+
+    def stop(self):
+        self.receiver.stop()
+        Clock.unschedule(self.update_gui)
+
+    def update_gui(self, dt):
+        #data = self.receiver.dataHandler.get_latest_emg_data(self.number_of_packages)
+        data = next(self.fake_data)
+        self.rawDataBox.update_graphs(data)
+        self.update_predictions(data)
+        self.update_prediction_graphs()
+
+    def update_predictions(self,data):
+        predictions = self.gestureModel.predict(data)
+        self.predictions.append(predictions)
+
+    def update_prediction_graphs(self):
+        predictions = self.predictions.get_predictions()
+        self.gestureBox.update_graphs(predictions)
+
+
+class GraphBox:
+    def __init__(self, box_id, number_of_graphs):
+        self.box_id = box_id
+        self.number_of_graphs = number_of_graphs
+        self.graphs = []
+        self.plots = []
+        self.ylim = (-150,150)
+
+    def init_graphs(self):
+        for g in range(self.number_of_graphs):
+            self.add_graph()
+
+    def add_graph(self):
+        graph = Graph(xmin=-0, xmax=100, ymin=self.ylim[0], ymax=self.ylim[1])
+        plot = LinePlot(color=[1, 0, 0, 1], line_width=1.1)
+        graph.add_plot(plot)
+        self.plots.append(plot)
+        self.graphs.append(graph)
+        self.box_id.add_widget(graph)
+
+    def update_graphs(self,data):
+        buffer_size = data.shape[0]  # data size
+        x_axis = [x for x in range(0, buffer_size)]  # x-axis data of graph
+        for graph in range(self.number_of_graphs):
+            single_graph = data[:, graph].tolist()
+            self.plots[graph].points = [(x, y) for (x, y) in zip(x_axis,single_graph)]
+            self.graphs[graph].xmax = buffer_size
+
+
+class SymbionicApp(App):
+    Window.size = (1000,600)
+
+    def build(self):
+        self.root = Builder.load_file("look.kv")
+        self.root.init_graphs()
+        return self.root
+
+
+if __name__ == "__main__":
+    SymbionicApp().run()

examples/graph/LICENSE renamed to symbionic/gui/graph/LICENSE

@@ -0,0 +1,38 @@
+#:import MeshLinePlot symbionic.gui.graph.MeshLinePlot
+Logic:
+    BoxLayout:
+        orientation: "vertical"
+        padding: 40
+        BoxLayout:
+            orientation: "horizontal"
+            size_hint: [1, .8]
+            BoxLayout:
+                orientation: "vertical"
+                id: raw_data_box
+                size_hint: [0.4, 1]
+            BoxLayout:
+                orientation: "vertical"
+                size_hint: [0.1, 1]
+            BoxLayout:
+                orientation: "vertical"
+                id: gesture_box
+                size_hint: [0.4, 1]
+        BoxLayout:
+            orientation: "horizontal"
+            size_hint: [1, .2]
+            BoxLayout:
+                orientation: "horizontal"
+            BoxLayout:
+                size_hint: None, None
+                size: 300, 50
+                orientation: "horizontal"
+                Button:
+                    id: start_button
+                    text: "START"
+                    bold: True
+                    on_press: root.start()
+                Button:
+                    id: stop_button
+                    text: "STOP"
+                    bold: True
+                    on_press: root.stop()