Data

The data resource publishes robot data to Qualia and downloads exact published trees. Uploaded bytes go straight to storage; Qualia verifies the manifests and publishes them together with the data, never handling the bulk bytes itself.

Install

pip install 'qualia-sdk==0.7.0rc4'

# or with uv
uv pip install 'qualia-sdk==0.7.0rc4'

This deployment runs ahead of the public release, so it needs the matching pre-release wheel — 0.7.0rc4. Pin it rather than upgrading: the upload protocol is negotiated exactly, so the client and the endpoint it talks to have to be the same generation. A plain pip install qualia-sdk resolves to the current public release and will be turned away by this deployment.

The native transfer engine that powers upload/download is bundled in the wheel — there is no separate step.

Upload a dataset

Stage a local dataset directory and atomically publish its validated source, Git history, and lakehouse rows in one call. Episodes and frames become visible only after the paired publication succeeds.

from qualia import Qualia

client = Qualia()

result = client.data.upload("/data/gongsta", name="gongsta")
print(result["dataset_id"], result["episode_count"], result["total_frames"])

Response

{
  "repo_path": "acct-01234567-89ab-4cde-8fab-0123456789ab/gongsta",
  "name": "gongsta",
  "file_count": 128,
  "committed_files": 128,
  "dataset_id": "…",
  "dataset_version_id": "…",
  "episode_count": 42,
  "total_frames": 51840
}

Parameters

Parameter	Default	Description
local_dir	—	Directory of dataset files (or an MCAP session) to upload.
name	—	Dataset name within your account (a single path segment).
source_format	"lerobot_v3"	Structured source format. One of lerobot_v2_1, lerobot_v3, or mcap_ros2; the two LeRobot layouts are distinct.
max_workers	4	Files read + chunked concurrently. Bounds CPU/RAM/disk so a large upload can’t saturate your machine.
num_threads	~half your cores	Native chunk/hash thread-pool size.

Re-uploading a dataset only sends the chunks that changed — unchanged files are skipped client-side, and edited files upload just their changed regions.

Chunk reuse across client restarts

Until now an ordinary upload() kept no record of how the previous version of a file was cut into chunks, so an edited file went through the ordinary cold upload path every time.

upload() now keeps a small on-disk cache of each file’s chunk boundaries (chunk IDs plus offsets and sizes; never bytes, credentials, URLs or manifest IDs) that survives client restarts. On the next upload of a changed file the SDK re-chunks the current file as usual, compares the unchanged prefix and suffix against the cached boundaries, and asks the server — under your current authorization — whether those chunks still exist. Only chunks the server confirms are kept; everything else uploads normally. A missing, damaged or unreadable cache is silently treated as a miss.

What this saves is bytes stored, not time. On the harness’s 36.3 MB changed-video fixture, re-uploading the edited file after a client restart stored 2,922,261 bytes without the cache and 26,741 bytes with it. No upload-time speedup is claimed: the file is still read and chunked in full.

Uploading a directory writes the cache once at the end, however many files it holds, so the database work does not grow with the file count. The work of preparing each eligible file’s hint does grow, though it stays small: on one development machine, offering 1,009 multi-chunk files took about 97 ms and 10,009 took about 301 ms.

The cache applies to changed files of at most 64 MiB in ordinary directory uploads. See Configuration for where it lives, how to disable it, and how it is bounded.

Import an exact Hugging Face dataset

The SDK sends only an exact lowercase 40-hex commit and optional token to the service. The import runs entirely server-side, streaming the pinned Hugging Face tree straight into storage; dataset bodies and scratch files never touch the SDK or any Qualia service:

from qualia import Qualia

client = Qualia()
repo = "qualia-robotics/openarm-cube-in-box-v1"
revision = "17c29c487d886878ddcb3c5649101b2860fbc047"

result = client.data.import_huggingface(
    repo,
    revision,
    "openarm-cube-in-box",
    source_format="lerobot_v3",
)
print(result["dataset_id"], result["dataset_version_id"])

For private or gated repositories, pass hf_token=os.environ["HF_TOKEN"]. The token appears only in the initial submission and is never returned in poll results or progress events. The worker authenticates the exact remote file identities and publishes an immutable DatasetVersion.

