Merge pull request #32 from PayalLakra/bio_amptool

Requirements updates, Chords baud rate update, ffteeg- moving window implementation

Author: lorforlinux

README.md

@@ -51,7 +51,7 @@ To use the script, run it from the command line with various options:
 ### Options
 
 - `-p`, `--port` <port>: Specify the serial port to use (e.g., COM5, /dev/ttyUSB0).
-- `-b`, `--baudrate` <baudrate>: Set the baud rate for serial communication (default is 230400).
+- `-b`, `--baudrate` <baudrate>: Set the baud rate for serial communication. By default the script will first attempt to use 230400, and if that fails, it will automatically fallback to 115200.
 - `--csv`: Enable CSV logging. Data will be saved to a timestamped file.
 - `--lsl`: Enable LSL streaming. Sends data to an LSL outlet.
 - `-v`, `--verbose`: Enable verbose output with detailed statistics and error reporting.
@@ -66,38 +66,8 @@ To use the script, run it from the command line with various options:
 
 - **Log Intervals**: The script logs data counts every second and provides a summary every 10 minutes, including the sampling rate and drift in seconds per hour.
 
-### LSL Streaming
-
-- **Stream Name**: `BioAmpDataStream`
-- **Stream Type**: `EXG`
-- **Channel Count**: `6`
-- **Sampling Rate**: `UNO-R3 : 250 Hz` , `UNO-R4 : 500 Hz`
-- **Data Format**: `float32`
-
-
-### Script Functions
-
-`auto_detect_arduino(baudrate, timeout=1)`: Detects an Arduino connected via serial port. Returns the port name if detected.
-
-`read_arduino_data(ser, csv_writer=None)`: Reads and processes data from the Arduino. Writes data to CSV and/or LSL stream if enabled.
-
-`start_timer()`: Initializes timers for 1-second and 10-minute intervals.
-
-`log_one_second_data(verbose=False)`: Logs and resets data for the 1-second interval.
-
-`log_ten_minute_data(verbose=False)`: Logs data and statistics for the 10-minute interval.
-
-`parse_data(port,baudrate,lsl_flag=False,csv_flag=False,verbose=False)`: Parses data from Arduino and manages logging, streaming, and GUI updates.
-
-`cleanup()`: Handles all the cleanup tasks.
-
-`main()`: Handles command-line argument parsing and initiates data processing.
-
 ## Applications
 
-> [!IMPORTANT]
- Before using the below Applications make sure you are in application folder.
-
 ### GUI
 
 - `python gui.py`: Enable the real-time data plotting GUI.
@@ -108,7 +78,7 @@ To use the script, run it from the command line with various options:
 
 ### HEART RATE
 
-- `python heartbeat.ecg.py`:Enable a GUI with real-time ECG and heart rate.
+- `python heartbeat_ecg.py`:Enable a GUI with real-time ECG and heart rate.
 
 ### EMG ENVELOPE
 
@@ -122,6 +92,38 @@ To use the script, run it from the command line with various options:
 
 - `python ffteeg.py`: Enable a GUI with real-time EEG data with its FFT.
 
+### Keystroke
+
+- `python keystroke.py`: On running, a pop-up opens for connecting, and on pressing Start, blinks are detected to simulate spacebar key presses.
+
+## Running All Applications Together
+
+To run all applications together:
+
+```bash
+python app.py
+```
+
+> [!NOTE] 
+> Before running, make sure to install all dependencies by running the command:
+```bash
+pip install -r app_requirements.txt
+```
+
+This will launch a Web interface. Use the interface to control the applications:
+
+1. Click the `Start LSL Stream` button to initiate the LSL stream.
+2. Then, click on any application button to run the desired module.
+
+### Available Applications
+- `ffteeg`: Real-time EEG analysis with FFT and brainwave power calculation.
+- `heartbeat_ecg`: Analyze ECG data and extract heartbeat metrics.
+- `eog`: Real-time EOG monitoring with blink detection.
+- `emgenvelope`: Real-time EMG monitor with filtering and RMS envelope.
+- `keystroke`: GUI for EOG-based blink detection triggering a keystroke.
+- `game`: Launch an EEG game for 2 players (Tug of War).
+- `csv_plotter`: Plot data from a CSV file.
+- `gui`: Launch the GUI for real time signal visualization.
 
 ## Troubleshooting
 

