gesture classifier: quiet-segment channel QC, ChannelQC class, label auto-discovery docs

Channel QC overhaul (1_build_dataset.py):
- Replace ChannelQC/assess_channel_quality with direct scipy bandpass+notch
- Quiet-segment method: rank 500ms windows by global RMS, use bottom 20%
  to assess per-channel noise floor during rest
- Dead threshold: filtered RMS < 0.5 uV
- Noise threshold: filtered RMS > --noise_threshold (default 30 uV)
- New --qc_quiet_sec START END flag to pin an explicit rest window
- New --noise_threshold CLI flag

Label auto-discovery docs:
- Document all accepted label filenames (emg.txt, labels.csv, events.csv,
  labels.txt, {recording}_emg.txt) in module docstring, param docstring,
  --labels_path help, and README
- Add docstring to find_event_for_file() in _file_utils.py

run_channel_qc.py example: fully rewritten to use correct ChannelQC API
(update/evaluate), demonstrating both realtime and batch patterns

README (gesture_classifier): updated Channel QC section, parameters table,
label auto-discovery table, Option B examples

Core fixes (earlier in session):
- _session_loader: fix _extract_bitvolts multiplying by 1e6 when units=uV
- _zmq_client: get_latest_window returns data [0] not timestamps [1]
- 3_predict_realtime: N_channels lazy-init workaround, channels=[] normalisation
- predict.py: metadata.json fallback path, manager_root derivation

Author: Jshulgach

examples/analysis/run_channel_qc.py

@@ -1,64 +1,144 @@
 """
 Channel Quality Control Example
--------------------------------
-Demonstrates how to use the ChannelQC class to identify bad channels 
-based on line noise, impedance (if available), and signal artifacts.
+--------------------------------
+Demonstrates two usage patterns of ChannelQC + QCParams:
+
+  1. Realtime (streaming) — update() on each incoming chunk, evaluate() each window.
+  2. Batch (offline)      — slide over a full recording; flag channels bad if
+                            >50 % of evaluations mark them bad.  Flat / zero-crossing
+                            checks are disabled because quiet rest-period windows are
+                            healthy, not stuck.
+
+API summary
+-----------
+    qc = ChannelQC(fs, n_channels, window_sec, params=QCParams(...))
+    qc.update(chunk)   # chunk: (samples, n_channels), any number of rows
+    out = qc.evaluate()
+    # out['bad']      – np.bool_[n_channels]  single-eval verdict
+    # out['excluded'] – set(int)              hysteresis-stabilised bad set
+    # out['metrics']  – dict with 'rms', 'robust_z', 'pl_ratio', etc.
 """
 
 import numpy as np
