Merge pull request #39 from PayalLakra/bio_amptool

Add code comments for better readability

Author: lorforlinux

app.py

@@ -357,8 +357,8 @@ def main():
     else:
         ser = detect_hardware(baudrate=args.baudrate)
         if ser is None:
-             sys.stderr.write("No\n")
-             sys.exit(1)  # Exit with a non-zero code to indicate failure
+            sys.stderr.write("Serial Connection not established properly.Try Again\n")
+            sys.exit(1)  # Exit with a non-zero code to indicate failure
     if ser is None:
         print("Arduino port not specified or detected. Exiting.")  # Notify if no port is available
         return

chords.py

@@ -1,3 +1,4 @@
 numpy==2.1.3
 pylsl==1.16.2
-pyserial==3.5
+pyserial==3.5
+bleak==0.22.3

chords_requirements.txt

@@ -4,12 +4,12 @@
 import plotly.graph_objects as go
 
 class CSVPlotterApp:
-    def __init__(self, root):
+    def __init__(self, root):                 # Initialize the main application window
         self.root = root
         self.root.title("CSV Plotter GUI")
-        self.filename = None
-        self.data = None
-        self.create_widgets()
+        self.filename = None                  # Variable to store the selected CSV file name
+        self.data = None                      # Variable to store the loaded data
+        self.create_widgets()                 # Call the method to create widgets
 
     def create_widgets(self):
         # Create a frame for buttons and file name display
@@ -41,19 +41,19 @@ def create_widgets(self):
         self.plot_button.pack(pady=10)
 
     def load_csv(self):
-        self.filename = filedialog.askopenfilename(filetypes=[("CSV files", "*.csv")])
+        self.filename = filedialog.askopenfilename(filetypes=[("CSV files", "*.csv")])   # Open file dialog to select CSV file
         if self.filename:
             try:
-                with open(self.filename, "r", encoding="utf-8") as f:
-                    lines = f.readlines()
+                with open(self.filename, "r", encoding="utf-8") as f:                    # Open the selected CSV file
+                    lines = f.readlines()                                                # Read all lines into a list for header detection
 
-                header_index = None        # Find the row where 'Counter' appears
-                for i, line in enumerate(lines):
+                header_index = None                                                      # Initialize the variable to track where (Header)'Counter' appears
+                for i, line in enumerate(lines):                                         # Iterate through lines to find the header line
                     if "Counter" in line:
-                        header_index = i
-                        break
+                        header_index = i                                                 # Store the index of the line containing 'Counter'
+                        break                                                            # If found, break the loop
 
-                if header_index is None:
+                if header_index is None:                                                 # If no header row with 'Counter' was found, show error and exit
                     messagebox.showerror("Error", "CSV file must contain a 'Counter' column.")
                     return
 
@@ -73,7 +73,7 @@ def load_csv(self):
                 messagebox.showerror("Error", f"Could not load CSV file: {e}")
 
     def setup_dropdown_menu(self):
-        # Get available channel columns (Channel1 to Channel6)
+        # Get available channel columns
         channel_columns = [col for col in self.data.columns if 'Channel' in col]
 
         # Populate dropdown menu with available channels
@@ -82,6 +82,7 @@ def setup_dropdown_menu(self):
             self.channel_selection.set(channel_columns[0])  # Default selection to the first channel
 
     def plot_data(self):
+        """Creates an interactive plot of the selected data channel using Plotly."""
         selected_channel = self.channel_selection.get()   # Get the selected channel
         if not selected_channel:
             messagebox.showerror("Error", "No channel selected for plotting")
@@ -96,9 +97,9 @@ def plot_data(self):
             yaxis_title="Value",
             template="plotly_white"
         )
-        fig.show()
+        fig.show()               # Display the plot in a new window
 
 if __name__ == "__main__":
-    root = tk.Tk()
-    app = CSVPlotterApp(root)
-    root.mainloop()
+    root = tk.Tk()               # Create the main Tkinter root window
+    app = CSVPlotterApp(root)    # Create an instance of the CSVPlotterApp class
+    root.mainloop()              # Start the Tkinter main loop

