Metadata-Version: 2.4
Name: current-data-py
Version: 0.1.10
Summary: Current Data protobuf messages and schema metadata
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: protobuf>=3.20
Requires-Dist: turbodata==0.1.0
Requires-Dist: grpcio>=1.51.1
Requires-Dist: alibabacloud-oss-v2<2,>=1.4.0
Provides-Extra: decode
Requires-Dist: av>=12; extra == "decode"
Requires-Dist: pillow>=9; extra == "decode"

# 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.

```bash
ENV=test python -m current_data_py.examples.read_remote_td
```

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

```python
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.

```python
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:

```python
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](https://www.alibabacloud.com/help/en/oss/user-guide/regions-and-endpoints).
`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:

```python
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`.

```python
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](../management_service/README.md#tos-preparation-for-v3).

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

> [!IMPORTANT]
> **Video decoding requires PyAV, Pillow and the FFmpeg native libraries used by
> PyAV.** `open_td()` enables decoding by default, but a base `current-data-py`
> installation does not install the video dependencies. Install the `[decode]`
> extra in every environment that decodes video, including DataLoader workers.

```bash
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](https://pyav.basswood.io/docs/stable/overview/installation.html)
  and [Pillow build](https://pillow.readthedocs.io/en/stable/installation/building-from-source.html).
- **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:

```bash
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.

```python
# 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:

```python
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.

```python
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

```python
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](https://www.alibabacloud.com/help/en/oss/developer-reference/upload-files-using-oss-sdk-for-python-v2/)); 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.

## 打包

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

## 测试

```bash
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
```

## 推送

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

# PyPI
python -m twine upload dist/*
```
