:py:mod:`viam.components.input.client` ====================================== .. py:module:: viam.components.input.client Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: viam.components.input.client.ControllerClient Attributes ~~~~~~~~~~ .. autoapisummary:: viam.components.input.client.LOGGER .. py:data:: LOGGER .. py:class:: ControllerClient(name: str, channel: grpclib.client.Channel) Bases: :py:obj:`viam.components.input.input.Controller`, :py:obj:`viam.resource.rpc_client_base.ReconfigurableResourceRPCClientBase` gRPC client for an Input Controller .. py:method:: get_controls(*, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **__) -> List[viam.components.input.input.Control] :async: Returns a list of Controls provided by the Controller :: # Get the controller from the machine. my_controller = Controller.from_robot( robot=myRobotWithController, name="my_controller") # Get the list of Controls provided by the controller. controls = await my_controller.get_controls() # Print the list of Controls provided by the controller. print(f"Controls: {controls}") :returns: List of controls provided by the Controller :rtype: List[Control] .. py:method:: get_events(*, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **__) -> Dict[viam.components.input.input.Control, viam.components.input.input.Event] :async: Returns the most recent Event for each input (which should be the current state) :: # Get the controller from the machine. my_controller = Controller.from_robot( robot=myRobotWithController, name="my_controller") # Get the most recent Event for each Control. recent_events = await my_controller.get_events() # Print out the most recent Event for each Control. print(f"Recent Events: {recent_events}") :returns: The most recent event for each input :rtype: Dict[Control, Event] .. py:method:: register_control_callback(control: viam.components.input.input.Control, triggers: List[viam.components.input.input.EventType], function: Optional[viam.components.input.input.ControlFunction], extra: Optional[Dict[str, Any]] = None, **__) Register a function that will fire on given EventTypes for a given Control :: # Define a function to handle pressing the Start Menu Button "BUTTON_START" on # your controller, printing out the start time. def print_start_time(event): print(f"Start Menu Button was pressed at this time: {event.time}") # Define a function that handles the controller. async def handle_controller(controller): # Get the list of Controls on the controller. controls = await controller.get_controls() # If the "BUTTON_START" Control is found, register the function # print_start_time to fire when "BUTTON_START" has the event "ButtonPress" # occur. if Control.BUTTON_START in controls: controller.register_control_callback( Control.BUTTON_START, [EventType.BUTTON_PRESS], print_start_time) else: print("Oops! Couldn't find the start button control! Is your " "controller connected?") exit() while True: await asyncio.sleep(1.0) async def main(): # ... < INSERT CONNECTION CODE FROM MACHINE'S CODE SAMPLE TAB > # Get your controller from the machine. my_controller = Controller.from_robot( robot=myRobotWithController, name="my_controller") # Run the handleController function. await handleController(my_controller) # ... < INSERT ANY OTHER CODE FOR MAIN FUNCTION > :param control: The control to register the function for :type control: Control :param triggers: The events that will trigger the function :type triggers: List[EventType] :param function: The function to run on specific triggers :type function: ControlFunction .. py:method:: reset_channel(channel: grpclib.client.Channel) Called when the RPC channel was reset. Passes in the new channel. :param channel: The new RPC Channel :type channel: Channel .. py:method:: trigger_event(event: viam.components.input.input.Event, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **__) :async: Directly send an Event (such as a button press) from external code :: # Define a "Button is Pressed" event for the control BUTTON_START. button_is_pressed_event = Event( time(), EventType.BUTTON_PRESS, Control.BUTTON_START, 1.0) # Trigger the event on your controller. Set this trigger to timeout if it has # not completed in 7 seconds. await myController.trigger_event(event=my_event, timeout=7.0) :param event: The event to trigger :type event: Event .. py:method:: do_command(command: Mapping[str, viam.utils.ValueTypes], *, timeout: Optional[float] = None, **__) -> Mapping[str, viam.utils.ValueTypes] :async: Send/Receive arbitrary commands to the Resource :: command = {"cmd": "test", "data1": 500} result = component.do(command) :param command: The command to execute :type command: Mapping[str, ValueTypes] :raises NotImplementedError: Raised if the Resource does not support arbitrary commands :returns: Result of the executed command :rtype: Mapping[str, ValueTypes] .. py:method:: get_geometries(*, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> List[viam.proto.common.Geometry] :async: Get all geometries associated with the Component, in their current configuration, in the frame of the Component. :: geometries = await component.get_geometries() if geometries: # Get the center of the first geometry print(f"Pose of the first geometry's centerpoint: {geometries[0].center}") :returns: The geometries associated with the Component. :rtype: List[Geometry] .. py:method:: from_robot(robot: viam.robot.client.RobotClient, name: str) -> typing_extensions.Self :classmethod: Get the component named ``name`` from the provided robot. :param robot: The robot :type robot: RobotClient :param name: The name of the component :type name: str :returns: The component, 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 = my_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()