Qualia intentionally rejects dot-prefixed Git-control paths. The importer permits only the exact root names .gitattributes and .gitignore: it downloads and verifies them, commits their path, size, Git blob OID, and content identity under excluded_repository_metadata, and removes them before publication. Any other hidden path, .lock path, or meta/qds/** attempt fails preflight. The provenance sidecar itself is published at the non-authoritative, meta/qualia/imports/huggingface-dataset.json path.

The pinned medium validation fixture TobiBrtnr/openarm_shirt_folding_new_converted@c34057e9cb7760f939dbd5759679c12818fa896b is LeRobot 2.1 and about 2.08 GB. Its pinned media inventory reports AV1 even though legacy metadata may label it avc1; Qualia treats validated media bytes as authoritative. The public metadata does not establish collection hardware, so do not describe it as real-robot data without separate provenance evidence.

Upload a robot MCAP session

Recorded robot sessions upload through the same call with source_format="mcap_ros2". Lay the session out as one subdirectory per episode, each holding exactly one .mcap bag. An episode.json sidecar may provide normalized task and mode fallbacks when the bag does not:

session-2026-07-20/
  episode-000/
    episode-000.mcap
    episode.json        # { "task": "fold the towel", "mode": "teleop" }
  episode-001/
    episode-001.mcap
    episode.json

result = client.data.upload(
    "/data/session-2026-07-20",
    name="dk1-kitchen",
    source_format="mcap_ros2",
)
print(result["dataset_id"], result["episode_count"], result["total_frames"])

The upload validates the session first and fails fast if a bag is missing or not a real MCAP file. A missing sidecar is valid; Qualia derives task and recording purpose from authenticated bag contents when available.

Download a dataset

Download a dataset into a local directory, preserving relative paths. Returns the list of written file paths. This is repository-tree download; for a pinned training dataset, use immutable version materialization below.

paths = client.data.download("gongsta", dest="/data/gongsta-copy")

Materialize an immutable DatasetVersion

Render one pinned DatasetVersion as an exact LeRobot v3 directory and download it directly from object storage. The API returns a durable operation and bounded pages of immutable artifact metadata; bulk bytes never pass through the Qualia. The SDK authenticates the complete tree, verifies every artifact, and atomically publishes dest only after the whole tree succeeds. Camera streams use Qualia’s versioned canonical AV1 profile; an already canonical, full-range AV1 source can be preserved byte-for-byte, while trims and other codecs are frame-exactly encoded and fully decoded for verification. The destination must not already exist.

result = client.data.materialize_version(
    "44444444-4444-4444-8444-444444444444",
    "/data/training-set-v7",
)
print(result.bundle.bundle_id, result.written_file_count, result.destination)

Interrupted rendering is recoverable: repeating the call addresses the same deterministic materialization identity. A failed download never exposes a partially published destination.

Explore the episode lake

Filter episodes across every source dataset in the authenticated account with one bounded query. The result includes one offset page, totals over the complete filtered set, and task, tier, and embodiment facets. Each facet axis excludes its own filter while retaining every other filter.

from qualia import LakeExploreQuery, LakeNumericPredicate

query = LakeExploreQuery(
    tasks=["fold shirt"],
    tiers=["retrainable"],
    tags=["reviewed"],
    numeric_predicates=[
        LakeNumericPredicate(field="jerk_mean", op="lt", value=0.4)
    ],
    limit=50,
)
page = client.data.explore(query)
print(page.total, page.total_frames, page.facets.by_task)
for episode in page.episodes:
    print(episode.dataset_name, episode.episode_index, episode.metrics)

Numeric fields are the lake columns duration_s, frame_count, and quality_score, plus Qualia’s typed computed metrics. Channel predicates support action, state, and telemetry streams with min, max, mean, or std. Every matching episode carries channel_predicate_evidence: the exact native channel, index, aggregate observation, operator, and threshold that satisfied each predicate. The SDK binds that evidence back to the ordered request and rejects missing, changed, non-finite, or non-satisfying observations.

Search is a literal, case-insensitive substring match over task and dataset name; % and _ have no wildcard meaning. A facet axis returns at most 512 values, while by_task_total_values, by_tier_total_values, and by_interface_hash_total_values report the complete cardinality. Active selections are retained when an axis is truncated.

Each response’s page, totals, and all facet axes come from one database statement and therefore one live MVCC snapshot. page_limit and page_offset bind the response to the request. Offset pagination is not mutation-stable between calls: after publishing data or changing review state, restart at offset zero. Use exploration for analysis and review, then publish a pinned DatasetVersion for reproducible training. The SDK rejects malformed rows, duplicate identities, inconsistent totals or facets, altered evidence, and truncated or mismatched pages.

Split episodes at their recorder markers

A continuous teleop session is uploaded as pause/resume-delimited sequences. The presses inside one — a foot pedal calling a take a success or an issue — are recorded as markers, not as cuts. Splitting turns those sequences into the takes they describe, without re-uploading or re-encoding anything.

markers = client.data.episode_markers(episode_id)
for marker in markers.markers or []:
    print(marker.at_us / 1e6, marker.label, marker.role)

split = client.data.split_episode(episode_id, cut_at=["success", "issue"])
for child in split.children:
    print(child.episode_index, child.frame_start, child.frame_end, child.tags)

cut_at names marker labels as the recording wrote them, never the role the ingest assigned. Each cut closes the span before it, and the child inherits that cut’s labels as tags — so a take’s verdict travels with it. Place cuts anywhere else with cuts=[{"at_us": 32_800_000, "label": "success"}], microseconds from the episode’s first frame; the label says “the recording called this instant that, but it belongs here”, which is a marker moved somewhere better. A request naming neither is refused.

Children are new episodes above the dataset’s highest index, tiling the parent’s frames exactly and sharing its video narrowed by frame range. The parent row is never mutated, so client.data.unsplit_episode(episode_id) reverses a split exactly. Pass dry_run=True to see the same plan without writing it.

To cut a whole session, apply one label set across many episodes:

batch = client.data.split_episodes(
    [episode.episode_id for episode in page.episodes],
    cut_at=["success", "issue"],
)
print(len(batch.splits), "split;", len(batch.skipped), "left alone")
for skipped in batch.skipped:
    print(skipped.episode_id, skipped.reason)

An episode the server declines — no marker matched, no frames to split, or children already exist — lands in skipped with the server’s own sentence, and the run continues. Every other failure stops it. Because each split is independently reversible, a stopped run leaves what it already split intact and re-running skips those rather than doubling them.

Marker instants are signed: a press a moment before the first usable frame lands slightly negative, which is information about the recording. markers is None means the ingest carried no marker topic at all, which is a different claim from an episode that carried one and holds no presses. Qualia privileges no marker vocabulary — the mapping from recorded values to roles is declared per upload.

Trim, name, and undo episode edits

Splitting carves an episode into children. The rest of the edit surface narrows or labels episodes in place, and every one of them appends a revision rather than mutating what is already published.

trimmed = client.data.trim_episodes(
    episode_ids,
    start_at_first=["ready"],
    end_at_last=["done"],
    note="drop the lead-in and the run-out",
)
restored = client.data.clear_trims(episode_ids, note="back to full extent")

Each episode is narrowed to start at ITS first marker so labelled and end at ITS last, so one call fits a whole selection whose presses landed at different instants. The base extent is never overwritten, which is why clear_trims restores it exactly.

name_subtasks uses the same boundary rule to label a span instead of carving it — useful when you want the structure recorded but the episode left whole:

named = client.data.name_subtasks(
    episode_ids,
    at=["grasp", "place"],
    boundary="ends_at",   # or "starts_at" to have each marker OPEN its span
    note="name the pick-and-place spans",
)

A batch split reverses the same way a single one does, episode by episode:

for undone in client.data.unsplit_episodes(episode_ids):
    print(undone.parent_episode_id, undone.deleted)

For a single episode, revise_episode appends one revision carrying any combination of trim bounds, task, subtasks, tier, tags, and review state. Pass expected_parent_rev to make the write conditional on the revision you read, and operation_id to make a retry idempotent:

history = client.data.list_episode_revisions(episode_id)
for revision in history.revisions:
    print(revision.rev, revision.note, revision.created_at)

commit = client.data.revise_episode(
    episode_id,
    note="mark reviewed",
    review_state="reviewed",
    expected_parent_rev=history.latest_rev,
)

The revision chain is append-stable and immutable, so it doubles as the audit trail for who changed an episode and why.

Analyze an immutable DatasetVersion

Run native action/state health analysis against one exact published version:

report = client.data.health_for_version(
    "44444444-4444-4444-8444-444444444444"
)
print(report.health_score, report.grade)
for flagged in report.flagged_episodes:
    print(flagged.member_ordinal, flagged.episode_revision_id, flagged.reasons)

Qualia re-proves the complete ordered membership and dataset-snapshot identity. It then fully proves and reads exact Parquet ranges for an evenly spaced, deterministic sample of at most 120 members. Every flag is bound to an immutable episode revision and is a review recommendation, not an automatic deletion.

The sample is explicit in report.sampling. This synchronous analysis is not a byte scrub of unsampled artifacts or video, and it does not claim model-based visual or behavioral metrics without a pinned model revision and output artifact; those fields are marked unavailable in evidence_completeness.

Review and apply bad-data suggestions

Analyze one or more immutable DatasetVersions without changing them. Every recommendation includes a risk score, machine-readable reasons, and supporting evidence. The response binds the complete recommendation set to an evidence digest and the exact paired Git/lakehouse head.

evidence = client.data.suggest_curation(
    ["44444444-4444-4444-8444-444444444444"]
)
for suggestion in evidence.suggestions:
    print(suggestion.episode_index, suggestion.risk_score, suggestion.reasons)

After review, publish the analysis’s sealed recommendation set as a new immutable DatasetVersion. apply_curation starts a durable publication and waits for its receipt; reuse the same idempotency key after a lost response or a timeout to resume it. The source version, source bytes, and Git history remain available.

from uuid import uuid4

receipt = client.data.apply_curation(
    evidence,
    idempotency_key=uuid4(),
    definition_id="55555555-5555-4555-8555-555555555555",
    note="Reviewed automated bad-data evidence",
)
print(receipt.version_id, receipt.membership_hash, receipt.git_oid)

Application fails closed if the analysis selection, evidence digest, repository head, target definition, membership counts, snapshot, or Git receipt changes between review and publication.

Run a durable curation analysis

health_for_version reads one version’s health directly. When you want the analysis to survive a disconnect — or to span several pins at once — start a durable one and read its progress back:

import time

analysis = client.data.start_curation_analysis(
    [version_id],
    idempotency_key=operation_uuid,
)

while analysis.status not in ("ready", "failed"):
    time.sleep(2)
    analysis = client.data.get_curation_analysis(analysis.analysis_id)

if analysis.status == "failed":
    raise RuntimeError(analysis.failure_detail or analysis.failure_code)

page = client.data.list_curation_suggestions(analysis.analysis_id)
for suggestion in page.suggestions:
    print(suggestion)

The analysis is keyed to exact DatasetVersion pins, so the same idempotency_key recovers the running analysis instead of starting a second one. Suggestions are only listable once status reaches ready — the server enforces that seal — and they page by opaque keyset — pass page_token to continue. Applying them is apply_curation, which publishes a new immutable DatasetVersion and deletes nothing.

List datasets

page = client.data.list()
for d in page.items:
    print(d.artifact_id, d.path)

Watch in-flight uploads

upload() narrates the server’s side of a transfer on stderr, but only in the process running it. The account’s pending feed, the one the dashboard’s pending band renders, is also readable directly, so an upload running detached on a rig can be watched from anywhere:

for op in client.data.uploads():
    print(op.name, op.phase, op.progress_line or op.message)

Rows in uploading carry the SDK’s heartbeat (transferred_bytes, staged_files); rows in preparing carry the worker’s live step in progress, refreshed with its lease every 30 s; a failed row survives 24 h with message, failure_retryable and, when ingest got that far, ingest_report. The model is tolerant on purpose: phase is a plain string and unknown fields are kept, because the platform adds steps without an SDK release.

Browse DatasetVersions and their sources

list returns canonical datasets. A DatasetVersion is an immutable pin of one — what training actually consumes — and versions hang off a definition:

versions = client.data.list_versions(definition_id)
for version in versions.versions:
    print(version.id, version.version, version.created_at)

version = client.data.get_version(version_id)
sources = client.data.list_version_sources(version_id)
for source in sources.sources:
    print(source.source_dataset_id, source.git_oid)

list_version_sources returns the exact Git/publication anchors that contributed the version’s members, which is how you answer “which upload did this training set actually come from”. All three page by opaque token via page_size / page_token.

Saved lake queries

A dataset definition is a saved lake query plus the pins published from it.

for definition in client.data.list_definitions().defs:
    print(definition.id, definition.name)

definition = client.data.get_definition(definition_id)

consumers = client.data.list_definitions_by_source(source_dataset_id)

list_definitions_by_source answers the reverse question — which definitions consume a given source, directly or through an immutable pin — so you can see what a re-upload would affect before you publish it.

Get immutable training features

Training accepts a pinned DatasetVersion, never a mutable repository name or external dataset identifier. Retrieve its exact camera descriptors and use their key values in camera_mappings.

features = client.data.get_training_features(
    "44444444-4444-4444-8444-444444444444"
)
for camera in features.camera_features:
    print(camera.key, camera.stream, camera.timebase_numerator,
          camera.timebase_denominator, camera.frame_timestamp_mapping)

Response

{
  "dataset_version_id": "44444444-4444-4444-8444-444444444444",
  "training_adapter_revision_id": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
  "camera_features": [
    {
      "key": "observation.images.wrist",
      "stream": "video/wrist",
      "timebase_numerator": 1,
      "timebase_denominator": 1000000,
      "frame_timestamp_mapping": "time_start_plus_frame_over_fps"
    }
  ]
}

Next steps

• Finetune — train a VLA on a dataset you’ve uploaded.