csvplotter.py

@@ -30,8 +30,8 @@ def plot_lsl_data():
     global inlet, num_channels, data
 
     print("Searching for available LSL streams...")
-    streams = resolve_streams()
-    available_streams = [s.name() for s in streams]
+    streams = resolve_streams()                         # Discover available LSL streams
+    available_streams = [s.name() for s in streams]     # Get list of stream names
 
     if not available_streams:
         print("No LSL streams found!")
@@ -43,7 +43,7 @@ def plot_lsl_data():
 
         if resolved_streams:
             print(f"Successfully connected to {stream_name}!")
-            inlet = StreamInlet(resolved_streams[0])
+            inlet = StreamInlet(resolved_streams[0])           # Create an inlet to receive data from the stream
             break
         else:
             print(f"Failed to connect to {stream_name}.")

gui.py

@@ -15,19 +15,21 @@ def __init__(self):
         self.setWindowTitle("Real-Time ECG Monitor")  # Set up GUI window
         self.setGeometry(100, 100, 800, 600)
 
-        self.plot_widget = PlotWidget(self)
-        self.plot_widget.setBackground('w')
-        self.plot_widget.showGrid(x=True, y=True)
+        self.plot_widget = PlotWidget(self)           # Create the plotting widget
+        self.plot_widget.setBackground('w')           # Set background color to white
+        self.plot_widget.showGrid(x=True, y=True)     # Show grid lines
 
         # Heart rate label at the bottom
         self.heart_rate_label = QLabel(self)
         self.heart_rate_label.setStyleSheet("font-size: 20px; font-weight: bold; color: black;")
         self.heart_rate_label.setAlignment(Qt.AlignCenter)
 
+        # Layout setup - vertical layout for plot and label
         layout = QVBoxLayout()
         layout.addWidget(self.plot_widget)
         layout.addWidget(self.heart_rate_label)
 
+        # Set the central widget that holds all other widgets
         central_widget = QWidget()
         central_widget.setLayout(layout)
         self.setCentralWidget(central_widget)
@@ -36,6 +38,7 @@ def __init__(self):
         print("Searching for available LSL streams...")
         available_streams = pylsl.resolve_streams()
 
+        # Exit if no streams are found
         if not available_streams:
             print("No LSL streams found! Exiting...")
             sys.exit(0)
@@ -53,21 +56,21 @@ def __init__(self):
             print("Unable to connect to any LSL stream! Exiting...")
             sys.exit(0)
 
-        # Sampling rate
+        # Get Sampling rate from the stream info.
         self.sampling_rate = int(self.inlet.info().nominal_srate())
         print(f"Sampling rate: {self.sampling_rate} Hz")
 
         # Data and buffers
         self.buffer_size = self.sampling_rate * 10  # Fixed-size buffer for 10 seconds
         self.ecg_data = np.zeros(self.buffer_size)  # Fixed-size array for circular buffer
         self.time_data = np.linspace(0, 10, self.buffer_size)  # Fixed time array for plotting
-        self.r_peaks = []  # Store the indices of R-peaks
-        self.heart_rate = None
+        self.r_peaks = []       # Store the indices of R-peaks
+        self.heart_rate = None  # Initialize heart rate variable
         self.current_index = 0  # Index for overwriting data
 
         self.b, self.a = butter(4, 20.0 / (0.5 * self.sampling_rate), btype='low')   # Low-pass filter coefficients
 
-        self.timer = pg.QtCore.QTimer()   # Timer for updating the plot
+        self.timer = pg.QtCore.QTimer()   # Timer for updating the plot (every 10 ms)
         self.timer.timeout.connect(self.update_plot)
         self.timer.start(10)
 
@@ -86,7 +89,7 @@ def __init__(self):
         self.moving_average_window_size = 5   # Initialize moving average buffer
         self.heart_rate_history = []          # Buffer to store heart rates for moving average
 
