refactor: Adapt scripts to work with new data structure

Signed-off-by: Gilbert Tanner <gilberttanner.work@gmail.com>

Author: TannerGilbert

.gitignore

@@ -160,4 +160,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
-.vscode/
+.vscode/
+recordings/

README.md

@@ -42,21 +42,25 @@ Data collection can be done using the [`record_data.py` script](uMyo_python_tool
 
 Usage:
 ```bash
-usage: record_data.py [-h] [--output OUTPUT] [--num_sensors NUM_SENSORS] [-p PORT]
+usage: record_data.py [-h] [--output_folder OUTPUT_FOLDER] [--subject SUBJECT] [--gesture GESTURE] [--num_sensors NUM_SENSORS]
+                      [-p PORT]
 
 Record data from uMyo
 
 options:
   -h, --help            show this help message and exit
-  --output OUTPUT       Output file to save sensor data
+  --output_folder OUTPUT_FOLDER
+                        Output file to save sensor data
+  --subject SUBJECT     Name of the subject
+  --gesture GESTURE     Name of the gesture
   --num_sensors NUM_SENSORS
                         Number of sensors to record
   -p PORT, --port PORT  USB receiving station port
 ```
 
 Example
 ```bash
-python uMyo_python_tools/record_data.py --num_sensors 5 --output recordings/23_02_25/data/person_open.csv -p /dev/ttyUSB0
+python uMyo_python_tools/record_data.py --num_sensors 5 --output_folder recordings/session-24_02_25  --subject person --gesture fist
 ```
 
 ## Training

requirements.txt

@@ -7,3 +7,4 @@ scikit_learn
 scipy
 torch
 datasets
+huggingface-hub

training/scripts/gesture-recognition.ipynb

@@ -94,7 +94,6 @@
     "GESTURES_TYPE = parameters.GESTURES_TYPE\n",
     "\n",
     "DATA_DIR = \"../../recordings/\"\n",
-    "SENSOR_PLACEMENT = 0\n",
     "\n",
     "lf = parameters.LF\n",
     "hf = parameters.HF\n",
@@ -203,7 +202,7 @@
     "        df = dataset['raw'].to_pandas()\n",
     "        df.columns = range(df.shape[1])\n",
     "        df_fft = dataset['fft'].to_pandas()\n",
-    "        df.columns = range(df.shape[1])\n",
+    "        df_fft.columns = range(df.shape[1])\n",
     "        dfs[GESTURES.index(recording[\"gesture\"])].append(df)\n",
     "        if USE_FFT:\n",
     "            fft_dfs[GESTURES.index(recording[\"gesture\"])].append(df_fft)"
@@ -220,25 +219,25 @@
     "    return len(df.columns) >= num_sensors * NUM_READINGS\n",
     "\n",
     "def load_local_dataset(dfs=dfs, fft_dfs=fft_dfs):\n",
-    "    # structure: recordings/date/data/*.csv\n",
     "    recordings = []\n",
-    "    baseline_file = None\n",
     "    \n",
-    "    pattern = re.compile(rf\"({'|'.join(TEST_SUBJECTS)})_({'|'.join(GESTURES)})\\d*_{SENSOR_PLACEMENT}.csv\")\n",
+    "    pattern = re.compile(rf\"({'|'.join(TEST_SUBJECTS)})_({'|'.join(GESTURES)}).csv\")\n",
     "    \n",
     "    for date_dir, _, filenames in os.walk(DATA_DIR):\n",
-    "        if date_dir.endswith('data') and any(day in date_dir for day in DAYS) and not 'static' in date_dir:\n",
+    "        if \"raw\" in date_dir and any(day in date_dir for day in DAYS):\n",
     "            for filename in filenames:\n",
     "                if pattern.match(filename):\n",
     "                    gesture = next((gesture for gesture in GESTURES if gesture in filename), None)\n",
-    "                                    \n",
+    "\n",
     "                    emg_data_file = os.path.join(date_dir, filename)\n",
+    "                    fft_data_file = os.path.join(date_dir.replace('raw', 'fft'), f'fft_{filename}')\n",
+    "                    imu_data_file = os.path.join(date_dir.replace('raw', 'imu'), f'imu_{filename}')\n",
     "                    if _check_sensors_present(emg_data_file):\n",
     "                        recordings.append({\n",
     "                            \"gesture\": gesture,\n",
     "                            \"raw_data_filepath\": emg_data_file,\n",
-    "                            \"imu_data_filepath\": os.path.join(date_dir, f'imu_{filename}') if os.path.exists(os.path.join(date_dir, f'imu_{filename}')) else None,\n",
-    "                            'fft_data_filepath': os.path.join(date_dir, f'fft_{filename}') if os.path.exists(os.path.join(date_dir, f'fft_{filename}')) else None,\n",
+    "                            \"imu_data_filepath\": imu_data_file if os.path.exists(imu_data_file) else None,\n",
+    "                            'fft_data_filepath': fft_data_file if os.path.exists(fft_data_file) else None,\n",
     "                        })\n",
     "                    else:\n",
     "                        print(f\"File {emg_data_file} has the wrong number of sensors. Should be {NUM_SENSORS}. Skipping.\")\n",
@@ -265,8 +264,22 @@
    "source": [
     "if DATASET_SOURCE == parameters.DATASET.HUGGING_FACE:\n",
     "    load_remote_dataset()\n",
+    "elif DATASET_SOURCE == parameters.DATASET.LOCAL:\n",
+    "    load_local_dataset()\n",
     "else:\n",
-    "    load_local_dataset()"
+    "    load_remote_dataset()\n",
+    "    dfs_remote = dfs\n",
+    "    fft_dfs_remote = fft_dfs\n",
+    "    \n",
+    "    # Reset global variables\n",
+    "    dfs = [[] for _ in range(NUM_CLASSES)]\n",
+    "    fft_dfs = [[] for _ in range(NUM_CLASSES)]\n",
+    "    \n",
+    "    load_local_dataset()\n",
+    "    # Combine\n",
+    "    for i in range(NUM_CLASSES):\n",
+    "        dfs[i] = dfs_remote[i] + dfs[i]\n",
+    "        fft_dfs[i] = fft_dfs_remote[i] + fft_dfs[i]"
    ]
   },
   {
@@ -1882,7 +1895,7 @@
    "sourceType": "notebook"
   },
   "kernelspec": {
-   "display_name": "base",
+   "display_name": "Python 3",
    "language": "python",
    "name": "python3"
   },
