viam.module.module

Attributes

RESOURCE_TYPE_COMPONENT

RESOURCE_TYPE_SERVICE

LOGGER

Exceptions

ResourceNotFoundError

Exception raised when a resource is not found in the registry

ValidationError

Exception raised when there is an error during module validation

Classes

ComponentConfig

Abstract base class for protocol messages.

AddResourceRequest

Abstract base class for protocol messages.

HandlerDefinition

Abstract base class for protocol messages.

HandlerMap

Abstract base class for protocol messages.

ReadyRequest

Abstract base class for protocol messages.

ReadyResponse

Abstract base class for protocol messages.

ReconfigureResourceRequest

Abstract base class for protocol messages.

RemoveResourceRequest

Abstract base class for protocol messages.

ValidateConfigRequest

Abstract base class for protocol messages.

ValidateConfigResponse

Abstract base class for protocol messages.

ResourceRPCSubtype

Abstract base class for protocol messages.

ResourceBase

The base requirements for a Resource.

Registry

The global registry of robotic parts.

Model

Represents a specific model within a family of models

ResourceName

Abstract base class for protocol messages.

Subtype

Represents a known component/service (resource) API

RobotClient

gRPC client for a Robot. This class should be used for all interactions with a robot.

DialOptions

Server

gRPC Server

ModuleRPCService

Helper class that provides a standard way to create an ABC using

Reconfigurable

The Reconfigurable protocol defines the requirements for making a resource Reconfigurable

Stoppable

The Stoppable protocol defines the requirements for making a resource Stoppable.

Module

Functions

resource_name_from_string(→ viam.proto.common.ResourceName)

Create a ResourceName from its string representation (namespace:resource_type:resource_subtype/<optional_remote:>name)

Module Contents

exception viam.module.module.ResourceNotFoundError(resource: str, name: str)[source]

Bases: ViamGRPCError

Exception raised when a resource is not found in the registry

exception viam.module.module.ValidationError(message: str)[source]

Bases: ViamGRPCError

Exception raised when there is an error during module validation

class viam.module.module.ComponentConfig(*, name: str = ..., namespace: str = ..., type: str = ..., model: str = ..., frame: global___Frame | None = ..., depends_on: collections.abc.Iterable[str] | None = ..., service_configs: collections.abc.Iterable[global___ResourceLevelServiceConfig] | None = ..., attributes: google.protobuf.struct_pb2.Struct | None = ..., api: str = ..., log_configuration: global___LogConfiguration | None = ...)

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.

name: str
namespace: str

deprecated; use api

type: str

deprecated; use api

model: str
api: str
property frame: global___Frame
property depends_on: google.protobuf.internal.containers.RepeatedScalarFieldContainer[str]
property service_configs: google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___ResourceLevelServiceConfig]
property attributes: google.protobuf.struct_pb2.Struct
property log_configuration: global___LogConfiguration
HasField(field_name: Literal['attributes', b'attributes', 'frame', b'frame', 'log_configuration', b'log_configuration']) 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.module.module.AddResourceRequest(*, config: viam.gen.app.v1.robot_pb2.ComponentConfig | None = ..., dependencies: collections.abc.Iterable[str] | None = ...)

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.

property config: viam.gen.app.v1.robot_pb2.ComponentConfig
property dependencies: google.protobuf.internal.containers.RepeatedScalarFieldContainer[str]
HasField(field_name: Literal['config', b'config']) 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.module.module.HandlerDefinition(*, subtype: viam.gen.robot.v1.robot_pb2.ResourceRPCSubtype | None = ..., models: collections.abc.Iterable[str] | None = ...)

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.

property subtype: viam.gen.robot.v1.robot_pb2.ResourceRPCSubtype
property models: google.protobuf.internal.containers.RepeatedScalarFieldContainer[str]
HasField(field_name: Literal['subtype', b'subtype']) 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.module.module.HandlerMap(*, handlers: collections.abc.Iterable[global___HandlerDefinition] | None = ...)

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.

