viam.services.discovery ======================= .. py:module:: viam.services.discovery Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/viam/services/discovery/client/index /autoapi/viam/services/discovery/discovery/index /autoapi/viam/services/discovery/service/index Classes ------- .. autoapisummary:: viam.services.discovery.DiscoveryClient viam.services.discovery.Discovery Package Contents ---------------- .. py:class:: DiscoveryClient(name: str, channel: grpclib.client.Channel) Bases: :py:obj:`viam.services.discovery.discovery.Discovery`, :py:obj:`viam.resource.rpc_client_base.ReconfigurableResourceRPCClientBase` Connect to the Discovery service, which allows you to discover resources on a machine. .. py:attribute:: client :type: viam.proto.service.discovery.DiscoveryServiceStub .. py:attribute:: channel .. py:method:: discover_resources(*, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.app.robot.ComponentConfig] :async: Get all component configs of discovered resources on a machine :: my_discovery = DiscoveryClient.from_robot(machine, "my_discovery") # Get the discovered resources result = await my_discovery.discover_resources( "my_discovery", ) discoveries = result.discoveries :param name: The name of the discover service :type name: str :returns: A list of ComponentConfigs that describe the components found by a discover service :rtype: List[ComponentConfig] .. py:method:: do_command(command: Mapping[str, viam.utils.ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, viam.utils.ValueTypes] :async: Send/receive arbitrary commands. :: service = SERVICE.from_robot(robot=machine, "builtin") # replace SERVICE with the appropriate class my_command = { "cmnd": "dosomething", "someparameter": 52 } # Can be used with any resource, using the motion service as an example await service.do_command(command=my_command) :param command: The command to execute :type command: Dict[str, ValueTypes] :returns: Result of the executed command :rtype: Dict[str, ValueTypes] .. py:method:: from_robot(robot: viam.robot.client.RobotClient, name: str) -> typing_extensions.Self :classmethod: Get the service named ``name`` from the provided robot. :: async def connect() -> RobotClient: # Replace "" (including brackets) with your API key and "" with your API key ID options = RobotClient.Options.with_api_key("", "") # Replace "" (included brackets) with your machine's connection URL or FQDN return await RobotClient.at_address("", options) async def main(): robot = await connect() # Can be used with any resource, using the motion service as an example motion = MotionClient.from_robot(robot=machine, name="builtin") robot.close() :param robot: The robot :type robot: RobotClient :param name: The name of the service :type name: str :returns: The service, if it exists on the robot :rtype: Self .. py:method:: get_resource_name(name: str) -> viam.proto.common.ResourceName :classmethod: 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 = Arm.get_resource_name("my_arm") :param name: The name of the Resource :type name: str :returns: The ResourceName of this Resource :rtype: ResourceName .. py:method:: get_operation(kwargs: Mapping[str, Any]) -> viam.operations.Operation 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. ). :param kwargs: The kwargs object containing the operation :type kwargs: Mapping[str, Any] :returns: The operation associated with this function :rtype: viam.operations.Operation .. py:method:: close() :async: 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() .. py:class:: Discovery(name: str, *, logger: Optional[logging.Logger] = None) Bases: :py:obj:`viam.services.service_base.ServiceBase` Discovery represents a Discovery service. This acts as an abstract base class for any drivers representing specific discovery implementations. This cannot be used on its own. If the ``__init__()`` function is overridden, it must call the ``super().__init__()`` function. .. py:attribute:: API :type: Final The API of the Resource .. py:method:: discover_resources(*, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.app.robot.ComponentConfig] :abstractmethod: :async: Get all component configs of discovered resources on a machine :: my_discovery = DiscoveryClient.from_robot(machine, "my_discovery") # Get the discovered resources result = await my_discovery.discover_resources( "my_discovery", ) discoveries = result.discoveries :param name: The name of the discover service :type name: str :returns: A list of ComponentConfigs that describe the components found by a discover service :rtype: List[ComponentConfig] .. py:method:: from_robot(robot: viam.robot.client.RobotClient, name: str) -> typing_extensions.Self :classmethod: Get the service named ``name`` from the provided robot. :: async def connect() -> RobotClient: # Replace "" (including brackets) with your API key and "" with your API key ID options = RobotClient.Options.with_api_key("", "") # Replace "" (included brackets) with your machine's connection URL or FQDN return await RobotClient.at_address("", options) async def main(): robot = await connect() # Can be used with any resource, using the motion service as an example motion = MotionClient.from_robot(robot=machine, name="builtin") robot.close() :param robot: The robot :type robot: RobotClient :param name: The name of the service :type name: str :returns: The service, if it exists on the robot :rtype: Self .. py:method:: do_command(command: Mapping[str, viam.utils.ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, viam.utils.ValueTypes] :abstractmethod: :async: Send/receive arbitrary commands. :: service = SERVICE.from_robot(robot=machine, "builtin") # replace SERVICE with the appropriate class my_command = { "cmnd": "dosomething", "someparameter": 52 } # Can be used with any resource, using the motion service as an example await service.do_command(command=my_command) :param command: The command to execute :type command: Dict[str, ValueTypes] :returns: Result of the executed command :rtype: Dict[str, ValueTypes] .. py:method:: get_resource_name(name: str) -> viam.proto.common.ResourceName :classmethod: 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 = Arm.get_resource_name("my_arm") :param name: The name of the Resource :type name: str :returns: The ResourceName of this Resource :rtype: ResourceName .. py:method:: get_operation(kwargs: Mapping[str, Any]) -> viam.operations.Operation 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. ). :param kwargs: The kwargs object containing the operation :type kwargs: Mapping[str, Any] :returns: The operation associated with this function :rtype: viam.operations.Operation .. py:method:: close() :async: 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()