airoa-lineage

OpenLineage-based data lineage tracking library for AIRoA robot data pipelines.

Overview

airoa-lineage is a Python library for tracking data lineage in robot data pipelines. Built on the OpenLineage standard, it integrates with Marquez server to visualize and manage data flows across your robotics data infrastructure.

This library is designed to be integrated into robot data collection systems, ETL pipelines, and batch processing jobs to provide transparent tracking of data provenance, transformations, and dependencies.

Key Features

• 🤖 Teleoperation Session Management: Track robot data collection sessions with TeleopSession
• 🔄 Data Conversion Session Management: Track data conversion pipelines with ConversionSession
• ☁️ S3 Upload Management: Track S3 object storage uploads with S3UploadSession (includes device and job tracking)
• 🔌 USB Data Copy Management: Track USB data copy operations with USBCopySession (includes device and job tracking)
• 📊 OpenLineage Integration: Full OpenLineage specification support (START/COMPLETE/RUNNING/FAIL/ABORT)
• 🏷️ Custom Run Facets: Robot metadata (robotId, location, repository info) via CommonRunFacet
• 🔍 Marquez Client: Query jobs, datasets, and lineage graphs via REST API
• 🌐 Namespace-scoped Operations: All operations bound to a single namespace for isolation

Core Concepts

Event Flow

OpenLineage events follow a standard lifecycle pattern:

• START: Job begins, includes input/output datasets and schemas
• RUNNING: Optional progress updates during long operations (recommended for multi-phase jobs)
• COMPLETE: Job ends successfully, includes processing statistics
• FAIL: Job ends with failure
• ABORT: Job was cancelled or interrupted

Each session automatically manages run IDs and event timestamps. See Architecture for detailed event flow patterns.

Namespace

Namespaces provide logical separation for different projects or environments:

• Production vs Development
• Different robot fleets
• Project-specific data pipelines

All MarquezClient operations are namespace-scoped, ensuring complete isolation between environments. All operations (emit, query, lineage) are scoped to the client's namespace. Examples in this repository use the "airoa_examples" namespace.

For namespace design patterns and best practices, see Architecture.

Installation

Using uv (Recommended)

# Install from PyPI (when published)
uv add airoa-lineage

# Or install from local source
uv add /path/to/airoa-lineage

# Or install from Git repository
uv add git+https://github.com/airoa-org/airoa-lineage.git

Using pip

# Install from PyPI (when published)
pip install airoa-lineage

# Or install from local source
pip install /path/to/airoa-lineage

# Or install from Git repository
pip install git+https://github.com/airoa-org/airoa-lineage.git

Requirements

• Python >= 3.8
• openlineage-python >= 1.31.0, < 1.32.0
• requests >= 2.31.0
• python-dotenv >= 1.0.0

Quick Start

Once you've installed airoa-lineage in your project, you can start tracking data lineage in your robot data pipelines.

Basic Usage

from openlineage.client.run import Dataset

from airoa_lineage.facets import CommonRunFacet
from airoa_lineage.teleop import TeleopSession

# Create session with robot metadata
common_facet = CommonRunFacet(
    robotId="hsr001",
    location="weblab",
    repositoryHash="df110d5",
    repositoryUri="https://github.com/user/repo.git",
    repositoryTag="v1.0.0",
    repositoryBranch="main"
)
session = TeleopSession(namespace="production", common_facet=common_facet)

# Define output dataset
output_ds = Dataset(namespace="production", name="teleop_rosbag")

# Track teleoperation session with nominal time
run_id = session.start(
    nominal_start_time="2025-10-22T00:00:00+00:00",
    nominal_end_time="2025-10-22T05:00:00+00:00"
)
print(f"Session started: {run_id}")

# Collect robot data (sensor readings, camera feeds, motor commands)...

session.complete(output_datasets=[output_ds])
# Or if interrupted: session.cancel()

# See examples/teleop_session.py for complete example with device facets and error handling
# See examples/teleop_cancel.py for cancellation handling

For more examples, see the examples/ directory.

Documentation

• CLI Usage Guide - Command-line interface reference

  • Installation and quick start
  • Configuration (file, environment variables, CLI arguments)
  • Commands (start, complete, cancel)
  • Output modes and exit codes
  • Shell script integration examples
  • Troubleshooting and common errors

• Architecture - System design and core components

  • TeleopSession, ConversionSession, MarquezClient, Timestamp Utilities
  • Event flow and RUNNING events
  • Nominal time support and namespace design patterns

• Development Guide - Setup, testing, and building

  • Environment setup and dependencies
  • Code quality checks (ruff, mypy)
  • Running tests and building packages

• Deployment Guide - Environment setup and production

  • Environment variables configuration
  • Marquez server setup (Docker, Kubernetes)
  • Security, monitoring, and high availability

Examples

Teleoperation Session

uv run python examples/teleop_session.py

Demonstrates a complete teleoperation session with nominal time tracking.

S3 Upload Session

uv run python examples/s3_upload_session.py

Demonstrates a complete S3 object storage upload session with nominal time tracking.

Batch ETL Job

uv run python examples/simple_batch_etl.py

Shows how to track a batch ETL pipeline with input/output datasets.

Development

For developers who want to contribute to airoa-lineage, see the Development Guide.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Related Links

• OpenLineage - Open standard for data lineage
• Marquez - Metadata server
• OpenLineage Python Client - Official Python client