@@ -1896,7 +1909,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.12.3"
+   "version": "3.11.4"
   }
  },
  "nbformat": 4,

uMyo_python_tools/offline_classifier.py

@@ -18,48 +18,34 @@
 USE_FFT = True
 
 
-def check_sensors_present(file, num_sensors=NUM_SENSORS):
+def _check_sensors_present(file, num_sensors=NUM_SENSORS):
     df = pd.read_csv(file)
     return len(df.columns) >= num_sensors * NUM_READINGS
 
 
-def load_data(data_dir, test_subjects, gestures, window_size, sensor_placement=0):
+def load_data(data_dir, test_subjects, gestures, window_size):
     recordings = []
 
-    pattern = re.compile(
-        rf"({'|'.join(test_subjects)})_({'|'.join(gestures)})\d*_{sensor_placement}.csv"
-    )
-
+    pattern = re.compile(rf"({'|'.join(test_subjects)})_({'|'.join(gestures)}).csv")
+    
     for date_dir, _, filenames in os.walk(data_dir):
-        if date_dir.endswith("data") and not "static" in date_dir:
+        if "raw" in date_dir:
             for filename in filenames:
                 if pattern.match(filename):
-                    gesture = next(
-                        (gesture for gesture in gestures if gesture in filename), None
-                    )
+                    gesture = next((gesture for gesture in gestures if gesture in filename), None)
 
                     emg_data_file = os.path.join(date_dir, filename)
