viam.rpc.dial

Attributes

LOGGER

Classes

Credentials	Credentials to connect to the robot and the Viam app.
DialOptions
AuthenticatedChannel	Represents a connection to the server, which can be used with generated
ViamChannel

Functions

dial(→ ViamChannel)
dial_direct(→ grpclib.client.Channel)

Module Contents

viam.rpc.dial.LOGGER

class viam.rpc.dial.Credentials

Credentials to connect to the robot and the Viam app.

type: Literal['robot-location-secret'] | Literal['robot-secret'] | Literal['api-key']

The type of credential

payload: str

The credential

class viam.rpc.dial.DialOptions(*, disable_webrtc: bool = False, auth_entity: str | None = None, credentials: Credentials | None = None, insecure: bool = False, allow_insecure_downgrade: bool = False, allow_insecure_with_creds_downgrade: bool = False, max_reconnect_attempts: int = 3, timeout: float = 20, initial_connection_attempts: int = 3, initial_connection_attempt_timeout: float | None = None)

disable_webrtc: bool

Bypass Web RTC and connect directly to the robot.

auth_entity: str | None

The URL to authenticate against. Should be used if the address passed in and FQDN of the server do not match.

credentials: Credentials | None

Credentials for connecting to the robot

insecure: bool = False

Determine if the RPC connection is TLS based. Must be provided to establish an insecure connection. Otherwise, a TLS based connection will be assumed.

allow_insecure_downgrade: bool = False

Allow the RPC connection to be downgraded to an insecure connection if detected. This is only used when credentials are not present.

allow_insecure_with_creds_downgrade: bool = False

Allow the RPC connection to be downgraded to an insecure connection if detected, even with credentials present. This is generally unsafe to use, but can be requested.

max_reconnect_attempts: int = 3

Max number of times the client attempts to reconnect when connection is lost

initial_connection_attempts: int = 3

Max number of times the client will attempt to establish an initial connection If set to a non-positive integer, then there will be no limit to initial connection attempts

initial_connection_attempt_timeout: float

Number of seconds before dial connection times out on initial connection attempts Defaults to whatever value is set in the timeout field

timeout: float = 20

Number of seconds before the dial connection times out Set to 20sec to match _defaultOfferDeadline in goutils/rpc/wrtc_call_queue.go

classmethod with_api_key(api_key: str, api_key_id: str) → typing_extensions.Self

Create DialOptions with an API key for credentials and default values for other arguments.

Parameters:

• api_key (str) – your API key

• api_key_id (str) – your API key ID. Must be a valid UUID

Raises:

ValueError – Raised if the api_key_id is not a valid UUID

Returns:

the DialOptions

Return type:

Self

class viam.rpc.dial.AuthenticatedChannel(host: str | None = None, port: int | None = None, *, ssl: None | bool | AuthenticatedChannel.__init__.ssl = None, server_hostname: str | None = None)

Bases: grpclib.client.Channel

Represents a connection to the server, which can be used with generated stub classes to perform gRPC calls.

channel = Channel()
client = cafe_grpc.CoffeeMachineStub(channel)

...

request = cafe_pb2.LatteOrder(
    size=cafe_pb2.SMALL,
    temperature=70,
    sugar=3,
)
reply: empty_pb2.Empty = await client.MakeLatte(request)

...

channel.close()

request(name: str, cardinality: grpclib.const.Cardinality, request_type: Type[grpclib.stream._SendType], reply_type: Type[grpclib.stream._RecvType], *, timeout: float | None = None, deadline: grpclib.metadata.Deadline | None = None, metadata: grpclib.metadata._MetadataLike | None = None) → grpclib.client.Stream[grpclib.stream._SendType, grpclib.stream._RecvType]

close() → None

Closes connection to the server.

class viam.rpc.dial.ViamChannel

channel: grpclib.client.Channel

release: Callable[[], None]

close()

__del__()

async __aenter__()

async __aexit__(exc_type, exc_value, traceback)

async viam.rpc.dial.dial(address: str, options: DialOptions | None = None) → ViamChannel

async viam.rpc.dial.dial_direct(address: str, options: DialOptions | None = None) → grpclib.client.Channel