<jupyter_start><jupyter_text>Tutorial Notebook for rASR implementation It consists in:- loading the eeg signal and events from the *xdf* and *set* files (- visualizing the power spectral density (check dead channels and line noise) - apply a sliding window on both training and test data- train rASR on the training data- apply rASR on test data with blending- compare outcomes with different rASR parameters  Set paths and loading data<jupyter_code>training_files   = '/Users/louis/Dropbox/rASR Data/calibration data/sme_1_1.xdf_filt.set'
test_files       = '/Users/louis/Dropbox/rASR Data/filtered/sme_1_1.xdf_filt.set'
rasr_matlab_file = '/Users/louis/Dropbox/rASR Data/Riemannian ASR out/sme_1_1.set'

channels = ["Fp1", "Fp2", "T8", "Cz", "O1"]

mne_eeg_training = mne.io.read_raw_eeglab(training_files, preload=True).pick_channels(channels)
print(f"training data loaded with {len(mne_eeg_training.info['ch_names'])} channels and {mne_eeg_training.times[-1]-mne_eeg_training.times[0]:.1f} seconds")
mne_eeg_test     = mne.io.read_raw_eeglab(test_files, preload=True).pick_channels(channels)
print(f"test data loaded with {len(mne_eeg_test.info['ch_names'])} channels and {mne_eeg_test.times[-1]-mne_eeg_test.times[0]:.1f} seconds")
print("Files LOADED")
plt.close("all")
plot_args = dict(sfreq=mne_eeg_test.info["sfreq"], ch_names=mne_eeg_test.info["ch_names"], linewidth=0.25, scalings=1e-4)
plotTimeSeries(mne_eeg_test.get_data().T, color="black", **plot_args);<jupyter_output>Reading /Users/louis/Dropbox/rASR Data/calibration data/sme_1_1.xdf_filt.fdt
Reading 0 ... 14999  =      0.000 ...    59.996 secs...
training data loaded with 5 channels and 60.0 seconds
Reading /Users/louis/Dropbox/rASR Data/filtered/sme_1_1.xdf_filt.fdt
Reading 0 ... 247499  =      0.000 ...   989.996 secs...
test data loaded with 5 channels and 990.0 seconds
Files LOADED<jupyter_text>Set RASR parameters, apply epoching and prepare pipelines<jupyter_code>window_len        = 0.5  # in seconds
window_overlap    = 0.9  # in ratio of window_len
rejection_cutoff  = 3    # in std from the estimated distribution

# PREPARE TRAINING AND TEST EPOCHS
window_size     = int(mne_eeg_training.info["sfreq"] * window_len)  # epoch length in samples
window_interval = int(window_size * (1 - window_overlap))           # step interval in samples


# convert training data into epochs
X_training = np.swapaxes(epoch(mne_eeg_training.get_data(), window_size, window_interval, axis=-1), 1, 2)
print(f"training set with epochs of shape {X_training.shape}")

# convert test data into epochs
X_test = np.swapaxes(epoch(mne_eeg_test.get_data(), window_size, window_interval, axis=-1), 1, 2)
print(f"test set with epochs of shape {X_test.shape}")

rASR_pipeline     = make_pipeline(RASR(rejection_cutoff=rejection_cutoff))
blending_pipeline = make_pipeline(Blending(window_overlap=window_size-window_interval, merge=True))
print("Pipelines built")<jupyter_output>training set with epochs of shape (1240, 125, 5)
test set with epochs of shape (20615, 125, 5)
Pipelines built<jupyter_text>Filter the epochs and reconstruct raw<jupyter_code>import time
start = time.time()
rASR_pipeline.fit(X_training)
fitting_time = time.time()-start

start = time.time()
X_test_cleaned = rASR_pipeline.transform(X_test)
data_cleaned = blending_pipeline.fit_transform(X_test_cleaned)
transform_time = time.time()-start

print(f"data cleaned and reconstructed \n fit in {fitting_time*1e3 :.0f}ms\n transform in {transform_time*1e3 :.0f}ms ({transform_time*1e3/X_test.shape[0] :.2f}ms/epoch)")<jupyter_output>data cleaned and reconstructed 
 fit in 657ms
 transform in 4759ms (0.23ms/epoch)<jupyter_text>Visualize comparison<jupyter_code># plot data with zoom effect
def plot_comparison(data, data_cleaned):
    plt.figure()

    ax2 = plt.subplot(212)
    plot_args["scalings"]=5e-5

    plotTimeSeries(data_cleaned, color="red", ax=ax2, **plot_args);
    plotTimeSeries(data, color="black", ax=ax2, **plot_args);
    
    ax1 = plt.subplot(211)

    plotTimeSeries(data_cleaned, color="red", ax=ax1, **plot_args);
    plotTimeSeries(data, color="black", ax=ax1, **plot_args);
    ax1.set_xlim(207,215)

    zoom_effect(ax1,ax2)
    # add a widget to scroll to the signal (update xlim)
    from ipywidgets import interact, FloatSlider
    def update_axis(xmin, xmax):
        if xmin<xmax:
            ax1.set_xlim(xmin,xmax)

    Tmax=data.shape[0] / mne_eeg_test.info["sfreq"]
    print(Tmax)
    i=FloatSlider(min=0, max=Tmax, step=10, continuous_update=False)
    ii=FloatSlider(min=0, max=Tmax, step=10, continuous_update=False)
    from ipywidgets import FloatSlider
    interact(update_axis,xmin=i, xmax=ii);
from utils.viz import zoom_effect
data = blending_pipeline.fit_transform(X_test)  # reconstruct test data for comparison (not cleaned)
plt.close("all")
plot_comparison(data, data_cleaned)
plt.title("before (black), RASR python (red)");

data_rasr_matlab     = mne.io.read_raw_eeglab(rasr_matlab_file, preload=True).pick_channels(channels).get_data()[:, :data_cleaned.shape[0]].T
print(data_rasr_matlab.shape)
plt.close("all")
plot_comparison(data_rasr_matlab, data_cleaned)
plt.title("RASR matlab (black), RASR python (red)");<jupyter_output>Reading /Users/louis/Dropbox/rASR Data/Riemannian ASR out/sme_1_1.fdt
Reading 0 ... 247499  =      0.000 ...   989.996 secs...
(247493, 5)
{'fraction_non_alphanumeric': 0.118384140639611, 'fraction_numerical': 0.03235459135964092, 'mean_word_length': 5.374255065554231, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '433745', 'license': 'MIT', 'n_tokens_mistral': 2080, 'n_tokens_neox': 1966, 'n_words': 425, 'path': 'notebooks/RASR_tutorial.ipynb', 'repo_name': 'timeflux/timeflux_rasr', 'size': '5347'}