diff --git a/.gitignore b/.gitignore index 11b7c50..605bcfe 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ *.asv *.fig -*.mat \ No newline at end of file +*.mat +__pycache__/ \ No newline at end of file diff --git a/ML/YES_281124.flb b/ML/YES_281124.flb new file mode 100644 index 0000000..a3874ff Binary files /dev/null and b/ML/YES_281124.flb differ diff --git a/ML/utils/README.md b/ML/utils/README.md new file mode 100644 index 0000000..2d448e6 --- /dev/null +++ b/ML/utils/README.md @@ -0,0 +1,126 @@ +# flb_utils + +Portable Python utilities for reading REKLAB `.flb` binary files, exporting +trials to CSV, and generating signal plots. Drop the `utils/` folder into any +project that needs to process `.flb` recordings. + +--- + +## Contents + +| File | Description | +|------|-------------| +| `flb_reader.py` | Low-level binary parser — `read_flb()` and `flb_to_csv()` | +| `main.py` | CLI entry point: reads a `.flb` file, exports CSVs, saves plots | +| `requirements.txt` | Python package dependencies | + +--- + +## Requirements + +Python 3.8+ is required. + +```bash +pip install -r requirements.txt +``` + +| Package | Purpose | +|---------|---------| +| `numpy` | Array operations and binary parsing | +| `pandas` | Trial data as DataFrames, CSV export | +| `matplotlib` | Signal plots | + +--- + +## Quick start + +```bash +# Minimal usage — exports all CSVs + plots trial 1 + overlay +python main.py YES_281124.flb + +# Plot a specific trial (e.g. trial 11) +python main.py YES_281124.flb --trial 11 + +# Plot trial 11, skip CSV export +python main.py YES_281124.flb --trial 11 --no_csv + +# Export CSVs only, skip all plots +python main.py YES_281124.flb --no_plots + +# Custom output directory +python main.py YES_281124.flb --output_dir ./results + +# IES01 subject (different raw channel order) +python main.py YES_281124.flb --subject_id IES01 --trial 5 + +# Reduce overlay plot density (default downsample = 10) +python main.py YES_281124.flb --downsample 20 --no_csv +``` + +### All options + +| Option | Default | Description | +|--------|---------|-------------| +| `flb_file` | *(required)* | Path to the `.flb` file | +| `--trial N` | `1` | Trial number to plot (1-based) | +| `--output_dir DIR` | `_output/` | Root output folder | +| `--subject_id ID` | `default` | Use `IES01` for that subject's channel order | +| `--downsample N` | `10` | Keep every N-th sample in the overlay plot | +| `--no_csv` | off | Skip CSV export | +| `--no_plots` | off | Skip plot generation | + +--- + +## Output structure + +``` +/ + csv/ + _trial001.csv + _trial002.csv + ... + plots/ + trial01_all_channels.png ← all 6 channels for Trial 1 + all_trials_overlay.png ← all trials overlaid per channel +``` + +--- + +## Using as a library + +```python +from flb_reader import read_flb, flb_to_csv + +# Read trials into a list of DataFrames +trials = read_flb('recording.flb') +df = trials[0] # columns: time, pos, tq, gm, gl, sol, ta +print(df.head()) + +# Export all trials to CSV +csv_paths = flb_to_csv('recording.flb', output_dir='./csv') +``` + +--- + +## Channel layout + +| Column | Signal | Units | +|--------|--------|-------| +| `time` | Time axis | s | +| `pos` | Ankle angle | rad | +| `tq` | Torque | Nm | +| `gm` | Gastrocnemius Medial EMG | mV | +| `gl` | Gastrocnemius Lateral EMG | mV | +| `sol` | Soleus EMG | mV | +| `ta` | Tibialis Anterior EMG | mV | + +> **Note:** Subject `IES01` has a different raw channel order. Pass +> `--subject_id IES01` (CLI) or `subject_id='IES01'` (API) to remap correctly. + +--- + +## FLB format reference + +The `.flb` format was reverse-engineered from `flbio.m` / `flb2mat.m` +(R. Kearney, McGill University). Supported versions: 2, 3, and 4. +See `flb_reader.py` for full format documentation. diff --git a/ML/utils/__init__.py b/ML/utils/__init__.py new file mode 100644 index 0000000..30ac437 --- /dev/null +++ b/ML/utils/__init__.py @@ -0,0 +1,2 @@ +# utils package – FLB file reader and signal processing helpers +from .flb_reader import read_flb, flb_to_csv diff --git a/ML/utils/flb_reader.py b/ML/utils/flb_reader.py new file mode 100644 index 0000000..706ab8f --- /dev/null +++ b/ML/utils/flb_reader.py @@ -0,0 +1,265 @@ +""" +flb_reader.py +------------- +Reads REKLAB .flb binary files and converts each trial to a pandas DataFrame. +Optionally exports all trials to CSV files. + +FLB binary format (reverse-engineered from flbio.m / flb2mat.m by R. Kearney, McGill): + Each trial block: + [int32] flb_version (2, 3, or 4) + [int32] nDim + [int32] nReal (number of realizations, usually 1) + [int32] nChan (number of channels, usually 6) + [int32] chanLen (number of samples per channel) + [int32] chanFormat (2 = int16, 4 = float32) + [int32] stringLen + [char×stringLen] domainName + [float32] domainIncr (sampling interval in seconds) + [float32] domainStart + [int32] commentLen + [char×commentLen] comment (or 'DEFAULT' if commentLen == 0) + for each channel: + [int32] nameLen + [char×nameLen] chanName (or 'DEFAULT' if nameLen == 0) + [float64 × nChan] chanMin + [float64 × nChan] chanMax + [float32 or int16 × (chanLen × nChan × nReal)] raw data + → reshaped to (chanLen, nChan, nReal), column-major order + +Channel layout (all subjects except IES01): + Ch1 = POS — ankle angle (rad) + Ch2 = TQ — torque (Nm) + Ch3 = GM — Gastrocnemius Medial EMG (mV) + Ch4 = GL — Gastrocnemius Lateral EMG (mV) + Ch5 = SOL — Soleus EMG (mV) + Ch6 = TA — Tibialis Anterior EMG (mV) +""" + +import struct +import os +import numpy as np +import pandas as pd + + +# ── Channel name map (MATLAB is 1-indexed, Python 0-indexed) ───────────── +DEFAULT_CHANNEL_NAMES = ['pos', 'tq', 'gm', 'gl', 'sol', 'ta'] + +# IES01 subject has a different channel order in the raw file +IES01_CHANNEL_ORDER = [0, 1, 3, 5, 4, 2] # reorder to: pos, tq, gm, gl, sol, ta + + +# ── Low-level binary helpers ────────────────────────────────────────────── + +def _read_int32(f): + raw = f.read(4) + if len(raw) < 4: + return None + return struct.unpack('= 6: + data = data[:, IES01_CHANNEL_ORDER] + + # Use channel names from the file header if available, else defaults + file_chan_names = h.get('chanName', []) + if file_chan_names and all(c not in ('DEFAULT', '') for c in file_chan_names): + col_names = [c.lower().replace(' ', '_') for c in file_chan_names] + else: + col_names = channel_names + + n_ch = min(data.shape[1], len(col_names)) + df = pd.DataFrame(data[:, :n_ch], columns=col_names[:n_ch]) + df.insert(0, 'time', time) + + # Attach metadata as DataFrame attributes + df.attrs['trial_index'] = trial_idx + df.attrs['domainIncr'] = Ts + df.attrs['domainStart'] = h['domainStart'] + df.attrs['comment'] = h['comment'] + df.attrs['chanName'] = h['chanName'] + df.attrs['source_file'] = os.path.basename(filepath) + + trials.append(df) + trial_idx += 1 + + print(f"Read {len(trials)} trials from '{os.path.basename(filepath)}'") + return trials + + +def flb_to_csv(filepath, output_dir=None, subject_id='default'): + """ + Convert all trials in an FLB file to individual CSV files. + + Parameters + ---------- + filepath : str + Path to the .flb file. + output_dir : str, optional + Directory to write CSV files. Defaults to a folder named after the + FLB file (without extension) next to the FLB file. + subject_id : str + Subject identifier (affects channel ordering for IES01). + + Returns + ------- + list of str + Paths to the written CSV files. + """ + base_name = os.path.splitext(os.path.basename(filepath))[0] + + if output_dir is None: + output_dir = os.path.join(os.path.dirname(filepath), base_name + '_csv') + + os.makedirs(output_dir, exist_ok=True) + + trials = read_flb(filepath, subject_id=subject_id) + csv_paths = [] + + for i, df in enumerate(trials): + csv_name = f"{base_name}_trial{i+1:03d}.csv" + csv_path = os.path.join(output_dir, csv_name) + df.to_csv(csv_path, index=False) + csv_paths.append(csv_path) + + print(f"Saved {len(csv_paths)} CSV files to '{output_dir}'") + return csv_paths diff --git a/ML/utils/main.py b/ML/utils/main.py new file mode 100644 index 0000000..0cc3953 --- /dev/null +++ b/ML/utils/main.py @@ -0,0 +1,285 @@ +""" +main.py +------- +Entry point: reads a REKLAB .flb file, exports every trial to CSV, and +generates signal plots. + +Usage +----- + python main.py [options] + +Options +------- + --output_dir DIR Root output directory (default: _output/) + --subject_id ID Subject ID; use 'IES01' for that channel order + (default: default) + --downsample N Downsample factor for the overlay plot (default: 10) + --no_csv Skip CSV export + --no_plots Skip plot generation + +Output structure +---------------- + / + csv/ + _trial001.csv + _trial002.csv + ... + plots/ + trial01_all_channels.png — all 6 channels for Trial 1 + all_trials_overlay.png — all trials overlaid per channel +""" + +import os +import argparse +import numpy as np +import matplotlib +matplotlib.use('Agg') # non-interactive backend; works without a display +import matplotlib.colors as mcolors +import matplotlib.pyplot as plt + +from flb_reader import read_flb, flb_to_csv + + +# ── Plotting helpers ────────────────────────────────────────────────────── + +# Per-channel display names and units (two separate strings for two-line y-label) +CHANNEL_META = { + 'pos': ('position', 'rad'), + 'tq': ('torque', 'Nm'), + 'gm': ('gm_emg', 'mV'), + 'gl': ('gl_emg', 'mV'), + 'sol': ('sol_emg', 'mV'), + 'ta': ('ta_emg', 'mV'), +} + +# One distinct colour per channel (matches reference image) +CHANNEL_COLORS = { + 'pos': '#1f77b4', # blue + 'tq': '#d62728', # red + 'gm': '#2ca02c', # dark green + 'gl': '#ff7f0e', # orange + 'sol': '#9467bd', # purple + 'ta': '#8c564b', # brown +} + + +def _ch_label(col: str): + """Return (name, unit) for y-axis label.""" + return CHANNEL_META.get(col.lower(), (col, '')) + + +# Ordered fallback palette — applied positionally when name doesn't match +_COLOR_PALETTE = list(CHANNEL_COLORS.values()) + + +def _ch_color(col: str, idx: int = 0) -> str: + """Return the hex colour for a channel by name, then by index.""" + c = CHANNEL_COLORS.get(col.lower()) + if c: + return c + return _COLOR_PALETTE[idx % len(_COLOR_PALETTE)] + + +def plot_trial(df, trial_num: int, output_path: str) -> None: + """ + Save a figure showing all signal channels for a single trial, + styled to match the reference image. + + Parameters + ---------- + df : pd.DataFrame + Trial DataFrame with a 'time' column plus one column per channel. + trial_num : int + 1-based trial index (used in the figure title). + output_path : str + Full path for the saved PNG file. + """ + signal_cols = [c for c in df.columns if c != 'time'] + n_ch = len(signal_cols) + n_smp = len(df) + fs = round(1.0 / df.attrs.get('domainIncr', 0.001)) + dur = n_smp / fs + subj = os.path.splitext(df.attrs.get('source_file', ''))[0] + comment = df.attrs.get('comment', '') + + # Build two-line title + trial_label = f'{subj} — Trial {trial_num} (first test trial)' if trial_num == 1 \ + else f'{subj} — Trial {trial_num}' + stats_label = f'Comment: "{comment}" | {n_smp} samples @ {fs} Hz = {dur:.1f} s' + + fig, axes = plt.subplots(n_ch, 1, figsize=(14, 2.5 * n_ch), + sharex=True, facecolor='white') + if n_ch == 1: + axes = [axes] + + fig.suptitle(f'{trial_label}\n{stats_label}', + fontsize=10, fontweight='bold', y=1.01, + fontfamily='monospace') + + for idx, (ax, col) in enumerate(zip(axes, signal_cols)): + name, unit = _ch_label(col) + color = _ch_color(col, idx) + + ax.plot(df['time'], df[col], color=color, linewidth=0.55) + ax.set_facecolor('white') + + # Two-line y-label: name on top, (unit) below + ax.set_ylabel(f'{name}\n({unit})', fontsize=8, labelpad=6, + rotation=90, va='center') + + # Tight y-limits with a small margin + ydata = df[col].values + ymin, ymax = ydata.min(), ydata.max() + margin = (ymax - ymin) * 0.08 if ymax != ymin else 0.1 + ax.set_ylim(ymin - margin, ymax + margin) + + ax.grid(True, linewidth=0.4, color='#cccccc', linestyle='-') + ax.tick_params(labelsize=7.5) + ax.spines[['top', 'right']].set_visible(False) + + axes[-1].set_xlabel('Time (s)', fontsize=9) + fig.tight_layout() + fig.savefig(output_path, dpi=150, bbox_inches='tight') + plt.close(fig) + + +def plot_all_trials_overlay(trials, output_path: str, downsample: int = 10) -> None: + """ + Save a figure with one subplot per channel, overlaying all trials. + + Parameters + ---------- + trials : list of pd.DataFrame + output_path : str + Full path for the saved PNG file. + downsample : int + Keep every N-th sample to reduce render time (default 10). + """ + if not trials: + return + + signal_cols = [c for c in trials[0].columns if c != 'time'] + n_ch = len(signal_cols) + n_trials = len(trials) + + # Each trial gets a shade along the channel's base colour + fig, axes = plt.subplots(n_ch, 1, figsize=(14, 2.5 * n_ch), + sharex=False, facecolor='white') + if n_ch == 1: + axes = [axes] + + fig.suptitle(f'All {n_trials} Trials Overlay', + fontsize=11, fontweight='bold', y=1.01) + + for idx, (ax, col) in enumerate(zip(axes, signal_cols)): + base_color = _ch_color(col, idx) + cmap = mcolors.LinearSegmentedColormap.from_list( + 'ch', ['#cccccc', base_color], N=n_trials) + name, unit = _ch_label(col) + for i, df in enumerate(trials): + t = df['time'].values[::downsample] + y = df[col].values[::downsample] + ax.plot(t, y, linewidth=0.4, alpha=0.55, color=cmap(i / max(n_trials - 1, 1)), + label=f'T{i+1}') + ax.set_facecolor('white') + ax.set_ylabel(f'{name}\n({unit})', fontsize=8, labelpad=6, + rotation=90, va='center') + ax.grid(True, linewidth=0.4, color='#cccccc', linestyle='-') + ax.tick_params(labelsize=7.5) + ax.spines[['top', 'right']].set_visible(False) + ax.set_xlabel('Time (s)', fontsize=8) + + # Single legend for the whole figure + handles, labels = axes[0].get_legend_handles_labels() + fig.legend(handles, labels, loc='upper right', fontsize=6, + ncol=max(1, n_trials // 10), bbox_to_anchor=(1.0, 1.0)) + + fig.tight_layout() + fig.savefig(output_path, dpi=150, bbox_inches='tight') + plt.close(fig) + print(f"Saved overlay plot → '{output_path}'") + + +# ── Main ────────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser( + description='Read a REKLAB .flb file, export CSVs, and generate plots.') + parser.add_argument('flb_file', + help='Path to the .flb file.') + parser.add_argument('--output_dir', default=None, + help='Root output directory ' + '(default: _output/ next to the .flb file).') + parser.add_argument('--subject_id', default='default', + help="Subject ID. Use 'IES01' for that subject's channel order.") + parser.add_argument('--downsample', type=int, default=10, + help='Downsample factor for the overlay plot (default: 10).') + parser.add_argument('--trial', type=int, default=1, + help='Trial number to plot (1-based, default: 1).') + parser.add_argument('--no_csv', action='store_true', + help='Skip CSV export.') + parser.add_argument('--no_plots', action='store_true', + help='Skip plot generation.') + args = parser.parse_args() + + flb_path = os.path.abspath(args.flb_file) + if not os.path.isfile(flb_path): + raise FileNotFoundError(f"File not found: {flb_path}") + + base_name = os.path.splitext(os.path.basename(flb_path))[0] + + # Resolve output directory + if args.output_dir: + output_dir = os.path.abspath(args.output_dir) + else: + output_dir = os.path.join(os.path.dirname(flb_path), f'{base_name}_output') + + csv_dir = os.path.join(output_dir, 'csv') + plots_dir = os.path.join(output_dir, 'plots') + + # ── Read all trials ─────────────────────────────────────────────────── + trials = read_flb(flb_path, subject_id=args.subject_id) + + if not trials: + print("No trials found in the file. Exiting.") + return + + # ── Export CSVs ─────────────────────────────────────────────────────── + if not args.no_csv: + os.makedirs(csv_dir, exist_ok=True) + csv_paths = [] + for i, df in enumerate(trials): + csv_name = f"{base_name}_trial{i+1:03d}.csv" + csv_path = os.path.join(csv_dir, csv_name) + df.to_csv(csv_path, index=False) + csv_paths.append(csv_path) + print(f"Saved {len(csv_paths)} CSV files → '{csv_dir}'") + + # Quick preview of trial 1 + import pandas as pd + print("\nTrial 1 preview (first 5 rows):") + print(pd.read_csv(csv_paths[0]).head().to_string(index=False)) + + # ── Generate plots ──────────────────────────────────────────────────── + if not args.no_plots: + os.makedirs(plots_dir, exist_ok=True) + + # Single-trial plot (--trial N, default 1) + t_idx = args.trial - 1 + if not (0 <= t_idx < len(trials)): + print(f"Warning: --trial {args.trial} out of range " + f"(file has {len(trials)} trials). Plotting trial 1.") + t_idx = 0 + trial_path = os.path.join(plots_dir, f'trial{t_idx+1:02d}_all_channels.png') + plot_trial(trials[t_idx], trial_num=t_idx + 1, output_path=trial_path) + print(f"Saved trial-{t_idx+1} plot → '{trial_path}'") + + # 2) All-trials overlay + overlay_path = os.path.join(plots_dir, 'all_trials_overlay.png') + plot_all_trials_overlay(trials, overlay_path, downsample=args.downsample) + + print("\nDone.") + + +if __name__ == '__main__': + main() diff --git a/ML/utils/requirements.txt b/ML/utils/requirements.txt new file mode 100644 index 0000000..4021870 --- /dev/null +++ b/ML/utils/requirements.txt @@ -0,0 +1,3 @@ +numpy +pandas +matplotlib