BIOSIGNALS: Real-Time Physiological Signal Acquisition for Human-Robot Interaction

BIOSIGNALS software interface: Real-time monitoring of BVP, GSR, Temperature, and Accelerometer data with PyQt5 GUI

The Challenge: Synchronised Multi-Modal Physiological Data in Real-Time

In affective computing and human-robot interaction research, understanding human emotions requires capturing multiple physiological signals simultaneously whilst maintaining precise temporal synchronisation with experimental events. However, most commercial solutions are either prohibitively expensive, closed-source, or lack the sub-100ms event synchronisation needed for dynamic HRI protocols.

During my PhD research on Emotional Human-Robot Interaction at Universidad Nacional de Educación a Distancia (UNED), I encountered a fundamental technical gap: no existing system could acquire heterogeneous physiological signals at different sampling rates whilst providing TCP/IP-based remote triggering for closed-loop robotic control.

The system needed to:

• Acquire 4 signal modalities simultaneously with per-channel thread-safe ring buffers
• Visualise in real-time (>30 FPS) without blocking data acquisition threads
• Respond to TCP/IP triggers with minimal latency for closed-loop HRI experiments
• Annotate events with microsecond timestamps embedded in EDF+ format
• Process HRV and GSR features online for real-time emotion classification
• Maintain modular architecture for integration with external ML pipelines

System Architecture: Multi-Threaded Event-Driven Design

BIOSIGNALS implements a multi-threaded event-driven architecture with strict separation of concerns to handle concurrent I/O, real-time visualisation, and remote control without race conditions.

Software architecture: Multi-threaded design with state machine, data managers, ring buffers, and TCP/IP trigger server

┌──────────────────────────────────────────────────────────────┐
│              BIOSIGNALS_APP_01.py (Main Thread)              │
│                                                               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │    State     │─▶│   Socket     │─▶│Data Managers │      │
│  │   Machine    │  │   Threads    │  │  (x4 types)  │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
│         │                 │                    │             │
│         ▼                 ▼                    ▼             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │  GUI Layer   │  │   Trigger    │  │Ring Buffers  │      │
│  │   (PyQt5)    │  │   Server     │  │(Thread-Safe) │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
└──────────────────────────────────────────────────────────────┘
                    

1. Finite State Machine for Lifecycle Management

The application implements an FSM to manage connection states and recording transitions:

class MyApp(QtWidgets.QApplication):
    """
    Main application with state machine control.
    
    States:
        SERVER → Disconnected from Empatica Server
        DEVICE → Connected, waiting for E4 device
        VIEW   → Device streaming data
    
    Substates (in VIEW):
        OFF        → Paused
        ON         → Recording
        WAIT_PAUSE → Transitioning
    """
    def __init__(self):
        self.state = "SERVER"
        self.substate = ""
        self.trigger_control = False  # Remote trigger flag
        self.pause_control = True     # Local pause flag
        
    @QtCore.pyqtSlot(str)
    def trigger_event(self, action):
        """Handle TCP/IP trigger events with FSM logic"""
        if self.state == "VIEW":
            if not self.trigger_control:  # Not recording
                if action == 'start':
                    # Create EDF files
                    for dmg in self.dmgs:
                        dmg.create_file()
                    self.start()
                    self.thread.flag.set()
                    self.trigger_control = True
                    
            else:  # Already recording
                if action in ['start', 'stop']:
                    # Stop and save
                    self.start()
                    for dmg in self.dmgs:
                        dmg.save_streamData()
                        dmg.reset_data_store()
                    self.trigger_control = False
                else:
                    # Annotate event during recording
                    for dmg in self.dmgs:
                        dmg.online_annotation(action)

2. Thread-Safe Ring Buffer Implementation

Each signal has its own circular buffer with mutex-protected concurrent access:

