Store Metadata as Values

.gitignore

@@ -45,4 +45,20 @@ test_*.py
 *.csv
 .vscode/*
 test_delsys_api.py
-resources/
+resources/
+*.csv 
+*.txt
+ContinuousTransitions/*
+FORS-EMG/*
+MyoDisCo/*
+NinaProDB1/*
+*.zip
+libemg/_datasets/__pycache__/*
+CIILData/*
+EMGEPN612.pkl
+OneSubjectMyoDataset/
+_3DCDataset/
+ContractionIntensity/
+CIILData/
+*.pkl
+LimbPosition/

dataset_tryout.py

@@ -0,0 +1,4 @@
+from libemg.datasets import *
+
+accs = evaluate('LDA', 300, 100, feature_list=['MAV','SSC','ZC','WL'], included_datasets=['FougnerLP'], save_dir='')
+print('\n' + str(accs))

libemg/_datasets/_3DC.py

@@ -0,0 +1,45 @@
+from libemg._datasets.dataset import Dataset
+from libemg.data_handler import OfflineDataHandler, RegexFilter
+
+class _3DCDataset(Dataset):
+    def __init__(self, dataset_folder="_3DCDataset/"):
+        Dataset.__init__(self, 
+                        1000, 
+                        10, 
+                        '3DC Armband (Prototype)', 
+                        22, 
+                        {0: "Neutral", 1: "Radial Deviation", 2: "Wrist Flexion", 3: "Ulnar Deviation", 4: "Wrist Extension", 5: "Supination", 6: "Pronation", 7: "Power Grip", 8: "Open Hand", 9: "Chuck Grip", 10: "Pinch Grip"}, 
+                        '8 (4 Train, 4 Test)',
+                        "The 3DC dataset including 11 classes.",
+                        "https://doi.org/10.3389/fbioe.2020.00158")
+        self.url = "https://github.com/libemg/3DCDataset"
+        self.dataset_folder = dataset_folder
+
+    def prepare_data(self, split = False, subjects_values = None, sets_values = None, reps_values = None,
+                     classes_values = None):
+        if subjects_values is None:
+            subjects_values = [str(i) for i in range(1,23)]
+        if sets_values is None:
+            sets_values = ["train", "test"]
+        if reps_values is None:
+            reps_values = ["0","1","2","3"]
+        if classes_values is None:
+            classes_values = [str(i) for i in range(11)]
+
+        print('\nPlease cite: ' + self.citation+'\n')
+        if (not self.check_exists(self.dataset_folder)):
+            self.download(self.url, self.dataset_folder)
+
+        regex_filters = [
+            RegexFilter(left_bound = "/", right_bound="/EMG", values = sets_values, description='sets'),
+            RegexFilter(left_bound = "_", right_bound=".txt", values = classes_values, description='classes'),
+            RegexFilter(left_bound = "EMG_gesture_", right_bound="_", values = reps_values, description='reps'),
+            RegexFilter(left_bound="Participant", right_bound="/",values=subjects_values, description='subjects')
+        ]
+        odh = OfflineDataHandler()
+        odh.get_data(folder_location=self.dataset_folder, regex_filters=regex_filters, delimiter=",")
+        data = odh
+        if split:
+            data = {'All': odh, 'Train': odh.isolate_data("sets", [0], fast=True), 'Test': odh.isolate_data("sets", [1], fast=True)}
+
+        return data

libemg/_datasets/__init__.py

@@ -0,0 +1,17 @@
+from libemg._datasets import _3DC
+from libemg._datasets import ciil
+from libemg._datasets import continous_transitions
+from libemg._datasets import dataset
+from libemg._datasets import emg_epn612
+from libemg._datasets import fors_emg
+from libemg._datasets import fougner_lp
+from libemg._datasets import grab_myo
+from libemg._datasets import hyser
+from libemg._datasets import intensity
+from libemg._datasets import kaufmann_md
+from libemg._datasets import myodisco
+from libemg._datasets import nina_pro
+from libemg._datasets import one_subject_emager
+from libemg._datasets import one_subject_myo
+from libemg._datasets import radmand_lp
+from libemg._datasets import tmr_shirleyryanabilitylab

libemg/_datasets/ciil.py

@@ -0,0 +1,148 @@
+from libemg._datasets.dataset import Dataset
+from libemg.data_handler import OfflineDataHandler, RegexFilter, FilePackager
+from pathlib import Path
+
+
+
+class CIIL_MinimalData(Dataset):
+    def __init__(self, dataset_folder='CIILData/'):
+        Dataset.__init__(self, 
+                         200, 
+                         8, 
+                         'Myo Armband', 
+                         11, 
+                         {0: 'Close', 1: 'Open', 2: 'Rest', 3: 'Flexion', 4: 'Extension'}, 
+                         '1 Train (1s), 15 Test',
+                         "The goal of this Myo dataset is to explore how well models perform when they have a limited amount of training data (1s per class).", 
+                         'https://ieeexplore.ieee.org/abstract/document/10394393')
+        self.url = "https://github.com/LibEMG/CIILData"
+        self.dataset_folder = dataset_folder
+
+    def prepare_data(self, split = False):
+        print('\nPlease cite: ' + self.citation+'\n')
+        if (not self.check_exists(self.dataset_folder)):
+            self.download(self.url, self.dataset_folder)
+
+        subfolder = 'MinimalTrainingData'
+        subjects = [str(i) for i in range(0, 11)]
+        classes_values = [str(i) for i in range(0,5)]
+        reps_values = ["0","1","2"]
+        sets = ["train", "test"]
+        regex_filters = [
+            RegexFilter(left_bound = "/", right_bound="/", values = sets, description='sets'),
+            RegexFilter(left_bound = "/subject", right_bound="/", values = subjects, description='subjects'),
+            RegexFilter(left_bound = "R_", right_bound="_", values = reps_values, description='reps'),
+            RegexFilter(left_bound = "C_", right_bound=".csv", values = classes_values, description='classes')
+        ]
+        odh = OfflineDataHandler()
+        odh.get_data(folder_location=self.dataset_folder + '/' + subfolder, regex_filters=regex_filters, delimiter=",")
+
+        data = odh
+        if split:
+            data = {'All': odh, 'Train': odh.isolate_data("sets", [0], fast=True), 'Test': odh.isolate_data("sets", [1], fast=True)}
+
+        return data
+
+class CIIL_ElectrodeShift(Dataset):
+    def __init__(self, dataset_folder='CIILData/'):
+        Dataset.__init__(self, 
+                         200, 
+                         8, 
+                         'Myo Armband', 
+                         21, 
+                         {0: 'Close', 1: 'Open', 2: 'Rest', 3: 'Flexion', 4: 'Extension'}, 
+                         '5 Train (Before Shift), 8 Test (After Shift)',
+                         "An electrode shift confounding factors dataset.", 
+                         'https://link.springer.com/article/10.1186/s12984-024-01355-4')
+        self.url = "https://github.com/LibEMG/CIILData"
+        self.dataset_folder = dataset_folder
+
+    def prepare_data(self, split = False):
+        print('\nPlease cite: ' + self.citation+'\n')
+        if (not self.check_exists(self.dataset_folder)):
+            self.download(self.url, self.dataset_folder)
+
+        subfolder = 'ElectrodeShift'
+        subjects = [str(i) for i in range(0, 21)]
+        classes_values = [str(i) for i in range(0,5)]
+        reps_values = ["0","1","2","3","4"]
+        sets = ["training", "trial_1", "trial_2", "trial_3", "trial_4"]
+        regex_filters = [
+            RegexFilter(left_bound = "/", right_bound="/", values = sets, description='sets'),
+            RegexFilter(left_bound = "/subject", right_bound="/", values = subjects, description='subjects'),
+            RegexFilter(left_bound = "R_", right_bound="_", values = reps_values, description='reps'),
+            RegexFilter(left_bound = "C_", right_bound=".csv", values = classes_values, description='classes')
+        ]
+        odh = OfflineDataHandler()
+        odh.get_data(folder_location=self.dataset_folder + '/' + subfolder, regex_filters=regex_filters, delimiter=",")
+
+        data = odh
+        if split:
+            data = {'All': odh, 'Train': odh.isolate_data("sets", [0], fast=True), 'Test': odh.isolate_data("sets", [1,2,3,4], fast=True)}
+
+        return data
+
+
+class CIIL_WeaklySupervised(Dataset):
+    def __init__(self, dataset_folder='CIIL_WeaklySupervised/'):
+        Dataset.__init__(self, 
+                         1000, 
+                         8, 
+                         'OyMotion gForcePro+ EMG Armband', 
+                         16, 
+                         {0: 'Close', 1: 'Open', 2: 'Rest', 3: 'Flexion', 4: 'Extension'}, 
+                         '30 min weakly supervised, 1 rep calibration, 14 reps test',
+                         "A weakly supervised environment with sparse supervised calibration.", 
+                         'In Submission')
+        self.url = "https://unbcloud-my.sharepoint.com/:u:/g/personal/ecampbe2_unb_ca/EaABHYybhfJNslTVcvwPPwgB9WwqlTLCStui30maqY53kw?e=MbboMd"
+        self.dataset_folder = dataset_folder
+
+    def prepare_data(self, split = False):
+        print('\nPlease cite: ' + self.citation+'\n')
+        if (not self.check_exists(self.dataset_folder)):
+            self.download_via_onedrive(self.url, self.dataset_folder)
+
+        # supervised odh loading
+        subjects = [str(i) for i in range(0, 16)]
+        classes_values = [str(i) for i in range(0,5)]
+        reps_values = [str(i) for i in range(0,15)]
+        setting_values     = [".csv", ""] # this is arbitrary to get a field that separates WS from S
+        regex_filters = [
+            RegexFilter(left_bound = "", right_bound="", values = setting_values, description='settings'),
+            RegexFilter(left_bound = "/S", right_bound="/", values = subjects, description='subjects'),
+            RegexFilter(left_bound = "R", right_bound=".csv", values = reps_values, description='reps'),
+            RegexFilter(left_bound = "C", right_bound="_R", values = classes_values, description='classes')
+        ]
+        odh_s = OfflineDataHandler()
+        odh_s.get_data(folder_location=self.dataset_folder+"CIIL_WeaklySupervised/",
+                       regex_filters=regex_filters,
+                       delimiter=",")
+
+        # weakly supervised odh loading
+        subjects = [str(i) for i in range(0, 16)]
+        reps_values      = [str(i) for i in range(3)]
+        setting_values     = ["", ".csv"] # this is arbitrary to get a field that separates WS from S
+        regex_filters = [
+            RegexFilter(left_bound = "", right_bound="", values = setting_values, description='settings'),
+            RegexFilter(left_bound = "/S", right_bound="/", values = subjects, description='subjects'),
+            RegexFilter(left_bound = "WS", right_bound=".csv", values = reps_values, description='reps'),
+        ]
+        metadata_fetchers = [
+            FilePackager(regex_filter=RegexFilter(left_bound="", right_bound="targets.csv", values=["_"], description="classes"),
+            package_function=lambda x, y: (x.split("WS")[1][0] == y.split("WS")[1][0]) and (Path(x).parent == Path(y).parent)
+            )
+        ]
+        odh_ws = OfflineDataHandler()
+        odh_ws.get_data(folder_location=self.dataset_folder+"CIIL_WeaklySupervised/", 
+                     regex_filters=regex_filters, 
+                     metadata_fetchers=metadata_fetchers,
+                     delimiter=",")
+
+        data = odh_s + odh_ws
+        if split:
+            data = {'All': data, 
+                    'Pretrain': odh_ws,
+                    'Train': odh_s.isolate_data("reps", [0], fast=True), 
+                    'Test': odh_s.isolate_data("reps", list(range(1,15)), fast=True)}
+
+        return data

libemg/_datasets/continous_transitions.py

@@ -0,0 +1,63 @@
+from libemg._datasets.dataset import Dataset
+from libemg.data_handler import OfflineDataHandler
+import h5py
+import numpy as np
+
+class ContinuousTransitions(Dataset):
+    def __init__(self, dataset_folder="ContinuousTransitions/"):
+        Dataset.__init__(self, 
+                        2000, 
+                        6, 
+                        'Delsys', 
+                        43, 
+                        {0: 'No Motion', 1: 'Wrist Flexion', 2: 'Wrist Extension', 3: 'Wrist Pronation', 4: 'Wrist Supination', 5: 'Hand Close', 6: 'Hand Open'}, 
+                        '6 Training (Ramp), 42 Transitions (All combinations of Transitions) x 6 Reps',
+                        "The testing set in this dataset has continuous transitions between classes which is a more realistic offline evaluation standard for myoelectric control.",
+                        "https://ieeexplore.ieee.org/document/10254242")
+        self.dataset_folder = dataset_folder
+
+    def prepare_data(self, split = False):
+        print('\nPlease cite: ' + self.citation+'\n')
+        if (not self.check_exists(self.dataset_folder)):
+            print("Please download the dataset from: ") #TODO: Update
+            return 
+
+        # Training ODH
+        odh_tr = OfflineDataHandler()
+        odh_tr.subjects = []
+        odh_tr.classes = []
+        odh_tr.extra_attributes = ['subjects', 'classes']
+
+        # Testing ODH
+        odh_te = OfflineDataHandler()
+        odh_te.subjects = []
+        odh_te.classes = []
+        odh_te.extra_attributes = ['subjects', 'classes']
+
+        for s_i, s in enumerate([2,3,4,5,6,7,8,9,10,11,12,13,14,15,17,18,19,20,21,22,23,25,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47]):
+            data = h5py.File('ContinuousTransitions/P' + f"{s:02}" + '.hdf5', "r")
+            cont_labels = data['continuous']['emg']['prompt'][()]
+            cont_labels = np.hstack([np.ones((1000)) * cont_labels[0], cont_labels[0:len(cont_labels)-1000]]) # Rolling about 0.5s as per Shri's suggestion
+            cont_emg = data['continuous']['emg']['signal'][()]
+            cont_chg_idxs = np.insert(np.where(cont_labels[:-1] != cont_labels[1:])[0], 0, -1)
+            cont_chg_idxs = np.insert(cont_chg_idxs, len(cont_chg_idxs), len(cont_emg))
+            for i in range(0, len(cont_chg_idxs)-1):
+                odh_te.data.append(cont_emg[cont_chg_idxs[i]+1:cont_chg_idxs[i+1]])
+                odh_te.classes.append(np.expand_dims(cont_labels[cont_chg_idxs[i]+1:cont_chg_idxs[i+1]]-1, axis=1))
+                odh_te.subjects.append(np.ones((len(odh_te.data[-1]), 1)) * s_i) 
+
+            ramp_emg = data['ramp']['emg']['signal'][()]
+            ramp_labels = data['ramp']['emg']['prompt'][()]
+            r_chg_idxs = np.insert(np.where(ramp_labels[:-1] != ramp_labels[1:])[0], 0, -1)
+            r_chg_idxs = np.insert(r_chg_idxs, len(r_chg_idxs), len(ramp_emg))
+            for i in range(0, len(r_chg_idxs)-1):
+                odh_tr.data.append(ramp_emg[r_chg_idxs[i]+1:r_chg_idxs[i+1]])
+                odh_tr.classes.append(np.expand_dims(ramp_labels[r_chg_idxs[i]+1:r_chg_idxs[i+1]]-1, axis=1))
+                odh_tr.subjects.append(np.ones((len(odh_tr.data[-1]), 1)) * s_i) 
+
+        odh_all = odh_tr + odh_te
+        data = odh_all
+        if split:
+            data = {'All': odh_all, 'Train': odh_tr, 'Test': odh_te}
+
+        return data

libemg/_datasets/dataset.py

@@ -0,0 +1,55 @@
+import os
+from libemg.data_handler import OfflineDataHandler
+from onedrivedownloader import download as onedrive_download
+# this assumes you have git downloaded (not pygit, but the command line program git)
+
+class Dataset:
+    def __init__(self, sampling, num_channels, recording_device, num_subjects, gestures, num_reps, description, citation):
+        # Every class should have this 
+        self.sampling=sampling
+        self.num_channels=num_channels 
+        self.recording_device=recording_device
+        self.num_subjects=num_subjects
+        self.gestures=gestures
+        self.num_reps=num_reps
+        self.description=description
+        self.citation=citation
+
+    def download(self, url, dataset_name):
+        clone_command = "git clone " + url + " " + dataset_name
+        os.system(clone_command)
+
+    def download_via_onedrive(self, url, dataset_name, unzip=True, clean=True):
+        onedrive_download(url=url,
+                          filename = dataset_name,
+                          unzip=unzip,
+                          clean=clean)
+
+    def remove_dataset(self, dataset_folder):
+        remove_command = "rm -rf " + dataset_folder
+        os.system(remove_command)
+
+    def check_exists(self, dataset_folder):
+        return os.path.exists(dataset_folder)
+
+    def prepare_data(self, split = False):
+        pass
+
+    def get_info(self):
+        print(str(self.description) + '\n' + 'Sampling Rate: ' + str(self.sampling) + '\nNumber of Channels: ' + str(self.num_channels) + 
+              '\nDevice: ' + self.recording_device + '\nGestures: ' + str(self.gestures) + '\nNumber of Reps: ' + str(self.num_reps) + '\nNumber of Subjects: ' + str(self.num_subjects) +
+              '\nCitation: ' + str(self.citation))
+
+# given a directory, return a list of files in that directory matching a format
+# can be nested
+# this is just a handly utility
+def find_all_files_of_type_recursively(dir, terminator):
+    files = os.listdir(dir)
+    file_list = []
+    for file in files:
+        if file.endswith(terminator):
+            file_list.append(dir+file)
+        else:
+            if os.path.isdir(dir+file):
+                file_list += find_all_files_of_type_recursively(dir+file+'/',terminator)
+    return file_list