-        # Connect double-click event
+        # Connect double-click event for zoom reset
         self.plot_widget.scene().sigMouseClicked.connect(self.on_double_click)
 
     def on_double_click(self, event):
@@ -139,11 +142,11 @@ def calculate_heart_rate(self):
                 # Update heart rate label with moving average & convert into int
                 self.heart_rate_label.setText(f"Heart Rate: {int(moving_average_hr)} BPM")
         else:
-            self.heart_rate_label.setText("Heart Rate: Calculating...") 
+            self.heart_rate_label.setText("Heart Rate: Calculating...")   # Display message if not enough R-peaks detected
 
     def plot_r_peaks(self, filtered_ecg):
-        r_peak_times = self.time_data[self.r_peaks]   # Extract the time of detected R-peaks
-        r_peak_values = filtered_ecg[self.r_peaks]
+        r_peak_times = self.time_data[self.r_peaks]             # Extract the time of detected R-peaks
+        r_peak_values = filtered_ecg[self.r_peaks]              # Get corresponding ECG values
         self.r_peak_curve.setData(r_peak_times, r_peak_values)  # Plot R-peaks as red dots
 
 if __name__ == "__main__":

heartbeat_ecg.py

@@ -9,23 +9,24 @@
 
 class EOGPeakDetector:
     def __init__(self, blink_button, keystroke_action, connect_button):
-        self.inlet = None
-        self.sampling_rate = None
-        self.buffer_size = None
-        self.eog_data = None
-        self.current_index = 0
-        self.b, self.a = None, None
-        self.blink_button = blink_button
-        self.keystroke_action = keystroke_action
-        self.connect_button = connect_button
-        self.blink_detected = False
-        self.running = False
-        self.connected = False
-        self.last_blink_time = 0
-        self.refractory_period = 0.2
-        self.stop_threads = False
+        self.inlet = None                         # LSL inlet for receiving data
+        self.sampling_rate = None                 # Sampling rate of the data stream
+        self.buffer_size = None                   # Size of the buffer for storing EOG data
+        self.eog_data = None                      # Buffer for EOG data
+        self.current_index = 0                    # Current index in the buffer
+        self.b, self.a = None, None               # Filter coefficients for low-pass filter
+        self.blink_button = blink_button          # Button to trigger blink action
+        self.keystroke_action = keystroke_action  # Action to perform on blink detection
+        self.connect_button = connect_button      # Button to connect to LSL stream
+        self.blink_detected = False               # Flag to indicate if a blink has been detected
+        self.running = False                      # Flag to control the detection loop
+        self.connected = False                    # Flag to indicate if connected to LSL stream
+        self.last_blink_time = 0                  # Last time a blink was detected
+        self.refractory_period = 0.2              # Refractory period to prevent multiple eye blink detections
+        self.stop_threads = False                 # Flag to stop threads
 
     def initialize_stream(self):
+        """Initialize the LSL stream connection and set up the buffer and filter coefficients."""
         print("Searching for available LSL streams...")
         available_streams = pylsl.resolve_streams()
 
@@ -42,10 +43,10 @@ def initialize_stream(self):
                 print(f"Sampling rate: {self.sampling_rate} Hz")
 
                 # Set buffer size and filter coefficients
-                self.buffer_size = self.sampling_rate * 1
-                self.eog_data = np.zeros(self.buffer_size)
-                self.b, self.a = butter(4, 10.0 / (0.5 * self.sampling_rate), btype='low')
-                self.connected = True
+                self.buffer_size = self.sampling_rate * 1      # Buffer size for 1 second of data
+                self.eog_data = np.zeros(self.buffer_size)     # Initialize buffer for EOG data
+                self.b, self.a = butter(4, 10.0 / (0.5 * self.sampling_rate), btype='low')  # Low-pass filter coefficients
+                self.connected = True                          # Set connected flag to True(LSL Stream connected)
                 print("LSL stream connected successfully.")
                 return True  # Stop trying after first successful connection
 
@@ -57,27 +58,31 @@ def initialize_stream(self):
         return False
 
     def start_detection(self):
