current-data-py

Current Data protobuf messages and schema metadata. Requires Python 3.9+.

pip install current-data-py

from current_data_py import SCHEMA_TO_CLASS, descriptor_set, topic_metadata
from current_data_py.schema import Pose

message = Pose()
message.position.y = 1.0
metadata = topic_metadata(Pose)
decoded = SCHEMA_TO_CLASS[metadata["schema_name"]].FromString(message.SerializeToString())

TD examples write/read a Pose on /head/pose/command:

python -m current_data_py.examples.write_td example.td
python -m current_data_py.examples.read_td example.td

Layout

• schema/: message classes, schema registry and protobuf metadata helpers.
• examples/: runnable TD reading/writing and schema-listing examples.
• remote/: gRPC client/credential handling in client.py, OSS Range source in source.py.
• tests/: installed-package, distribution and remote integration checks.
• Root: package exports and build configuration.

current_data_py.schema and current_data_py.remote imports are preserved. Metadata helpers also remain available from current_data_py; their module is now current_data_py.schema.message_metadata. Run examples with python -m current_data_py.examples.read_td, write_td or list_schemas. Generated protobuf sources stay in the shared proto_gen/python tree and are bundled under the private _proto namespace during packaging.

Development and releases

Packaging configuration lives in current_data_py/. Generated messages live only in the repository's proto_gen/python (message schemas in msg/, management RPC bindings in management/). Regenerate them using the existing proto/generate.sh workflow after changing their proto definitions. Python RPC generation uses protoc and grpc_python_plugin (tested with 3.21.12 and 1.51.1); these are development tools and are not required when installing the package. The build reads those files and adjusts their Python imports in the build directory, placing them under current_data_py._proto in the wheel. It does not modify or duplicate generated sources in the checkout. The source distribution includes the original generated files so it can also build without protoc. Installation requires neither the repository nor protoc.

cd current_data_py  # from the repository root
python -m pip install build twine
python -m build
python -m twine check dist/*
python tests/check_distribution.py

Artifacts are written to current_data_py/dist/. The build adds the shared protobuf inputs to the source archive without copying them into this directory.

For development, use a regular pip install . and run from outside the checkout; reinstall after changes. Editable installs are not supported by this build step.

Install the wheel in a fresh environment and run python /absolute/path/to/current_data_py/tests/check_package.py from outside the repository. Use a clean dist for each release and increment the version in pyproject.toml. Confirm the package name, license and supported Python versions before publishing. No license has been selected yet.

python -m twine upload --repository testpypi dist/*
# After validating the test release:
python -m twine upload dist/*

Remote TD reads

Run the example against data.current-robotics.work:443 (TLS by default):

pip install 'current-data-py[remote]'
python -m current_data_py.examples.read_remote_td \
  --name-prefix recording --topic /head/pose/command

For a local plaintext service, pass localhost:50051 --insecure. It opens the first matching TD and prints at most ten decoded messages. Use --start-timestamp and --end-timestamp for an inclusive time range in nanoseconds. The client obtains and reuses STS credentials automatically.

Install current-data-py[remote]. Import the optional API from current_data_py.remote; basic schema imports do not load gRPC or the OSS SDK.

from contextlib import closing
from current_data_py import SCHEMA_TO_CLASS
from current_data_py.remote import DataClient, MetadataFilter

with DataClient() as client:
    with closing(client.iter_data(MetadataFilter(name_prefix="recording"))) as items:
        for metadata in items:
            with client.open_td(metadata) as reader:
                schemas = {
                    topic.name: topic.metadata["schema_name"]
                    for group in reader.summary().topics_infos
                    for topic in group.topic_metadatas
                }
                for message in reader.read_messages(topic_names=["/head/pose/command"]):
                    decoded = SCHEMA_TO_CLASS[schemas[message.topic_name]].FromString(message.data)
                    print(message.timestamp, decoded)
            break

iter_data(filter=None, timeout=None) yields partial Metadata objects with only ID and storage populated. It consumes gRPC batches incrementally; close the iterator when stopping early. A failed stream raises its gRPC error and is not replayed. DataClient() defaults to data.current-robotics.work:443 with TLS. Use DataClient(address="other-host:443") to override the endpoint, or pass an existing channel as DataClient(channel) (the channel takes precedence). The context manager or close() closes only client-created channels; supplied channels remain caller-owned. The read credential RPC has a 30-second deadline; one client shares credentials across files and refreshes on demand within five minutes of expiry. Only explicit token expiry triggers one refresh and retry; other errors propagate. Credentials remain in memory.

open_td(metadata) supplies a regular TurboData Reader. Use its existing topic, time-range and strategy options. The source performs HEAD once on opening and Range reads with If-Match; changed objects and incomplete responses fail. Each reading thread reuses its own HTTP connection pool. Finish all reads before leaving the context, which releases those pools. If server endpoint or scope changes, create a new DataClient. Objects must be finalized and directly readable.

Run python /absolute/path/to/tests/check_remote.py against the installed wheel with the remote extra. It uses local gRPC and HTTP servers with the real Python stub, OSS SDK and TD Reader; no cloud credentials are needed. Actual Go/OSS production integration remains a separate deployment check.