current-data-py

Environment

Set ENV=test before importing the SDK to use the test service. ENV=prod, an unset ENV, or an empty value selects production. Other values raise ValueError at import time.

ENV=test python -m current_data_py.examples.read_remote_td

In Python, set the environment before the first SDK import:

import os
os.environ["ENV"] = "test"

from current_data_py.management import ManagementClient
from current_data_py.remote import DataClient

with ManagementClient() as management, DataClient() as data:
    pass  # Both clients use the test service over TLS.

The environment initializes current_data_py.DEFAULT_ADDRESS once at import. Changing ENV after import does not switch clients. An explicit address= overrides the default; a caller-provided channel takes precedence over both. Changing current_data_py.DEFAULT_ADDRESS still affects newly created clients. OSS uploads and reads use credentials from the selected management service; explicit bucket arguments remain unchanged.

Management

ManagementClient connects over TLS. Pass address= to configure the service, or supply your own gRPC channel.

The examples below read deployment-specific values from environment variables. Set MANAGEMENT_ADDRESS to your service address and UPLOAD_DATA_TYPE to a valid integer enum value supplied by your deployment administrator. Set the bucket and object-key variables to resources you are authorized to access. These variables are read explicitly by the examples; they are not automatic SDK configuration.

import os

from current_data_py.management import ManagementClient, Metadata, MetadataFilter

with ManagementClient(address=os.environ["MANAGEMENT_ADDRESS"]) as client:
    upload = client.get_upload_credentials(data_type=int(os.environ["UPLOAD_DATA_TYPE"]))
    read = client.get_read_credentials(bucket=os.environ["READ_BUCKET"])

    # Register an already uploaded file; replace the example file details.
    record = client.register_data(Metadata(
        name="recording",
        data_type=int(os.environ["UPLOAD_DATA_TYPE"]),
        storages=[{"bucket": os.environ["UPLOADED_BUCKET"], "object_key": os.environ["UPLOADED_OBJECT_KEY"]}],
        file_info={"size_bytes": 1234, "checksum": "sha256:<file-hash>"},
    ))
    for item in client.stream_data_references(
        MetadataFilter(name_prefix="recording"),
        return_fields=["name", "data_type", "recording_info.task.name", "recording_info.recording_time"],
    ):
        print(item.id, item.name, item.data_type, item.storages, item.recording_info)

Metadata, MetadataFilter, and responses are Protobuf objects exported by the SDK. Registration only saves metadata; it does not upload files. Credential calls return fresh credentials without caching. Calls accept timeout= and metadata= for gRPC headers.

Remote TD reads

Payload decoding

python -m pip install --only-binary=av,pillow 'current-data-py[decode]'

• Prebuilt wheels: PyAV wheels bundle FFmpeg libraries on supported platforms; no separate FFmpeg system installation is normally needed. The command above fails if compatible PyAV/Pillow wheels are unavailable instead of silently compiling them. Wheel availability depends on Python, OS and architecture.
• Source builds: install a compiler, Python development headers, pkg-config and compatible FFmpeg development libraries: libavcodec, libavformat, libavutil, libavdevice, libavfilter, libswscale and libswresample. Runtime shared libraries must also remain available in the deployed environment. Follow the requirements for your chosen PyAV version and Pillow build.
• The ffmpeg executable alone is insufficient: this SDK calls PyAV's native library bindings, not a command-line subprocess. Decoding uses the CPU and does not require CUDA or a GPU.
• Protobuf-only decoding and raw reads (enable_decode=False) do not require PyAV, Pillow or FFmpeg for decoding.

Check the video dependencies in the environment that will run the reader:

python -c "import av; from PIL import Image; av.CodecContext.create('h264', 'r'); print(av.library_versions)"

Decoding applies to both sequential reads and time sampling; it does not depend on a robot dataset format.

# Read all messages in an inclusive time range, without constructing sample queries.
with client.open_td(reference, bucket=bucket) as reader:
    messages = reader.read_messages(
        topic_names=topics,
        start_timestamp=start_ns,
        end_timestamp=end_ns,
    )
    try:
        for result in messages:
            print(result.topic_name, result.timestamp, type(result.value))
    finally:
        messages.close()  # Also release iterator state when stopping early.

