Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,13 @@ Use `*` to match one level and `**` to match any depth:

## Teleoperation

Connect to Adamo and stream data in real time.
Connect to Adamo and stream data in real time. By default the SDK uses **QUIC** to reach the cloud router. You can pass `protocol="udp"` or `protocol="tcp"` when QUIC is unavailable or UDP is blocked on your network.

```python
import adamo

session = adamo.connect(api_key="ak_...")
# session = adamo.connect(api_key="ak_...", protocol="udp")

# Publish joint states from your robot
session.put("arm-01/control/joint_states", payload)
Expand Down
50 changes: 43 additions & 7 deletions src/adamo/operate/_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import json
from typing import Literal, cast

import zenoh

Expand All @@ -17,14 +18,38 @@

DEFAULT_API_URL = "https://q14iirks46.execute-api.eu-west-2.amazonaws.com"

_ZENOH_PROTOCOLS = ("quic", "udp", "tcp")
ZenohProtocol = Literal["quic", "udp", "tcp"]

def _build_zenoh_config(info: ConnectionInfo) -> zenoh.Config:
"""Build a Zenoh Config targeting the Adamo router via QUIC."""

def _validate_protocol(protocol: str) -> ZenohProtocol:
if protocol not in _ZENOH_PROTOCOLS:
allowed = ", ".join(repr(p) for p in _ZENOH_PROTOCOLS)
raise ValueError(f"protocol must be one of {allowed}, got {protocol!r}")
return cast(ZenohProtocol, protocol)


def _zenoh_connect_endpoint(info: ConnectionInfo, protocol: ZenohProtocol) -> str:
if protocol == "quic":
return info.quic_endpoint
if protocol == "udp":
if not info.udp_endpoint:
raise ValueError("UDP endpoint missing from connection config")
return info.udp_endpoint
# tcp — derive from UDP URL (same rule as adamohq/adamo-network)
if "udp/" not in info.udp_endpoint:
raise ValueError(
"Cannot derive TCP endpoint: UDP URL does not contain 'udp/'"
)
return info.udp_endpoint.replace("udp/", "tcp/", 1)


def _build_zenoh_config(info: ConnectionInfo, protocol: ZenohProtocol) -> zenoh.Config:
"""Build a Zenoh client config targeting the Adamo router."""
endpoint = _zenoh_connect_endpoint(info, protocol)
config = zenoh.Config()
config.insert_json5("mode", json.dumps("client"))
config.insert_json5(
"connect/endpoints", json.dumps([info.quic_endpoint])
)
config.insert_json5("connect/endpoints", json.dumps([endpoint]))
return config


Expand All @@ -34,6 +59,7 @@ def connect(
token: str | None = None,
org_id: str | None = None,
api_url: str = DEFAULT_API_URL,
protocol: ZenohProtocol = "quic",
) -> Session:
"""Connect to Adamo and return a Session.

Expand All @@ -44,12 +70,18 @@ def connect(
token: A Supabase JWT token. Used for user-authenticated sessions.
org_id: Organization ID (only used with ``token``).
api_url: Override the Adamo API base URL.
protocol: Zenoh transport to the cloud router: ``\"quic\"`` (default),
``\"udp\"``, or ``\"tcp\"``. QUIC matches existing behavior. UDP uses
the API ``adamo_udp_url``; TCP is derived from that URL by replacing
``udp/`` with ``tcp/``. On restrictive networks, UDP may be blocked
by firewalls or NAT; TCP or QUIC may be more reliable.

Returns:
A connected :class:`Session`.
"""
p = _validate_protocol(protocol)
info = _resolve_auth(api_key=api_key, token=token, org_id=org_id, api_url=api_url)
config = _build_zenoh_config(info)
config = _build_zenoh_config(info, p)
zenoh_session = zenoh.open(config)
return Session(zenoh_session, info)

Expand All @@ -60,16 +92,20 @@ async def connect_async(
token: str | None = None,
org_id: str | None = None,
api_url: str = DEFAULT_API_URL,
protocol: ZenohProtocol = "quic",
) -> Session:
"""Async version of :func:`connect`.

The Zenoh session itself is synchronous (the Python binding doesn't expose
an async open), but the API config fetch is done asynchronously.

Accepts the same arguments as :func:`connect`, including ``protocol``.
"""
p = _validate_protocol(protocol)
info = await _resolve_auth_async(
api_key=api_key, token=token, org_id=org_id, api_url=api_url
)
config = _build_zenoh_config(info)
config = _build_zenoh_config(info, p)
zenoh_session = zenoh.open(config)
return Session(zenoh_session, info)

Expand Down