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

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: Optional[str] = None, credentials: Optional[Credentials] = None, insecure: bool = False, allow_insecure_downgrade: bool = False, allow_insecure_with_creds_downgrade: bool = False, max_reconnect_attempts: int = 3, timeout: float = 20)

disable_webrtc: bool

Bypass Web RTC and connect directly to the robot.

auth_entity: Optional[str]

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

credentials: Optional[Credentials]

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

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: Optional[str] = None, port: Optional[int] = None, *, ssl: Union[None, bool, AuthenticatedChannel.__init__.ssl] = None, server_hostname: Optional[str] = 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: Optional[float] = None, deadline: Optional[grpclib.metadata.Deadline] = None, metadata: Optional[grpclib.metadata._MetadataLike] = 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: Optional[DialOptions] = None) → ViamChannel

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