Sequential results are DecodedMessage(topic_name, timestamp, value). Reads are lazy: all selected topics share one raw read iterator, with a separate video decoder per topic. Results are yielded as decoding completes; delayed video frames can arrive after newer messages from other topics. Use each result's timestamp for alignment. GOP prefix frames are consumed for decoding but excluded from returned results outside the requested time bounds. Reverse-order decoded reads are not supported; use raw for reverse encoded reads.

To request particular timestamps instead:

from turbodata import SampleQuery

# reference, bucket, topic and timestamps_ns come from your selected recording.
with client.open_td(reference, bucket=bucket) as reader:
    samples_by_topic = reader.sample([SampleQuery(topic, timestamps_ns)])
    for sample in samples_by_topic[0]:
        if sample.found:
            print(sample.timestamp, type(sample.value))

value is a Protobuf message, or an RGB PIL image for H.264 video. Missing samples have found=False and value=None. Timestamps are actual floor-match times, not necessarily the requested times. Repeated video samples may share the same image object; copy it before modifying it in place. Decoding errors propagate; no normalization, resizing or alignment tolerance is applied. Only schemas registered in the SDK are decoded. An unregistered schema raises KeyError with the schema name, even if the TD metadata includes a descriptor. Unsupported video codecs also raise errors.

Decoding is enabled by default. Set enable_decode=False to return the original TurboData Reader when your code needs encoded payloads in .data. The decoded reader exposes summary(), read_messages(), sample() and raw; use raw for encoded reads and other low-level APIs. Video sampling automatically requests GOP dependencies. Keep all reads inside the open_td() context.

For a local Reader, use current_data_py.decoding.DecodedReader(reader). The same module exports decode_messages(payloads, schema) and decode_video(results) for already-read payloads. Supply a complete ordered row from Reader.sample(video_decodable=True) to decode_video; decoder state is local to each call.

Storage and credentials

DataClient.open_td() accepts an optional bucket argument. Without it, reads use the management service's OSS_READ_BUCKET and OSS_READ_PREFIX defaults. With it, the SDK requests whole-bucket read credentials, limited by the server's RAM role, and selects the first matching storage in the reference. An explicit bucket does not inherit the default prefix. Default and per-bucket credentials are cached separately and refreshed automatically.

import os

from current_data_py.remote import DataClient

with DataClient(address=os.environ["MANAGEMENT_ADDRESS"]) as client:
    for reference in client.iter_data():
        # Use the service's default bucket and prefix.
        with client.open_td(reference) as reader:
            print(reader.summary())

        # Request read access to the selected bucket.
        with client.open_td(reference, bucket=os.environ["READ_BUCKET"]) as reader:
            print(reader.summary())

The explicit bucket must contain a storage listed in the reference. This uses the existing bucket-aware management API; no protocol changes are required.

OSS

import os

from current_data_py.management import ManagementClient
from current_data_py.oss import OSSClient

with ManagementClient(address=os.environ["MANAGEMENT_ADDRESS"]) as management:
    with OSSClient(management, permission="read", bucket=os.environ["READ_BUCKET"]) as source:
        source.download_file(os.environ["SOURCE_OBJECT_KEY"], "/tmp/input.h5")

    # Convert the input into /tmp/output.td before uploading.
    with OSSClient(management, permission="upload", data_type=int(os.environ["UPLOAD_DATA_TYPE"])) as destination:
        key = destination.object_prefix + "recording/output.td"
        destination.upload_file(key, "/tmp/output.td")
        print(destination.bucket, key)  # Use these values when registering metadata.

• Read access covers the requested bucket, subject to server permissions. The server chooses the upload bucket and prefix based on the requested data type.
• Credentials refresh automatically. Expired tokens trigger one refresh and retry.
• Uploads reject overwrites by default; set forbid_overwrite=False to replace. Downloads replace the local file only on success; its parent directory must exist.
• Uploads use a single PutObject request (maximum 5 GiB); multipart upload is not supported.
• Supply region= for custom upload endpoints. Deploy the current management credential API before using the new bucket and data-type arguments.

打包

cd current_data_py  # 从仓库根目录执行
# 发布新版本前，更新 pyproject.toml 中的 version
python -m pip install build twine
rm -rf build dist
python -m build

测试

python -m twine check dist/*
python tests/check_distribution.py
python -m pip install --force-reinstall dist/*.whl
python -I tests/check_package.py
python -I tests/check_environment.py
python -I tests/check_management.py
python -I tests/check_remote.py
python -I tests/check_oss.py

推送

# TestPyPI
python -m twine upload --repository testpypi dist/*

# PyPI
python -m twine upload dist/*