Adding comments to keystroke.py

Author: PayalLakra

keystroke.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()