-from pyoephys.processing import ChannelQC
-
-def generate_bad_data(n_channels=8, n_samples=10000):
-    """
-    Generates data where:
-    - Ch 0 is good
-    - Ch 1 is railed (saturated)
-    - Ch 2 has high 60Hz noise
-    - Ch 3 is disconnected (approx zero)
-    """
-    t = np.linspace(0, 5, n_samples)
-    data = np.random.randn(n_channels, n_samples) * 10 # Good baseline (10 uV)
-    
-    # Ch 1: Railed
-    data[1, :] = 5000.0 
-    
-    # Ch 2: 60Hz Noise
-    data[2, :] += 500 * np.sin(2 * np.pi * 60 * t)
-    
-    # Ch 3: Dead/Disconnected
-    data[3, :] = np.random.randn(n_samples) * 0.1
-    
-    return data
-
-def main():
-    fs = 2000.0
-    print("Generating sample data with 8 channels...")
-    data = generate_bad_data()
-    
-    print("Initializing Channel QC...")
-    qc = ChannelQC(fs=fs)
-    
-    print("Running QC analysis...")
-    # Analyze a window
-    results = qc.compute_qc(data, window_idx=0)
-    
-    print("\n--- QC Results ---")
-    print(f"Total Channels: {len(results)}")
-    
-    for ch, metrics in results.items():
-        status = "BAD" if not metrics['status'] else "GOOD"
-        print(f"Channel {ch}: {status}")
-        if not metrics['status']:
-            print(f"  Issues: {metrics.get('reasons', [])}")
-            
-    # Visualize if requested (requires matplotlib)
-    try:
-        import matplotlib.pyplot as plt
-        print("\nPlotting results...")
-        qc.plot_qc_summary(data)
-        plt.show()
-    except ImportError:
-        print("\nMatplotlib not found, skipping plot.")
+from pyoephys.processing import ChannelQC, QCParams
+
+
+# ---------------------------------------------------------------------------
+# Synthetic data helpers
+# ---------------------------------------------------------------------------
+
+def make_test_recording(n_channels: int = 16, duration_sec: float = 5.0, fs: float = 2000.0):
+    """Return (samples, n_channels) array with a few planted bad channels."""
+    n = int(duration_sec * fs)
+    t = np.arange(n) / fs
+    rng = np.random.default_rng(42)
+
+    # Good EMG: band-limited noise, ~30 µV RMS
+    data = rng.normal(0, 30, (n, n_channels)).astype(np.float32)
+
+    # Ch 1 – railed / saturated
+    data[:, 1] = 4500.0
+
+    # Ch 2 – heavy 60 Hz powerline
+    data[:, 2] += 800 * np.sin(2 * np.pi * 60 * t)
+
+    # Ch 3 – dead electrode (~0.05 µV RMS)
+    data[:, 3] = rng.normal(0, 0.05, n)
+
+    return data, fs
+
+
+# ---------------------------------------------------------------------------
+# 1. Realtime pattern
+# ---------------------------------------------------------------------------
+
+def demo_realtime(data: np.ndarray, fs: float):
+    """Process data chunk-by-chunk as it would arrive from hardware."""
+    n_samples, n_ch = data.shape
+    chunk_ms = 200
+    chunk_size = int(fs * chunk_ms / 1000)
+
+    params = QCParams(
+        robust_z_bad=3.0,
+        pl_ratio_thresh=0.30,
+        flat_std_min=1.0,   # enabled in realtime — sustained flat = stuck ADC
+        zc_min_hz=3.0,
+    )
+    qc = ChannelQC(fs=int(fs), n_channels=n_ch, window_sec=chunk_ms / 1000, params=params)
+
+    print("\n=== Realtime pattern ===")
+    for start in range(0, n_samples - chunk_size, chunk_size):
+        chunk = data[start : start + chunk_size]
+        qc.update(chunk)
+        out = qc.evaluate()
+
+    # Final stabilised verdict
+    excluded = out["excluded"]
+    m = out["metrics"]
+    print(f"  Excluded channels (hysteresis): {sorted(excluded)}")
+    print(f"  Median RMS: {m['median_rms']:.1f} µV")
+    for ch in sorted(excluded):
+        print(f"    ch {ch:3d}  rms={m['rms'][ch]:.1f}  z={m['robust_z'][ch]:.2f}  pl={m['pl_ratio'][ch]:.3f}")
+
+
+# ---------------------------------------------------------------------------
+# 2. Batch / offline pattern
+# ---------------------------------------------------------------------------
+
+def demo_batch(data: np.ndarray, fs: float):
+    """Score every channel over the full recording; vote across windows."""
+    n_samples, n_ch = data.shape
+    chunk_size = int(fs * 0.5)  # 500 ms windows
+
+    # Disable robust Z (HD-EMG has a wide biological RMS spread; active muscle
+    # channels would be flagged as Z-score outliers at ±3 SD, which is wrong).
+    # Disable flatline/ZC checks too — quiet rest-period channels are healthy.
+    # Only powerline ratio + absolute dead-channel threshold are kept.
+    dead_rms_uv = 0.5
+    params = QCParams(
+        robust_z_bad=999.0,     # effectively disabled
+        robust_z_warn=999.0,
+        pl_ratio_thresh=0.30,
+        flat_std_min=0.0,       # disabled
+        zc_min_hz=0.0,          # disabled
+    )
+    qc = ChannelQC(fs=int(fs), n_channels=n_ch, window_sec=0.5, params=params)
+
+    bad_vote = np.zeros(n_ch, dtype=int)
+    n_evals = 0
+    for start in range(0, n_samples - chunk_size, chunk_size):
+        qc.update(data[start : start + chunk_size])
+        out = qc.evaluate()
+        bad_vote += out["bad"].astype(int)
+        n_evals += 1
+
+    bad_channels = sorted(i for i in range(n_ch) if bad_vote[i] > n_evals * 0.5)
+
+    # Also catch truly dead electrodes via absolute RMS threshold
+    dead = [i for i in range(n_ch) if out["metrics"]["rms"][i] < dead_rms_uv]
+    bad_channels = sorted(set(bad_channels) | set(dead))
+    good_channels = sorted(set(range(n_ch)) - set(bad_channels))
+    m = out["metrics"]
+
+    print("\n=== Batch / offline pattern ===")
+    print(f"  {len(good_channels)} good, {len(bad_channels)} bad channels")
+    print(f"  Median RMS: {m['median_rms']:.1f} µV")
+    print(f"  Bad channels: {bad_channels}")
+    for ch in bad_channels:
+        print(f"    ch {ch:3d}  rms={m['rms'][ch]:.1f}  z={m['robust_z'][ch]:.2f}  pl={m['pl_ratio'][ch]:.3f}")
+    return good_channels
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    main()
+    data, fs = make_test_recording()
+    print(f"Data shape: {data.shape}  (samples × channels),  fs={fs:.0f} Hz")
+    print("Planted bad channels: 1 (railed), 2 (60 Hz), 3 (dead)")
+
+    demo_realtime(data, fs)
+    good = demo_batch(data, fs)
+    print(f"\nChannels safe for downstream use: {good}")
+

