Metadata-Version: 2.4
Name: current-data-py
Version: 0.1.9
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.

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

```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. This uses
the existing bucket-aware management API; no protocol changes are required.

## 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"])) 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](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 custom upload endpoints. Deploy the current management
  credential API before using the new bucket and data-type arguments.

## 打包

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