diff --git a/eegnb/analysis/streaming_utils.py b/eegnb/analysis/streaming_utils.py index bd7d301f..07fbfd9f 100644 --- a/eegnb/analysis/streaming_utils.py +++ b/eegnb/analysis/streaming_utils.py @@ -85,7 +85,7 @@ def check(eeg: EEG, n_samples=256) -> pd.Series: ------ from eegnb.devices.eeg import EEG - from eegnb.analysis.utils import check + from eegnb.analysis.streaming_utils import check eeg = EEG(device='museS') check(eeg, n_samples=256) @@ -123,7 +123,7 @@ def check_report(eeg: EEG, n_times: int=60, pause_time=5, thres_std_low=None, th Usage: ------ from eegnb.devices.eeg import EEG - from eegnb.analysis.utils import check_report + from eegnb.analysis.streaming_utils import check_report eeg = EEG(device='museS') check_report(eeg) diff --git a/eegnb/analysis/utils.py b/eegnb/analysis/utils.py index ea42613b..0f87b13b 100644 --- a/eegnb/analysis/utils.py +++ b/eegnb/analysis/utils.py @@ -6,7 +6,7 @@ from collections import OrderedDict from glob import glob from typing import Union, List -from time import sleep, time +from time import time import os import pandas as pd @@ -23,9 +23,7 @@ from scipy.signal import lfilter, lfilter_zi from eegnb import _get_recording_dir -from eegnb.devices.eeg import EEG from eegnb.devices.utils import EEG_INDICES, SAMPLE_FREQS -from eegnb.utils.cancel import wait_for_cancel # this should probably not be done here sns.set_context("talk") @@ -418,124 +416,6 @@ def channel_filter( return filt_samples -def check(eeg: EEG, n_samples=256) -> pd.Series: - """ - Usage: - ------ - - from eegnb.devices.eeg import EEG - from eegnb.analysis.utils import check - eeg = EEG(device='museS') - check(eeg, n_samples=256) - - """ - - df = eeg.get_recent(n_samples=n_samples) - - # seems to be necessary to give brainflow cnxn time to settle - if len(df) != n_samples: - sleep(10) - df = eeg.get_recent(n_samples=n_samples) - - assert len(df) == n_samples - - n_channels = eeg.n_channels - sfreq = eeg.sfreq - device_backend = eeg.backend - device_name = eeg.device_name - - vals = df.values[:, :n_channels] - df.values[:, :n_channels] = channel_filter(vals, - n_channels, - sfreq, - device_backend, - device_name) - - std_series = df.std(axis=0) - - return std_series - - - -def check_report(eeg: EEG, n_times: int=60, pause_time=5, thres_std_low=None, thres_std_high=None, n_goods=2,n_inarow=5): - """ - Usage: - ------ - from eegnb.devices.eeg import EEG - standard deviation for a quality recording. - - thresholds = { - standard deviation for a quality recording. - - thresholds = { - bad: 15, - good: 10, - great: 1.5 // Below 1 usually indicates not connected to anything - } - """ - - # If no upper and lower std thresholds set in function call, - # set thresholds based on the following per-device name defaults - edn = eeg.device_name - flag = False - if thres_std_low is None: - if edn in thres_stds.keys(): - thres_std_low = thres_stds[edn][0] - if thres_std_high is None: - if edn in thres_stds.keys(): - thres_std_high = thres_stds[edn][1] - - print("\n\nRunning signal quality check...") - print(f"Accepting threshold stdev between: {thres_std_low} - {thres_std_high}") - - CHECKMARK = "√" - CROSS = "x" - - print(f"running check (up to) {n_times} times, with {pause_time}-second windows") - print(f"will stop after {n_goods} good check results in a row") - - good_count=0 - - n_samples = int(pause_time*eeg.sfreq) - - sleep(5) - - for loop_index in range(n_times): - print(f'\n\n\n{loop_index+1}/{n_times}') - std_series = check(eeg, n_samples=n_samples) - - indicators = "\n".join( - [ - f" {k:>4}: {CHECKMARK if v >= thres_std_low and v <= thres_std_high else CROSS} (std: {round(v, 1):>5})" - for k, v in std_series.items() - ] - ) - print("\nSignal quality:") - print(indicators) - - bad_channels = [k for k, v in std_series.items() if v < thres_std_low or v > thres_std_high ] - if bad_channels: - print(f"Bad channels: {', '.join(bad_channels)}") - good_count=0 # reset good checks count if there are any bad chans - else: - print('No bad channels') - good_count+=1 - - if good_count==n_goods: - print("\n\n\nAll good! You can proceed on to data collection :) ") - break - - # after every n_inarow trials ask user if they want to cancel or continue - 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") - - 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 - def fix_musemissinglines(orig_f,new_f=''): #if new_f == '': new_f = orig_f.replace('.csv', '_fml.csv') diff --git a/eegnb/cli/__main__.py b/eegnb/cli/__main__.py index 8c5c938f..465d929c 100644 --- a/eegnb/cli/__main__.py +++ b/eegnb/cli/__main__.py @@ -8,7 +8,7 @@ from .utils import run_experiment from eegnb import generate_save_fn from eegnb.devices.eeg import EEG -from eegnb.analysis.utils import check_report +from eegnb.analysis.streaming_utils import check_report from eegnb.analysis.pipelines import load_eeg_data, make_erp_plot, analysis_report, example_analysis_report from typing import Optional @@ -139,7 +139,7 @@ def checksigqual(eegdevice: str): """ from eegnb.devices.eeg import EEG - from eegnb.analysis.utils import check_report + from eegnb.analysis.streaming_utils import check_report eeg = EEG(device=eegdevice)