app_requirements.txt

@@ -7,4 +7,6 @@ neurokit2==0.2.10
 plotly==5.24.1
 pandas==2.2.3
 tk==0.1.0
-PyAutoGUI==0.9.54
+PyAutoGUI==0.9.54
+Flask==3.1.0
+psutil==6.1.1

chords.py

@@ -54,10 +54,17 @@
 board = ""          # Variable for Connected Arduino Board
 supported_boards = {
     "UNO-R3": {"sampling_rate": 250, "Num_channels": 6},
-    "UNO-CLONE": {"sampling_rate": 250, "Num_channels": 6},           # Baud Rate 115200
+    "UNO-CLONE": {"sampling_rate": 250, "Num_channels": 6},
+    "GENUINO-UNO": {"sampling_rate": 250, "Num_channels": 6}, 
     "UNO-R4": {"sampling_rate": 500, "Num_channels": 6},
     "RPI-PICO-RP2040": {"sampling_rate": 500, "Num_channels": 3},
-    "NANO-CLONE": {"sampling_rate": 250, "Num_channels": 8},          # Baud Rate 115200
+    "NANO-CLONE": {"sampling_rate": 250, "Num_channels": 8},
+    "NANO-CLASSIC": {"sampling_rate": 250, "Num_channels": 8},
+    "STM32F4-BLACK-PILL": {"sampling_rate": 500, "Num_channels": 8},
+    "STM32G4-CORE-BOARD": {"sampling_rate": 500, "Num_channels": 16},
+    "MEGA-2560-R3": {"sampling_rate": 250, "Num_channels": 16},
+    "MEGA-2560-CLONE": {"sampling_rate": 250, "Num_channels": 16},
+    "GIGA-R1": {"sampling_rate": 500, "Num_channels": 6},
 }
 
 # Initialize gloabal variables for Incoming Data
@@ -80,14 +87,18 @@ def connect_hardware(port, baudrate, timeout=1):
         ser = serial.Serial(port, baudrate=baudrate, timeout=timeout)  # Try opening the port
         response = None
         retry_counter = 0
-        while response is None or retry_counter < retry_limit:
+        while response not in supported_boards and retry_counter < retry_limit:
             ser.write(b'WHORU\n') # Check board type
-            response = ser.readline().strip().decode()  # Try reading from the port
+            try:
+                response = ser.readline().strip().decode()  # Attempt to decode the response
+            except UnicodeDecodeError as e:
+                print(f"Decode error: {e}. Ignoring this response.")
+                response = None
             retry_counter += 1
             if response in supported_boards:  # If response is received, assume it's the Arduino
                 global board, sampling_rate, data, num_channels, packet_length
                 board = response # Set board type
-                print(f"{response} detected at {port}")  # Notify the user
+                print(f"{response} detected at {port} with baudrate {baudrate}.")  # Notify the user
                 sampling_rate = supported_boards[board]["sampling_rate"]
                 num_channels = supported_boards[board]["Num_channels"]
                 packet_length = (2 * num_channels) + HEADER_LENGTH + 1
@@ -97,17 +108,19 @@ def connect_hardware(port, baudrate, timeout=1):
         ser.close()  # Close the port if no response
     except (OSError, serial.SerialException):  # Handle exceptions if the port can't be opened
         pass
-    print("Unable to connect to any hardware!")  # Notify if no Arduino is found
+    print(f"Unable to connect to any hardware at baudrate {baudrate}")  # Notify if no Arduino is found
     return None  # Return None if not found
 
-# Function to automatically detect the Arduino's serial port
-def detect_hardware(baudrate, timeout=1):
+def detect_hardware(baudrate=None, timeout=1):
     ports = serial.tools.list_ports.comports()  # List available serial ports
     ser = None
+    baudrates = [baudrate] if baudrate else [230400, 115200]
     for port in ports:  # Iterate through each port
-        ser = connect_hardware(port.device, baudrate)
-        if ser is not None:
-            return ser
+        for baud_rate in baudrates:  # Iterate through all baud rates
+            print(f"Trying {port.device} at Baudrate {baud_rate}...")
+            ser = connect_hardware(port.device, baud_rate, timeout)
+            if ser is not None:
+                return ser
     print("Unable to detect hardware!")  # Notify if no Arduino is found
     return None  # Return None if not found
 