-                    if check_sensors_present(emg_data_file):
-                        recordings.append(
-                            {
-                                "gesture": gesture,
-                                "raw_data_filepath": emg_data_file,
-                                "imu_data_filepath": (
-                                    os.path.join(date_dir, f"imu_{filename}")
-                                    if os.path.exists(
-                                        os.path.join(date_dir, f"imu_{filename}")
-                                    )
-                                    else None
-                                ),
-                                "fft_data_filepath": (
-                                    os.path.join(date_dir, f"fft_{filename}")
-                                    if os.path.exists(
-                                        os.path.join(date_dir, f"fft_{filename}")
-                                    )
-                                    else None
-                                ),
-                            }
-                        )
+                    fft_data_file = os.path.join(date_dir.replace('raw', 'fft'), f'fft_{filename}')
+                    imu_data_file = os.path.join(date_dir.replace('raw', 'imu'), f'imu_{filename}')
+                    if _check_sensors_present(emg_data_file):
+                        recordings.append({
+                            "gesture": gesture,
+                            "raw_data_filepath": emg_data_file,
+                            "imu_data_filepath": imu_data_file if os.path.exists(imu_data_file) else None,
+                            'fft_data_filepath': fft_data_file if os.path.exists(fft_data_file) else None,
+                        })
+                    else:
+                        print(f"File {emg_data_file} has the wrong number of sensors. Should be {NUM_SENSORS}. Skipping.")
 
     dfs = [[] for _ in range(len(gestures))]
     fft_dfs = [[] for _ in range(len(gestures))]

uMyo_python_tools/parameters.py

@@ -16,6 +16,7 @@
 class DATASET(Enum):
     HUGGING_FACE = 1
     LOCAL = 2
+    BOTH = 3
 
 GESTURES_TYPE = ("dynamic_gestures", "static_gestures")
 GESTURES = ("baseline", "fist", "peace", "up", "down", "lift")

uMyo_python_tools/record_data.py

@@ -4,23 +4,32 @@
 
 import serial
 import umyo_parser
-
 from parameters import IDS
 
 MAX_DATA_LAG = 400
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Record data from uMyo")
     parser.add_argument(
-        "--output",
+        "--output_folder",
         type=str,
-        default="sensor_data.csv",
+        default="../recordings",
         help="Output file to save sensor data",
     )
+    parser.add_argument(
+        "--subject",
+        type=str,
+        help="Name of the subject",
+    )
+    parser.add_argument(
+        "--gesture",
+        type=str,
+        help="Name of the gesture",
+    )
     parser.add_argument(
         "--num_sensors",
         type=int,
-        default=4,
+        default=5,
         help="Number of sensors to record",
     )
     parser.add_argument("-p", "--port", type=str, default="COM7", help="USB receiving station port")
@@ -35,9 +44,10 @@
     timeout=0,
     )
 
-    output_path = pathlib.Path(args.output)
-    if output_path.parent.exists() is False:
-        output_path.parent.mkdir(parents=True)
+    output_folder = pathlib.Path(args.output_folder)
+    (output_folder / args.subject / 'raw').mkdir(parents=True, exist_ok=True)
+    (output_folder / args.subject / 'fft').mkdir(parents=True, exist_ok=True)
+    (output_folder / args.subject / 'imu').mkdir(parents=True, exist_ok=True)
 
     recordings = 0
     last_data_ids = [0] * args.num_sensors
@@ -93,11 +103,11 @@
                 flattened_fft = [item for sublist in spectrum_data for item in sublist]
                 flattened_imu = [item for sublist in imu_data for item in sublist]
 
-                with open(output_path, "a") as f:
+                with open(output_folder / args.subject / "raw" / f"{args.subject}_{args.gesture}.csv", "a") as f:
                     f.write(",".join(map(str, flattened_data)) + "\n")
-                with open(output_path.parent / ("fft_" + output_path.name), "a") as f:
+                with open(output_folder / args.subject / "fft" / f"fft_{args.subject}_{args.gesture}.csv", "a") as f:
                     f.write(",".join(map(str, flattened_fft)) + "\n")
-                with open(output_path.parent / ("imu_" + output_path.name), "a") as f:
+                with open(output_folder / args.subject / "imu" / f"imu_{args.subject}_{args.gesture}.csv", "a") as f:
                     f.write(",".join(map(str, flattened_imu)) + "\n")
                 recordings += 1
                 print("Recordings done: ", recordings)