+        """Start the peak detection process"""
         print("Starting peak detection...")
-        self.running = True
+        self.running = True    # Flag to control the detection loop
         while self.running:
             try:
                 samples, _ = self.inlet.pull_chunk(timeout=1.0, max_samples=1)
                 if samples:
                     for sample in samples:
-                        self.eog_data[self.current_index] = sample[0]
-                        self.current_index = (self.current_index + 1) % self.buffer_size
+                        self.eog_data[self.current_index] = sample[0]                     # Store sample in circular buffer at current position
+                        self.current_index = (self.current_index + 1) % self.buffer_size  # Update index with wrap-around using modulo
 
                     filtered_eog = lfilter(self.b, self.a, self.eog_data)
-                    self.detect_blinks(filtered_eog)
+                    self.detect_blinks(filtered_eog)             # Run blink detection on the filtered signal
             except Exception as e:
                 print(f"Error in detection: {e}")
                 break
 
     def stop_detection(self):
+        """Stop the peak detection process"""
         print("Stopping peak detection...")
-        self.running = False
+        self.running = False               # Set running flag to False to stop the detection loop
 
     def detect_blinks(self, filtered_eog):
+        """Detect blinks in the filtered EOG signal using a threshold-based method."""
+        # Calculate dynamic threshold based on signal statistics
         mean_signal = np.mean(filtered_eog)
         stdev_signal = np.std(filtered_eog)
         threshold = mean_signal + (1.7 * stdev_signal)
@@ -87,8 +92,8 @@ def detect_blinks(self, filtered_eog):
         if start_index < 0:
             start_index = 0
 
-        filtered_window = filtered_eog[start_index:self.current_index]
-        peaks = self.detect_peaks(filtered_window, threshold)
+        filtered_window = filtered_eog[start_index:self.current_index]  # Get the current window of filtered EOG data
+        peaks = self.detect_peaks(filtered_window, threshold)           # Detect peaks above threshold in the current window
 
         current_time = time.time()
         if peaks and (current_time - self.last_blink_time > self.refractory_period):
@@ -105,6 +110,7 @@ def detect_peaks(self, signal, threshold):
         return peaks
 
     def trigger_action(self):
+        """Trigger the keystroke action when a blink is detected."""
         if not self.blink_detected:
             self.blink_detected = True
             print("Triggering action...")
@@ -121,17 +127,20 @@ def update_button_color(self):
         self.blink_button.after(100, lambda: self.blink_button.config(bg="SystemButtonFace"))
 
 def quit_action(detector):
+    """Handle the quit action for the GUI."""
     print("Quit button pressed. Exiting program.")
     detector.stop_threads = True
     detector.stop_detection()
     popup.quit()
     popup.destroy()
 
 def keystroke_action():
+    """Perform the keystroke action (press spacebar)."""
     print("Spacebar pressed!")
     pyautogui.press('space')
 
 def connect_start_stop_action(detector, connect_button):
+    """Handle the connect/start/stop action for the GUI."""
     if not detector.connected:
         print("Connect button pressed. Starting connection in a new thread.")
         threading.Thread(target=connect_to_stream, args=(detector, connect_button), daemon=True).start()
@@ -152,6 +161,7 @@ def connect_to_stream(detector, connect_button):
         print("Failed to connect to LSL stream.")
 
 def create_popup():
+    """Create the popup window for the EOG keystroke emulator."""
     global popup
     popup = tk.Tk()
     popup.geometry("300x120")
@@ -196,8 +206,8 @@ def move(event):
 
     detector = EOGPeakDetector(blink_button, keystroke_action, connect_button)
 
-    connect_button.config(command=lambda: connect_start_stop_action(detector, connect_button))
-    quit_button.config(command=lambda: quit_action(detector))
+    connect_button.config(command=lambda: connect_start_stop_action(detector, connect_button))   # Connect/Start/Stop action
+    quit_button.config(command=lambda: quit_action(detector))    # Quit action
 
     popup.mainloop()