property handlers: google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___HandlerDefinition]
class viam.module.module.ReadyRequest(*, parent_address: str = ..., webrtc_offer: 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.

parent_address: str
webrtc_offer: str
class viam.module.module.ReadyResponse(*, ready: bool = ..., handlermap: global___HandlerMap | None = ..., webrtc_answer: 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.

ready: bool
webrtc_answer: str
property handlermap: global___HandlerMap
HasField(field_name: Literal['handlermap', b'handlermap']) 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.module.module.ReconfigureResourceRequest(*, config: viam.gen.app.v1.robot_pb2.ComponentConfig | None = ..., dependencies: collections.abc.Iterable[str] | None = ...)

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.

property config: viam.gen.app.v1.robot_pb2.ComponentConfig
property dependencies: google.protobuf.internal.containers.RepeatedScalarFieldContainer[str]
HasField(field_name: Literal['config', b'config']) 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.module.module.RemoveResourceRequest(*, name: 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.

name: str
class viam.module.module.ValidateConfigRequest(*, config: viam.gen.app.v1.robot_pb2.ComponentConfig | None = ...)

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.

property config: viam.gen.app.v1.robot_pb2.ComponentConfig
HasField(field_name: Literal['config', b'config']) 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.module.module.ValidateConfigResponse(*, dependencies: collections.abc.Iterable[str] | None = ...)

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.

property dependencies: google.protobuf.internal.containers.RepeatedScalarFieldContainer[str]
class viam.module.module.ResourceRPCSubtype(*, subtype: viam.gen.common.v1.common_pb2.ResourceName | None = ..., proto_service: 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.

proto_service: str
property subtype: viam.gen.common.v1.common_pb2.ResourceName
HasField(field_name: Literal['subtype', b'subtype']) 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.module.module.ResourceBase[source]

Bases: Protocol

The base requirements for a Resource.

SUBTYPE: ClassVar[viam.resource.types.Subtype]

The Subtype of the Resource

name: str

The name of the Resource

classmethod get_resource_name(name: str) viam.proto.common.ResourceName[source]

Get the ResourceName for this Resource with the given name

# Can be used with any resource, using an arm as an example
my_arm_name = my_arm.get_resource_name("my_arm")
Parameters:

name (str) – The name of the Resource

Returns:

The ResourceName of this Resource

Return type:

ResourceName

classmethod from_robot(robot: viam.robot.client.RobotClient, name: str) typing_extensions.Self[source]
Abstractmethod:

Get the Resource named name from the provided robot.

# Can be used with any resource, using an arm as an example
my_arm = Arm.from_robot(robot, "my_arm")
Parameters:
  • robot (RobotClient) – The robot

  • name (str) – The name of the Resource

Returns:

The Resource, if it exists on the robot

Return type:

Self

abstract do_command(command: Mapping[str, viam.utils.ValueTypes], *, timeout: float | None = None, **kwargs) Mapping[str, viam.utils.ValueTypes][source]
Async:

Send/Receive arbitrary commands to the Resource

command = {"cmd": "test", "data1": 500}
result = component.do(command)
Parameters:

command (Mapping[str, ValueTypes]) – The command to execute

Raises:

NotImplementedError – Raised if the Resource does not support arbitrary commands

Returns:

Result of the executed command

Return type:

Mapping[str, ValueTypes]

get_operation(kwargs: Mapping[str, Any]) viam.operations.Operation[source]

Get the Operation associated with the currently running function.

When writing custom resources, you should get the Operation by calling this function and check to see if it’s cancelled. If the Operation is cancelled, then you can perform any necessary (terminating long running tasks, cleaning up connections, etc. ).

Parameters:

kwargs (Mapping[str, Any]) – The kwargs object containing the operation

Returns:

The operation associated with this function

Return type:

viam.operations.Operation

async close()[source]

Safely shut down the resource and prevent further use.

Close must be idempotent. Later configuration may allow a resource to be “open” again. If a resource does not want or need a close function, it is assumed that the resource does not need to return errors when future non-Close methods are called.

await component.close()
class viam.module.module.Registry[source]

The global registry of robotic parts.

NB The Registry should almost never be used directly

The Registry keeps track of the types of Resources that are available on robots using this SDK. All the base resource types are pre-registered (for example Arm, Motor).

If you create a new resource type that is not an extension of any of the existing base resource types, then you must register said resource using Registry.register(...).

classmethod register_subtype(registration: ResourceRegistration[Resource])[source]

Register a Subtype with the Registry

Parameters:

registration (ResourceRegistration) – Object containing registration data for the subtype

Raises:
classmethod register_resource_creator(subtype: viam.resource.types.Subtype, model: viam.resource.types.Model, registration: ResourceCreatorRegistration)[source]

Register a specific Model and validator function for the specific resource Subtype with the Registry

Parameters:
Raises:
classmethod lookup_subtype(subtype: viam.resource.types.Subtype) ResourceRegistration[source]

Lookup and retrieve a registered Subtype by its name

Parameters:

subtype (str) – The subtype of the resource

Raises:

ResourceNotFoundError – Raised if the Subtype is not registered

Returns:

The registration object of the resource

Return type:

ResourceRegistration

classmethod lookup_resource_creator(subtype: viam.resource.types.Subtype, model: viam.resource.types.Model) viam.resource.types.ResourceCreator[source]

Lookup and retrieve a registered resource creator by its subtype and model

Parameters:
  • subtype (Subtype) – The Subtype of the resource

  • model (Model) – The Model of the resource

Raises:

ResourceNotFoundError – Raised if the Subtype Model pairing is not registered

Returns:

The function to create the resource

Return type:

ResourceCreator

classmethod lookup_validator(subtype: viam.resource.types.Subtype, model: viam.resource.types.Model) viam.resource.types.Validator[source]

Lookup and retrieve a registered validator function by its subtype and model. If there is none, return None

Parameters:
  • subtype (Subtype) – The Subtype of the resource

  • model (Model) – The Model of the resource

Returns:

The function to validate the resource

Return type:

Validator

classmethod REGISTERED_SUBTYPES() Mapping[viam.resource.types.Subtype, ResourceRegistration][source]

The dictionary of all registered resources - Key: Subtype of the resource - Value: The registration object for the resource

Returns:

All registered resources

Return type:

Mapping[Subtype, ResourceRegistration]

classmethod REGISTERED_RESOURCE_CREATORS() Mapping[str, ResourceCreatorRegistration][source]

The dictionary of all registered resources - Key: subtype/model - Value: The ResourceCreatorRegistration for the resource

Returns:

All registered resources

Return type:

Mapping[str, ResourceCreatorRegistration]

viam.module.module.RESOURCE_TYPE_COMPONENT = 'component'
viam.module.module.RESOURCE_TYPE_SERVICE = 'service'
class viam.module.module.Model(model_family: ModelFamily, name: str)[source]

Represents a specific model within a family of models

model_family: ModelFamily

The family to which this model belongs

name: str

The name of the model

__str__() str[source]

Return str(self).

__repr__() str[source]

Return repr(self).

__hash__() int[source]

Return hash(self).

__eq__(other: object) bool[source]

Return self==value.

classmethod from_string(model: str, *, ignore_errors=False) typing_extensions.Self[source]

Create a `Model` from its string representation (namespace:family:name).

Parameters:
  • model (str) – The Model as a string

  • ignore_errors (bool, optional) – If namespace or family are not found in the string, default to empty string rather than raise an exception. Defaults to False.

Raises:

ValueError – Raised if the provided string is not a valid Model

Returns:

The Model

Return type:

Self

class viam.module.module.ResourceName(*, namespace: str = ..., type: str = ..., subtype: str = ..., name: 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.

namespace: str
type: str
subtype: str
name: str
class viam.module.module.Subtype(namespace: str, resource_type: str, resource_subtype: str)[source]

Represents a known component/service (resource) API

namespace: str

The namespace of the resource

resource_type: str

The type of the resource, for example component or service

resource_subtype: str

The subtype of the resource for example servo, arm, vision

__str__() str[source]

Return str(self).

__repr__() str[source]

Return repr(self).

__hash__() int[source]

Return hash(self).

__eq__(other: object) bool[source]

Return self==value.

classmethod from_resource_name(resource_name: viam.proto.common.ResourceName) typing_extensions.Self[source]

Convert a `ResourceName` into a `Subtype`

Parameters:

resource_name (viam.proto.common.ResourceName) – The ResourceName to convert

Returns:

A new Subtype

Return type:

Self

classmethod from_string(string: str) typing_extensions.Self[source]

Create a `Subtype` from its string representation (namespace:resource_type:resource_subtype)

Parameters:

string (str) – The Subtype as a string

Raises:

ValueError – Raised if the string does not represent a valid Subtype

Returns:

A new Subtype

Return type:

Self

viam.module.module.resource_name_from_string(string: str) viam.proto.common.ResourceName[source]

Create a ResourceName from its string representation (namespace:resource_type:resource_subtype/<optional_remote:>name)

Parameters:

string (str) – The ResourceName as a string

Raises:

ValueError – Raised if the provided string is not a valid ResourceName

Returns:

The new ResourceName

Return type:

viam.proto.common.ResourceName

class viam.module.module.RobotClient[source]

gRPC client for a Robot. This class should be used for all interactions with a robot.

There are 2 ways to instantiate a robot client:

RobotClient.at_address(...)
RobotClient.with_channel(...)

You can use the client standalone or within a context:

robot = await RobotClient.at_address(...)
async with await RobotClient.with_channel(...) as robot: ...

You must close() the robot to release resources.

Note: Robots used within a context are automatically closed UNLESS created with a channel. Robots created using with_channel are not automatically closed.

Establish a Connection:

import asyncio

from viam.rpc.dial import DialOptions, Credentials
from viam.robot.client import RobotClient


async def connect():
    opts = RobotClient.Options.with_api_key(
        # Replace "<API-KEY>" (including brackets) with your machine's API key
        api_key='<API-KEY>',
        # Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
        api_key_id='<API-KEY-ID>'
    )
    return await RobotClient.at_address('<ADDRESS-FROM-THE-VIAM-APP>', opts)


async def main():
    # Make a RobotClient
    robot = await connect()
    print('Resources:')
    print(robot.resource_names)
    await robot.close()

if __name__ == '__main__':
    asyncio.run(main())

For more information, see Machine Management API.

class Options[source]
refresh_interval: int = 0

How often to refresh the status/parts of the robot in seconds. If not set, the robot will not be refreshed automatically

dial_options: viam.rpc.dial.DialOptions | None = None

Options used to connect clients to gRPC servers

log_level: int

The log level to output

check_connection_interval: int = 10

The frequency (in seconds) at which to check if the robot is still connected. 0 (zero) signifies no connection checks

attempt_reconnect_interval: int = 1

The frequency (in seconds) at which to attempt to reconnect a disconnected robot. 0 (zero) signifies no reconnection attempts

disable_sessions: bool = False

Whether sessions are disabled

classmethod with_api_key(api_key: str, api_key_id: str, **kwargs) typing_extensions.Self[source]

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

# Replace "<API-KEY>" (including brackets) with your machine's API key
api_key = '<API-KEY>'
# Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
api_key_id = '<API-KEY-ID>'

opts = RobotClient.Options.with_api_key(api_key, api_key_id)

robot = await RobotClient.at_address('<ADDRESS-FROM-THE-VIAM-APP>', opts)
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 RobotClient.Options

Return type:

Self

For more information, see Establish a connection.

classmethod at_address(address: str, options: Options) typing_extensions.Self[source]
Async:

Create a robot client that is connected to the robot at the provided address.

async def connect():

    opts = RobotClient.Options.with_api_key(
        # Replace "<API-KEY>" (including brackets) with your machine's API key
        api_key='<API-KEY>',
        # Replace "<API-KEY-ID>" (including brackets) with your machine's API key ID
        api_key_id='<API-KEY-ID>'
    )
    return await RobotClient.at_address('ADDRESS FROM THE VIAM APP', opts)


async def main():
    # Make a RobotClient
    robot = await connect()
Parameters:
  • address (str) – Address of the robot (IP address, URL, etc.)

  • options (Options) – Options for connecting and refreshing

Returns:

the RobotClient

Return type:

Self

For more information, see Establish a connection.

classmethod with_channel(channel: grpclib.client.Channel | viam.rpc.dial.ViamChannel, options: Options) typing_extensions.Self[source]
Async:

Create a robot that is connected to a robot over the given channel.

Any robots created using this method will NOT automatically close the channel upon exit.

from viam.robot.client import RobotClient
from viam.rpc.dial import DialOptions, dial


async def connect_with_channel() -> RobotClient:
    async with await dial('ADDRESS', DialOptions()) as channel:
        return await RobotClient.with_channel(channel, RobotClient.Options())

robot = await connect_with_channel()
Parameters:
  • channel (ViamChannel) – The channel that is connected to a robot, obtained by viam.rpc.dial

  • options (Options) – Options for refreshing. Any connection options will be ignored.

Returns:

the RobotClient

Return type:

Self

For more information, see Establish a connection.

async refresh()[source]

Manually refresh the underlying parts of this robot

await robot.refresh()

For more information, see Machine Management API.

get_component(name: viam.proto.common.ResourceName) viam.components.component_base.ComponentBase[source]

Get a component using its ResourceName.

This function should not be called directly except in specific cases. The method Component.from_robot(...) is the preferred method for obtaining components.

arm = Arm.from_robot(robot=robot, name="my_arm")

Because this function returns a generic ComponentBase rather than the specific component type, it will be necessary to cast the returned component to the desired component. This can be done using a few different methods:

  • Assertion:

    arm = robot.get_component(Arm.get_resource_name("my_arm"))
    assert isinstance(arm, Arm)
    end_pos = await arm.get_end_position()
    
  • Explicit cast:

    from typing import cast
    arm = robot.get_component(Arm.get_resource_name("my_arm"))
    arm = cast(Arm, arm)
    end_pos = await arm.get_end_position()
    
  • Declare type on variable assignment.

    • Note: If using an IDE, a type error may be shown which can be ignored.

    arm: Arm = robot.get_component(Arm.get_resource_name("my_arm"))  # type: ignore
    end_pos = await arm.get_end_position()
    
Parameters:

name (viam.proto.common.ResourceName) – The component’s ResourceName

Raises:
  • ValueError – Raised if the requested resource is not a component

  • ComponentNotFoundError – Error if component with the given type and name does not exist in the registry

Returns:

The component

Return type:

ComponentBase

For more information, see Machine Management API.

get_service(name: viam.proto.common.ResourceName) viam.services.service_base.ServiceBase[source]

Get a service using its ResourceName

This function should not be called directly except in specific cases. The method Service.from_robot(...) is the preferred method for obtaining services.

service = MyService.from_robot(robot=robot, name="my_service")

Because this function returns a generic ServiceBase rather than a specific service type, it will be necessary to cast the returned service to the desired service. This can be done using a few methods:

  • Assertion:

    service = robot.get_service(MyService.get_resource_name("my_service"))
    assert isinstance(service, MyService)
    
  • Explicit cast:

    from typing import cast
    service = robot.get_service(MyService.get_resource_name("my_service"))
    service = cast(MyService, my_service)
    
  • Declare type on variable assignment

    • Note: If using an IDE, a type error may be shown which can be ignored.

    service: MyService = robot.get_service(MyService.get_resource_name("my_service"))  # type: ignore
    
Parameters:

name (viam.proto.common.ResourceName) – The service’s ResourceName

Raises:
  • ValueError – Raised if the requested resource is not a component

  • ComponentNotFoundError – Error if component with the given type and name does not exist in the registry

Returns:

The service

Return type:

ServiceBase

For more information, see Machine Management API.

property resource_names: List[viam.proto.common.ResourceName]

Get a list of all resource names

resource_names = robot.resource_names
Returns:

The list of resource names

Return type:

List[viam.proto.common.ResourceName]

For more information, see Machine Management API.

async close()[source]

Cleanly close the underlying connections and stop any periodic tasks.

await robot.close()

For more information, see Machine Management API.

async __aenter__()[source]
async __aexit__(exc_type, exc_value, traceback)[source]
async get_status(components: List[viam.proto.common.ResourceName] | None = None)[source]

Get the status of the robot’s components. You can optionally provide a list of ResourceName for which you want statuses.

# Get the status of the resources on the machine.
statuses = await robot.get_status()
Parameters:

components (Optional[List[viam.proto.common.ResourceName]]) – Optional list of ResourceName for components you want statuses.

For more information, see Machine Management API.

async get_operations() List[viam.proto.robot.Operation][source]

Get the list of operations currently running on the robot.

operations = await robot.get_operations()
Returns:

The list of operations currently running on a given robot.

Return type:

List[viam.proto.robot.Operation]

For more information, see Machine Management API.

async cancel_operation(id: str)[source]

Cancels the specified operation on the robot.

await robot.cancel_operation("INSERT OPERATION ID")
Parameters:

id (str) – ID of operation to cancel.

For more information, see Machine Management API.

async block_for_operation(id: str)[source]

Blocks on the specified operation on the robot. This function will only return when the specific operation has finished or has been cancelled.

await robot.block_for_operation("INSERT OPERATION ID")
Parameters:

id (str) – ID of operation to block on.

For more information, see Machine Management API.

async get_frame_system_config(additional_transforms: List[viam.proto.common.Transform] | None = None) List[viam.proto.robot.FrameSystemConfig][source]

Get the configuration of the frame system of a given robot.

# Get a list of each of the reference frames configured on the machine.
frame_system = await robot.get_frame_system_config()
print(f"frame system configuration: {frame_system}")
Returns:

The configuration of a given robot’s frame system.

Return type:

List[viam.proto.robot.FrameSystemConfig]

For more information, see Machine Management API.

async transform_pose(query: viam.proto.common.PoseInFrame, destination: str, additional_transforms: List[viam.proto.common.Transform] | None = None) viam.proto.common.PoseInFrame[source]

Transform a given source Pose from the reference frame to a new specified destination which is a reference frame.

pose = await robot.transform_pose(PoseInFrame(), "origin")
Parameters:
  • query (viam.proto.common.PoseInFrame) – The pose that should be transformed.

  • destination (str) – The name of the reference frame to transform the given pose to.

Returns:

The pose and the reference frame for the new destination.

Return type:

PoseInFrame

For more information, see Machine Management API.

abstract transform_point_cloud()[source]
Async:

async discover_components(queries: List[viam.proto.robot.DiscoveryQuery]) List[viam.proto.robot.Discovery][source]

Get the list of discovered component configurations.

# Define a new discovery query.
q = robot.DiscoveryQuery(subtype=acme.API, model="some model")

# Define a list of discovery queries.
qs = [q]

# Get component configurations with these queries.
component_configs = await robot.discover_components(qs)
Parameters:

queries (List[viam.proto.robot.DiscoveryQuery]) – The list of component models to lookup configurations for.

Returns:

A list of discovered component configurations.

Return type:

List[Discovery]

For more information, see Machine Management API.

async stop_all(extra: Dict[viam.proto.common.ResourceName, Dict[str, Any]] = {})[source]

Cancel all current and outstanding operations for the robot and stop all actuators and movement.

# Cancel all current and outstanding operations for the robot and stop all actuators and movement.
await robot.stop_all()
await robot.stop_all()
Parameters:

extra (Dict[viam.proto.common.ResourceName, Dict[str, Any]]) – Any extra parameters to pass to the resources’ stop methods, keyed on the resource’s ResourceName.

For more information, see Machine Management API.

async log(name: str, level: str, time: datetime.datetime, log: str, stack: str)[source]

Send log from Python module over gRPC.

Create a LogEntry object from the log to send to RDK.

Parameters:
  • name (str) – The logger’s name.

  • level (str) – The level of the log.

  • time (datetime) – The log creation time.

  • log (str) – The log message.

  • stack (str) – The stack information of the log.

For more information, see Machine Management API.

async get_cloud_metadata() viam.proto.robot.GetCloudMetadataResponse[source]

Get app-related information about the robot.

metadata = machine.get_cloud_metadata()
print(metadata.machine_id)
print(metadata.machine_part_id)
print(metadata.primary_org_id)
print(metadata.location_id)
Returns:

App-related metadata.

Return type:

viam.proto.robot.GetCloudMetadataResponse

For more information, see Machine Management API.

async shutdown()[source]

Shutdown shuts down the machine.

Raises:

GRPCError – Raised with DeadlineExceeded status if shutdown request times out, or if robot server shuts down before having a chance to send a response. Raised with status Unavailable if server is unavailable, or if robot server is in the process of shutting down when response is ready.

For more information, see Machine Management API.

class viam.module.module.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.module.module.Server(resources: List[viam.resource.base.ResourceBase], *, module_service: viam.module.service.ModuleRPCService | None = None)[source]

Bases: viam.resource.manager.ResourceManager

gRPC Server

async serve(host: str | None = 'localhost', port: int | None = 9090, log_level: int | None = logging.INFO, *, path: str | None = None)[source]

Server the gRPC server on the provided host and port

Parameters:
  • host (Optional[str], optional) – Desired hostname of the server. Defaults to “localhost”.

  • port (Optional[int], optional) – Desired port of the server. Defaults to 9090.

  • log_level (Optional[int], optional) – The minimum log level. To not receive any logs, set to None. Defaults to logging.INFO.

  • path (Optional[str], optional) – UNIX socket path. Takes precedence over host and port if set. Defaults to None.

classmethod create_and_serve(components: List[viam.resource.base.ResourceBase], host: str | None = 'localhost', port: int | None = 9090, log_level: int = logging.INFO, *, path: str | None = None)[source]
Async:

Convenience method to create and start the server.

Parameters:
  • components (List[ComponentBase]) – List of components to manage

  • host (str, optional) – Desired hostname. Defaults to “localhost”.

  • port (int, optional) – Desired port. Defaults to 9090.

  • log_level (int, optional) – The minimum log level. To not receive any logs, set to None. Defaults to logging.INFO

  • path (Optional[str], optional) – UNIX socket path. Takes precedence over host and port if set. Defaults to None.

async close()

Close the resourcce manager by removing all resources. Please note that any errors will not raise an exception. Errors will still be logged.

class viam.module.module.ModuleRPCService(module: viam.module.module.Module)[source]

Bases: viam.proto.module.ModuleServiceBase

Helper class that provides a standard way to create an ABC using inheritance.

async AddResource(stream: grpclib.server.Stream[viam.proto.module.AddResourceRequest, viam.proto.module.AddResourceResponse]) None[source]
async ReconfigureResource(stream: grpclib.server.Stream[viam.proto.module.ReconfigureResourceRequest, viam.proto.module.ReconfigureResourceResponse]) None[source]
async RemoveResource(stream: grpclib.server.Stream[viam.proto.module.RemoveResourceRequest, viam.proto.module.RemoveResourceResponse]) None[source]
async Ready(stream: grpclib.server.Stream[viam.proto.module.ReadyRequest, viam.proto.module.ReadyResponse]) None[source]
async ValidateConfig(stream: grpclib.server.Stream[viam.proto.module.ValidateConfigRequest, viam.proto.module.ValidateConfigResponse]) None[source]
class viam.module.module.Reconfigurable[source]

Bases: Protocol

The Reconfigurable protocol defines the requirements for making a resource Reconfigurable

reconfigure(config: viam.proto.app.robot.ComponentConfig, dependencies: Mapping[viam.proto.common.ResourceName, viam.resource.base.ResourceBase])[source]
class viam.module.module.Stoppable[source]

Bases: Protocol

The Stoppable protocol defines the requirements for making a resource Stoppable.

All resources that physically move should be Stoppable.

stop(*, extra: Mapping[str, Any] | None = None, timeout: float | None = None, **kwargs)[source]
viam.module.module.LOGGER
class viam.module.module.Module(address: str, *, log_level: int = logging.INFO)[source]
parent: viam.robot.client.RobotClient | None = None
server: viam.rpc.server.Server
classmethod from_args() typing_extensions.Self[source]

Create a new Module with the args provided in the command line. The first argument after the command must be the socket path. If the second argument after the command is “–log-level=debug”, the Module’s logger will be DEBUG level. Otherwise, it will be INFO level. See LogLevel documentation in the RDK for more information on how to start modules with a “log-level” commandline argument.

Raises:

Exception – If there is no socket path provided in the command line argument

Returns:

a new Module instance

Return type:

Module

classmethod run_with_models(*models: viam.resource.base.ResourceBase)[source]
Async:

Module entrypoint that takes a list of ResourceBase implementations. In most cases you’ll want to use run_from_registry instead (see below).

classmethod run_from_registry()[source]
Async:

Module entrypoint that automatically includes all the resources you’ve created in your program.

Example:

if __name__ == ‘__main__’:

asyncio.run(Module.run_from_registry())

Full example at examples/easy_resource/main.py.

async start()[source]

Start the module service and gRPC server

async stop()[source]

Stop the module service and gRPC server

set_ready(ready: bool)[source]

Set the module’s ready state. The module automatically sets to READY on load. Setting to False can be useful in instances where the module is not instantly ready (for example waiting on hardware)

Parameters:

ready (bool) – Whether the module is ready

async add_resource(request: viam.proto.module.AddResourceRequest)[source]
async reconfigure_resource(request: viam.proto.module.ReconfigureResourceRequest)[source]
async remove_resource(request: viam.proto.module.RemoveResourceRequest)[source]
async ready(request: viam.proto.module.ReadyRequest) viam.proto.module.ReadyResponse[source]
add_model_from_registry(subtype: viam.resource.types.Subtype, model: viam.resource.types.Model)[source]

Add a pre-registered model to this Module

async validate_config(request: viam.proto.module.ValidateConfigRequest) viam.proto.module.ValidateConfigResponse[source]