Quick Start

Get up and running with SiFi Bridge Python in just a few minutes. This guide covers the essential steps to connect to a device and start acquiring data.

Basic Workflow

Working with SiFi Bridge Python follows this pattern:

1. Create a SifiBridge instance
2. Connect to a device
3. Configure sensors
4. Start data acquisition
5. Read data packets
6. Stop and disconnect

Your First Connection

Here's a minimal example to connect and stream ECG data:

import sifi_bridge_py as sbp

# Create bridge instance
sb = sbp.SifiBridge()

# Connect to any available BioPoint device
sb.connect(sbp.DeviceType.BIOPOINT_V1_3)

# Configure ECG sensor (500 Hz sampling rate)
sb.configure_ecg(state=True, fs=500)

# Disable low latency mode for simpler data reception logic
sb.set_low_latency_mode(False)

# Start acquisition
sb.start()

# Read 10 ECG packets
for i in range(10):
    packet = sb.get_ecg()
    ecg_data = packet['data']['ecg']
    print(f"Packet {i}: {len(ecg_data)} samples, first value: {ecg_data[0]:.3f}")

# Stop and disconnect
sb.stop()
sb.disconnect()

Understanding Data Packets

Sensor data has the exact same structure as SiFi Bridge's, deserialized as a dictionary:

{
    'id': 'device',           # Device identifier
    'packet_type': 'ecg',          # Type of data (ecg, emg, imu, ppg, etc.)
    'timestamp': 1234567890.123,   # Unix timestamp
    'status': 'ok',                # Packet status
    'data': {
        'ecg': [0.001, 0.002, ...]  # Sensor data arrays
    }
}

Connect to Specific Device

Connect to a device by its MAC address (Linux/Windows) or UUID (macOS):

# List available devices
devices = sb.list_devices(sbp.ListSources.BLE)
print(f"Found devices: {devices}")

# Connect to specific device by address
sb.connect("XX:XX:XX:XX:XX:XX")  # Use actual MAC/UUID

Configure Multiple Sensors

Enable and configure multiple sensors at once:

# Configure multiple sensors
sb.configure_ecg(state=True, fs=500)
sb.configure_emg(state=True, fs=2000)
sb.configure_imu(state=True, fs=100)

# Start acquisition
sb.start()

# Read data from any sensor
ecg_packet = sb.get_ecg()
emg_packet = sb.get_emg()
imu_packet = sb.get_imu()

Simple Data Collection Loop

Collect data continuously with a loop:

import sifi_bridge_py as sbp
import time

sb = sbp.SifiBridge()
sb.connect()

# Enable sensors
sb.configure_sensors(ecg=True, emg=True, imu=True)

sb.start()

# Collect for 10 seconds
start_time = time.time()
data_buffer = []

try:
    while time.time() - start_time < 10:
        packet = sb.get_data()  # Get any packet
        data_buffer.append(packet)
        print(f"Received {packet['packet_type']} packet")

except KeyboardInterrupt:
    print("Stopped by user")

finally:
    sb.stop()
    sb.disconnect()
    print(f"Collected {len(data_buffer)} packets")

Save Data to CSV

Stream data directly to CSV files:

# Create bridge with CSV output
sb = sbp.SifiBridge(publishers="csv://./data/")

sb.connect()
sb.configure_ecg(state=True, fs=500)
sb.start()

# Data automatically saves to ./data/*.csv files
time.sleep(60)  # Record for 1 minute

sb.stop()
sb.disconnect()

CSV files are created per sensor:

• ecg_DEVICENAME_TIMESTAMP.csv
• emg_DEVICENAME_TIMESTAMP.csv
• imu_DEVICENAME_TIMESTAMP.csv

Using Threading

Since get_data() blocks, use threading for responsive applications:

import threading
import sifi_bridge_py as sbp

def data_collection_thread(sb):
    """Background thread for data collection"""
    while running:
        try:
            packet = sb.get_data(timeout=1.0)
            if packet:
                # Process packet
                print(f"Got {packet['packet_type']}")
        except:
            pass

# Setup
sb = sbp.SifiBridge()
sb.connect()
sb.configure_ecg(state=True)
sb.start()

# Start background thread
running = True
thread = threading.Thread(target=data_collection_thread, args=(sb,))
thread.start()

# Do other work in main thread
time.sleep(30)

# Cleanup
running = False
thread.join()
sb.stop()
sb.disconnect()

Common Patterns

Check Connection Status

info = sb.show()
if info['connected']:
    print(f"Connected to {info['device']}")
else:
    print("Not connected")

Retry Connection

import time

max_retries = 5
for attempt in range(max_retries):
    if sb.connect():
        print("Connected!")
        break
    print(f"Retry {attempt + 1}/{max_retries}")
    time.sleep(2)

Wait for Specific Packet Type

# Wait for ECG packet
ecg_packet = sb.get_ecg()

# Wait for any packet with specific key
packet = sb.get_data_with_key("timestamp")

Error Handling

Always include error handling for robust applications:

import sifi_bridge_py as sbp

try:
    sb = sbp.SifiBridge()

    # Try to connect
    if not sb.connect(sbp.DeviceType.BIOPOINT_V1_3):
        raise ConnectionError("Failed to connect to device")

    sb.configure_ecg(state=True)
    sb.start()

    # Collect data
    for _ in range(100):
        packet = sb.get_ecg()
        # Process packet...

except ConnectionError as e:
    print(f"Connection error: {e}")
except KeyboardInterrupt:
    print("Interrupted by user")
finally:
    # Always cleanup
    try:
        sb.stop()
        sb.disconnect()
    except:
        pass

Quick Reference

# Import
import sifi_bridge_py as sbp

# Create bridge
sb = sbp.SifiBridge()

# Connection
sb.connect()                              # Connect to any device
sb.connect(sbp.DeviceType.BIOPOINT_V1_3)  # Connect by device type
sb.connect("MAC_ADDRESS")                 # Connect by address
sb.disconnect()                           # Disconnect

# Configuration
sb.configure_ecg(state=True, fs=500)      # Configure ECG
sb.configure_emg(state=True, fs=2000)     # Configure EMG
sb.configure_sensors(ecg=True, emg=True)  # Enable multiple sensors
sb.set_low_latency_mode(False)            # Set low latency

# Control
sb.start()                                # Start acquisition
sb.stop()                                 # Stop acquisition

# Data
packet = sb.get_data()                    # Get any packet
ecg = sb.get_ecg()                        # Get ECG packet
emg = sb.get_emg()                        # Get EMG packet
imu = sb.get_imu()                        # Get IMU packet

# Device commands
sb.send_command(sbp.DeviceCommand.OPEN_LED_1)
sb.send_command(sbp.DeviceCommand.START_MOTOR)