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: 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)[source]
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[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: Optional[str] = None, port: Optional[int] = None, *, ssl: Union[None, bool, AuthenticatedChannel.__init__.ssl] = None, server_hostname: Optional[str] = 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: Optional[float] = None, deadline: Optional[grpclib.metadata.Deadline] = None, metadata: Optional[grpclib.metadata._MetadataLike] = 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: Optional[DialOptions] = None) ViamChannel[source]
async viam.rpc.dial.dial_direct(address: str, options: Optional[DialOptions] = None) grpclib.client.Channel[source]