class RingBuffer(QtCore.QThread):
    emitter = QtCore.pyqtSignal()
    
    def __init__(self, channels, num_samples, sample_rate):
        self.max = num_samples
        self.data = np.zeros((self.max, channels))
        self.cur = 0  # Current write position
        self.cur_show = self.max  # Display offset
        
        # Visualisation control
        self.seconds = 6
        self.control = sample_rate * self.seconds
        
    def append(self, x):
        """O(1) circular insertion"""
        self.cur = self.cur % self.max
        self.data[self.cur, :] = np.array(x)
        self.cur += 1
        
        if self.cur_show > 0:
            self.cur_show -= 1
            
        # Emit signal every N seconds for GUI update
        if (self.cur_show == 0) and ((self.cur % self.control) == 0):
            self.emitter.emit()
            
    def get(self):
        """Return ordered data (oldest → newest)"""
        data = np.vstack((self.data[self.cur:, :], 
                          self.data[:self.cur, :]))
        return data[self.cur_show:, :]

3. TCP/IP Trigger Server for Remote Control

A dedicated thread handles remote commands for experimental automation:

class TriggerServer(QtCore.QThread):
    socket_emitter = QtCore.pyqtSignal(str)
    
    def create_socket(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_address = ('localhost', 10000)
        self.sock.bind(self.server_address)
        self.activated = True
        
    def run(self):
        self.sock.listen(1)
        while self.activated:
            connection, client = self.sock.accept()
            while True:
                data = connection.recv(128)
                if data:
                    # Emit event to main thread
                    self.socket_emitter.emit(data.decode())

# Client usage (from any Python script)
from COM.trigger_client import trigger_client

tc = trigger_client('192.168.1.100', 10000)
tc.create_socket()
tc.connect()

tc.send_msg(b'start')           # Begin recording
tc.send_msg(b'stimulus_happy')  # Annotate event
tc.send_msg(b'stop')            # Stop and save

Heart Rate Variability: Robust Multi-Stage Peak Detection

Extracting clean NN intervals (Normal-to-Normal) from photoplethysmography signals is challenging due to motion artifacts, low SNR, and inter-individual morphology variations. Our pipeline implements a 6-stage robust algorithm:

def compute_nni(hrdata, sample_rate=64, sliding_window=0.5, 
                prominence=0.1, dist_q1=50, dist_q2=120,  
                std_window=6, std_th=130, method='remove'):
    """
    Compute NN intervals from BVP signal with robust artifact handling.
    
    Args:
        hrdata: Raw BVP signal (numpy array)
        sample_rate: Sampling frequency (Hz)
        sliding_window: Window for adaptive envelope (seconds)
        prominence: Minimum peak prominence (normalized)
        dist_q1, dist_q2: Valid HR range (bpm) [50-120]
        std_window: STD convolution window (samples)
        std_th: Artifact threshold (STD threshold)
        method: 'remove', 'iqr', 'modified_z' for artifact handling
    
    Returns:
        nni_revised: Clean NN intervals (milliseconds)
    """
    
    # Step 1: Signal inversion and smoothing
    hrdata_inv = hrdata * (-1)
    roll_mean = savgol_filter(hrdata_inv, 81, 2)  # Order-2 S-G
    
    # Step 2: Adaptive upper envelope
    windowsize = int(sliding_window * sample_rate)
    add = np.zeros(int(windowsize / 2))
    add[:] = np.nan
    hrdata_ext = np.concatenate((add, hrdata_inv, add))
    
    roll_max = []
    for i in range(len(hrdata)):
        roll_max.append(np.nanmax(hrdata_ext[i:i+windowsize]))
    
    sroll_max = savgol_filter(roll_max, 51, 2)
    mn = 0.3 * np.std(sroll_max)
    sroll_max = sroll_max + mn
    
    # Step 3: Peak enhancement (simplified representation)
    simpleHR_1 = (hrdata_inv - roll_mean) * (hrdata_inv > roll_mean)
    envoltorio = minmax(sroll_max - roll_mean)
    simpleHR_2_raw = sigmoid(0, 2, 5, envoltorio) * simpleHR_1
    simpleHR_2 = savgol_filter(simpleHR_2_raw, 31, 2)
    
    # Step 4: Peak detection
    peaksx = np.where((simpleHR_2 > 0))[0]
    peaksy = simpleHR_2[peaksx]
    peaks, _ = find_peaks(peaksy, prominence=prominence)
    
    # Step 5: NN intervals + physiological filter
    nni = tools.nn_intervals((peaksx[peaks] / sample_rate) * 1000)
    hr = tools.heart_rate(nni)
    
    index = np.logical_and((hr >= dist_q1), (hr <= dist_q2))
    nni_revised = nni[index]
    
    # Step 6: Convolutional STD artifact detection (O(n))
    std = std_convoluted(nni_revised, std_window)
    index_std = [i for i in range(len(std)) if std[i] > std_th]
    
    # Remove artifact segments
    if method == 'remove':
        nni_revised[index_std] = np.nan
    elif method == 'iqr':
        nni_revised[index_std] = outliers_iqr_method(nni_revised)
    
    return nni_revised[~np.isnan(nni_revised)]


def std_convoluted(nni, N):
    """
    Compute local STD via convolution (O(n) vs O(n*k) sliding window).
    
    Uses: Var(X) = E[X²] - E[X]²
    Convolves both X and X² with uniform kernel.
    """
    im = np.array(nni, dtype=np.float64)
    im2 = im**2
    kernel = np.ones(2*N+1)
    
    s = convolve(im, kernel, mode="same")      # Sum of X
    s2 = convolve(im2, kernel, mode="same")    # Sum of X²
    ns = convolve(np.ones(im.shape), kernel, mode="same")
    
    return np.sqrt((s2 - s**2 / ns) / ns)  # STD formula

HRV Feature Extraction: Temporal, Frequency, Non-Linear

Once clean NN intervals are obtained, we extract a comprehensive feature set spanning three domains:

def compute_features(nni):
    """Extract comprehensive HRV features."""
    features = {}
    
    # Temporal domain
    features['mean_hr'] = tools.heart_rate(nni).mean()
    features['sdnn'] = td.sdnn(nni)[0]      # Global variability
    features['rmssd'] = td.rmssd(nni)[0]    # Vagal tone indicator
    features['pnn50'] = td.nn50(nni)[1]     # % NNI > 50ms diff
    
    # Frequency domain (Welch PSD)
    psd = fd.welch_psd(nni, show=False)
    features['hf_lf_ratio'] = psd['fft_ratio']  # Sympatho-vagal
    features['lf'] = psd['fft_peak'][1]         # 0.04-0.15 Hz
    features['hf'] = psd['fft_peak'][2]         # 0.15-0.4 Hz
    features['log_lf'] = psd['fft_log'][1]
    features['log_hf'] = psd['fft_log'][2]
    
    # Non-linear
    features['sampen'] = nl.sample_entropy(nni)[0]  # Complexity
    
    return features

GSR Decomposition: Tonic and Phasic Components

Galvanic Skin Response (GSR), also known as Electrodermal Activity (EDA), measures skin conductance controlled by sympathetic nervous system activity. The signal comprises two components:

Decomposition Algorithm

def extract_gsr_components(gsr_data):
    """
    Decompose GSR into tonic and phasic components.
    
    Process:
    1. Upsample to 8 Hz (standard for GSR analysis)
    2. Rolling mean (window=20) for tonic extraction
    3. Subtraction for phasic component
    4. Savitzky-Golay refinement of tonic
    
    Args:
        gsr_data: DataFrame ['datetime', 'EDA']
    
    Returns:
        DataFrame ['EDA', 'phasic', 'tonic']
    """
    gsr_data = pd.DataFrame(gsr_data, columns=['datetime', 'EDA'])
    sampleRate = 4  # Empatica E4 native rate
    startTime = gsr_data.iloc[0, 0]
    
    # Interpolate to 8 Hz
    gsr_data = interpolateDataTo8Hz(gsr_data, sampleRate, startTime)
    
    # Tonic: rolling mean
    rolling_mean = gsr_data.EDA.rolling(window=20).mean()
    
    # Phasic: signal - tonic
    gsr_data['phasic'] = gsr_data.EDA - rolling_mean
    
    # Refine tonic with Savitzky-Golay
    window_length = int(len(gsr_data) / 100) * 2 + 1
    gsr_data['tonic'] = savgol_filter(gsr_data.EDA, 
                                      window_length, 2)
    
    return gsr_data


def compute_phasic_features(gsr_data):
    """
    Detect and characterise Skin Conductance Responses (SCRs).
    
    SCR Detection Criteria:
    - Amplitude threshold: >0.1 μS
    - Must return to baseline
    - Minimum 2 inflection points
    
    Features per SCR:
        - start/end: Temporal bounds (sample indices)
        - peak_locs: Peak location (sample index)
        - amp: Amplitude (μS)
        - rise_time: Onset → peak (samples)
        - recovery_time: Peak → baseline (samples)
    
    Returns:
        DataFrame with all detected SCRs
    """
    aux1 = np.diff(gsr_data.phasic > 0.1)  # Activation
    aux2 = np.diff(gsr_data.phasic < 0)     # Return
    true_list = np.where(aux2)[0]
    
    peaks = {
        'start': [], 'end': [], 'peak_locs': [], 
        'amp': [], 'rise_time': [], 'recovery_time': []
    }
    
    for ini, end in zip(true_list, true_list[1:]):
        indx_onsets = np.where(aux1[ini:end])[0]
        
        if len(indx_onsets) >= 2:  # Valid SCR
            start = ini + indx_onsets[0]
            finish = end
            
            peaks['start'].append(start)
            peaks['end'].append(end)
            
            # Amplitude
            segment = gsr_data.phasic[start:finish]
            peak_amp = segment.max()
            peaks['amp'].append(np.abs(peak_amp - gsr_data.phasic[start]))
            
            # Peak location
            peak_loc = np.where(segment == peak_amp)[0][0]
            peaks['peak_locs'].append(start + peak_loc)
            
            # Temporal characteristics
            peaks['rise_time'].append(peak_loc)
            peaks['recovery_time'].append((finish - start) - peak_loc)
    
    return pd.DataFrame.from_dict(peaks)


def compute_tonic_features(gsr_data, fs, seconds, overlap=0.9):
    """
    Tonic features with overlapping windows.
    
    Features:
        - offset: Linear regression intercept (baseline level)
        - slope: Linear regression slope (trend)
        - std: Standard deviation (stability)
    
    Args:
        overlap: Window overlap fraction (0.9 = 90%)
    """
    step = int((1 - overlap) * fs * seconds)
    length = fs * seconds
    windows = int((len(gsr_data) - length) / step) + 1
    
    tonic = {'offset': [], 'slope': [], 'std': []}
    
    for i in range(windows):
        ini = i * step
        end = ini + length
        
        # Linear trend
        offset, slope = estimate_coefs(
            np.arange(0, length), 
            gsr_data[ini:end]
        )
        
        tonic['offset'].append(offset)
        tonic['slope'].append(slope)
        tonic['std'].append(np.std(gsr_data[ini:end]))
    
    return tonic

Real-World Application: Affective Robot Storytelling

BIOSIGNALS was the cornerstone of our Affective Robot Story-telling research, where a NAO humanoid robot adapted its narrative delivery based on children's real-time emotional responses detected through physiological signals.

"Fusing GSR-derived arousal with HRV-derived valence achieved 74% accuracy in dynamic emotion classification. Crucially, BIOSIGNALS' synchronised event markers revealed that GSR peaks occurred 1-2 seconds after emotionally intense narrative points—a finding that informed our classifier's optimal temporal window of 5 seconds."

Automated Experimental Script

The TCP/IP trigger system enabled fully automated experiments with millisecond-precision event marking:

# Automated multi-trial experimental protocol
from COM.trigger_client import trigger_client
import time
import random

tc = trigger_client('192.168.1.100', 10000)
tc.create_socket()
tc.connect()

stimuli = ['neutral', 'happy', 'sad', 'fear', 'anger']
trials = 20

for trial in range(trials):
    # Start recording with baseline
    tc.send_msg(b'start')
    time.sleep(2)  # 2s baseline
    
    # Present randomised stimulus
    stimulus = random.choice(stimuli)
    tc.send_msg(stimulus.encode())
    print(f"Trial {trial}: {stimulus}")
    time.sleep(5)  # 5s stimulus presentation
    
    # Recovery period
    tc.send_msg(b'recovery')
    time.sleep(3)
    
    # Stop and auto-save with trial number
    tc.send_msg(b'stop')
    time.sleep(1)  # Inter-trial interval

print("Experiment completed! EDF files saved with timestamps.")

Technical Implementation: Critical Design Decisions

1. Why PyQt5 + QwtPlot Over Modern Web Frameworks?

In 2018, the natural choice for data visualisation might have been Electron, React, or Plotly Dash. However, for real-time physiological signal processing, PyQt5 offered decisive advantages:

# Precision timer for 64 Hz BVP plotting
self.bvp_timer = QtCore.QTimer()
self.bvp_timer.setTimerType(QtCore.Qt.PreciseTimer)
self.bvp_timer.timeout.connect(self.bvp_update)
self.bvp_timer.start(int((1 / 64) * 1000))  # 15.625 ms

def bvp_update(self):
    """Update BVP plot with minimal latency"""
    data = self.dmgs[0].getSamples()  # Thread-safe buffer access
    self.bvp_curve.setData(data[:, 0], data[:, 1])
    self.bio_graph.qwtPlot_bvp.replot()  # Native Qt replot

2. EDF+ Format: Why Not HDF5 or Parquet?

Decision rationale: For a research tool targeting the neuroscience community, compatibility with established pipelines (EEGLAB, MNE-Python) was paramount. EDF+ provides this whilst maintaining embedded annotations—critical for event-related analysis.

class edf_writter:
    def annotation(self, instant, duration, event):
        """Write timestamped annotation to EDF+ file"""
        self.file.writeAnnotation(instant, duration, event)
        
    def save_streamData(self):
        """Incremental write during recording"""
        for i in range(len(self.data_store)):
            self.file.writeSamples(self.data_store[i])
        self.file.close()

3. Thread Architecture: Separating Concerns

BIOSIGNALS employs 5 concurrent threads with minimal contention:

class data_manager(QtCore.QThread):
    """Thread-safe per-signal data management"""
    def __init__(self, signal, sample_rate):
        super().__init__()
        self.mutexBuffer = Lock()
        self.buffer = RingBuffer(...)
        
    def appendSample(self, sample):
        """Thread-safe insertion"""
        self.mutexBuffer.acquire()
        try:
            self.buffer.append(sample)
            self.cur_index += 1
        finally:
            self.mutexBuffer.release()
            
    def getSamples(self):
        """Thread-safe retrieval for plotting"""
        self.mutexBuffer.acquire()
        try:
            return self.buffer.get()
        finally:
            self.mutexBuffer.release()

Lessons Learned & Future Enhancements

What Worked Exceptionally Well

• ✅ TCP/IP trigger system: Enabled seamless integration with PsychoPy, Unity, ROS, and custom experiment controllers—adopted by 3+ external research groups
• ✅ Modular architecture: MODULES/ directory allowed dynamic loading of custom processing pipelines without modifying core code
• ✅ Open-source release (GPL-3.0): 50+ stars on GitHub, cited in 5+ publications, DOI: 10.5281/zenodo.3759262
• ✅ Real-time HRV processing: 200ms latency for 60s windows enabled closed-loop biofeedback experiments

Trade-offs & Known Limitations

• ⚠️ Hardware dependency: Tightly coupled to Empatica E4 protocol (migration to Polar H10, BITalino, or Shimmer requires refactoring empatica_client.py)
• ⚠️ No cloud synchronisation: Multi-site collaborative studies require manual EDF file sharing (no real-time data federation)
• ⚠️ Limited preprocessing: Artifact removal (motion, electromagnetic interference) performed post-experiment—online filtering would improve classification accuracy
• ⚠️ Windows size trade-off: HRV frequency-domain analysis requires ≥60s windows, limiting temporal resolution for dynamic emotion tracking

If I Were to Rebuild Today (2025 Perspective)

Performance Metrics & Validation

Benchmarked on Intel i7-8550U (4 cores, 1.8-4 GHz) with 16GB RAM, Ubuntu 18.04:

Impact & Scientific Validation

BIOSIGNALS has been released as open-source software with a permanent DOI for academic citation, and directly contributed to multiple peer-reviewed publications in affective computing and human-robot interaction:

Getting Started with BIOSIGNALS

System Requirements

Python 3.6+
PyQt5 >= 5.12
pyhrv >= 0.4.0
PythonQwt >= 0.8.0
scipy >= 1.5.0
pandas >= 1.0.0
matplotlib >= 3.0.0
numpy >= 1.18.0
pyEDFlib >= 0.1.20

Installation & Quick Start

# 1. Clone repository
git clone https://github.com/mikelval82/BIOSIGNALS.git
cd BIOSIGNALS

# 2. Install dependencies
pip install -r requirements.txt

# 3. Configure Empatica Server (config.py)
E4_server_ADDRESS = 'localhost'  # Or remote IP
EMPATICA_PORT = 8000

# 4. Launch application
python BIOSIGNALS_APP_01.py

# GUI Workflow:
# - Set participant ID and output directory
# - Click "Server" to connect to Empatica Server
# - Click "Refresh" to discover E4 devices
# - Select device from dropdown
# - Click "Connect" to start streaming
# - Click "Trigger" to enable remote control (port 10000)

# 5. Remote control from another terminal/script
python
>>> from COM.trigger_client import trigger_client
>>> tc = trigger_client('localhost', 10000)
>>> tc.create_socket()
>>> tc.connect()
>>> tc.send_msg(b'start')    # Begin recording
>>> tc.send_msg(b'baseline') # Annotate event
>>> tc.send_msg(b'stop')     # Save and increment trial

Citation & Licence

@software{biosignals2020,
  author       = {Val Calvo, Mikel},
  title        = {{BIOSIGNALS: Real-time Physiological Signal 
                   Acquisition System for Emotion Recognition}},
  year         = 2020,
  publisher    = {Zenodo},
  version      = {v1.0},
  doi          = {10.5281/zenodo.3759262},
  url          = {https://doi.org/10.5281/zenodo.3759262},
  note         = {Open-source software for multi-modal 
                  biosignal acquisition with TCP/IP 
                  synchronisation and EDF+ export}
}

@phdthesis{valcalvo2021emotional,
  author = {Val Calvo, Mikel},
  title  = {{Emotional Human-Robot Interaction Using 
            Physiological Signals}},
  school = {Universidad Nacional de Educación a Distancia},
  year   = {2021},
  url    = {https://espacio-pre.uned.es/entities/publication/
            0e85194e-6187-4e8d-a34c-ca07e5880bd8/full}
}

📜 Licence: GNU General Public Licence v3.0

You are free to:

• ✓ Use commercially and privately
• ✓ Modify and distribute
• ✓ Use for patent claims
• ⚠️ Derivative works must maintain GPL-3.0
• ⚠️ Source code disclosure required for distributions

Resources & Community

Related Open-Source Projects

BIOSIGNALS is part of a larger ecosystem for multi-modal physiological computing:

Future blog posts will explore MULTI_GEERT architecture, real-time artifact removal with knowledge graphs, and deep learning for neuroprosthetics. Stay tuned!

About the Author

Mikel Val Calvo, PhD

AI Research Scientist specialising in affective computing, neuroprosthetics, and human-robot interaction. Former researcher at Universidad Miguel Hernández de Elche's NeuraViPeR (H2020) project. Currently developing LLM-powered solutions for digital health at LabLENI-UPV.