/* * ECG Data Collection using the AD8232 module with ESP32. * This code collects the ECG signal and applies a digital notch filter to eliminate * 60Hz power line noise. * The built-in LED lights up when the AD8232 module detects a loose electrode. * * Notch filter based on: * CHOY, T. T. C.; LEUNG, P. M. Real-time microprocessor-based * 50 Hz notch filter for ECG. Journal of Biomedical Engineering, * v. 10, n. 3, p. 285-288, 1988. * * Filter options vary with the sampling frequency (Fs): * For Fs = 120Hz, use "N_POINTS 2" and "N 1" * For Fs = 240Hz, use "N_POINTS 3" and "N 2" * For Fs = 360Hz, use "N_POINTS 4" and "N 3" * "a" should be between 0 and 1: * - "a" close to 0 distorts the ECG signal more but is less susceptible to variations * in noise and sampling frequencies. * - "a" close to 1 distorts the ECG signal less but is more susceptible to variations * in noise and sampling frequencies. * * Created by: Erick Dario León Bueno de Camargo, 2023 * Adapted by: [Your Name], 2023 */// Desired sampling frequency in Hz:#define SAMPLE_FREQUENCY 120// Constants for the notch filter#define N_POINTS 2#define N 1#define a 0.05// Pin definitions based on your hardware setup#define ANALOG_PIN 39 // Use GPIO39 (Port VN in) (ADC1_CH0) for analog input#define LO_MINUS_PIN 19 // GPIO19 for "Lo-"#define LO_PLUS_PIN 18 // GPIO18 for "Lo+"#define LED_PIN 2 // Built-in LED (adjust if necessary)unsigned long sample_interval_us = 1000000UL / SAMPLE_FREQUENCY;int X[N_POINTS]; // Array for current measurementsint Y[N_POINTS]; // Array for filtered measurementsunsigned long current_time, previous_time;int counter = 0;void setup() { Serial.begin(115200); pinMode(LO_MINUS_PIN, INPUT); pinMode(LO_PLUS_PIN, INPUT); pinMode(LED_PIN, OUTPUT); previous_time = micros();}void loop() { current_time = micros(); // Check for loose electrodes if ((digitalRead(LO_MINUS_PIN) == HIGH) || (digitalRead(LO_PLUS_PIN) == HIGH)) { digitalWrite(LED_PIN, HIGH); } else { digitalWrite(LED_PIN, LOW); // Sampling at precise intervals if (current_time - previous_time >= sample_interval_us) { previous_time += sample_interval_us; // Maintain consistent sampling rate // Read analog value from the ECG module X[counter] = analogRead(ANALOG_PIN); int previous_index = (counter + N_POINTS - N) % N_POINTS; // Apply notch filter Y[counter] = ((X[counter] + X[previous_index]) / 2) + a * (((X[counter] + X[previous_index]) / 2) - Y[previous_index]); // Output the filtered signal Serial.println(Y[counter]); // Update counter counter = (counter + 1) % N_POINTS; } }}
Exemplo de dados coletados
Python para coleta dos dados da porta serial usando a lib pyserial.
import serialimport time# Adjust the serial port and baud rate to match your setup# ser = serial.Serial('COM3', 115200) # For Windows, e.g., 'COM3'# ser = serial.Serial('/dev/ttyUSB0', 115200) # For Linuxser = serial.Serial('/dev/tty.usbserial-112220', 115200) # For macOS# Create or open a file to write the datawith open('./ecg_data.csv', 'w') as file: # Write header if needed file.write('timestamp,data\n') # Collect data for a specified duration or until stopped try: while True: # Read a line from the serial port line = ser.readline().decode('utf-8').strip() # Get the current timestamp timestamp = time.time() # Write timestamp and data to the file file.write(f'{timestamp},{line}\n') # Print to the console for monitoring (optional) print(f'{timestamp},{line}') except KeyboardInterrupt: print("Data collection stopped by user.")# Close the serial portser.close()
Código pandas com hvplot usado para “plotar” os dados:
import pandas as pdimport hvplot.pandas # hvPlot extension# Load the ECG datafile_path = 'ecg_data.csv'ecg_data = pd.read_csv(file_path)# Convert the timestamp to a more readable format if necessary# Assuming timestamp is in seconds, convert it to datetimeecg_data['timestamp'] = pd.to_datetime(ecg_data['timestamp'], unit='s')# Convert 'data' to numeric, forcing errors to NaN (e.g., "S1774" is invalid)ecg_data['data'] = pd.to_numeric(ecg_data['data'], errors='coerce')# Create an interactive plot using hvPlotecg_data.hvplot(x='timestamp', y='data', title='ECG Data Over Time', width=800, height=400)
import neurokit2 as nk# Load ECG signal (example)ecg_signal = nk.ecg_simulate(duration=10, heart_rate=60)# Process the ECG signalecg_signals, info = nk.ecg_process(ecg_signal, sampling_rate=500)# Extract R-peaksr_peaks = info['ECG_R_Peaks']# Visualize the ECG signal and R-peaksnk.ecg_plot(ecg_signals)
