Drop pynput dependency to fix Linux installs

eegnb/analysis/streaming_utils.py

@@ -7,9 +7,10 @@
 from glob import glob
 from typing import Union, List
 from time import sleep, time
-from pynput import keyboard
 import os
 
+from eegnb.utils.cancel import wait_for_cancel
+
 import pandas as pd
 import numpy as np
 import matplotlib.pyplot as plt
@@ -192,21 +193,10 @@ def check_report(eeg: EEG, n_times: int=60, pause_time=5, thres_std_low=None, th
         if (loop_index+1) % n_inarow == 0:
             print(f"\n\nLooks like you still have {len(bad_channels)} bad channels after {loop_index+1} tries\n")
 
-            prompt_time = time()
-            print(f"Starting next cycle in 5 seconds, press C and enter to cancel")
-            c_key_pressed = False
-
-            def update_key_press(key):
-                if key.char == 'c':
-                    globals().update(c_key_pressed=True)
-            listener = keyboard.Listener(on_press=update_key_press)
-            listener.start()
-            while time() < prompt_time + 5:
-                if c_key_pressed:
-                    print("\nStopping signal quality checks!")
-                    flag = True
-                    break
-            listener.stop()
+            print("Starting next cycle in 5 seconds, press C and enter to cancel")
+            if wait_for_cancel(timeout=5.0, cancel_key="c"):
+                print("\nStopping signal quality checks!")
+                flag = True
         if flag: 
             break  
 

eegnb/analysis/utils.py

@@ -25,7 +25,7 @@
 from eegnb import _get_recording_dir
 from eegnb.devices.eeg import EEG
 from eegnb.devices.utils import EEG_INDICES, SAMPLE_FREQS
-from pynput import keyboard
+from eegnb.utils.cancel import wait_for_cancel
 
 # this should probably not be done here
 sns.set_context("talk")
@@ -529,21 +529,10 @@ def check_report(eeg: EEG, n_times: int=60, pause_time=5, thres_std_low=None, th
         if (loop_index+1) % n_inarow == 0:
             print(f"\n\nLooks like you still have {len(bad_channels)} bad channels after {loop_index+1} tries\n")
 
-            prompt_time = time()
-            print(f"Starting next cycle in 5 seconds, press C and enter to cancel")
-            c_key_pressed = False
-
-            def update_key_press(key):
-                if key.char == 'c':
-                    globals().update(c_key_pressed=True)
-            listener = keyboard.Listener(on_press=update_key_press)
-            listener.start()
-            while time() < prompt_time + 5:  
-                if c_key_pressed:
-                    print("\nStopping signal quality checks!")
-                    flag = True
-                    break
-            listener.stop()
+            print("Starting next cycle in 5 seconds, press C and enter to cancel")
+            if wait_for_cancel(timeout=5.0, cancel_key="c"):
+                print("\nStopping signal quality checks!")
+                flag = True
         if flag: 
             break
 

eegnb/utils/cancel.py

@@ -0,0 +1,69 @@
+"""Cross-platform stdin-based cancel prompt.
+
+Replaces ``pynput.keyboard.Listener`` for the simple case of "give the
+user N seconds to press a key + Enter to cancel an operation". Uses a
+daemon thread reading from stdin so it works on Linux / macOS / Windows
+and in terminals without a ``DISPLAY`` (e.g. headless rigs over SSH, or CI).
+
+pynput was dropped because it pulls in evdev (Linux) which currently
+fails to build from source under several common toolchains.
+"""
+
+from __future__ import annotations
+
+import queue
+import sys
+import threading
+import time
+
+# stdin.readline() has no timeout, so one daemon thread reads lines into this
+# queue and each prompt waits on the queue with its own deadline.
+_lines: queue.Queue[str] = queue.Queue()
+_reader_started = threading.Event()
+
+
+def _ensure_reader() -> None:
+    if _reader_started.is_set():
+        return
+    _reader_started.set()
+
+    def _pump() -> None:
+        while True:
+            try:
+                line = sys.stdin.readline()
+            except (OSError, ValueError):
+                return
+            if line == "":  # EOF
+                return
+            _lines.put(line)
+
+    threading.Thread(target=_pump, name="eegnb-stdin-cancel", daemon=True).start()
+
+
+def wait_for_cancel(timeout: float, cancel_key: str = "c") -> bool:
+    """Block for up to ``timeout`` seconds waiting for the user to type
+    ``cancel_key`` + Enter on stdin.
+
+    Returns True if cancel was requested, False if the timeout elapsed.
+    """
+    _ensure_reader()
+    key = cancel_key.strip().lower()
+
+    # discard input typed before this prompt
+    while True:
+        try:
+            _lines.get_nowait()
+        except queue.Empty:
+            break
+
+    deadline = time.monotonic() + timeout
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            return False
+        try:
+            line = _lines.get(timeout=remaining)
+        except queue.Empty:
+            return False
+        if line.strip().lower() == key:
+            return True

requirements.txt

@@ -33,8 +33,11 @@ pyserial>=3.5
 h5py>=3.1.0
 pytest-shutil
 pyobjc>=8.0; sys_platform == 'darwin'
-#Removed keyboard dependency due segmentation fault on Apple Silicon: https://github.com/boppreh/keyboard/issues/507
-pynput
+#Removed pynput dependency: it pulls in evdev, which fails to build from
+#source on modern Linux under the conda toolchain (missing kernel header
+#symbols). The two callsites that used pynput.keyboard.Listener have been
+#replaced with a threading + stdin "press enter to cancel" helper in
+#eegnb/utils/cancel.py — cross-platform, zero new deps.
 airium>=0.1.0
 attrdict>=2.0.1
 attrdict3