qualia-sdk — Python client for Qualia robot data: upload, curate, version, export

Status: beta pre-release (0.8.0rc15) · Published: qualia-sdk wheels for Python 3.10+

Qualia keeps robot recordings (MCAP or LeRobot) as datasets of episodes. From Python you upload a folder of takes and read what preparation found in it. You link the episodes to a Task (what the robot was asked to do), run checks that flag failed or broken takes, and keep or reject each take. The kept takes become an immutable version, which downloads as LeRobot v3 or v2.1, ROS 2 MCAP or a partner layout. The upload engine is compiled into the wheel, so there is nothing else to install. The qualia command line is the separate qualia-cli package.

Install

pip install --pre qualia-sdk

The client and the deployment it talks to negotiate the upload protocol exactly, so pin the version your deployment's docs name, for example pip install 'qualia-sdk==0.8.0rc15'. client.preparation first shipped in 0.8.0rc12; 0.8.0rc11 raises AttributeError. client.preparation.wait first shipped in 0.8.0rc14.

Usage

Set QUALIA_API_KEY (and QUALIA_BASE_URL for a deployment other than the default), or log in once with qualia auth login. An API key names its account, so no account setting is needed.

from qualia import Qualia

client = Qualia()

dataset = client.data.upload("./takes", "box-pick")["dataset_id"]
# Preparation starts by itself. Wait for it to end.
prep = client.preparation.wait(dataset)
print(prep.state, prep.warnings)  # what preparation found

# The VLM check reads each episode against the published Task it is linked to.
task = client.tasks.create(
    "Pick a box", draft={"description": "Move the box from the crate to the table"}
)
task = client.tasks.update(task.id, success="The box stands on the table")
client.tasks.publish(task.id, etag=task.etag)
client.datasets.update(dataset, linked_task=task.id)

estimate = client.checks.estimate(dataset)  # free: nothing is reserved
started = client.checks.run(dataset, max_cost=estimate.credits)
client.checks.wait(started.check_runs[0])  # a VLM check starts one run
for issue in client.checks.findings(dataset).items:
    print(issue.text)

client.episodes.reject(dataset, ["<episode-id>"], reason="the box slipped")
client.episodes.keep(dataset, filter="state = UNREVIEWED")
client.versions.create(dataset, "1.0")  # the takes a person kept
client.data.download(f"{dataset}:1.0", "./box-pick-1.0", format="lerobot_v2_1")

An API key names its account automatically. Set an account override only when needed; the API guide describes the server requirement and release history. An account= (or QUALIA_ACCOUNT) that is not an account UUID is refused with a ValueError; leave it unset to use the key's own account. Publishing a preview (client.versions.create(dataset, tag, preview=...), or qualia dataset pin <dataset> <tag> --preview <preview>) needs no override either: the tag the server answers under the key's account UUID is the one asked for.

client.data.download("<dataset>") with no version takes stable, the newest tag. A fresh upload has no tag yet, so it takes the newest version, v0 (the upload as it arrived), and reports {"step": "newest", "version": "<slug>:v0"} to on_progress. Only a dataset with no version at all is refused. While the format renders, a status read or start that times out, drops its connection or gets a 429, 503 or 504 is tried again, up to 4 tries in a row, and none starts past timeout. The listing that signs the links again during a long transfer gets the same 4 tries when it has no answer or a 503 or 504. The API guide's Data section has the examples and the waits.

A handheld (UMI) recording converts at a UMI rig of its workspace. Make one from a gripper template with client.robots.make_umi_rig(template, name) (or qualia gripper umi-rig), then give it its calibration with client.robots.import_gripper_calibration(rig.name, "tcp_calibration.json"); a workspace may have several. Name the rig with rig= on upload. Without it, an upload into a merged dataset whose uploads agree on one rig takes it, and a tree that declares [rig] kind = "umi-handheld" takes the workspace's one UMI rig: none is refused (404), several are refused asking which (409). client.datasets.resolve_poses(dataset, gripper="<name>") names the rig a resolve uses, and asks which when there are several and none is named. The API guide's Data section has the example.

A merged dataset moves its uploads with precheck_merged_poses and move_merged_poses. With several UMI rigs of one kind, name the rig of each upload that asks. The rig of a move that lands is stored for the upload:

