viam.robot.client ================= .. py:module:: viam.robot.client Attributes ---------- .. autoapisummary:: viam.robot.client.LOGGER Classes ------- .. autoapisummary:: viam.robot.client.RobotClient Module Contents --------------- .. py:data:: LOGGER .. py:class:: RobotClient gRPC client for a machine. This class should be used for all interactions with a machine. 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:: machine = await RobotClient.at_address(...) async with await RobotClient.with_channel(...) as machine: ... You must ``close()`` the machine to release resources. Note: Machines used within a context are automatically closed UNLESS created with a channel. Machines 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 "" (including brackets) with your machine's API key api_key='', # Replace "" (including brackets) with your machine's API key ID api_key_id='' ) return await RobotClient.at_address('', opts) async def main(): # Make a RobotClient machine = await connect() print('Resources:') print(machine.resource_names) await machine.close() if __name__ == '__main__': asyncio.run(main()) For more information, see `Machine Management API `_. .. py:class:: Options .. py:attribute:: refresh_interval :type: int :value: 0 How often to refresh the status of the parts of the machine in seconds. If not set, the machine will not be refreshed automatically .. py:attribute:: dial_options :type: Optional[viam.rpc.dial.DialOptions] :value: None Options used to connect clients to gRPC servers .. py:attribute:: log_level :type: int The log level to output .. py:attribute:: check_connection_interval :type: int :value: 10 The frequency (in seconds) at which to check if the machine is still connected. 0 (zero) signifies no connection checks .. py:attribute:: attempt_reconnect_interval :type: int :value: 1 The frequency (in seconds) at which to attempt to reconnect a disconnected machine. 0 (zero) signifies no reconnection attempts .. py:attribute:: disable_sessions :type: bool :value: False Whether sessions are disabled .. py:method:: with_api_key(api_key: str, api_key_id: str, **kwargs) -> typing_extensions.Self :classmethod: Create RobotClient.Options with an API key for credentials and default values for other arguments. :: # Replace "" (including brackets) with your machine's API key api_key = '' # Replace "" (including brackets) with your machine's API key ID api_key_id = '' opts = RobotClient.Options.with_api_key(api_key, api_key_id) machine = await RobotClient.at_address('', opts) :param api_key: your API key :type api_key: str :param api_key_id: your API key ID. Must be a valid UUID :type api_key_id: str :raises ValueError: Raised if the api_key_id is not a valid UUID :returns: the RobotClient.Options :rtype: Self For more information, see `Establish a connection `_. .. py:method:: at_address(address: str, options: Options) -> typing_extensions.Self :classmethod: :async: Create a robot client that is connected to the machine at the provided address. :: async def connect(): opts = RobotClient.Options.with_api_key( # Replace "" (including brackets) with your machine's API key api_key='', # Replace "" (including brackets) with your machine's API key ID api_key_id='' ) return await RobotClient.at_address('MACHINE ADDRESS', opts) async def main(): # Make a RobotClient machine = await connect() :param address: Address of the machine (IP address, URL, etc.) :type address: str :param options: Options for connecting and refreshing :type options: Options :returns: the RobotClient :rtype: Self For more information, see `Establish a connection `_. .. py:method:: with_channel(channel: Union[grpclib.client.Channel, viam.rpc.dial.ViamChannel], options: Options) -> typing_extensions.Self :classmethod: :async: Create a machine that is connected to a machine over the given channel. Any machines 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()) machine = await connect_with_channel() :param channel: The channel that is connected to a machine, obtained by ``viam.rpc.dial`` :type channel: ViamChannel :param options: Options for refreshing. Any connection options will be ignored. :type options: Options :returns: the RobotClient :rtype: Self For more information, see `Establish a connection `_. .. py:method:: refresh() :async: Manually refresh the underlying parts of this machine. :: await machine.refresh() For more information, see `Machine Management API `_. .. py:method:: get_component(name: viam.proto.common.ResourceName) -> viam.components.component_base.ComponentBase 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=machine, 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 = machine.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 = machine.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 = machine.get_component(Arm.get_resource_name("my_arm")) # type: ignore end_pos = await arm.get_end_position() :param name: The component's ResourceName :type name: viam.proto.common.ResourceName :raises ValueError: Raised if the requested resource is not a component :raises ComponentNotFoundError: Error if component with the given type and name does not exist in the registry :returns: The component :rtype: ComponentBase For more information, see `Machine Management API `_. .. py:method:: get_service(name: viam.proto.common.ResourceName) -> viam.services.service_base.ServiceBase 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=machine, 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 = machine.get_service(MyService.get_resource_name("my_service")) assert isinstance(service, MyService) - Explicit cast:: from typing import cast service = machine.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 = machine.get_service(MyService.get_resource_name("my_service")) # type: ignore :param name: The service's ResourceName :type name: viam.proto.common.ResourceName :raises ValueError: Raised if the requested resource is not a component :raises ComponentNotFoundError: Error if component with the given type and name does not exist in the registry :returns: The service :rtype: ServiceBase For more information, see `Machine Management API `_. .. py:property:: resource_names :type: List[viam.proto.common.ResourceName] Get a list of all resource names :: resource_names = machine.resource_names :returns: The list of resource names :rtype: List[viam.proto.common.ResourceName] For more information, see `Machine Management API `_. .. py:method:: close() :async: Cleanly close the underlying connections and stop any periodic tasks. :: await machine.close() For more information, see `Machine Management API `_. .. py:method:: __aenter__() :async: .. py:method:: __aexit__(exc_type, exc_value, traceback) :async: .. py:method:: get_operations() -> List[viam.proto.robot.Operation] :async: Get the list of operations currently running on the machine. :: operations = await machine.get_operations() :returns: The list of operations currently running on a given machine. :rtype: List[viam.proto.robot.Operation] For more information, see `Machine Management API `_. .. py:method:: cancel_operation(id: str) :async: Cancels the specified operation on the machine. :: await machine.cancel_operation("INSERT OPERATION ID") :param id: ID of operation to cancel. :type id: str For more information, see `Machine Management API `_. .. py:method:: block_for_operation(id: str) :async: Blocks on the specified operation on the machine. This function will only return when the specific operation has finished or has been cancelled. :: await machine.block_for_operation("INSERT OPERATION ID") :param id: ID of operation to block on. :type id: str For more information, see `Machine Management API `_. .. py:method:: get_frame_system_config(additional_transforms: Optional[List[viam.proto.common.Transform]] = None) -> List[viam.proto.robot.FrameSystemConfig] :async: Get the configuration of the frame system of a given machine. :: # Get a list of each of the reference frames configured on the machine. frame_system = await machine.get_frame_system_config() print(f"frame system configuration: {frame_system}") :returns: The configuration of a given machine's frame system. :rtype: List[viam.proto.robot.FrameSystemConfig] For more information, see `Machine Management API `_. .. py:method:: transform_pose(query: viam.proto.common.PoseInFrame, destination: str, additional_transforms: Optional[List[viam.proto.common.Transform]] = None) -> viam.proto.common.PoseInFrame :async: Transform a given source Pose from the reference frame to a new specified destination which is a reference frame. :: from viam.proto.common import Pose, PoseInFrame pose = Pose( x=1.0, # X coordinate in mm y=2.0, # Y coordinate in mm z=3.0, # Z coordinate in mm o_x=0.0, # X component of orientation vector o_y=0.0, # Y component of orientation vector o_z=0.0, # Z component of orientation vector theta=0.0 # Orientation angle in degrees ) pose_in_frame = PoseInFrame( reference_frame="world", pose=pose ) transformed_pose = await machine.transform_pose(pose_in_frame, "world") :param query: The pose that should be transformed. :type query: viam.proto.common.PoseInFrame :param destination: The name of the reference frame to transform the given pose to. :type destination: str :returns: The pose and the reference frame for the new destination. :rtype: PoseInFrame For more information, see `Machine Management API `_. .. py:method:: transform_point_cloud() :abstractmethod: :async: .. py:method:: get_models_from_modules() -> List[viam.proto.robot.ModuleModel] :async: Get a list of all models provided by local and registry modules on the machine. This includes models that are not currently configured on the machine. :: # Get module models module_models = await machine.get_models_from_modules(qs) Args: :returns: A list of discovered models. :rtype: List[ModuleModel] .. py:method:: stop_all(extra: Dict[viam.proto.common.ResourceName, Dict[str, Any]] = {}) :async: Cancel all current and outstanding operations for the machine and stop all actuators and movement. :: # Cancel all current and outstanding operations for the machine and stop all actuators and movement. await machine.stop_all() :param extra: Any extra parameters to pass to the resources' ``stop`` methods, keyed on the resource's ``ResourceName``. :type extra: Dict[viam.proto.common.ResourceName, Dict[str, Any]] For more information, see `Machine Management API `_. .. py:method:: log(name: str, level: str, time: datetime.datetime, message: str, stack: str) :async: Send log from Python module over gRPC. Create a LogEntry object from the log to send to RDK. :param name: The logger's name. :type name: str :param level: The level of the log. :type level: str :param time: The log creation time. :type time: datetime :param message: The log message. :type message: str :param stack: The stack information of the log. :type stack: str For more information, see `Machine Management API `_. .. py:method:: get_cloud_metadata() -> viam.proto.robot.GetCloudMetadataResponse :async: Get app-related information about the machine. :: metadata = await 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. :rtype: viam.proto.robot.GetCloudMetadataResponse For more information, see `Machine Management API `_. .. py:method:: shutdown() :async: Shutdown shuts down the machine. :: await machine.shutdown() :raises GRPCError: Raised with DeadlineExceeded status if shutdown request times out, or if the machine server shuts down before having a chance to send a response. Raised with status Unavailable if server is unavailable, or if machine server is in the process of shutting down when response is ready. For more information, see `Machine Management API `_. .. py:method:: get_version() -> viam.proto.robot.GetVersionResponse :async: Get version information about the machine. :: result = await machine.get_version() print(result.platform) print(result.version) print(result.api_version) :returns: Machine version related information. :rtype: viam.proto.robot.GetVersionResponse For more information, see `Machine Management API `_. .. py:method:: get_machine_status() -> viam.proto.robot.GetMachineStatusResponse :async: Get status information about the machine's resources and configuration. :: machine_status = await machine.get_machine_status() machine_state = machine_status.state resource_statuses = machine_status.resources cloud_metadata = machine_status.resources[0].cloud_metadata config_status = machine_status.config :returns: current status of the machine (initializing or running), resources (List[ResourceStatus]) and config of the machine. :rtype: viam.proto.robot.GetMachineStatusResponse For more information, see `Machine Management API `_.