viam.services.vision ==================== .. py:module:: viam.services.vision Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/viam/services/vision/client/index /autoapi/viam/services/vision/service/index /autoapi/viam/services/vision/vision/index Classes ------- .. autoapisummary:: viam.services.vision.Classification viam.services.vision.Detection viam.services.vision.VisionClient viam.services.vision.CaptureAllResult viam.services.vision.Vision Package Contents ---------------- .. py:class:: Classification(*, class_name: str = ..., confidence: float = ...) Bases: :py:obj:`google.protobuf.message.Message` the general form of the output from a classifier .. py:attribute:: class_name :type: str the class name .. py:attribute:: confidence :type: float the confidence score of the classification .. py:class:: Detection(*, x_min: int | None = ..., y_min: int | None = ..., x_max: int | None = ..., y_max: int | None = ..., confidence: float = ..., class_name: str = ..., x_min_normalized: float | None = ..., y_min_normalized: float | None = ..., x_max_normalized: float | None = ..., y_max_normalized: float | None = ...) Bases: :py:obj:`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. .. py:attribute:: x_min :type: int the four corners of the box .. py:attribute:: y_min :type: int .. py:attribute:: x_max :type: int .. py:attribute:: y_max :type: int .. py:attribute:: confidence :type: float the confidence of the detection .. py:attribute:: class_name :type: str label associated with the detected object .. py:attribute:: x_min_normalized :type: float the four corners of the box, in proportion to the respective image dimension .. py:attribute:: y_min_normalized :type: float .. py:attribute:: x_max_normalized :type: float .. py:attribute:: y_max_normalized :type: float .. py:method:: HasField(field_name: Literal['_x_max', b'_x_max', '_x_max_normalized', b'_x_max_normalized', '_x_min', b'_x_min', '_x_min_normalized', b'_x_min_normalized', '_y_max', b'_y_max', '_y_max_normalized', b'_y_max_normalized', '_y_min', b'_y_min', '_y_min_normalized', b'_y_min_normalized', 'x_max', b'x_max', 'x_max_normalized', b'x_max_normalized', 'x_min', b'x_min', 'x_min_normalized', b'x_min_normalized', 'y_max', b'y_max', 'y_max_normalized', b'y_max_normalized', 'y_min', b'y_min', 'y_min_normalized', b'y_min_normalized']) -> 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, :exc:`ValueError` will be raised. :param field_name: The name of the field to check for presence. :type field_name: str :returns: Whether a value has been set for the named field. :rtype: bool :raises ValueError: if the `field_name` is not a member of this message. .. py:method:: WhichOneof(oneof_group: Literal['_x_max', b'_x_max']) -> Literal['x_max'] | None WhichOneof(oneof_group: Literal['_x_max_normalized', b'_x_max_normalized']) -> Literal['x_max_normalized'] | None WhichOneof(oneof_group: Literal['_x_min', b'_x_min']) -> Literal['x_min'] | None WhichOneof(oneof_group: Literal['_x_min_normalized', b'_x_min_normalized']) -> Literal['x_min_normalized'] | None WhichOneof(oneof_group: Literal['_y_max', b'_y_max']) -> Literal['y_max'] | None WhichOneof(oneof_group: Literal['_y_max_normalized', b'_y_max_normalized']) -> Literal['y_max_normalized'] | None WhichOneof(oneof_group: Literal['_y_min', b'_y_min']) -> Literal['y_min'] | None WhichOneof(oneof_group: Literal['_y_min_normalized', b'_y_min_normalized']) -> Literal['y_min_normalized'] | None Returns the name of the field that is set inside a oneof group. If no field is set, returns None. :param oneof_group: the name of the oneof group to check. :type oneof_group: str :returns: The name of the group that is set, or None. :rtype: str or None :raises ValueError: no group with the given name exists .. py:class:: VisionClient(name: str, channel: grpclib.client.Channel) Bases: :py:obj:`viam.services.vision.vision.Vision`, :py:obj:`viam.resource.rpc_client_base.ReconfigurableResourceRPCClientBase` Connect to the Vision service, which allows you to access various computer vision algorithms (like detection, segmentation, tracking, etc) that usually only require a camera or image input. .. py:attribute:: client :type: viam.proto.service.vision.VisionServiceStub .. py:attribute:: channel .. py:method:: capture_all_from_camera(camera_name: str, return_image: bool = False, return_classifications: bool = False, return_detections: bool = False, return_object_point_clouds: bool = False, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> viam.services.vision.vision.CaptureAllResult :async: Get the next image, detections, classifications, and objects all together, given a camera name. Used for visualization. :: my_detector = VisionClient.from_robot(machine, "my_detector") # Get the captured data for a camera result = await my_detector.capture_all_from_camera( "my_camera", return_image=True, return_detections=True, ) image = result.image detections = result.detections :param camera_name: The name of the camera to use for detection :type camera_name: str :param return_image: Ask the vision service to return the camera's latest image :type return_image: bool :param return_classifications: Ask the vision service to return its latest classifications :type return_classifications: bool :param return_detections: Ask the vision service to return its latest detections :type return_detections: bool :param return_object_point_clouds: Ask the vision service to return its latest 3D segmentations :type return_object_point_clouds: bool :returns: A class that stores all potential returns from the vision service. It can return the image from the camera along with its associated detections, classifications, and objects, as well as any extra info the model may provide. :rtype: vision.CaptureAllResult For more information, see `Computer Vision service `_. .. py:method:: get_detections_from_camera(camera_name: str, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.service.vision.Detection] :async: Get a list of detections in the next image given a camera and a detector :: my_detector = VisionClient.from_robot(robot=machine, "my_detector") # Get detections for the next image from the specified camera detections = await my_detector.get_detections_from_camera("my_camera") :param camera_name: The name of the camera to use for detection :type camera_name: str :raises ViamError: Raised if given an image without a specified width and height :returns: A list of 2D bounding boxes, their labels, and the confidence score of the labels, around the found objects in the next 2D image from the given camera, with the given detector applied to it. :rtype: List[viam.proto.service.vision.Detection] For more information, see `Computer Vision service `_. .. py:method:: get_detections(image: viam.media.video.ViamImage, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.service.vision.Detection] :async: Get a list of detections in the given image using the specified detector :: my_camera = Camera.from_robot(robot=machine, "my_camera") my_detector = VisionClient.from_robot(robot=machine, "my_detector") # Get an image from the camera img = await my_camera.get_image() # Get detections for that image detections = await my_detector.get_detections(img) :param image: The image to get detections for :type image: ViamImage :raises ViamError: Raised if given an image without a specified width and height :returns: A list of 2D bounding boxes, their labels, and the confidence score of the labels, around the found objects in the next 2D image from the given camera, with the given detector applied to it. :rtype: List[viam.proto.service.vision.Detection] For more information, see `Computer Vision service `_. .. py:method:: get_classifications_from_camera(camera_name: str, count: int, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.service.vision.Classification] :async: Get a list of classifications in the next image given a camera and a classifier :: my_classifier = VisionClient.from_robot(robot=machine, "my_classifier") # Get the 2 classifications with the highest confidence scores for the next image from the camera classifications = await my_classifier.get_classifications_from_camera( "my_camera", 2) :param camera_name: The name of the camera to use for detection :type camera_name: str :param count: The number of classifications desired :type count: int :returns: The list of Classifications :rtype: List[viam.proto.service.vision.Classification] For more information, see `Computer Vision service `_. .. py:method:: get_classifications(image: viam.media.video.ViamImage, count: int, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.service.vision.Classification] :async: Get a list of classifications in the given image using the specified classifier :: my_camera = Camera.from_robot(robot=machine, "my_camera") my_classifier = VisionClient.from_robot(robot=machine, "my_classifier") # Get an image from the camera img = await my_camera.get_image() # Get the 2 classifications with the highest confidence scores for the image classifications = await my_classifier.get_classifications(img, 2) :param image: The image to get detections for :type image: ViamImage :param count: The number of classifications desired :type count: int :returns: The list of Classifications :rtype: List[viam.proto.service.vision.Classification] For more information, see `Computer Vision service `_. .. py:method:: get_object_point_clouds(camera_name: str, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[viam.proto.common.PointCloudObject] :async: Returns a list of the 3D point cloud objects and associated metadata in the latest picture obtained from the specified 3D camera (using the specified segmenter). To deserialize the returned information into a numpy array, use the Open3D library. :: import numpy as np import open3d as o3d my_segmenter = VisionClient.from_robot(robot=machine, "my_segmenter") # Get the objects from the camera output objects = await my_segmenter.get_object_point_clouds("my_camera") # write the first object point cloud into a temporary file with open("/tmp/pointcloud_data.pcd", "wb") as f: f.write(objects[0].point_cloud) pcd = o3d.io.read_point_cloud("/tmp/pointcloud_data.pcd") points = np.asarray(pcd.points) :param camera_name: The name of the camera :type camera_name: str :returns: The pointcloud objects with metadata :rtype: List[viam.proto.common.PointCloudObject] For more information, see `Computer Vision service `_. .. py:method:: get_properties(*, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> viam.services.vision.vision.Vision.Properties :async: Get info about what vision methods the vision service provides. Currently returns boolean values that state whether the service implements the classification, detection, and/or 3D object segmentation methods. :: my_detector = VisionClient.from_robot(robot=machine, "my_detector") properties = await my_detector.get_properties() detections_supported = properties.detections_supported classifications_supported = properties.classifications_supported :returns: The properties of the vision service :rtype: Properties For more information, see `Computer Vision service `_. .. 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:: CaptureAllResult(image: Optional[viam.media.video.ViamImage] = None, classifications: Optional[List[viam.proto.service.vision.Classification]] = None, detections: Optional[List[viam.proto.service.vision.Detection]] = None, objects: Optional[List[viam.proto.common.PointCloudObject]] = None, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None) CaptureAllResult represents the collection of things that you have requested from the CaptureAllFromCamera method. This is used most often for visualization purposes, since normally, returning the image on every call to a classifier/detector/etc would be costly and unnecessary. The default result for each field is None rather than the empty list to distinguish between "there was no request for the classifier/detector to return a result" vs. "the classifier/detector was requested, but there were no results". .. py:attribute:: image .. py:attribute:: classifications .. py:attribute:: detections .. py:attribute:: objects .. py:attribute:: extra .. py:class:: Vision(name: str, *, logger: Optional[logging.Logger] = None) Bases: :py:obj:`viam.services.service_base.ServiceBase` Vision represents a Vision service. This acts as an abstract base class for any drivers representing specific vision implementations. This cannot be used on its own. If the ``__init__()`` function is overridden, it must call the ``super().__init__()`` function. For more information, see `Computer Vision service `_. .. py:attribute:: API :type: Final The API of the Resource .. py:attribute:: Properties :type: TypeAlias :value: GetPropertiesResponse Properties is a class that states what features are supported on the associated vision service. Currently, these are the following properties: - classifications_supported (bool): GetClassifications and GetClassificationsFromCamera are implemented. - detections_supported (bool): GetDetections and GetDetectionsFromCamera are implemented. - object_point_clouds_supported (bool): GetObjectPointClouds is implemented. .. py:method:: capture_all_from_camera(camera_name: str, return_image: bool = False, return_classifications: bool = False, return_detections: bool = False, return_object_point_clouds: bool = False, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> CaptureAllResult :abstractmethod: :async: Get the next image, detections, classifications, and objects all together, given a camera name. Used for visualization. :: my_detector = VisionClient.from_robot(machine, "my_detector") # Get the captured data for a camera result = await my_detector.capture_all_from_camera( "my_camera", return_image=True, return_detections=True, ) image = result.image detections = result.detections :param camera_name: The name of the camera to use for detection :type camera_name: str :param return_image: Ask the vision service to return the camera's latest image :type return_image: bool :param return_classifications: Ask the vision service to return its latest classifications :type return_classifications: bool :param return_detections: Ask the vision service to return its latest detections :type return_detections: bool :param return_object_point_clouds: Ask the vision service to return its latest 3D segmentations :type return_object_point_clouds: bool :returns: A class that stores all potential returns from the vision service. It can return the image from the camera along with its associated detections, classifications, and objects, as well as any extra info the model may provide. :rtype: vision.CaptureAllResult For more information, see `Computer Vision service `_. .. py:method:: get_detections_from_camera(camera_name: str, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.service.vision.Detection] :abstractmethod: :async: Get a list of detections in the next image given a camera and a detector :: my_detector = VisionClient.from_robot(robot=machine, "my_detector") # Get detections for the next image from the specified camera detections = await my_detector.get_detections_from_camera("my_camera") :param camera_name: The name of the camera to use for detection :type camera_name: str :raises ViamError: Raised if given an image without a specified width and height :returns: A list of 2D bounding boxes, their labels, and the confidence score of the labels, around the found objects in the next 2D image from the given camera, with the given detector applied to it. :rtype: List[viam.proto.service.vision.Detection] For more information, see `Computer Vision service `_. .. py:method:: get_detections(image: viam.media.video.ViamImage, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.service.vision.Detection] :abstractmethod: :async: Get a list of detections in the given image using the specified detector :: my_camera = Camera.from_robot(robot=machine, "my_camera") my_detector = VisionClient.from_robot(robot=machine, "my_detector") # Get an image from the camera img = await my_camera.get_image() # Get detections for that image detections = await my_detector.get_detections(img) :param image: The image to get detections for :type image: ViamImage :raises ViamError: Raised if given an image without a specified width and height :returns: A list of 2D bounding boxes, their labels, and the confidence score of the labels, around the found objects in the next 2D image from the given camera, with the given detector applied to it. :rtype: List[viam.proto.service.vision.Detection] For more information, see `Computer Vision service `_. .. py:method:: get_classifications_from_camera(camera_name: str, count: int, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.service.vision.Classification] :abstractmethod: :async: Get a list of classifications in the next image given a camera and a classifier :: my_classifier = VisionClient.from_robot(robot=machine, "my_classifier") # Get the 2 classifications with the highest confidence scores for the next image from the camera classifications = await my_classifier.get_classifications_from_camera( "my_camera", 2) :param camera_name: The name of the camera to use for detection :type camera_name: str :param count: The number of classifications desired :type count: int :returns: The list of Classifications :rtype: List[viam.proto.service.vision.Classification] For more information, see `Computer Vision service `_. .. py:method:: get_classifications(image: viam.media.video.ViamImage, count: int, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.service.vision.Classification] :abstractmethod: :async: Get a list of classifications in the given image using the specified classifier :: my_camera = Camera.from_robot(robot=machine, "my_camera") my_classifier = VisionClient.from_robot(robot=machine, "my_classifier") # Get an image from the camera img = await my_camera.get_image() # Get the 2 classifications with the highest confidence scores for the image classifications = await my_classifier.get_classifications(img, 2) :param image: The image to get detections for :type image: ViamImage :param count: The number of classifications desired :type count: int :returns: The list of Classifications :rtype: List[viam.proto.service.vision.Classification] For more information, see `Computer Vision service `_. .. py:method:: get_object_point_clouds(camera_name: str, *, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> List[viam.proto.common.PointCloudObject] :abstractmethod: :async: Returns a list of the 3D point cloud objects and associated metadata in the latest picture obtained from the specified 3D camera (using the specified segmenter). To deserialize the returned information into a numpy array, use the Open3D library. :: import numpy as np import open3d as o3d my_segmenter = VisionClient.from_robot(robot=machine, "my_segmenter") # Get the objects from the camera output objects = await my_segmenter.get_object_point_clouds("my_camera") # write the first object point cloud into a temporary file with open("/tmp/pointcloud_data.pcd", "wb") as f: f.write(objects[0].point_cloud) pcd = o3d.io.read_point_cloud("/tmp/pointcloud_data.pcd") points = np.asarray(pcd.points) :param camera_name: The name of the camera :type camera_name: str :returns: The pointcloud objects with metadata :rtype: List[viam.proto.common.PointCloudObject] For more information, see `Computer Vision service `_. .. py:method:: get_properties(*, extra: Optional[Mapping[str, viam.utils.ValueTypes]] = None, timeout: Optional[float] = None) -> Properties :abstractmethod: :async: Get info about what vision methods the vision service provides. Currently returns boolean values that state whether the service implements the classification, detection, and/or 3D object segmentation methods. :: my_detector = VisionClient.from_robot(robot=machine, "my_detector") properties = await my_detector.get_properties() detections_supported = properties.detections_supported classifications_supported = properties.classifications_supported :returns: The properties of the vision service :rtype: Properties For more information, see `Computer Vision service `_. .. 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()