check = client.datasets.precheck_merged_poses(merged.name)
moved = client.datasets.move_merged_poses(
    merged.name,
    membership_revision=check.membership_revision,
    rigs={upload.name: rig.name},  # upload to the UMI rig it was recorded on
)

The API guide has the full example. An upload sends its transfer session id when it begins, so the UMI rig composed for it stays its own. In a workspace with several UMI rigs, an upload from an SDK too old to send it reads as not composed and is refused, so upgrade. Against a gateway that predates the field, the SDK drops it and begins again (QDS service: UMI rigs).

A LeRobot dataset can say who recorded each episode, on which station and in which shift: put meta/qualia/episodes.jsonl in the folder before you upload it, one line per episode with its episode_index (a number, from 0) and operator, scene (the station) and session_id (the shift) as quoted strings:

{"episode_index": 0, "operator": "ana", "scene": "station-2", "session_id": "2026-11-09-am"}
{"episode_index": 1, "operator": "ben", "scene": "station-1", "session_id": "2026-11-09-am"}
{"episode_index": 2, "operator": "ana", "scene": "station-2", "session_id": "2026-11-09-pm"}

An episode without a line keeps them empty, and a line that cannot be applied is left out with a finding, never a failed upload. rig_id there means the robot, as in episode.json, so leave stations out of it. Filter and count with client.episodes.list(dataset, filter='operator = "ana" AND session = "2026-11-09-am"') and client.episodes.summarize(dataset, facets=["operator", "scene", "session"]), and correct a station with client.episodes.update_many(dataset, [{"episode": "<EPISODE>", "scene": "station-1"}]), which wins over the file (data collection guide).

client.datasets.setup() takes no solver: a dataset follows its robot's solver, set on the robot's Solver page (setting a dataset up). client.checks.reach_status(dataset).calibration_mix is set when the dataset's uploads hold more than one capture calibration (or the solver is refused for a mix), and says which calibrations and rigs, whether each is current and whether the solver will not run on the mix or runs (API guide).

Upgrading from 0.3.6: 0.8.0rc11 removed the fine-tuning calls, client.finetune and client.jobs. Training isn't available on your account's plan yet. We're working on it. To request access, write to support@qualiastudios.dev.

Catalog checks

Use client.checks.estimate(dataset, kind="heuristic", categories=["heuristic"]) to see which free checks would run, then client.checks.run(dataset, kind="heuristic", all=True) to start them without a credit ceiling. kind can also be "vlm"; without it the catalog selects runnable heuristics (VLM catalog checks are unavailable here). A run contains only one kind. VLM runs need max_cost covering the estimated bound.

Automatic checks

Free quality checks default on, paid video checks default off. An organization Admin/Owner can set the org default, a workspace Admin/Owner its default, and a dataset writer an override. Each half inherits independently from dataset → workspace → org → platform. Reading returns the effective on/off value and its source. Switching on starts nothing; existing episodes need a manual run.

client.checks.automatic_policy("<ORG_UUID>", org=True)
client.checks.set_automatic_policy("<ORG_UUID>", org=True, graph=False)
client.checks.set_automatic_policy("<WORKSPACE_UUID>", workspace=True, heuristic=False)
client.checks.set_automatic_policy("<DATASET_UUID>", reset=("graph",))

Automatic policy contract and permissions.

Pipeline templates

A workspace keeps its own pipeline templates as TOML files that pin public templates. client.pipelines.catalog() lists what may be pinned; validate and apply take the file's text, and apply writes the next version only when the graph changed. estimate gives the most a launch may cost; launch takes that bound as max_cost and is refused if the run no longer prices as quoted.

from pathlib import Path

applied = client.pipelines.apply(Path("our-act.toml").read_text(encoding="utf-8"))
values = {"dataset": "<VERSION_ID>"}
bound = client.pipelines.estimate(
    applied.pipeline_template_id, project_id="<PROJECT_ID>", values=values
)
client.pipelines.launch(
    applied.pipeline_template_id,
    project_id="<PROJECT_ID>",
    values=values,
    max_cost=bound.credits,
    quote_digest=bound.quote_digest,
)

versions, version and retire read and end a template's history. The file format and every rule.

API

• SDK docs: every resource, with examples

In the repository (these links do not resolve on PyPI):

• API guide, examples and release history
• Upload preparation: waiting for it, what prepare found, and confirming task names (client.preparation)
• Episode drafts and version publication
• Server contracts and operations

License

MIT