fix(docs): drop duplicate check/check_report from analysis.utils

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)
 

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')

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)