# Project export: Flexion Detection 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: What if we could predict when muscle failure will occur in surgeons, athletes, etc? We used electromyography to measure muscle activation and predict task failure up to 40 seconds in advance. - Devpost: https://devpost.com/software/the-boys-are-flexing - GitHub: https://github.com/ericyin23/Flexion-Detection-TreeHacks-2026.git - Video: https://www.youtube.com/embed/SL0jbPdoO6Y?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — Eric Yin (45 commits), biswasdhruv01 (1 commits) ## Devpost submission (written by the team) ### Overview What it does Our system predicts muscle failure up to 40 seconds before it happens. We use an electromyography (EMG) sensor to measure muscle activation, and process this data using Python. We calculate the Root Mean Square (RMS) of the voltage and Median Domain Frequency (MDF), and using these values we can estimate the time duration before a muscle fails. These values and the estimated time are provided in a web dashboard. ### Inspiration Since we each have medical backgrounds, we explored medical devices to address unsolved problems. We found an opportunity in muscle fatigue detectors: there are expensive options available for athletes, but none targeted towards other professions. We decided to focus on high precision occupations where the likelihood of muscle fatigue is high and may lead to dangerous outcomes. With our tool, we hope that injuries or mistakes can be prevented before they occur! ### How we built it The hardware includes an ESP32, Myoware 2.0, ECG pads, and code written in Python. At a high level, a calibration step records the strength of an individual, and then python records the EMG signal in distinct batches. From there we obtain Root Mean Square values (the number of motor units firing), and the Median Domain Frequency (the frequency at which motor units fire). These are used to create a global trend which is then updated further using Bayesian statistics. The global trend allows us to infer the Time to Failure. ### Challenges we ran into The biggest challenges included dealing with a weak signal, tuning parameters, how to measure TTF appropriately, and defining what failure is. Experimentation and iteration allowed us to identify solutions to each of these issues, but there is still space for improvement. We are excited to expand our prediction model to non-isometric movements in the future. ### Accomplishments we're proud of Since none of us have software backgrounds, we were nervous to take on a project that would push us far outside our comfort zones. However, we were able to work together effectively and use the provided tools to make a product that works better than we imagined. Additionally, it was the first hackathon experience for all of our team members! We had lots of fun and made great memories! ### What we learned We have powerful tools around us A lot can be accomplished within a weekend Go outside of your comfort zone Splitting up work and trying different strategies at the same time can help for quick iteration Fitting for data without a trained model is difficult Variation between subjects can be hard to account for ### What's next We are all graduating from Stanford following this quarter! Nathan will be working at SpaceX where he will help send the next generation of rockets to space as a Components Engineer. Eric will be pursuing medical school at the University of Alberta where he will bridge the gap between cutting edge research and clinical practice. Lastly, Dhruv will be entering into the medical devices industry, hoping to help push the boundaries of bio-innovation through cutting edge medical devices. ## README (from the GitHub repository) # Treehacks 2026 - EMG Fatigue Analysis Dashboard This project implements a real-time EMG fatigue analysis dashboard using Streamlit. It captures EMG data, processes it to calculate RMS and Median Frequency (MDF), and visualizes muscle fatigue indicators. ## Features - **Real-time EMG Monitoring**: visualizes raw EMG signals. - **Fatigue detection**: Calculates RMS and MDF to detect muscle fatigue. - **Time-to-Failure Prediction**: Estimates time remaining before muscle failure based on MDF trends. - **Calibration**: Automates calibration based on the first 5 seconds of stable data. - **Data Recording**: Saves session data to CSV files for further analysis. ## Hardware Requirements - ESP32 with EMG sensor - Serial connection to the PC ## Installation 1. Clone the repository. 2. Install the required dependencies: ```bash pip install -r requirements.txt ``` ## Usage To start the dashboard, run: ```bash streamlit run src/dashboard.py ``` ## Project Structure - `src/`: Contains the source code (`dashboard.py`, `core_logic.py`, `session.py`). - `data/`: Directory for storing recorded EMG data (ignored by git). - Contains CSV files with raw EMG recordings. - Columns include: `Timestamp`, `Value` (ADC reading). - Used for playback mode and data analysis. - `agents/`: Contains agent-related documentation. - `firmware/`: Contains ESP32 firmware code. ## License [Creative Commons Attribution-NonCommercial 4.0 International License](LICENSE) ## Detected evidence (automated analysis) Indexed codebase: 7 recognized source files, 39 KB. - Python (language) — detected in the code - Streamlit (technology) — detected in the code - Google Gemini (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (14 of 14) ``` .gitignore agents/agent_hackathon.md data/Clenching.csv data/dummy_fatigue_test.csv data/Failure.csv firmware/esp32_emg.ino LICENSE README.md requirements.txt src/core_logic.py src/dashboard.py src/record_data.py src/session.py src/utils.py ``` ### Dependencies - requirements.txt: matplotlib, numpy, openpyxl, pandas, pyserial, scikit-learn, scipy, streamlit ### Recent commits (newest first) - Merge pull request #18 from ericyin23/architecture - add license - Delete data/LICENSE - Create LICENSE - Merge pull request #17 from ericyin23/architecture - update readme - cleaned up architecture - Merge pull request #16 from ericyin23/update-calibration-window - updated 7 second calibration - Merge pull request #15 from ericyin23/skip2plots - update damping from 10 to 5% - Merge pull request #14 from ericyin23/skip2plots - cleaned up post graphs - Merge pull request #13 from ericyin23/ui-newman - final yaxis adjustment - update y axis for ui - Merge pull request #11 from ericyin23/bayesian-update - adding damping factor to decrease influence of later updates - made sampling deterministic - Merge pull request #10 from ericyin23/recorddata ## Key source files (fetched from GitHub, selected and truncated for size) ### agents/agent_hackathon.md ```markdown # Agent: Live Hackathon Dashboard ## Data Input * Source: `pyserial` (ESP32 via USB). * Format: 12-bit Raw ADC values with a delimiter. ## Logic Flow 1. **Calibration Mode:** A Streamlit button starts a 5-second timer. User pulls MVC. * Store `ref_mdf` and `ref_rms`. 2. **Live Mode:** Continuous 1000Hz stream. * Apply `preprocess()` and `extract_features()` from `core_logic.py`. 3. **Visualization:** * **RMS Gauge:** Show %MVC (0-100%). * **MDF Plot:** Live spectral shift. * **Countdown:** Large "Seconds until Failure" indicator based on the MDF slope. 4. **Safety:** If the signal clips at 4095, show a "Gain Too High" warning. ``` ### requirements.txt ``` numpy scipy pandas streamlit pyserial scikit-learn matplotlib openpyxl ``` ### src/utils.py ```python import serial.tools.list_ports def get_serial_ports(): """Returns a list of available serial ports.""" return [p.device for p in serial.tools.list_ports.comports()] ``` ### src/record_data.py ```python import serial import csv import time import argparse import os from datetime import datetime # Import utils to find ports try: from utils import get_serial_ports except ImportError: # Fallback if run from src directly without package context import sys sys.path.append(os.path.dirname(__file__)) from utils import get_serial_ports def record_data(port, baud=115200, output_dir="data"): """ Connects to the specified serial port and saves data to a CSV file. """ if not os.path.exists(output_dir): os.makedirs(output_dir) timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") filename = f"recording_{timestamp}.csv" filepath = os.path.join(output_dir, filename) print(f"Connecting to {port} at {baud} baud...") try: ser = serial.Serial(port, baud, timeout=1) # Clear buffer ser.reset_input_buffer() print("Connected! Recording... Press Ctrl+C to stop.") with open(filepath, 'w', newline='') as csvfile: writer = csv.writer(csvfile) writer.writerow(["Timestamp", "Value"]) # Header start_time = time.time() count = 0 while True: if ser.in_waiting: line = ser.readline() try: decoded = line.decode('utf-8').strip() if decoded: val = int(decoded) # Record relative time from start current_time = time.time() - start_time writer.writerow([f"{current_time:.6f}", val]) count += 1 if count % 1000 == 0: print(f"Recorded {count} samples...", end='\r') except ValueError: pass # Ignore garbled data except KeyboardInterrupt: print("\nRecording stopped by user.") except Exception as e: print(f"\nError: {e}") finally: if 'ser' in locals() and ser.is_open: ser.close() print(f"Data saved to {filepath}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Record EMG data from serial port.") parser.add_argument("--port", help="Serial port to connect to (e.g., COM3)") parser.add_argument("--baud", type=int, default=115200, help="Baud rate (default: 115200)") args = parser.parse_args() port = args.port if not port: ports = get_serial_ports() if not ports: print("No serial ports found!") # Default to manual input if no ports detected port = input("Enter port name manually (e.g. COM3): ") else: print("Available ports:") for i, p in enumerate(ports): print(f"{i+1}: {p}") selection = input("Select port number (or press Enter for first): ") if selection.isdigit() and 1 <= int(selection) <= len(ports): port = ports[int(selection)-1] else: port = ports[0] record_data(port, args.baud) ``` ### src/core_logic.py ```python import numpy as np from scipy import signal, stats import time class FatigueEngine: def __init__(self, fs=1000): self.fs = fs # 60Hz Notch + 20-450Hz Bandpass (Nyquist is 500Hz) self.b_notch, self.a_notch = signal.iirnotch(60, 30, fs) self.b_band, self.a_band = signal.butter(4, [20, 450], btype='bandpass', fs=fs) self.ref_mdf = None self.ref_rms = None self.mdf_history = [] self.ref_voltage_limit = 2.5 # Default to 2.5V # Bayesian TTF State (Total Session Duration D) self.mu = 60.0 # Prior mean: 60s self.sigma_sq = 30.0**2 # Prior variance self.obs_sigma_sq = 30.0**2 # High Observation noise variance to smooth spikes self.update_count = 0 def set_references(self, rms, mdf, voltage_limit=2.5): self.ref_rms = rms self.ref_mdf = mdf self.ref_voltage_limit = voltage_limit self.mdf_history = [] # Reset history on new calibration # Reset Bayesian state self.mu = 60.0 self.sigma_sq = 30.0**2 self.update_count = 0 def preprocess(self, raw_adc): """Standardizes ESP32 12-bit ADC or Mendeley integers to Centered Voltage.""" # 1. Reject local spikes (outliers) using Median Filter # Kernel size 3 or 5 is usually enough for sharp transients at 1kHz spikeless = signal.medfilt(np.array(raw_adc, dtype=float), kernel_size=3) # 2. Convert to 5.0V scale and remove DC offset voltage = (spikeless / 4095.0) * 5.0 centered = voltage - np.mean(voltage) notched = signal.lfilter(self.b_notch, self.a_notch, centered) output = signal.lfilter(self.b_band, self.a_band, notched) # Determine if signal is active (simple threshold on RMS) # This prevents noise processing when not flexing if np.sqrt(np.mean(output**2)) < 0.05: # 50mV noise floor guess return output return output def extract_features(self, filtered_window): """Calculates RMS (%MVC) and Median Frequency.""" windowed = filtered_window * np.hanning(len(filtered_window)) rms = np.sqrt(np.mean(windowed**2)) # PSD extraction freqs, psd = signal.welch(windowed, fs=self.fs, nperseg=len(windowed)) # Avoid minor errors if PSD is empty or sum is 0 if np.sum(psd) == 0: mdf = 0 else: mdf = freqs[np.where(np.cumsum(psd) >= np.sum(psd) / 2)[0][0]] return rms, mdf def get_time_remaining(self, current_mdf): """Calculates stable Time To Failure using smoothed trends.""" if self.ref_mdf is None: return None self.mdf_history.append(current_mdf) # Keep history manageable (reverted to 20) if len(self.mdf_history) > 20: self.mdf_history.pop(0) # We need enough points for smoothing and regression (reverted to 5) if len(self.mdf_history) < 5: return None # 1. Smooth the MDF history using Savitzky-Golay try: # Reverted to smaller win_size for shorter history win_size = min(len(self.mdf_history) // 2 * 2 - 1, 5) if win_size >= 3: smoothed_history = signal.savgol_filter(self.mdf_history, win_size, polyorder=2) else: smoothed_history = np.array(self.mdf_history) except: smoothed_history = np.array(self.mdf_history) threshold = 0.6 * self.ref_mdf # 2. Reverted to 5 points for slope estimation reg_points = min(5, len(smoothed_history)) y = smoothed_history[-reg_points:] x = np.arange(reg_points) slope, intercept, r_value, p_value, std_err = stats.linregress(x, y) # 3. Predict TTF: (Threshold - Current_Smoothed) / Slope current_smooth = smoothed_history[-1] # Only predict if slope is negative (fatiguing) and somewhat significant if slope < -0.005: ttf = (threshold - current_smooth) / slope # Clamp to 300s (5 mins) to prevent massive spikes on low slopes return min(300.0, max(0.0, float(ttf))) return None def update_bayesian_estimate(self, elapsed_time, local_ttf): """Updates the posterior estimate of total session duration.""" if local_ttf is None: # If no local prediction, we still have a prior/posterior estimate # Returning mu - elapsed gives the current "best guess" countdown return max(0.0, self.mu - elapsed_time), np.sqrt(self.sigma_sq) # Observation of total duration: D_obs = t + TTF_local # t is time specifically since the START of post-calibration monitoring d_obs = elapsed_time + local_ttf self.update_count += 1 # Dynamic observation noise: increases with time to trust new data less # Start at 1.0 * base, increase by 10% per update dynamic_obs_sigma_sq = self.obs_sigma_sq * (1 + 0.5 * self.update_count) # Kalman-like update for the mean of a Gaussian with known variance denom = self.sigma_sq + dynamic_obs_sigma_sq self.mu = (dynamic_obs_sigma_sq * self.mu + self.sigma_sq * d_obs) / denom self.sigma_sq = (self.sigma_sq * dynamic_obs_sigma_sq) / denom return max(0.0, self.mu - elapsed_time), np.sqrt(self.sigma_sq) ``` ### src/dashboard.py ```python import streamlit as st import numpy as np import time import pandas as pd import sys import os import altair as alt # Ensure src is in path sys.path.append(os.path.dirname(__file__)) from session import EMGSession from utils import get_serial_ports # Configuration UPDATE_DELAY = 0.01 # Faster update for serial st.set_page_config(page_title="Hackathon Dashboard", layout="wide") # Optimized Layout CSS st.markdown(""" """, unsafe_allow_html=True) # --- State Initialization --- if 'emg_session' not in st.session_state: st.session_state.emg_session = EMGSession() if 'show_analysis' not in st.session_state: st.session_state.show_analysis = False st.session_state.final_elapsed_time = 0 # --- Sidebar --- with st.sidebar: st.header("Settings") # Data Source Selection data_source = st.radio("Data Source", ["Live Serial", "Playback"]) if data_source == "Live Serial": # Port Selection ports = get_serial_ports() port = st.selectbox("Port", ports) if ports else st.text_input("Port (Manual)", "COM3") baud = st.number_input("Baud", 115200) record_data = st.checkbox("Record to CSV", value=True) st.divider() # Control Buttons col_start, col_stop = st.columns(2) with col_start: if st.button("Start Live", type="primary"): st.session_state.emg_session.connect(port, baud, record=record_data) with col_stop: if st.button("Stop"): st.session_state.emg_session.disconnect() else: # Playback st.markdown("### Playback Files") data_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data") if not os.path.exists(data_dir): os.makedirs(data_dir) files = [f for f in os.listdir(data_dir) if f.endswith(".csv")] selected_file = st.selectbox("Select Recording", files) if files else st.text("No CSV files found in /data") st.divider() col_start, col_stop = st.columns(2) with col_start: if st.button("Start Playback", type="primary", disabled=not files): if selected_file: st.session_state.emg_session.connect_playback(os.path.join(data_dir, selected_file)) if st.button("Skip to End"): if st.session_state.emg_session.mode == 'playback': st.session_state.emg_session.complete_playback() st.session_state.final_elapsed_time = st.session_state.emg_session.get_current_time() st.session_state.emg_session.disconnect() st.session_state.show_analysis = True st.rerun() if st.button("Mark Failure (Manual) & Analyze", type="primary"): # Capture end time relative to start st.session_state.final_elapsed_time = st.session_state.emg_session.get_current_time() st.session_state.emg_session.disconnect() st.session_state.show_analysis = True st.rerun() st.divider() with col_stop: if st.button("Stop"): st.session_state.emg_session.disconnect() st.session_state.show_analysis = False # --- Main Layout --- st.title("⚡ Muscle Fatigue Monitor") col1, col2, col3, col4 = st.columns(4) with col1: rms_metric = st.empty() with col2: mdf_metric = st.empty() with col3: ttf_metric = st.empty() with col4: elapsed_metric = st.empty() status_alert = st.empty() # Charts col_left, col_right = st.columns(2) with col_left: st.markdown("##### RMS Signal (Voltage)") rms_chart = st.empty() with col_right: st.markdown("##### MDF History") mdf_chart = st.empty() # --- Main Loop --- if st.session_state.emg_session.is_monitoring: # Reset analysis flag when monitoring starts st.session_state.show_analysis = False while st.session_state.emg_session.is_monitoring: # 1. Acquire Data has_data = st.session_state.emg_session.read_data() # Check for End of Playback if not has_data and st.session_state.emg_session.is_playback_complete(): st.session_state.emg_session.disconnect() # Calculate final time st.session_state.final_elapsed_time = st.session_state.emg_session.get_current_time() st.session_state.show_analysis = True st.rerun() has_new_metrics = st.session_state.emg_session.process_buffer() # Check for auto-detected failure if st.session_state.emg_session.failure_detected: st.session_state.final_elapsed_time = st.session_state.emg_session.get_current_time() st.session_state.emg_session.disconnect() st.session_state.show_analysis = True st.rerun() # 3. UI Updates metrics = st.session_state.emg_session.metrics status = st.session_state.emg_session.status_message status_type = st.session_state.emg_session.status_type # Status & Time elapsed = st.session_state.emg_session.get_current_time() elapsed_metric.metric("Time Elapsed", f"{elapsed:.1f}s") if status_type == "error": status_alert.error(status) elif status_type == "warning": status_alert.warning(status) elif status_type == "success": status_alert.success(statu [truncated — 7563 more characters] ``` ### src/session.py ```python from collections import deque import numpy as np import time from datetime import datetime from core_logic import FatigueEngine import serial import csv import os class EMGSession: def __init__(self, window_size=500): self.engine = FatigueEngine() self.window_size = window_size self.processing_queue = [] # Unbounded list to hold incoming data self.data_buffer = deque(maxlen=window_size) # For UI visualization self.is_monitoring = False self.is_calibrated = False self.calibration_accumulator = [] self.calibration_sample_index = 0 # Sample count when calibration finished self.status_message = "Ready" self.status_type = "info" # info, success, warning, error self.metrics = {"rms": 0, "mdf": 0, "ttf": None, "ttf_error": None} self.serial = None # Playback State self.mode = None # 'serial' or 'playback' self.playback_data = [] # List of values self.playback_idx = 0 self.last_playback_time = 0 # Recording State self.is_recording = False self.recording_file = None self.csv_writer = None # Session Tracking self.session_start_time = 0.0 # Time when connect/playback started (Machine time) self.total_samples_processed = 0 # Deterministic "Time" self.step_size = 50 # Process a window every 50 samples (20Hz update rate) self.prediction_history = [] # List of tuples (elapsed_time, predicted_ttf, std_dev) self.rms_history = [] # List of tuples (elapsed_time, rms) self.mdf_history = [] # List of tuples (elapsed_time, mdf) self.failure_detected = False def get_current_time(self): """Returns deterministic time based on samples processed.""" # t = samples / fs return self.total_samples_processed / self.engine.fs def add_data(self, chunk): """Adds raw data chunk to buffer.""" if len(chunk) > 0: self.processing_queue.extend(chunk) # Also update UI buffer self.data_buffer.extend(chunk) # Auto-calibration accumulation (both live and playback modes) if not self.is_calibrated and self.mode in ['serial', 'playback']: self.calibration_accumulator.extend(chunk) def process_buffer(self): """Runs the fatigue engine on the processing queue in deterministic steps.""" has_updated = False # Auto-calibration check if not self.is_calibrated and self.mode in ['serial', 'playback']: required_samples = int(7 * self.engine.fs) # 7 seconds if len(self.calibration_accumulator) >= required_samples: self._perform_calibration(self.calibration_accumulator[:required_samples]) self.calibration_accumulator = [] # Clear after calibration # We align the "start" of monitoring with the current sample count self.calibration_sample_index = self.total_samples_processed # Process all available full windows in the queue while len(self.processing_queue) >= self.window_size: # 1. Extract Window window = self.processing_queue[:self.window_size] data_array = np.array(window) # 2. Process processed = self.engine.preprocess(data_array) rms, mdf = self.engine.extract_features(processed) # 3. Compute Deterministic Time # The "time" for this window is effectively the end of the window? # Or the beginning? Let's say end of window = current head + window_size # But simpler: track 'total_samples_processed' as the leading edge # Actually, let's define time as 'time at the end of this window' current_time = (self.total_samples_processed + self.window_size) / self.engine.fs # Only compute TTF if calibrated ttf = None if self.is_calibrated: local_ttf = self.engine.get_time_remaining(mdf) # Deterministic time since calibration # If calibration finished at sample X, and we are at sample Y (end of window) # t_from_cal = (Y - X) / fs window_end_sample = self.total_samples_processed + self.window_size t_from_cal = (window_end_sample - self.calibration_sample_index) / self.engine.fs if local_ttf is not None: # Bayesian Update ttf_bayesian, ttf_std = self.engine.update_bayesian_estimate(t_from_cal, local_ttf) self.prediction_history.append((current_time, ttf_bayesian, ttf_std)) ttf = ttf_bayesian else: # Bayesian Update with no observation ttf_bayesian, ttf_std = self.engine.update_bayesian_estimate(t_from_cal, None) self.prediction_history.append((current_time, ttf_bayesian, ttf_std)) ttf = ttf_bayesian # Update History self.rms_history.append((current_time, rms)) self.mdf_history.append((current_time, mdf)) self.metrics = { "rms": rms, "mdf": mdf, "ttf": ttf, "ttf_error": self.metrics.get('ttf_error') # Keep old if not updated? Or None? } if ttf is not None: self.metrics["ttf_error"] = self.engine.sigma_sq**0.5 # approximate std dev # 4. Stride Forward # Remove 'step_size' samples from the front # If step_size < window_size, we have overlap. self.processing_queue = self.processing_queue[self.step_size:] self.total_samples_processed += self.step_ [truncated — 9330 more characters] ```