@@ -157,7 +170,7 @@ def read_arduino_data(ser, csv_writer=None, inverted=False):
                 total_packet_count += 1  # Increment total packet count for the current second
                 cumulative_packet_count += 1  # Increment cumulative packet count for the last 10 minutes
 
-                # Extract channel data (6 channels, 2 bytes per channel)
+                # Extract channel data (num_channels, 2 bytes per channel)
                 channel_data = []
                 for channel in range(num_channels):  # Loop through channel data bytes
                     high_byte = packet[2*channel + HEADER_LENGTH]
@@ -241,7 +254,7 @@ def parse_data(ser, lsl_flag=False, csv_flag=False, verbose=False, run_time=None
             csv_writer.writerow([f"Arduino Board: {board}"])
             csv_writer.writerow([f"Sampling Rate (samples per second): {supported_boards[board]['sampling_rate']}"])
             csv_writer.writerow([])  # Blank row for separation
-            csv_writer.writerow(['Counter', 'Channel1', 'Channel2', 'Channel3', 'Channel4', 'Channel5', 'Channel6'])  # Write header
+            csv_writer.writerow(['Counter'] + [f'Channel{i+1}' for i in range(num_channels)])  # Write header
 
         end_time = time.time() + run_time if run_time else None
         send_command(ser, 'START')
@@ -319,7 +332,7 @@ def main():
     global verbose,ser
     parser = argparse.ArgumentParser(description="Upside Down Labs - Chords-Python Tool",allow_abbrev = False)  # Create argument parser
     parser.add_argument('-p', '--port', type=str, help="Specify the COM port")  # Port argument
-    parser.add_argument('-b', '--baudrate', type=int, default=230400, help="Set baud rate for the serial communication")  # Baud rate 
+    parser.add_argument('-b', '--baudrate', type=int, help="Set baud rate for the serial communication")  # Baud rate 
     parser.add_argument('--csv', action='store_true', help="Create and write to a CSV file")  # CSV logging flag
     parser.add_argument('--lsl', action='store_true', help="Start LSL stream")  # LSL streaming flag
     parser.add_argument('-v', '--verbose', action='store_true', help="Enable verbose output with statistical data")  # Verbose flag

ffteeg.py

@@ -27,7 +27,7 @@ def __init__(self):
         self.eeg_plot_widget.showGrid(x=True, y=True)
         self.eeg_plot_widget.setLabel('bottom', 'EEG Plot')
         self.eeg_plot_widget.setYRange(-5000, 5000, padding=0)
-        self.eeg_plot_widget.setXRange(0, 2, padding=0)
+        self.eeg_plot_widget.setXRange(0, 4, padding=0)
         self.eeg_plot_widget.setMouseEnabled(x=False, y=True)  # Disable zoom
         self.main_layout.addWidget(self.eeg_plot_widget)
 
@@ -39,9 +39,9 @@ def __init__(self):
         self.fft_plot.setBackground('w')
         self.fft_plot.showGrid(x=True, y=True)
         self.fft_plot.setLabel('bottom', 'FFT')
-        # self.fft_plot.setYRange(0, 25000, padding=0)
+        # self.fft_plot.setYRange(0, 500, padding=0)
         self.fft_plot.setXRange(0, 50, padding=0)  # Set x-axis to 0 to 50 Hz
-        self.fft_plot.setMouseEnabled(x=False, y=False)  # Disable zoom
+        # self.fft_plot.setMouseEnabled(x=False, y=False)  # Disable zoom
         self.fft_plot.setAutoVisible(y=True)  # Allow y-axis to autoscale
         self.bottom_layout.addWidget(self.fft_plot)
 
@@ -74,13 +74,8 @@ def __init__(self):
         print(f"Sampling rate: {self.sampling_rate} Hz")
 
         # Data and Buffers
-        self.one_second_buffer = deque(maxlen=self.sampling_rate)  # 1-second buffer
-        self.buffer_size = self.sampling_rate * 10
-        self.moving_window_size = self.sampling_rate * 2  # 2-second window
-
-        self.eeg_data = np.zeros(self.buffer_size)
-        self.time_data = np.linspace(0, 10, self.buffer_size)
-        self.current_index = 0
+        self.eeg_data = deque(maxlen=500)       # Initialize moving window with 500 samples
+        self.moving_window = deque(maxlen=500)  # 500 samples for FFT and power calculation (sliding window)
 
         self.b_notch, self.a_notch = iirnotch(50, 30, self.sampling_rate)
         self.b_band, self.a_band = butter(4, [0.5 / (self.sampling_rate / 2), 48.0 / (self.sampling_rate / 2)], btype='band')
@@ -93,7 +88,7 @@ def __init__(self):
         self.timer.timeout.connect(self.update_plot)
         self.timer.start(20) 
 
-        self.eeg_curve = self.eeg_plot_widget.plot(self.time_data, self.eeg_data, pen=pg.mkPen('b', width=1))  #EEG Colour is blue
+        self.eeg_curve = self.eeg_plot_widget.plot(pen=pg.mkPen('b', width=1))
         self.fft_curve = self.fft_plot.plot(pen=pg.mkPen('r', width=1))  # FFT Colour is red
 
     def update_plot(self):
@@ -106,32 +101,26 @@ def update_plot(self):
                 band_filtered, self.zi_band = lfilter(self.b_band, self.a_band, notch_filtered, zi=self.zi_band)
                 band_filtered = band_filtered[-1]  # Get the current filtered point
 
-                # Update the EEG plot
-                self.eeg_data[self.current_index] = band_filtered
-                self.current_index = (self.current_index + 1) % self.buffer_size
+                # Update EEG data buffer
+                self.eeg_data.append(band_filtered)
 
-                if self.current_index == 0:
-                    plot_data = self.eeg_data
+                if len(self.moving_window) < 500:
+                    self.moving_window.append(band_filtered)
                 else:
-                    plot_data = np.concatenate((self.eeg_data[self.current_index:], self.eeg_data[:self.current_index]))
+                    self.process_fft_and_brainpower()
 
-                recent_data = plot_data[-self.moving_window_size:]
-                recent_time = np.linspace(0, len(recent_data) / self.sampling_rate, len(recent_data))
-                self.eeg_curve.setData(recent_time, recent_data)
+                    self.moving_window = deque(list(self.moving_window)[50:] + [band_filtered], maxlen=500)
 
-                self.one_second_buffer.append(band_filtered)           # Add the filtered point to the 1-second buffer
-                if len(self.one_second_buffer) == self.sampling_rate:  # Process FFT and brainwave power
-                    self.process_fft_and_brainpower()
-                    self.one_second_buffer.clear() 
-                    
-    def process_fft_and_brainpower(self):
-        window = np.hanning(len(self.one_second_buffer))       # Apply Hanning window to the buffer
-        buffer_windowed = np.array(self.one_second_buffer) * window
+            plot_data = np.array(self.eeg_data)
+            time_axis = np.linspace(0, 4, len(plot_data))
+            self.eeg_curve.setData(time_axis, plot_data)
 
-        # Perform FFT
-        fft_result = np.abs(fft(buffer_windowed))[:len(buffer_windowed) // 2]
+    def process_fft_and_brainpower(self):
+        window = np.hanning(len(self.moving_window))
+        buffer_windowed = np.array(self.moving_window) * window
+        fft_result = np.abs(np.fft.rfft(buffer_windowed))
         fft_result /= len(buffer_windowed)
-        freqs = np.fft.fftfreq(len(buffer_windowed), 1 / self.sampling_rate)[:len(buffer_windowed) // 2]
+        freqs = np.fft.rfftfreq(len(buffer_windowed), 1 / self.sampling_rate)
         self.fft_curve.setData(freqs, fft_result)
 
         brainwave_power = self.calculate_brainwave_power(fft_result, freqs)

templates/index.html

@@ -67,7 +67,7 @@ <h1>Chords-Python Applications</h1>
                     <button type="submit" name="app_name" value="gui" 
                             class="{% if 'gui' in running_apps %}running{% else %}not-running{% endif %}" 
                             {% if not lsl_started %}disabled{% endif %}>
-                        GUI of 6 Channels
+                        GUI of Channels
                     </button>
                     <button type="submit" name="app_name" value="keystroke" 
                             class="{% if 'keystroke' in running_apps %}running{% else %}not-running{% endif %}"