viam.rpc.dial

Module Contents

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)

Attributes

LOGGER

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

Credentials to connect to the robot and the Viam app.

type: Union[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]