Python API Reference¶

Beyond the CLI, the pipeline can be driven and inspected from Python. This page covers the public surface exported from the resxr package.

Public exports¶

from resxr import (
    run,             # run the full pipeline
    PipelineConfig,  # the validated configuration model
    Session,         # per-recording state object
    SessionMetadata, # parsed session metadata
    TrackingStream,  # one tracking system's data
    QualityFlag,     # a detected quality issue
    TrackingSystem,  # enum of tracking systems
    __version__,
)

run¶

def run(config_path: str = "config/pipeline_config.yaml") -> None

Runs the complete pipeline for every session named in the config (or auto-discovered). This is the same code path as the resxr CLI.

import resxr

resxr.run(config_path="config/pipeline_config.yaml")

It returns None; the result is the BIDS dataset written to output.bids_root. Individual session failures are logged and skipped.

PipelineConfig¶

The validated configuration. Load it from YAML to inspect or build on:

from resxr import PipelineConfig

config = PipelineConfig.from_yaml("config/pipeline_config.yaml")
print(config.input.data_dir)
print(config.output.bids_root)
print([name for name, sys in config.systems.items() if sys.enabled])

from_yaml(path) raises a ConfigurationError if the file is missing, unparseable, empty, or fails validation. See the Configuration Reference for every field. Helpers include is_system_enabled(name) and, on the validation sub-config, get_column_groups(check_name, default_columns=...).

Core data structures¶

These mirror the objects that flow through the pipeline stages.

TrackingSystem¶

An enum of the supported systems: HEAD, HANDS, EYES, FACE, BODY, CONTROLLERS. Each member's .value is the BIDS tracksys- label ("Head", "Hands", …).

Session¶

The per-recording carrier. Key attributes: session_id, subject_id, session_label, metadata, streams (dict[TrackingSystem, TrackingStream]), the raw_*_data frames, custom_tables/custom_tables_data, and merged_events_data. Useful members:

TrackingStream¶

One system's data. Attributes include system, data (raw), clean_data (post-masking), quality_flags, sampling_frequency (expected), sampling_frequency_effective (measured), and channel_count. Useful properties: duration_seconds, row_count, warning_count/error_count, and get_output_data() (returns clean_data if set, else data).

QualityFlag¶

A detected issue over a time segment — see Validation for the full field list and the from_mask(...) constructor used by checks.

SessionMetadata¶

The parsed SessionMetadata.json: timing (utc_start, fixed_delta_time), feature flags (face_enabled, hands_enabled, …), schema counts, device fields, and the engine-agnostic software_versions map.

Loading data without running the pipeline¶

You can load a single session's raw data directly:

from resxr import PipelineConfig
from resxr.io.readers import load_session

config = PipelineConfig.from_yaml("config/pipeline_config.yaml")
session = load_session("/path/to/session_dir", config.input)
print(session.session_id)
print(session.raw_continuous_data.shape)

Exceptions¶

All pipeline errors derive from ResXRError (importable from resxr.core). Catching it handles any pipeline failure:

from resxr import run
from resxr.core import ResXRError

try:
    run("config/pipeline_config.yaml")
except ResXRError as exc:
    print(f"Pipeline failed: {exc}")