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.

For uploads that distinguish raw and converted data, use V2 with both arguments:

with ManagementClient(address=os.environ["MANAGEMENT_ADDRESS"]) as client:
    raw = client.get_upload_credentials_v2(data_type="YAM", data_kind="RAW")
    converted = client.get_upload_credentials_v2(
        data_type="YAM", data_kind="CONVERTED", use_internal_endpoint=True,
    )

V2 accepts EGOSUITE, YAM, and CURRENT_V2 data types and RAW or CONVERTED data kinds. Each argument also accepts its full enum name (DATA_TYPE_YAM, DATA_KIND_RAW) or integer value. DataKind is exported from current_data_py.management; data type constants are in metadata_pb2. Unspecified or unknown enum values are rejected. RAW is not a V2 data type. The server must have the selected environment/type/kind destination configured; unfilled placeholders return FAILED_PRECONDITION. V2 requires a deployed GetUploadCredentialsV2 RPC and never falls back to the old RPC on failure. The existing get_upload_credentials() API and its defaults remain unchanged. V2 responses include the selected destination's oss_endpoint and region, along with its bucket and prefix; raw and converted destinations may use different OSS regions. The server derives the standard Alibaba Cloud endpoint from region: https://oss-<region>.aliyuncs.com by default, or https://oss-<region>-internal.aliyuncs.com with use_internal_endpoint=True. Use the internal endpoint from Alibaba Cloud resources in the same region. See the Alibaba Cloud endpoint reference. OSSClient accepts the same option for V2 uploads and retains it when refreshing credentials. It uses the returned endpoint and region for connections and signing. An explicit region= must match the returned region. If the endpoint or region changes during credential refresh, create a new OSSClient.

For reads in the bucket's own region, use the new read V2 RPC:

with ManagementClient(address=os.environ["MANAGEMENT_ADDRESS"]) as client:
    read = client.get_read_credentials_v2(
        bucket=os.environ["READ_BUCKET"], use_internal_endpoint=False,
    )

The bucket is required and must appear in the service's environment-specific bucket-to-region map, shared with V2 uploads. Reads grant whole-bucket GetObject access bounded by the read RAM role. Public networking is the default; set use_internal_endpoint=True for internal networking. Unconfigured buckets return FAILED_PRECONDITION; empty or malformed buckets return INVALID_ARGUMENT. The existing get_read_credentials() API keeps its configured default endpoint, region and scope. V2 calls never fall back to it on failure.

Multi-cloud credentials (V3)

Use V3 to explicitly choose OSS or native TOS. StorageProvider and StorageCredentials are exported from current_data_py.management.

from current_data_py.management import ManagementClient

with ManagementClient() as client:
    upload = client.get_upload_credentials_v3(
        provider="TOS", data_type="YAM", data_kind="RAW",
        use_internal_endpoint=False,
    )
    read = client.get_read_credentials_v3(
        provider="TOS", bucket=upload.bucket,
        use_internal_endpoint=False,
    )
    print(upload.provider, upload.bucket, upload.endpoint, upload.region)

provider is required and accepts OSS/TOS, full enum names such as STORAGE_PROVIDER_TOS, or enum integer values. Unspecified/unknown providers are rejected. Uploads require the same explicit type/kind as V2; reads require an explicitly configured bucket on the selected provider. OSS V3 uses the same routes and roles as OSS V2. TOS configuration and IAM preparation are described in the service deployment guide.

Both calls return provider, endpoint, region, bucket, object_prefix, access_key_id, access_key_secret, security_token, and expires_at. object_prefix is empty for whole-bucket reads. Use endpoint instead of V2's oss_endpoint. TOS public endpoints use tos-<region>.volces.com; internal endpoints use tos-<region>.ivolces.com. Internal networking requires access to the destination cloud and region's private network.

V3 calls accept timeout and metadata, do not cache credentials, and never retry or fall back to earlier RPC versions. Old servers return UNIMPLEMENTED; unconfigured destinations return FAILED_PRECONDITION. A response whose provider differs from the request raises ValueError.

These methods only fetch credentials. Pass TOS credentials to a native TOS client and refresh them before expiration. OSSClient, DataClient, metadata registration and signed URLs have not been extended to TOS by this change.

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 uses GetReadCredentialsV2 for whole-bucket credentials in the bucket's region, 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. Set use_internal_endpoint=True with an explicit bucket to use internal networking. Credentials are cached separately for each (bucket, use_internal_endpoint) pair, apart from the legacy default scope, 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 and have a configured region on the server. Deploy GetReadCredentialsV2 before upgrading clients that use explicit buckets; older servers return UNIMPLEMENTED and the SDK does not fall back to a potentially incorrect legacy endpoint.

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"]),
                   data_kind="CONVERTED") 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 uses V2 and covers the requested bucket, subject to server permissions. The server chooses the upload bucket and prefix based on the requested data type and kind. Supplying data_kind selects V2 and requires data_type; omit data_kind to retain the legacy upload API. Read clients reject both arguments.
• Reads and V2 uploads use the public endpoint by default. Set use_internal_endpoint=True on OSSClient to select internal networking. Legacy uploads reject this option.
• 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 legacy custom upload endpoints; V2 returns its own region. Deploy the V2 credential APIs and configure bucket regions before using reads or the upload data_kind option.

打包

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/*