uMyo_python_tools/visualize_data.py

@@ -196,9 +196,9 @@ def preprocess_data(data):
     data = _remove_outliers(data)
     return data
 
-def show_and_get_baseline(folder, subject, placement):
+def show_and_get_baseline(folder, subject):
     global power_noise, power_noise_filtered, baseline
-    file = os.path.join(folder, subject + "_baseline_" + str(placement) + ".csv")
+    file = os.path.join(folder, f"{subject}_baseline.csv")
     df = pd.read_csv(file, header=None)
     sensor1 = np.array(df.iloc[:, :8]).flatten() #* (1/fs)
     sensor1 = sensor1 - np.mean(sensor1)
@@ -248,8 +248,8 @@ def show_and_get_baseline(folder, subject, placement):
     plt.subplots_adjust(hspace=0.5)
     return np.vstack([sensor1, sensor2, sensor3, sensor4])
 
-def show_all_gestures(gestures_files, folder, subject, placement):
-    baseline = show_and_get_baseline(folder, subject, placement) # show neutral gesture for reference
+def show_all_gestures(gestures_files, folder, subject):
+    baseline = show_and_get_baseline(folder, subject) # show neutral gesture for reference
     gestures_data = [[] for _ in range(len(gestures_files))]
 
     for i, file in enumerate(gestures_files):
@@ -292,16 +292,16 @@ def show_all_gestures(gestures_files, folder, subject, placement):
         sensor4 = preprocess_data(gestures_data[i][3])
 
         b1 = baseline[0]
-        b1 = np.append(b1, np.zeros(len(sensor1)-len(b1)))
+        b1 = np.append(b1, np.zeros(len(sensor1)-len(b1))) if len(sensor1) > len(b1) else b1[:len(sensor1)]
         sensor1 -= b1
         b2 = baseline[1]
-        b2 = np.append(b2, np.zeros(len(sensor2)-len(b2)))
+        b2 = np.append(b2, np.zeros(len(sensor2)-len(b2))) if len(sensor2) > len(b2) else b2[:len(sensor2)]
         sensor2 -= b2
         b3 = baseline[2]
-        b3 = np.append(b3, np.zeros(len(sensor3)-len(b3)))
+        b3 = np.append(b3, np.zeros(len(sensor3)-len(b3))) if len(sensor3) > len(b3) else b3[:len(sensor3)]
         sensor3 -= b3
         b4 = baseline[3]
-        b4 = np.append(b4, np.zeros(len(sensor4)-len(b4)))
+        b4 = np.append(b4, np.zeros(len(sensor4)-len(b4))) if len(sensor4) > len(b4) else b4[:len(sensor4)]
         sensor4 -= b4
 
         axs[4].set_title("Filtered sensor 1")
@@ -417,22 +417,18 @@ def show_psd(gesture_files):
     parser = argparse.ArgumentParser(description="Visualize data from uMyo")
 
     parser.add_argument(
-        "--folder", type=str, default="../recordings/01_12_24_initial_placement_test/data", help="Folder with recordings"
+        "--folder", type=str, default="../recordings/session-24_02_25", help="Folder with recordings"
     )
     parser.add_argument(
         "--subject", type=str, default="nad", help="Test subject name"
     )
-    parser.add_argument(
-        "--placement", type=int, default=0, help="Placement version of the sensors"
-    )
 
     args = parser.parse_args()
 
-    folder = args.folder
+    raw_folder = os.path.join(args.folder, args.subject, 'raw')
     subject = args.subject
-    placement = args.placement
     gestures_files = [
-        os.path.join(folder, file) for file in os.listdir(folder) if file.endswith(str(placement)+".csv") and file.startswith(subject)
+        os.path.join(raw_folder, file) for file in os.listdir(raw_folder) if file.endswith(".csv") and file.startswith(subject)
     ]
 
-    show_all_gestures(gestures_files, folder, subject, placement)
+    show_all_gestures(gestures_files, raw_folder, subject)