Skip to main content

auki-domain-py

Synced from the repository

This page mirrors bindings/python/auki-domain-py/README.md in the auki-sdk repo (branch develop). The repository is the source of truth.

PyO3 bindings for auki-domain — the cluster lifecycle crate. Exposes the cluster membership document, the full ClusterManager lifecycle (create / join / bootstrap), the post-#216 ResourceEntry catalog shape, and the ReadFrom / StreamRequest stream subscription types to Python daemons.

Status: Active — post-#216 schema (#230).

Python module: auki_domain​

Value types​

Python classRole
ClusterMemberOne peer row in ClusterMembership
ClusterMembershipCluster membership document (JSON round-trip)
DaemonInfoDaemon-supplied identity fields for participant_info
ParticipantInfoSDK-produced identity wire shape, exchanged over libp2p /auki/info/0.0.1 (.to_json() for local/debug surfaces)
SensorEntryOne row in a peer's sensor catalog
ResourceEntryPost-#216 resource catalog row. variant discriminates sensor_log | pose_log | time_transform_log | detection_log. Nested blocks (head, extent, available, sensor, pose, manifest) returned as Python dicts. Construct via ResourceEntry.from_dict(d) or ResourceEntry.from_json(s) (see below).
MapLogResourceResource Catalog v0.4 Map Log row. Construct via MapLogResource.from_dict(d) or MapLogResource.from_json(s).
MessageEventOne live opaque message with resource_id, authenticated sender_peer_id, type, timestamp_ns, and byte payload
MessageChannelResourceExact remote receiver identity discovered through Resource Catalog v0.3
MessageChannelReceiverBlocking receiver returned by ClusterManager.register_message_channel
MessageChannelSenderPersistent outbound handle opened from a discovered MessageChannelResource
ReadFromStream start position: .latest() / .from_start() / .from_timestamp(ns)
StreamRequestConsumer → Producer handshake: resource_id, source_peer_id, from_
ClusterTargetPolicy enum for ClusterManager.bootstrap
StreamManifestBuilderProducer-side helper for building stream manifests

ClusterManager​

Daemon-side cluster handle. All constructors are synchronous (block on an internal multi-thread tokio runtime).

Constructors (static methods):

  • bootstrap(target, wallet_seed, discovery_url, …) — policy-driven; headless daemons use this
  • create_cluster(wallet_seed, cluster_name, discovery_url, …)
  • create_cluster_with_relay_multiaddrs(…, relay_multiaddrs, …)
  • create_cluster_with_relay_reservation(…, relay_dial_multiaddr, relay_advertise_multiaddr, …)
  • join_cluster(wallet_seed, cluster_name, discovery_url, …)
  • list_clusters(discovery_url) → list[ClusterEntry]

Stream methods:

  • open_stream(peer_id, resource_id) — auto-dispatches by catalog variant and sensor kind, including non-spatial scalar Sensor Logs
  • open_stream_with_request(peer_id, request: StreamRequest) — full post-#216 control; resolves the payload kind by the request's source_peer_id + resource_id
  • open_camera_stream / open_pointcloud_stream / open_joint_encoders_stream / open_audio_stream / open_pose_stream
  • open_map_stream(peer_id, resource, from_=ReadFrom.from_start()) — opens an exact discovered MapLogResource and validates the producer manifest

Catalog / registry:

  • fetch_resources_catalog(peer_id, variants=None) → list[ResourceEntry]
  • fetch_map_catalog(peer_id) → list[MapLogResource]
  • fetch_sensor_entry / fetch_clock_entry / fetch_frame_entry → canonical JSON

Live typed messaging:

  • register_message_channel(resource_id, capacity=64) advertises a Resource Catalog v0.3 message channel owned by this peer, tied to the SDK-declared local session clock, and returns a MessageChannelReceiver.
  • fetch_message_channels(peer_id) returns the peer's exact v0.3 MessageChannelResource rows without falling back to the v0.2 catalog.
  • open_message_channel(resource) validates the discovered owner and clock, then returns a persistent MessageChannelSender.
  • sender.send(type, timestamp_ns, payload) blocks until transport accepts the event into the remote bounded receiver queue. That ACK is not application acceptance; a delivery error is indeterminate and the same operation must not be retried automatically. sender.close() closes the Python handle.
  • receiver.recv() blocks until a live MessageEvent arrives or returns None after the channel closes.

The binding preserves the Rust API's live, ephemeral semantics. It does not store, replay, retry, or interpret messages.

Constructing ResourceEntry from Python​

Use ResourceEntry.from_dict(d) or ResourceEntry.from_json(s) to mint catalog rows from Python — e.g. to feed into ClusterManager.set_resource_catalog_provider. The dict / JSON shape must match the /auki/resources/0.2.0 wire format: a flat object with a variant discriminator field plus the common fields and variant-specific manifest block.

from auki_domain import ResourceEntry

entry = ResourceEntry.from_dict({
    "variant": "pose_log",
    "source_peer_id": "galbot",
    "writer_peer_id": "galbot",
    "resource_id": "base_link->head_left_camera_color_optical_frame",
    "state": "live",
    "head": {"kind": "rolling", "retention_ns": 5_000_000_000},
    "available": {"bytes": 0, "entries": 0, "duration_ns": 0},
    "pose": {"writer_mode": "movable"},
    "manifest": {
        "from_frame": {"peer_id": "galbot", "id": "base_link",   "hash": "..."},
        "to_frame":   {"peer_id": "galbot", "id": "head_left_camera_color_optical_frame", "hash": "..."},
        "clock":      {"peer_id": "galbot", "id": "sdk_clock",   "hash": "..."},
        "source":     {"kind": "manual"},
        "expected_rate_hz": 10,
    },
})

# or from a JSON string:
entry2 = ResourceEntry.from_json('{"variant": "pose_log", ...}')

# pass to ClusterManager:
manager.set_resource_catalog_provider(lambda: [entry])

set_resource_catalog_provider stores the callable, not the callable's return value. The callable is invoked for each inbound /auki/resources/0.2.0 fetch, so Python producers can return a live catalog that changes as backing streams become ready or unavailable. Return only resources that can currently accept stream opens; omit unavailable resources and re-add them later with the same stable resource_id.

All four variants (sensor_log, pose_log, time_transform_log, detection_log) are accepted. Serde discriminates by the variant field. Invalid input raises ValueError.

Post-#216 schema changes​

The old SensorStreamResource / TransformEdgeResource / PoseStreamResource pyclasses from v0.0.52 are deleted. They are replaced by the single flat ResourceEntry type, which mirrors the /auki/resources/0.2.0 wire shape.

Cross-reference:

  • ResourceEntry shape: docs/superpowers/specs/2026-05-27-216-schema-and-api-placement-design.md §1
  • StreamRequest / ReadFrom shape: spec §5
  • Companion binding: auki-session-py owns the declarative Session API (sensor/clock/frame registration, domain join/leave). These two bindings are complementary, not exclusive.

Depends on​