examples/gesture_classifier/.gesture_config.example

@@ -1,50 +1,48 @@
 # EMG Gesture Classifier Configuration (Open Ephys)
 # =================================================
+# Copy this file to .gesture_config and edit as needed.
+# All three pipeline scripts read settings from .gesture_config
+# (CLI arguments override these values).
 
-# Root directory containing Open Ephys recordings (parent of data subfolders)
-root_dir=./data
+# ── Shared ───────────────────────────────────────────────────────────────────
 
-# Model label (used for naming output files, e.g., 'model/128ch_model.pkl')
-label=demo_model
+# Path to the EMG recording (CSV, .npz, or Open Ephys folder / .oebin).
+# Defaults to the included example data when not set.
+data_path=./data/gestures
 
-# Enable verbose logging
-verbose=true
+# Path to the labels/events file (CSV with Sample Index,Label columns).
+# Auto-discovered alongside data_path when not set.
+labels_path=./data/labels.csv
 
-# ============================================================================
-# Dataset Building (1_build_dataset.py)
-# ============================================================================
+# Directory where the trained model will be saved and loaded from.
+# This is the *root* data directory — model files go in a model/ sub-folder.
+root_dir=./data/gesture_model
 
-# Output path for the training dataset (.npz)
-# If not specified, defaults to {root_dir}/training_dataset.npz
-save_path=./data/training_dataset.npz
+# Model label / name tag (used for output file naming).
+label=gesture_model
 
-# Feature extraction settings
-window_ms=200
-step_ms=50
+# Enable verbose logging.
+verbose=false
 
-# ============================================================================
-# Training (2_train_model.py)
-# ============================================================================
+# ── Dataset Building (1_build_dataset.py) ────────────────────────────────────
 
-# Number of training epochs
-epochs=100
+# Output path for the windowed feature dataset (.npz).
+dataset_path=./data/training_dataset.npz
 
-# Batch size
-batch_size=32
+# Window and step lengths in milliseconds.
+window_ms=200
+step_ms=50
 
-# Learning rate
-learning_rate=0.001
+# ── Training (2_train_model.py) ──────────────────────────────────────────────
 
-# Use K-Fold Cross Validation? (true/false)
+# Use K-Fold Cross Validation (true/false).
 kfold=true
 
-# ============================================================================
-# Prediction (3_predict_realtime.py)
-# ============================================================================
+# ── Real-Time Prediction (3_predict_realtime.py) ─────────────────────────────
 
-# ZMQ connection
-zmq_ip=tcp://127.0.0.1
-zmq_port=5556
+# Open Ephys machine IP and ZMQ data port.
+host=127.0.0.1
+port=5556
 
-# Inference Smoothing
+# Majority-vote smoothing window (number of consecutive predictions).
 smooth_k=5