# Project export: NeuroFlex: Neuroplastic Rehabilitation This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2026 - Tagline: NeuroFlex integrates EEG-based BCI technology with EMG biofeedback to provide direct, real-time feedback on re-establishing the brain-muscle connection for motor rehabilitation. - Devpost: https://devpost.com/software/neuroflex-paresis-rehabilitation - GitHub: https://github.com/mhtruong1031/TreeHacks2026 - Video: https://www.youtube.com/embed/lIZioBU2Jk4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — mhtruong1031 (9 commits), Andrew N. Oldag (7 commits) ## Devpost submission (written by the team) ### Inspiration Damage to motor pathways in the brain, spinal cord, or peripheral nerves can result in complete or partial paralysis, often arising from conditions such as stroke, cerebral palsy, multiple sclerosis, or traumatic brain/spinal cord injury. Neuroplasticity refers to the brain’s ability to reorganize its circuitry by strengthening active neural connections, pruning unused ones, and forming new neural pathways, enabling functional recovery after injury. Rehabilitation therefore emphasizes task-specific repetition to help rebuild motor function, yet early-stage neuroplasticity progress remains largely invisible, offering patients and clinicians little real-time, tangible feedback. This gap presents an opportunity for innovative AI-driven solutions that can visualize recovery and personalize rehabilitation, motivating individuals to remain engaged in their early rehabilitation efforts. Our vision is clear, providing a tool for patients and healthcare professionals to quantify motor recovery. What NeuroFlex does NeuroFlex makes rehabilitation more measurable, adaptive, and effective for individuals recovering from paresis. Provides users with real-time feedback on the timing and strength of their brain and muscle activation using entirely noninvasive measurements. Integrates EEG-based technology with EMG (muscle electrical activity) biofeedback. The model generates a coordination metric that quantifies how effectively muscle activation aligns with the user’s intended movement. Makes neuroplasticity visible and measurable* via a similarity score that changes over time. This is especially valuable during early rehabilitation (after the first month) and during **plateau phases, when improvements are often subtle and difficult to detect using conventional methods. Allows both users and clinicians to track recovery over time. AI-driven insights. The model recommends personalized rehabilitation exercises and visualization strategies based on the user’s current progress. Users can select the tone and personality of the system’s feedback. Provides guidance during relapse for the most optimal approaches on returning to a previous state. ### How we built it We utilized a BCI system connecting an OpenBCI headband EEG and EMG electrode to an OpenBCI Ganglion board, allowing measurements of brain and muscle activity simultaneously. These signals were transmitted via Bluetooth to our laptop through the BrainFlow API. For demonstration, we focused on fist clenching and relaxation of the right hand. Our platform, NeuroFlex, processes these signals through several steps. First, it calibrates a baseline by recording one minute of resting activity, which is used to remove background noise. The data is then filtered through a fast Fourier transform (FFT) to eliminate artifacts such as eye blinks and sweat interference, converted into frequency components, and downsampled to improve efficiency while preserving meaningful signals. To quantify recovery progress, we created a coordination index that measures how well EEG and EMG signals are synchronized during movement through singular value decomposition (SVD) coupling, projecting each point onto the mode of movement for the bimodality coefficient, then taking the norm of the coupling and bimodality metrics, with respect to the highest observed coordination value. These attempts are stored and ranked, with greater emphasis on recent performance to reflect ongoing neuroplastic changes. Using dynamic time warping of signals from previous attempts, we produce a similarity score that quantifies changes in brain-muscle coordination over time. The UI built using pyqtgraph highlights the user's five best recent attempts, considering both performance and recency to reveal meaningful trends in neuroplastic recovery. We also trained a Long Short-Term Memory (LSTM) model on sequences of past coordination data. This model predicts the user’s target coordination state and provides personalized feedback, displayed on our UI, to help them reach or regain their best performance. To enable increased functionality and enhance user experience, we utilized a series of context-dependent prompts pushed to a Gemini-based API. These prompts are designed to encourage the user to continue along promising paths and to introduce novel ideas when progress stalls. By adapting to each individual’s progress, NeuroFlex optimizes rehabilitation by making recovery measurable, personalized, and actionable. ### Challenges we ran into Surface electrode technology enables noninvasive collection of neural and muscle activity data, however, it presents challenges due to weaker and lower-resolution signals compared to invasive methods, especially when compounded with hardware limitations. As OpenBCI systems rely on complete hardware integration, missing the specific signal transmission dongle for the Ganglion board was a significant barrier that we overcame by transmitting our signals to a BrainFlow python API instead of the OpenBCI GUI, whose strict protocol misinterpreted our data. ### Accomplishments we're proud of NeuroFlex’s core innovation lies in its interdisciplinary pipeline that integrates EEG and EMG biofeedback into a unified coordination metric, transforming neuroplasticity into a clear, measurable signal. Despite significant hardware challenges, we successfully designed and built a fully functional system, from signal acquisition to real-time analysis and user feedback. Our platform includes an interactive UI that visualizes brain and muscle signals, displays similarity scores to quantify progress, and provides real-time prompts and feedback to guide rehabilitation. This demonstrates not only the analytical feasibility of our approach, but also its practical usability as a complete rehabilitation tool. Most of all, we are especially proud of our team’s perseverance, creative problem-solving, and collaborative effort. ### What we learned From dozens of iterations to troubleshoot hardware constraints to integrating multi-stream inputs, we translated a conceptual framework combining classical statistical methods, neural signal tracking, and AI into a robust, user-centered system. ### What's next NeuroFlex represents a step toward more accessible, intelligent rehabilitation by combining noninvasive sensing with AI-driven feedback and personalized recommendations. Our next steps include validating the system with improved hardware integration, increasing EEG electrode density to capture activity across more brain regions, and scaling channel capacity to enable more robust independent component analysis and signal separation. ## README (from the GitHub repository) No README available. ## Detected evidence (automated analysis) Indexed codebase: 34 recognized source files, 262 KB. - Python (language) — detected in the code - PyTorch (technology) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - TensorFlow (technology) — claimed on Devpost, not found in the code - AI coding agent: Claude Code — evidence: config files committed to the repository ## Codebase structure (from repository index) ### Files (120 of 340) ``` .claude/settings.local.json .gitignore analysis/__init__.py analysis/BaselineCalibration.py analysis/LLMPipeline.py analysis/MainPipeline.py analysis/PredictionPipeline.py analysis/PreprocessingPipeline.py analysis/PresentPipeline.py analysis/ProgressMonitor.py CLAUDE.md data_sources/__init__.py data_sources/base.py data_sources/simulated_source.py data_sources/udp_source.py generate_test_data.py gui/__init__.py gui/live_gui.py IMPLEMENTATION_SUMMARY.md main_runtime.py QUICK_START.md requirements.txt run_live_gui.py simulation/__init__.py simulation/compare_filter_configs.py simulation/dataset/class_map.json simulation/dataset/cocontraction/sub-001_ses-001_trial-041_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-041_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-042_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-042_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-043_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-043_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-044_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-044_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-045_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-045_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-046_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-046_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-047_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-047_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-048_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-048_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-049_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-049_events.tsv simulation/dataset/cocontraction/sub-001_ses-001_trial-050_eegemg.npz simulation/dataset/cocontraction/sub-001_ses-001_trial-050_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-041_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-041_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-042_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-042_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-043_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-043_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-044_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-044_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-045_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-045_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-046_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-046_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-047_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-047_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-048_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-048_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-049_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-049_events.tsv simulation/dataset/cocontraction/sub-002_ses-001_trial-050_eegemg.npz simulation/dataset/cocontraction/sub-002_ses-001_trial-050_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-041_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-041_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-042_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-042_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-043_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-043_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-044_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-044_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-045_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-045_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-046_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-046_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-047_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-047_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-048_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-048_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-049_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-049_events.tsv simulation/dataset/cocontraction/sub-003_ses-001_trial-050_eegemg.npz simulation/dataset/cocontraction/sub-003_ses-001_trial-050_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-031_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-031_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-032_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-032_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-033_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-033_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-034_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-034_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-035_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-035_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-036_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-036_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-037_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-037_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-038_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-038_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-039_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-039_events.tsv simulation/dataset/grip_release/sub-001_ses-001_trial-040_eegemg.npz simulation/dataset/grip_release/sub-001_ses-001_trial-040_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-031_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-031_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-032_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-032_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-033_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-033_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-034_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-034_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-035_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-035_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-036_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-036_events.tsv simulation/dataset/grip_release/sub-002_ses-001_trial-037_eegemg.npz simulation/dataset/grip_release/sub-002_ses-001_trial-037_events.tsv [220 more files omitted for size] ``` ### Dependencies - requirements.txt: google-genai, matplotlib, numpy, scipy, torch, tslearn ### Recent commits (newest first) - removed old api key - 4:22 am - updated platue calculation - Merge branch 'main' of https://github.com/mhtruong1031/TreeHacks2026 - updates to llm pipeline - top n - Merge branch 'main' of https://github.com/mhtruong1031/TreeHacks2026 - ayo - fixed simulation to 3 channel nueron 1 channel muscle - Merge branch 'main' of https://github.com/mhtruong1031/TreeHacks2026 - fixes - reorganizing simulation data and proccessing - revised preproccessing - im ngl i forgot what i changed trust me i made progress though - uhhh yeah - Merge branch 'main' of https://github.com/mhtruong1031/TreeHacks2026 - added udp signal reciever - Merge branch 'main' of https://github.com/mhtruong1031/TreeHacks2026 - im ngl i was in the middle of ill make the full commit later - add new preprocessing script ## Key source files (fetched from GitHub, selected and truncated for size) ### QUICK_START.md ```markdown # Quick Start Guide - Baseline Calibration System ## Installation No additional dependencies required! The system uses existing packages: - numpy - scipy - torch - (matplotlib for plotting - optional) ## Usage ### 1. Live EEG/EMG Session (OpenBCI) **Step 1: Start OpenBCI GUI and configure UDP streaming** - Set IP: 127.0.0.1 (or 0.0.0.0 for network streaming) - Set Port: 12345 - Start streaming **Step 2: Run the runtime system** ```bash python main_runtime.py --live --calibration-time 60 ``` **What happens:** 1. **Calibration Phase (60 seconds):** - Sit still and relax - System collects baseline data - Computes channel statistics 2. **Streaming Phase (continuous):** - Processes packets in real-time - Detects movement attempts with adaptive thresholds - Rejects artifacts automatically - Displays progress every 1000 packets **Output example:** ``` PHASE 1: BASELINE CALIBRATION Please sit still and relax for 60 seconds... [ 26.7%] 160 / 600 samples | 159.8 Hz | 1.0s | ~ 2.8s remaining [ 53.0%] 318 / 600 samples | 158.4 Hz | 2.0s | ~ 1.8s remaining [ 79.0%] 474 / 600 samples | 157.6 Hz | 3.0s | ~ 0.8s remaining ✓ Calibration complete PHASE 2: STREAMING ANALYSIS [ 1.0s] Packets: 157 | Rate: 156.7 Hz | Attempts: 0 | Model: Pending (0/30) [ 2.0s] Packets: 317 | Rate: 159.4 Hz | Attempts: 0 | Model: Pending (0/30) [ 3.0s] Packets: 545 | Rate: 184.3 Hz | Attempts: 6 | Model: Pending (6/30) Top attempts: 1. Coord: 0.4391 | Similarity: 0.0000 2. Coord: 0.4287 | Similarity: 0.1599 ``` **Note:** Status updates print every **1 second** for real-time feedback. ### 2. Testing with Simulated Data **For fast testing (unlimited speed):** ```bash python main_runtime.py --simulate test_data.csv --speed 0 --calibration-time 5 ``` **For realistic simulation (real-time):** ```bash python3 main_runtime.py --simulate test_data.csv --speed 1.0 ``` ### 3. Advanced Options **Enable spatial whitening:** ```bash python main_runtime.py --live --whitening ``` **Adjust adaptive threshold sensitivity:** ```bash python main_runtime.py --live --threshold-n-std 3.0 # More conservative (3σ) ``` **Skip calibration (legacy fixed threshold mode):** ```bash python main_runtime.py --simulate data.csv --skip-calibration ``` **Custom UDP configuration:** ```bash python main_runtime.py --live --ip 192.168.1.100 --port 8080 ``` ## Command-Line Options ``` Required (choose one): --live Use live UDP data from OpenBCI --simulate CSV_PATH Use CSV file for testing UDP options: --ip IP UDP IP address (default: 0.0.0.0) --port PORT UDP port (default: 12345) Simulation options: --speed SPEED Playback speed (0=unlimited, 1=real-time, 10=10x) Calibration options: --calibration-time SECS Calibration duration (default: 60s) --skip-calibration Use fixed thresholds instead --whitening Enable spatial whi [truncated — 4719 more characters] ``` ### IMPLEMENTATION_SUMMARY.md ```markdown # Baseline Calibration Integration - Implementation Summary ## Overview Successfully implemented a unified runtime system with baseline calibration for adaptive EEG/EMG signal processing. The system now supports two-phase execution (calibration → streaming) with both live UDP and simulated data sources. ## What Was Implemented ### Phase 1: DataSource Abstraction ✅ Created a unified interface for data acquisition: **Files Created:** - `data_sources/__init__.py` - Package exports - `data_sources/base.py` - Abstract DataSource class - `data_sources/udp_source.py` - Live UDP implementation (reuses parse_packet from udp_receiver.py) - `data_sources/simulated_source.py` - CSV playback with background thread and queue **Key Features:** - Common interface: `get_packet()`, `is_running()`, `close()` - UDP source with configurable timeout and socket options - Simulated source with adjustable playback speed (0=unlimited, 1=real-time, 10=10x) - Producer-consumer pattern with queue for smooth playback ### Phase 2: MainPipeline Calibration Support ✅ Modified `analysis/MainPipeline.py` to support baseline calibration: **Changes Made:** 1. **Added calibration parameters to `__init__`** (line 14-30): - `calibration`: Optional BaselineCalibration instance - `use_whitening`: Enable spatial whitening - `adaptive_threshold_n_std`: Multiplier for adaptive thresholds (default: 2.5σ) 2. **Added artifact rejection in `run()`** (line 57-66): - Checks `calibration.is_artifact()` before processing - Rejects packets exceeding 5σ from baseline - Optional spatial whitening via `apply_whitening()` - Tracks `artifact_count` for monitoring 3. **Modified `check_activation()` for adaptive thresholds** (line 139-161): - **Adaptive mode**: Per-channel thresholds = `mean + N*std` from calibration - **Fixed mode**: Legacy threshold (backward compatible) - Automatically switches based on calibration availability 4. **Added `get_calibration_info()` monitoring** (line 343-374): - Returns calibration status, mode, thresholds - Artifact rejection statistics - Whitening status **Backward Compatibility:** - All changes are additive with defaults - `MainPipeline(window_size_s=0.2, activation_threshold=0.2)` still works - If `calibration=None`, uses legacy fixed threshold behavior ### Phase 3: RuntimeOrchestrator ✅ Created `main_runtime.py` with comprehensive CLI and orchestration: **Key Components:** 1. **RuntimeOrchestrator Class:** - `_run_calibration_phase()`: Collects 60s rest data, runs calibration - `_run_streaming_phase()`: Initializes MainPipeline, processes packets - Progress monitoring every 1000 packets - Real-time statistics display 2. **CLI Interface:** ```bash # Live UDP with calibration python main_runtime.py --live --calibration-time 60 # Simulated unlimited speed python main_runtime.py --simulate trial_001.csv --speed 0 # With whitening python main_runtime.py --simulate trial.csv --whit [truncated — 8955 more characters] ``` ### requirements.txt ``` numpy scipy torch tslearn matplotlib google-genai ``` ### test_full_run.sh ```shell #!/bin/bash # Run full test and capture last 60 lines python3 main_runtime.py --simulate test_data.csv --speed 0 --calibration-time 5 2>&1 | tail -60 ``` ### test_backward_compat.py ```python """ Test backward compatibility: old MainPipeline code should still work. """ import numpy as np from analysis.MainPipeline import MainPipeline # Test 1: Old initialization (without calibration) print("Test 1: Old-style MainPipeline initialization") pipeline = MainPipeline(window_size_s=0.2, activation_threshold=0.2) print("✓ Pipeline created successfully") # Test 2: Process some packets print("\nTest 2: Process packets without calibration") for i in range(100): packet = np.random.randn(4) * 0.1 pipeline.run(packet) print(f"✓ Processed 100 packets") # Test 3: Check that it's using fixed threshold mode info = pipeline.get_calibration_info() print(f"\nTest 3: Verify fixed threshold mode") print(f" Calibrated: {info['calibrated']}") print(f" Mode: {info['mode']}") print(f" Fixed threshold: {info['fixed_threshold']}") assert info['calibrated'] == False assert info['mode'] == 'fixed_threshold' print("✓ Fixed threshold mode confirmed") # Test 4: With calibration print("\nTest 4: New-style initialization with calibration") from analysis.BaselineCalibration import BaselineCalibration # Create calibration calib = BaselineCalibration(fs=200.0) rest_data = np.random.randn(12000, 4) * 0.05 # 60 seconds of rest data calib.calibrate(rest_data) # Create pipeline with calibration pipeline2 = MainPipeline(calibration=calib, adaptive_threshold_n_std=2.5) print("✓ Pipeline with calibration created") # Process packets for i in range(100): packet = np.random.randn(4) * 0.1 pipeline2.run(packet) print(f"✓ Processed 100 packets with calibration") # Check adaptive mode info2 = pipeline2.get_calibration_info() print(f"\nTest 5: Verify adaptive threshold mode") print(f" Calibrated: {info2['calibrated']}") print(f" Mode: {info2['mode']}") print(f" Adaptive thresholds: {info2['adaptive_thresholds']}") assert info2['calibrated'] == True assert info2['mode'] == 'adaptive' print("✓ Adaptive threshold mode confirmed") # Test 6: Artifact rejection print("\nTest 6: Test artifact rejection") artifact_packet = np.array([5.0, 5.0, 5.0, 5.0]) # Large artifact normal_packet = np.array([0.05, 0.05, 0.05, 0.05]) pipeline2.run(artifact_packet) artifacts_1 = pipeline2.artifact_count pipeline2.run(normal_packet) artifacts_2 = pipeline2.artifact_count print(f" Artifacts after large packet: {artifacts_1}") print(f" Artifacts after normal packet: {artifacts_2}") assert artifacts_1 > 0, "Large packet should be rejected as artifact" print("✓ Artifact rejection working") print("\n" + "=" * 60) print("✅ All backward compatibility tests passed!") print("=" * 60) ``` ### test_live_gui.py ```python #!/usr/bin/env python3 """ Quick test of LiveGUI with simulated data. """ import sys from PyQt5.QtWidgets import QApplication from PyQt5.QtCore import QTimer from gui import LiveGUI import numpy as np app = QApplication(sys.argv) # Create GUI gui = LiveGUI(window_size_s=10.0, sampling_rate=200.0, update_rate_hz=60) gui.show() # Simulate data generation at 200 Hz packet_count = 0 def generate_data(): """Generate simulated EEG/EMG data.""" global packet_count # Generate 10 samples (50ms worth at 200 Hz) n = 10 t = packet_count + np.arange(n) # Simulate different frequencies for different channels data = np.column_stack([ 50 * np.sin(2 * np.pi * 10 * t / 200) + np.random.randn(n) * 5, # EEG 1: 10 Hz 50 * np.sin(2 * np.pi * 12 * t / 200) + np.random.randn(n) * 5, # EEG 2: 12 Hz 50 * np.sin(2 * np.pi * 8 * t / 200) + np.random.randn(n) * 5, # EEG 3: 8 Hz 100 * np.sin(2 * np.pi * 20 * t / 200) + np.random.randn(n) * 10, # EMG: 20 Hz, higher amplitude ]) gui.add_data(data) packet_count += n # Timer to simulate data at 200 Hz (batch of 10 every 50ms) data_timer = QTimer() data_timer.timeout.connect(generate_data) data_timer.start(50) # 50ms = 20 Hz batch rate, but 10 samples per batch = 200 Hz # Simulate coordination attempts after 2 seconds def show_coordination(): print("Updating coordination circles...") # Attempts in random order - will be sorted by GUI (best first) attempts = [ {'coordination_index': 0.2341, 'similarity_score': 0.48, 'attempt_id': 27}, # Fair {'coordination_index': 0.5124, 'similarity_score': 0.03, 'attempt_id': 8}, # Very poor {'coordination_index': 0.0845, 'similarity_score': 0.92, 'attempt_id': 52}, # Best! {'coordination_index': 0.3892, 'similarity_score': 0.19, 'attempt_id': 15}, # Poor {'coordination_index': 0.1523, 'similarity_score': 0.71, 'attempt_id': 41}, # Good ] # Predicted ideal - no coordination index shown, no attempt ID predicted_ideal = { 'coordination_index': 0.0723, # Not displayed 'similarity_score': 1.0 # Perfect match } gui.update_coordination_attempts(attempts, predicted_ideal) QTimer.singleShot(2000, show_coordination) # Auto-close after 10 seconds for testing QTimer.singleShot(10000, app.quit) print("Starting GUI test (will run for 10 seconds)...") print("You should see:") print(" - Blue: EEG Channel 1 (10 Hz)") print(" - Green: EEG Channel 2 (12 Hz)") print(" - Cyan: EEG Channel 3 (8 Hz)") print(" - Red: EMG (20 Hz, larger amplitude)") print(" - After 2s: 5 circles showing top attempts + 1 predicted ideal circle") print(" (Greener = higher similarity, darker = lower similarity)") print(" (6th circle should be brightest green - perfect similarity)") print() sys.exit(app.exec_()) ``` ### test.py ```python import argparse import logging import pyqtgraph as pg from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds from brainflow.data_filter import DataFilter, FilterTypes, DetrendOperations from pyqtgraph.Qt import QtGui, QtCore class Graph: def __init__(self, board_shim): self.board_id = board_shim.get_board_id() self.board_shim = board_shim self.exg_channels = BoardShim.get_exg_channels(self.board_id) self.sampling_rate = BoardShim.get_sampling_rate(self.board_id) self.update_speed_ms = 50 self.window_size = 4 self.num_points = self.window_size * self.sampling_rate self.app = QtGui.QApplication([]) self.win = pg.GraphicsWindow(title='BrainFlow Plot', size=(800, 600)) self._init_timeseries() timer = QtCore.QTimer() timer.timeout.connect(self.update) timer.start(self.update_speed_ms) QtGui.QApplication.instance().exec_() def _init_timeseries(self): self.plots = list() self.curves = list() for i in range(len(self.exg_channels)): p = self.win.addPlot(row=i, col=0) p.showAxis('left', False) p.setMenuEnabled('left', False) p.showAxis('bottom', False) p.setMenuEnabled('bottom', False) if i == 0: p.setTitle('TimeSeries Plot') self.plots.append(p) curve = p.plot() self.curves.append(curve) def update(self): data = self.board_shim.get_current_board_data(self.num_points) for count, channel in enumerate(self.exg_channels): # plot timeseries DataFilter.detrend(data[channel], DetrendOperations.CONSTANT.value) DataFilter.perform_bandpass(data[channel], self.sampling_rate, 3.0, 45.0, 2, FilterTypes.BUTTERWORTH.value, 0) DataFilter.perform_bandstop(data[channel], self.sampling_rate, 48.0, 52.0, 2, FilterTypes.BUTTERWORTH.value, 0) DataFilter.perform_bandstop(data[channel], self.sampling_rate, 58.0, 62.0, 2, FilterTypes.BUTTERWORTH.value, 0) self.curves[count].setData(data[channel].tolist()) self.app.processEvents() def main(): BoardShim.enable_dev_board_logger() logging.basicConfig(level=logging.DEBUG) parser = argparse.ArgumentParser() # use docs to check which parameters are required for specific board, e.g. for Cyton - set serial port parser.add_argument('--timeout', type=int, help='timeout for device discovery or connection', required=False, default=0) parser.add_argument('--ip-port', type=int, help='ip port', required=False, default=0) parser.add_argument('--ip-protocol', type=int, help='ip protocol, check IpProtocolType enum', required=False, default=0) parser.add_argument('--ip-address', type=str, help='ip address', required=False, default='') parser.add_argument('--serial-port', type=str, help='serial port', required=False, default='') parser.add_argument('--mac-address', type=str, help='mac address', required=False, default='') parser.add_argument('--other-info', type=str, help='other info', required=False, default='') parser.add_argument('--streamer-params', type=str, help='streamer params', required=False, default='') parser.add_argument('--serial-number', type=str, help='serial number', required=False, default='') parser.add_argument('--board-id', type=int, help='board id, check docs to get a list of supported boards', required=False, default=BoardIds.SYNTHETIC_BOARD) parser.add_argument('--file', type=str, help='file', required=False, default='') args = parser.parse_args() params = BrainFlowInputParams() params.ip_port = args.ip_port params.serial_port = args.serial_port params.mac_address = args.mac_address params.other_info = args.other_info params.serial_number = args.serial_number params.ip_address = args.ip_address params.ip_protocol = args.ip_protocol params.timeout = args.timeout params.file = args.file try: board_shim = BoardShim(BoardIds.GANGLION_NATIVE_BOARD, params) board_shim.prepare_session() board_shim.start_stream(450000, args.streamer_params) Graph(board_shim) except BaseException: logging.warning('Exception', exc_info=True) finally: logging.info('End') if board_shim.is_prepared(): logging.info('Releasing session') board_shim.release_session() if __name__ == '__main__': main() ``` ### test_comprehensive.py ```python """ Comprehensive integration test demonstrating all features: 1. DataSource abstraction (simulated) 2. Baseline calibration 3. Adaptive thresholds 4. Artifact rejection 5. Whitening 6. RuntimeOrchestrator """ import numpy as np from data_sources import SimulatedDataSource from analysis.BaselineCalibration import BaselineCalibration from analysis.MainPipeline import MainPipeline print("=" * 70) print(" COMPREHENSIVE INTEGRATION TEST") print("=" * 70) print() # Test 1: DataSource abstraction print("Test 1: DataSource Abstraction") print("-" * 70) # Create test data with artifacts test_data = np.random.randn(500, 4) * 0.05 # Add artifacts at specific locations test_data[100:105, :] = 5.0 # Large artifact test_data[200:205, :] = -5.0 # Another artifact np.savetxt('test_comprehensive.csv', test_data, delimiter=',', fmt='%.6f') source = SimulatedDataSource('test_comprehensive.csv', fs=200.0, speed=0) print(f"✓ Created SimulatedDataSource") print() # Test 2: Baseline calibration print("Test 2: Baseline Calibration") print("-" * 70) # Collect calibration data calib_data = [] for i in range(200): # 1 second @ 200 Hz packet = source.get_packet() if packet is not None: calib_data.append(packet) calib_data_array = np.array(calib_data) print(f"✓ Collected {len(calib_data)} calibration samples") # Calibrate calib = BaselineCalibration(fs=200.0) calib.calibrate(calib_data_array) print(f"✓ Calibration complete") print(f" Channel means: {calib.channel_mean}") print(f" Channel stds: {calib.channel_std}") print() # Test 3: Adaptive thresholds print("Test 3: Adaptive Thresholds") print("-" * 70) adaptive_thresholds = calib.channel_mean + 2.5 * calib.channel_std print(f" Adaptive thresholds (2.5σ): {adaptive_thresholds}") print(f"✓ Adaptive thresholds computed") print() # Test 4: MainPipeline with calibration print("Test 4: MainPipeline with Calibration") print("-" * 70) pipeline = MainPipeline( window_size_s=0.2, calibration=calib, use_whitening=False, adaptive_threshold_n_std=2.5 ) print(f"✓ Created MainPipeline with calibration") # Process remaining packets processed_count = 0 artifact_initial = 0 while source.is_running(): packet = source.get_packet() if packet is None: break # Check if this would be an artifact if processed_count == 0: artifact_initial = pipeline.artifact_count pipeline.run(packet) processed_count += 1 artifact_final = pipeline.artifact_count artifacts_in_stream = artifact_final - artifact_initial print(f"✓ Processed {processed_count} packets") print(f" Artifacts rejected: {artifacts_in_stream}") print() # Test 5: Calibration info print("Test 5: Calibration Info") print("-" * 70) info = pipeline.get_calibration_info() print(f" Calibrated: {info['calibrated']}") print(f" Mode: {info['mode']}") print(f" Adaptive threshold multiplier: {info['adaptive_threshold_n_std']}σ") print(f" Channel means: {[f'{m:.4f}' for m in info['channel_means']]}") print(f" Channel stds: {[f'{s:.4f}' for s in info['channel_stds']]}") print(f" Thresholds: {[f'{t:.4f}' for t in info['adaptive_thresholds']]}") print(f" Artifacts rejected: {info['artifacts_rejected']} / {info['total_packets']}") print(f" Artifact rate: {info['artifact_rate']*100:.2f}%") print(f"✓ Calibration info retrieved") print() # Test 6: Whitening print("Test 6: Spatial Whitening") print("-" * 70) # Create new source and pipeline with whitening source2 = SimulatedDataSource('test_comprehensive.csv', fs=200.0, speed=0) # Skip calibration data for i in range(200): source2.get_packet() pipeline2 = MainPipeline( window_size_s=0.2, calibration=calib, use_whitening=True, # Enable whitening adaptive_threshold_n_std=2.5 ) print(f"✓ Created MainPipeline with whitening enabled") # Process a few packets for i in range(50): packet = source2.get_packet() if packet is not None: pipeline2.run(packet) print(f"✓ Processed packets with spatial whitening") print() # Test 7: Compare modes print("Test 7: Compare Fixed vs Adaptive Modes") print("-" * 70) # Fixed threshold mode pipeline_fixed = MainPipeline(window_size_s=0.2, activation_threshold=0.2) info_fixed = pipeline_fixed.get_calibration_info() # Adaptive mode info_adaptive = pipeline.get_calibration_info() print(f"Fixed threshold mode:") print(f" Mode: {info_fixed['mode']}") print(f" Threshold: {info_fixed['fixed_threshold']}") print() print(f"Adaptive threshold mode:") print(f" Mode: {info_adaptive['mode']}") print(f" Thresholds: {[f'{t:.4f}' for t in info_adaptive['adaptive_thresholds']]}") print(f" Artifact rejection rate: {info_adaptive['artifact_rate']*100:.2f}%") print() # Cleanup source.close() source2.close() print("=" * 70) print("✅ ALL COMPREHENSIVE TESTS PASSED!") print("=" * 70) print() print("Summary:") print(f" ✓ DataSource abstraction working") print(f" ✓ Baseline calibration functional") print(f" ✓ Adaptive thresholds computed correctly") print(f" ✓ Artifact rejection active ({artifacts_in_stream} artifacts rejected)") print(f" ✓ Spatial whitening available") print(f" ✓ Backward compatibility maintained") print(f" ✓ RuntimeOrchestrator ready for use") print() ``` ### generate_test_data.py ```python """ Generate realistic test data for EEG/EMG pipeline testing. Creates a 2-minute dataset with: - 60 movement attempts (0.5 Hz) - Realistic baseline with low-amplitude noise - Varying coordination quality across attempts - Some artifacts for rejection testing """ import numpy as np # Parameters fs = 200.0 # Sampling rate (Hz) duration_s = 120.0 # 2 minutes n_samples = int(fs * duration_s) n_channels = 4 # EMG, Left EEG, Right EEG, Additional # Target: 60 attempts over 2 minutes n_attempts = 60 attempt_interval = duration_s / n_attempts # ~2 seconds between attempts print(f"Generating test data:") print(f" Duration: {duration_s} seconds ({duration_s/60:.1f} minutes)") print(f" Samples: {n_samples:,} @ {fs} Hz") print(f" Channels: {n_channels}") print(f" Movement attempts: {n_attempts}") print(f" Attempt interval: ~{attempt_interval:.1f}s") print() # Initialize with baseline noise np.random.seed(42) baseline_mean = np.array([0.01, -0.005, 0.002, 0.008]) # Small DC offset baseline_std = np.array([0.05, 0.045, 0.048, 0.046]) # Realistic noise level # Generate baseline (low-amplitude Gaussian noise) data = np.random.randn(n_samples, n_channels) * baseline_std + baseline_mean # Add 60 movement attempts with varying characteristics attempt_positions = [] for i in range(n_attempts): # Position: evenly spaced with small jitter center_pos = int((i + 0.5) * attempt_interval * fs) jitter = np.random.randint(-int(0.2 * fs), int(0.2 * fs)) # ±0.2s jitter start_pos = max(0, center_pos + jitter) # Duration: 200-500ms (40-100 samples) duration = np.random.randint(40, 100) end_pos = min(n_samples, start_pos + duration) if end_pos >= n_samples: break # Amplitude: varies to create different coordination qualities # Good attempts: 0.3-0.5, Poor attempts: 0.15-0.25 if i % 3 == 0: # Every 3rd attempt is "good" amplitude = np.random.uniform(0.3, 0.5) coordination_quality = "good" else: amplitude = np.random.uniform(0.15, 0.3) coordination_quality = "moderate" # Create movement pattern (smooth rise and fall) movement_length = end_pos - start_pos t = np.linspace(0, np.pi, movement_length) envelope = np.sin(t) # Smooth activation envelope # Add movement to all channels with slight variations for ch in range(n_channels): channel_amplitude = amplitude * np.random.uniform(0.8, 1.2) movement_signal = channel_amplitude * envelope # Add to baseline data[start_pos:end_pos, ch] += movement_signal attempt_positions.append((start_pos, end_pos, coordination_quality)) print(f"✓ Generated {len(attempt_positions)} movement attempts") # Add some artifacts (blinks, motion artifacts) n_artifacts = 10 artifact_positions = np.random.choice(n_samples, n_artifacts, replace=False) for pos in artifact_positions: # Large amplitude spike (should be rejected as artifact) artifact_amplitude = np.random.uniform(3.0, 8.0) * np.random.choice([-1, 1]) artifact_width = np.random.randint(5, 15) # 25-75ms start = max(0, pos - artifact_width // 2) end = min(n_samples, pos + artifact_width // 2) # Affect random subset of channels affected_channels = np.random.choice(n_channels, size=np.random.randint(1, n_channels+1), replace=False) for ch in affected_channels: data[start:end, ch] += artifact_amplitude * np.exp(-np.linspace(-2, 2, end-start)**2) print(f"✓ Added {n_artifacts} artifacts") # Add low-frequency drift (realistic baseline variation) drift_freq = 0.02 # Hz (50 second period) t = np.arange(n_samples) / fs drift = 0.02 * np.sin(2 * np.pi * drift_freq * t) data += drift[:, np.newaxis] print(f"✓ Added baseline drift") # Save to CSV output_file = 'test_data.csv' np.savetxt(output_file, data, delimiter=',', fmt='%.6f') print() print(f"✓ Saved to {output_file}") print() # Summary statistics print("Data Statistics:") print(f" Channel means: {np.mean(data, axis=0)}") print(f" Channel stds: {np.std(data, axis=0)}") print(f" Min value: {np.min(data):.4f}") print(f" Max value: {np.max(data):.4f}") print() # Attempt statistics good_attempts = sum(1 for _, _, q in attempt_positions if q == "good") moderate_attempts = len(attempt_positions) - good_attempts print("Movement Attempt Distribution:") print(f" Good coordination: {good_attempts} attempts") print(f" Moderate coordination: {moderate_attempts} attempts") print(f" Total: {len(attempt_positions)} attempts") print() # Expected detection expected_detectable = sum(1 for start, end, q in attempt_positions if np.max(np.abs(data[start:end])) > 0.2) print(f"Expected detectable (with threshold=0.2): {expected_detectable} attempts") print(f"Expected detectable (with adaptive ~0.25): ~{len(attempt_positions)} attempts") print() # Timeline preview print("Timeline Preview (first 20 seconds):") for start, end, quality in attempt_positions[:10]: time_s = start / fs duration_ms = (end - start) / fs * 1000 peak_amplitude = np.max(np.abs(data[start:end])) print(f" {time_s:5.1f}s: {duration_ms:3.0f}ms burst, " f"peak={peak_amplitude:.3f}, quality={quality}") print() print("✅ Test data generation complete!") print(f" Run with: python main_runtime.py --simulate {output_file} --speed 0 --calibration-time 5") ``` ### run_live_gui.py ```python #!/usr/bin/env python3 """ Integration script to run LiveGUI with the existing pipeline. Usage: # With simulated data python run_live_gui.py --simulate test_data.csv --speed 1.0 # With live UDP data python run_live_gui.py --live --ip 0.0.0.0 --port 12345 # Skip calibration python run_live_gui.py --simulate test_data.csv --skip-calibration """ import sys import argparse import threading import time import numpy as np from PyQt5.QtWidgets import QApplication from gui import LiveGUI from data_sources import UDPDataSource, SimulatedDataSource from analysis.MainPipeline import MainPipeline from analysis.BaselineCalibration import BaselineCalibration class GUIController: """Controller to connect data pipeline to GUI.""" def __init__(self, gui, data_source, calibration_time_s=60.0, skip_calibration=False): self.gui = gui self.data_source = data_source self.calibration_time_s = calibration_time_s self.skip_calibration = skip_calibration self.pipeline = None self.calibration = None self.running = True def run_calibration(self): """Run calibration phase.""" if self.skip_calibration: print("Skipping calibration...") return print(f"Starting calibration ({self.calibration_time_s}s)...") n_samples_needed = int(self.calibration_time_s * 200.0) rest_data = [] while len(rest_data) < n_samples_needed and self.data_source.is_running(): packet = self.data_source.get_packet() if packet is not None: # Use only first 4 channels (EMG + 3 EEG); ignore extra channels rest_data.append(np.asarray(packet).reshape(-1)[:4]) else: time.sleep(0.001) if len(rest_data) >= n_samples_needed: rest_data_array = np.array(rest_data) self.calibration = BaselineCalibration(fs=200.0) self.calibration.calibrate(rest_data_array) print("Calibration complete!") def run_streaming(self): """Run main data processing loop.""" print("Starting data streaming...") # Initialize pipeline self.pipeline = MainPipeline( window_size_s=0.2, activation_threshold=0.2, calibration=self.calibration ) # Set up callbacks for immediate GUI updates when async operations complete self.pipeline.on_prediction_ready = self._on_prediction_ready self.pipeline.on_similarity_ready = self._on_similarity_ready packet_count = 0 batch = [] coord_update_counter = 0 while self.running and self.data_source.is_running(): packet = self.data_source.get_packet() if packet is None: time.sleep(0.001) continue # Use only first 4 channels (EMG + 3 EEG); ignore extra channels packet = np.asarray(packet).reshape(-1)[:4] # Process through pipeline self.pipeline.run(packet) packet_count += 1 # Accumulate processed data if len(self.pipeline.processed_data) > 0: # Get latest processed sample latest = self.pipeline.processed_data[-1] batch.append(latest) # Send batch to GUI every 10 samples (~50ms @ 200Hz) if len(batch) >= 10: self.gui.add_data(np.array(batch)) batch = [] # Update coordination attempts display every 100 packets (~0.5s @ 200Hz) coord_update_counter += 1 if coord_update_counter >= 100: self._update_coordination_display() coord_update_counter = 0 print("Streaming stopped.") def _on_prediction_ready(self): """Callback when prediction completes - immediately update GUI.""" print("[DEBUG] Prediction ready - triggering immediate GUI update") self._update_coordination_display() def _on_similarity_ready(self): """Callback when similarity scores update - immediately update GUI.""" print("[DEBUG] Similarity scores ready - triggering immediate GUI update") self._update_coordination_display() def _update_coordination_display(self): """Extract top attempts from pipeline and update GUI.""" if not hasattr(self.pipeline, 'max_n_coord_cache'): print("[DEBUG] Pipeline doesn't have max_n_coord_cache yet") return try: # Lock the cache while reading with self.pipeline.cache_lock: top_nodes = self.pipeline.max_n_coord_cache.get_top_n_nodes(5) predicted_ideal_node = self.pipeline.max_n_coord_cache.predicted_ideal print(f"[DEBUG] Retrieved {len(top_nodes)} top nodes from cache") # Convert top attempts to dict format for GUI top_attempts = [] for i, node in enumerate(top_nodes, 1): print(f"[DEBUG] Node {i}: coord={node.coordination_index:.4f}, sim={node.similarity_score}") top_attempts.append({ 'coordination_index': node.coordination_index, 'similarity_score': node.similarity_score if node.similarity_score is not None else 0.0, 'attempt_id': i # Use position as ID for now }) # Convert predicted ideal to dict format predicted_ideal = None if predicted_ideal_node is not None: print(f"[DEBUG] Predicted ideal: coord={predicted_ideal_node.coordination_index:.4f}, sim={predicted_ideal_node.similarity_score}") predicted_ideal = { 'coordination_index': predicted_ideal_node.coordination_index, 'similarity_score': predicted_ideal_node.similarity_score if predicted_ideal_no [truncated — 3206 more characters] ``` [24 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]