Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 6 additions & 16 deletions eegnb/analysis/streaming_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down
21 changes: 5 additions & 16 deletions eegnb/analysis/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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

Expand Down
13 changes: 12 additions & 1 deletion eegnb/devices/eeg.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,13 @@

from serial import Serial, EIGHTBITS, PARITY_NONE, STOPBITS_ONE

import pyxid2
# pyxid2 is only needed for Cedrus XID response boxes (nirsport2 here).
# It has C-build issues on some platforms and is not required by the
# common Muse / OpenBCI / Unicorn paths, so make it optional.
try:
import pyxid2
except ImportError:
pyxid2 = None

from eegnb.devices.utils import (
get_openbci_usb,
Expand Down Expand Up @@ -621,6 +627,11 @@ def _serial_open_port(self,PORT_ID="COM4", BAUD=115200):


def _init_xid(self):
if pyxid2 is None:
raise ImportError(
"pyxid2 is required for Cedrus XID response boxes. "
"Install with: pip install pyxid2"
)
if self.xid_num is not None: # if an xis device number is supplied, open and init that device
xids_list = pyxid2.get_xid_devices()
xid = xids_list[self.xid_num]
Expand Down
Empty file added eegnb/utils/__init__.py
Empty file.
38 changes: 38 additions & 0 deletions eegnb/utils/cancel.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
"""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``.

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 sys
import threading


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.
"""
cancel_event = threading.Event()
cancel_key = cancel_key.strip().lower()

def _reader() -> None:
try:
line = sys.stdin.readline()
except (OSError, ValueError):
return
if line and line.strip().lower() == cancel_key:
cancel_event.set()

thread = threading.Thread(target=_reader, daemon=True)
thread.start()
cancel_event.wait(timeout=timeout)
return cancel_event.is_set()
7 changes: 5 additions & 2 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,11 @@ pyo>=1.0.3; platform_system == "Linux"
#pynput requires pyobjc, psychopy requires a version less than 8, setting pyobjc to
# a specific version prevents an endless dependency resolution loop.
pyobjc==7.3; 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
Expand Down
Loading