Skip to content

Multi-IMU recording with task markers (.xdf)

Converts a multi-IMU XDF recording with task markers into BIDS, writing one *_events.tsv per sensor from merged marker streams.

This example shows how to:

  • Load an XDF file with several independent Motion streams and several independent Markers streams
  • Recompute the sampling rate from timestamps when the nominal rate declared in the XDF header doesn't match what was actually recorded
  • Treat two unsynchronized IMU sensors as two separate BIDS recordings, each with its own tracksys label, rather than resampling them onto a shared clock
  • Merge multiple XDF marker streams into one *_events.tsv per recording with events_from_xdf_markers()

Requirements

pip install motionbids pyxdf numpy
examples/from_xdf_events.py
"""
Example: Convert a multi-IMU + task-marker XDF recording to BIDS format

This example demonstrates:
- Loading a real-world XDF file with two independent 'Motion' streams and
  two independent 'Markers' streams
- Recomputing sampling rate from timestamps when the nominal_srate declared
  in the XDF header doesn't match what was actually recorded (common with
  LSL clocks) - see from_xdf_movella.py for the single-sensor version
- Treating two unsynchronized IMU sensors as two separate BIDS recordings,
  each with its own tracksys label, rather than resampling them onto a
  shared clock - that resampling decision belongs to you, see the
  "Import is your job" note in the README
- Merging multiple XDF marker streams into one *_events.tsv per recording
  with events_from_xdf_markers()

Requirements:
    pip install motionbids pyxdf numpy

Data:
    Points at a local recording and is not bundled with the package. Update
    XDF_PATH below to point at your own file. It expects the same stream
    layout: one or more 6-channel (Accel_x/y/z, Gyro_x/y/z) IMU 'Motion'
    streams, plus one or more string 'Markers' streams.
"""

import re
import urllib.request
import numpy as np
import pyxdf
from pathlib import Path
from motionbids import (
    MotionData,
    Channel,
    export_bids_motion,
    events_from_xdf_markers,
    create_bids_directory_structure,
    export_dataset_description,
)

EXAMPLE_DATA_URL = (
    "https://raw.githubusercontent.com/JuliusWelzel/motionbids/main/"
    "examples/data/example_imu_events.xdf"
)

# Configuration
bids_root = Path(__file__).parent / "bids_dataset"
data_folder = Path(__file__).parent / "data"
SUBJECT_ID = "01"
TASK_NAME = "freeHandMovement"
# Channel label -> (BIDS channel_type, units)
CHANNEL_KIND = {
    "Accel": ("ACCEL", "m/s^2"),
    "Gyro": ("GYRO", "rad/s"),
}


# Ensure the example recording is available locally
data_folder.mkdir(exist_ok=True)
example_file = data_folder / "example_imu_events.xdf"
if example_file.exists():
    print(f"Example data already present: {example_file}")
else:
    print(f"Downloading example data from {EXAMPLE_DATA_URL}")
    urllib.request.urlretrieve(EXAMPLE_DATA_URL, example_file)
    print(f"Saved to {example_file}")

# Get all XDF files
xdf_files = sorted(data_folder.glob("*events.xdf"))

# Create base directory and dataset description once
bids_root.mkdir(exist_ok=True)

print("✓ Dataset description created")
streams, header = pyxdf.load_xdf(str(xdf_files[0]), synchronize_clocks=True)

motion_streams = [s for s in streams if s["info"]["type"][0] == "Motion"]
marker_streams = [s for s in streams if s["info"]["type"][0] == "Markers"]
print(f"Found {len(motion_streams)} IMU stream(s): "
      f"{[s['info']['name'][0] for s in motion_streams]}")
print(f"Found {len(marker_streams)} marker stream(s): "
      f"{[s['info']['name'][0] for s in marker_streams]}")


for imu_stream in motion_streams:
    sensor_name = imu_stream["info"]["name"][0]
    print(f"\nProcessing IMU stream: {sensor_name}")

    timestamps = imu_stream["time_stamps"]
    raw_data = np.asarray(imu_stream["time_series"], dtype=np.float64)

    # The XDF header's nominal_srate is not always what was actually
    # recorded (dropped samples, clock drift, ...) - recompute from the
    # timestamps and use that instead.
    nominal_srate = float(imu_stream["info"]["nominal_srate"][0])
    sampling_rate = len(timestamps) / (timestamps[-1] - timestamps[0])
    if abs(sampling_rate - nominal_srate) > 1.0:
        print(f"   Nominal rate {nominal_srate:.1f} Hz != effective rate "
              f"{sampling_rate:.1f} Hz (from timestamps) - using effective rate")

    channel_labels = [
        c["label"][0]
        for c in imu_stream["info"]["desc"][0]["channels"][0]["channel"]
    ]

    # Each IMU gets its own tracksys label (e.g. "CodeCell2_IMU" -> "imucodecell2")
    sensor_id = re.sub(r"_?imu$", "", sensor_name, flags=re.IGNORECASE).lower()
    tracksys = f"imu{sensor_id}"
    tracked_point = sensor_id  # Adjust to the sensor's actual anatomical placement

    channels = []
    for label in channel_labels:
        kind, axis = label.split("_")
        channel_type, units = CHANNEL_KIND[kind]
        channels.append(Channel(
            channel_name=label.lower(),
            channel_component=axis.lower(),
            channel_type=channel_type,
            channel_tracked_point=tracked_point,
            channel_units=units,
        ))

    # Combine every marker stream into one events.tsv, aligned to this
    # sensor's own first sample and sampling rate
    events = []
    for marker_stream in marker_streams:
        events.extend(events_from_xdf_markers(
            marker_stream,
            motion_first_timestamp=timestamps[0],
            sampling_frequency=sampling_rate,
        ))
    print(f"   {len(events)} events merged from {len(marker_streams)} marker stream(s)")

    motion = MotionData(
        subject=SUBJECT_ID,
        session=None,
        task_name=TASK_NAME,
        tracksys=tracksys,
        sampling_frequency=sampling_rate,
        tracked_points_count=1,
        recording_duration=timestamps[-1] - timestamps[0],
        manufacturer="Unknown",  # Adjust to your actual IMU hardware vendor
        manufacturers_model_name=sensor_name,
        recording_type="continuous",
        data=raw_data,
        channels=channels,
        events=events,
    )

    files = export_bids_motion(motion, out_dir=bids_root, overwrite=True)
    print(f"   Exported: {sorted(p.name for p in files.values())}")

print(f"\nBIDS dataset: {bids_root.absolute()}/")

Next Steps