Merge pull request #29 from SoftcodingForYou/make_gui_modular

V2.90.0 - Make GUI modular

Author: SoftcodingForYou

README.md

@@ -1,6 +1,6 @@
 # Neuri GUI
 
-print(«Hello Volt»)
+print("Hello, Volt!")
 
 A graphical interface for electrophysiological biosensors that allows you to:
 1. Set multiple parameters for real-time signal visualization
@@ -13,6 +13,12 @@ Watch the introduction on Youtube:
 
 [![Watch the introdcution](https://img.youtube.com/vi/8DbGR9KUszQ/hqdefault.jpg)](https://www.youtube.com/embed/8DbGR9KUszQ)
 
+## Compatible devices
+
+- Neuri 1.x by Helment
+- Neuri-Lolin S3-PRO by Helment
+- BioAmp EXG Pill by Upside Down Labs 
+
 ## Setup
 
 The GUI is distributed as a Python program. The GUI should work with **Python versions 3.9 or higher**.
@@ -55,11 +61,6 @@ ng.Run()
 
 Note that your settings are stored in a "settings.cfg" file inside the current workspace directory of the IDE or terminal.
 
-## Compatible devices
-
-- Neuri 1.x
-- Neuri-Lolin S3-PRO
-
 ## Troubleshooting
 
 1. If the GUI seems to be stuck, most probably the wrong ports have been selected Check the port names (ie COM12 vs COM13). Pay special attention to the fact that if a port file is specified in the settings.conf, the GUI will default to that port even if there is no device connected to it.

dist/neurigui-2.82.0.tar.gz

@@ -0,0 +1,53 @@
+try:
+    from compatible_boards import COMPATIBLE_BOARDS
+except:
+    from .compatible_boards import COMPATIBLE_BOARDS
+
+
+class BoardAgnosticUtils():
+        
+    def __init__(self, parameter):
+
+        # We can not just assign all parameters to for example self.params
+        # since Python's Process() does not allow to parse several of the
+        # stored objects in "parameters"
+        self.streaming_parameter = {
+            "baud_rate":    parameter.baud_rate,
+            "time_out":     parameter.time_out,
+            "port":         parameter.port,
+            "start_code":   parameter.start_code, # Start code that boards reads as sign to startt sampling (int)
+            "max_chans":    parameter.max_chans, # Specifies how many channels the incoming signals has (int)
+            "buffer_length":parameter.buffer_length, # Length of constructed signal buffer (int)
+            "buffer_add":   parameter.buffer_add, # Length of padding to avoid filter artifacts  (int)
+            "sample_rate":  parameter.sample_rate, # Sampling rate (Hz, int)
+            "PGA":          parameter.PGA, # Programmable gain (int)
+            "saving_interval": parameter.saving_interval, # how often the data gets written to disk (seconds, int)
+            "udp_ip":       parameter.udp_ip, # IP the data shall be forwarded to
+            "udp_port":     parameter.udp_port, # Port of signal forwarding
+            "set_customsession": parameter.set_customsession, # Whether or not to have a custom output file name (bool)
+            "sessionName":  parameter.sessionName, # Custom file name for output (str)
+        }
+
+
+    def get_board_specific_utils(self, board):
+        """Returns the board_specific functions that allows us to read and 
+        process incoming signals
+
+        Args:
+            board (str): Unique name of board. Defined in parameters.py
+
+        Returns:
+            (object): Class object containing all necessary functions to
+                work with given board.
+        """
+        for _, board_name in enumerate(list(COMPATIBLE_BOARDS.keys())):
+            if board == board_name:
+                return COMPATIBLE_BOARDS[board_name][3]
+        raise Exception("Could not get board-specific functions collection")
+
+
+    def headless_sampling(self):
+        pass # This function only exists in order to make timer call something
+
+    
+         

dist/neurigui-2.82.0-py3-none-any.whl dist/neurigui-2.90.0-py3-none-any.whldist/neurigui-2.82.0-py3-none-any.whl renamed to dist/neurigui-2.90.0-py3-none-any.whl

@@ -0,0 +1,22 @@
+"""Manual board management
+For every board, we collect here the sampling functions and store them 
+inside the same class. For this, you need to manage the imports and
+dictionnary manually, adding the imported object in fourth position of 
+the board's [list].
+"""
+try:
+    from signal_sampling_neuri_v1       import SamplingUtilsNeuriV1Helment
+    from signal_sampling_neuri_lolin    import SamplingUtilsNeuriLolinHelment
+    from signal_sampling_bioamp         import SamplingUtilsBioAmpUpsideDownLabs
+except: # Necessary if running as Python package
+    from .signal_sampling_neuri_v1      import SamplingUtilsNeuriV1Helment
+    from .signal_sampling_neuri_lolin   import SamplingUtilsNeuriLolinHelment
+    from .signal_sampling_bioamp        import SamplingUtilsBioAmpUpsideDownLabs
+
+
+COMPATIBLE_BOARDS = {
+    # Unique name: [Start code (int, None if no code required), Sampling rate (Hz), baud rate (int), Board_specific class (object)]
+    "Neuri V1 by Helment":                  [2, 200, 115200, SamplingUtilsNeuriV1Helment],
+    "Neuri-Lolin S3-PRO by Helment":        [2, 200, 115200, SamplingUtilsNeuriLolinHelment],
+    "BioAmp EXG Pill by Upside Down Labs":  [2, 125, 115200, SamplingUtilsBioAmpUpsideDownLabs],
+}

dist/neurigui-2.90.0.tar.gz

@@ -0,0 +1,112 @@
+import serial
+from threading                          import Thread
+from datetime                           import datetime
+
+class IOManager():
+
+    def __init__(self, parameter):
+        #Output
+        t0              = str(datetime.now())
+        t0              = t0.replace(':', '_')
+        if parameter["set_customsession"]:
+            file_name   = parameter["sessionName"] + '.txt'
+        else:
+            file_name   = 'Neuri ' + t0 + '.txt'
+
+        # Prepare data output
+        self.output_file= file_name
+
+        with open(self.output_file, 'w', encoding= "utf_8") as file:
+            file.write("Session Started\n")
+            file.close() # Important for data to get written
+
+
+    def set_up_serial_entry_point(self, baud_rate, time_out, port):
+        # =================================================================
+        # Connects to device, preferrably by USB
+        # INPUT
+        #   baud_rate       = baud rate of the port (scalar)
+        #   time_out        = how long a message should be waited for
+        #                     default = None (infinite), scalar or None
+        #   port            = port to connect to ("COMxy", str)
+        # OUTPUT
+        #   ser             = Serial connection (object)
+        # =================================================================
+
+        # Open communication protocol
+        ser            = serial.Serial()
+        ser.baudrate   = baud_rate
+        ser.timeout    = time_out
+        ser.port       = port
+        print('Ready to connect to board')
+        return ser
+    
+
+    def master_write_data(self, eeg_data, time_stamps, saving_interval):
+        # =================================================================
+        # Input:
+        #   eeg_data            2D numpy array [channels x samples] (float)
+        #   time_stamps         1D numpy array (float)
+        #   saving_interval     Scalar
+        # Output:
+        #   No output
+        # =================================================================
+        new_buffer          = eeg_data[:, -saving_interval:]
+        new_time_stamps     = time_stamps[-saving_interval:]
+            
+        save_thread = Thread(target=self.write_data_thread,
+            args=(new_buffer, new_time_stamps))
+        save_thread.start()
+
+
+    def calc_sample_rate(self, curr_time, prev_iter_time, sample_rate, time_stamps):
+
+        time_diff           = curr_time - prev_iter_time
+        # It took (time_diff) ms to fetch (sample_rate) samples.
+        # Calculate actual sampling rate
+        actual_sr           = int((sample_rate / (time_diff / 1000)))
+        print('%d ms: Writing data (sampling rate = %d Hz)' %
+            (time_stamps[-1], actual_sr))
+
+
+    def write_data_thread(self, eeg_data, time_stamps):
+        # =================================================================
+        # Input:
+        #   eeg_data            2D numpy array [channels x samples] (float)
+        #   time_stamps         1D numpy array (float)
+        # Output:
+        #   No output
+        # =================================================================
+        # Append the data points to the end of the file
+        with open(self.output_file, 'a', encoding= "utf_8") as file:
+            buffer_length = time_stamps.shape[0]
+
+            for sample_index in range(buffer_length):
+                # Format time stamp
+                time_stamp      = time_stamps[sample_index]
+                # format eeg data
+                eeg_data_points = eeg_data[:,sample_index].tolist()
+                eeg_data_points = [str(value) for value in eeg_data_points]
+                eeg_data_points = ",".join(eeg_data_points)
+                
+                # Write line to file
+                file.write(f"{time_stamp}, {eeg_data_points} \n")
+            file.close() # Important for data to get written
+
+
+    def build_standard_relay_message(self, s_chans):
+        """Build a relay jsonized message string that gets forwarded to any
+        third-party application.
+
+        Args:
+            s_chans (int): Number of channels (specific for boards)
+
+        Returns relay_array (str):json.format message string with keys for:
+            1. timestamps (t)
+            2. channels (c)
+        """
+        relay_array         = {}
+        relay_array["t"]    = ''
+        for iC in range(s_chans):
+            relay_array["".join(["c", str(iC+1)])] = ''
+        return relay_array

neuri/backend/board_agnostic_utils.py

@@ -1,6 +1,6 @@
 import tkinter                              as tk
 from PIL                                    import Image, ImageTk
-import os
+
 
 class ParamVal():