viam.rpc.dial

Attributes

LOGGER

Exceptions

InsecureConnectionError

Exception raised when establishing an insecure connection

ViamError

Base Viam Error

Classes

AuthenticateRequest

An AuthenticateRequest contains the credentials used to authenticate.

AuthServiceStub

PBCredentials

Abstract base class for protocol messages.

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

to_thread(→ _R)

Asynchronously run a function in a separate thread.

dial(→ ViamChannel)

dial_direct(→ grpclib.client.Channel)

Module Contents

exception viam.rpc.dial.InsecureConnectionError(address: str, authenticated: bool = False)[source]

Bases: ViamError

Exception raised when establishing an insecure connection

address

The endpoint of the connection

authenticated

True if the insecure connection provided credentials

message

Explanation of the error

exception viam.rpc.dial.ViamError[source]

Bases: Exception

Base Viam Error

class viam.rpc.dial.AuthenticateRequest(*, entity: str = ..., credentials: global___Credentials | None = ...)

Bases: google.protobuf.message.Message

An AuthenticateRequest contains the credentials used to authenticate.

entity: str
property credentials: global___Credentials
HasField(field_name: Literal['credentials', b'credentials']) bool

Checks if a certain field is set for the message.

For a oneof group, checks if any field inside is set. Note that if the field_name is not defined in the message descriptor, ValueError will be raised.

Parameters:

field_name (str) – The name of the field to check for presence.

Returns:

Whether a value has been set for the named field.

Return type:

bool

Raises:

ValueError – if the field_name is not a member of this message.

class viam.rpc.dial.AuthServiceStub(channel: grpclib.client.Channel)[source]
class viam.rpc.dial.PBCredentials(*, type: str = ..., payload: str = ...)

Bases: google.protobuf.message.Message

Abstract base class for protocol messages.

Protocol message classes are almost always generated by the protocol compiler. These generated types subclass Message and implement the methods shown below.

type: str

type is the type of credentials being used.

payload: str

payload is an opaque string used that are of the given type above.

async viam.rpc.dial.to_thread(func: collections.abc.Callable[_P, _R], *args: _P, **kwargs: _P) _R[source]

Asynchronously run a function in a separate thread.

This is a copy of the function defined in the python source, which is only available in python >= 3.9.

See: https://github.com/python/cpython/blob/main/Lib/asyncio/threads.py

viam.rpc.dial.LOGGER
class viam.rpc.dial.Credentials[source]

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)[source]
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

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[source]

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)[source]

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][source]
close() None

Closes connection to the server.

class viam.rpc.dial.ViamChannel[source]
channel: grpclib.client.Channel
release: Callable[[], None]
close()[source]
__del__()[source]
async __aenter__()[source]
async __aexit__(exc_type, exc_value, traceback)[source]
async viam.rpc.dial.dial(address: str, options: DialOptions | None = None) ViamChannel[source]
async viam.rpc.dial.dial_direct(address: str, options: DialOptions | None = None) grpclib.client.Channel[source]