Act based on inferences

viam module generate --language python --model-name intelligent-resource \
  --name vision-intelligence --public-namespace <ORGANIZATION-ID> --public

from typing import cast
from viam.services.vision import *

    @classmethod
    def validate_config(
        cls, config: ComponentConfig
    ) -> Tuple[Sequence[str], Sequence[str]]:
        req_deps = []
        fields = config.attributes.fields
        if "camera_name" not in fields:
            raise Exception("missing required camera_name attribute")
        elif not fields["camera_name"].HasField("string_value"):
            raise Exception("camera_name must be a string")
        camera_name = fields["camera_name"].string_value
        if not camera_name:
            raise ValueError("camera_name cannot be empty")
        req_deps.append(camera_name)
        if "vision_name" not in fields:
            raise Exception("missing required vision_name attribute")
        elif not fields["vision_name"].HasField("string_value"):
            raise Exception("vision_name must be a string")
        vision_name = fields["vision_name"].string_value
        if not vision_name:
            raise ValueError("vision_name cannot be empty")
        req_deps.append(vision_name)
        return req_deps, []

    @classmethod
    def validate_config(
        cls, config: ComponentConfig
    ) -> Tuple[Sequence[str], Sequence[str]]:
        req_deps = []
        fields = config.attributes.fields
        # ...
        if "arm_name" not in fields:
            raise Exception("missing required arm_name attribute")
        elif not fields["arm_name"].HasField("string_value"):
            raise Exception("arm_name must be a string")
        arm_name = fields["arm_name"].string_value
        if not arm_name:
            raise ValueError("arm_name cannot be empty")
        req_deps.append(arm_name)
        return req_deps, []

    def reconfigure(
        self, config: ComponentConfig,
        dependencies: Mapping[ResourceName, ResourceBase]
    ):
        camera_name = config.attributes.fields["camera_name"].string_value
        vision_name = config.attributes.fields["vision_name"].string_value

        # Get the full resource name for the vision service
        # (rdk:service:vision/object-detector)
        vision_resource_name = VisionClient.get_resource_name(vision_name)

        # Check if the vision resource exists in dependencies
        if vision_resource_name not in dependencies:
            raise KeyError(f"Vision service '{vision_name}' not found in "
                           f"dependencies. Available resources: "
                           f"{list(dependencies.keys())}")

        vision_resource = dependencies[vision_resource_name]
        self.vision_service = cast(VisionClient, vision_resource)
        self.camera_name = camera_name

        return super().reconfigure(config, dependencies)

    def reconfigure(
        self, config: ComponentConfig,
        dependencies: Mapping[ResourceName, ResourceBase]
    ):
        # ...
        arm_name = config.attributes.fields["arm_name"].string_value
        arm_resource_name = Arm.get_resource_name(arm_name)

        if arm_resource_name not in dependencies:
            raise KeyError(f"Arm component '{arm_name}' not found in "
                           f"dependencies. Available resources: "
                           f"{list(dependencies.keys())}")

        arm_resource = dependencies[arm_resource_name]
        self.arm = cast(Arm, arm_resource)

        return super().reconfigure(config, dependencies)

In the vision-intelligence/src/models/intelligent-resource.py file, find the API methods that you wish to change and add the logic.

Depending on if your vision service implements GetDetections or GetClassifications, use the relevant method and implement the logic.

detections = await self.vision_service.get_detections_from_camera(self.camera_name)
for d in detections:
    if d.confidence > 0.6 and d.class_name == "LABEL":
        self.logger.info(f"Detection {d.class_name} with confidence {d.confidence}.")
        # DO SOMETHING

classifications = await self.vision_service.get_classifications_from_camera(
    self.camera_name,
    4)
for c in classifications:
    if c.confidence > 0.6 and c.class_name == "LABEL":
        self.logger.info(f"Classification {c.class_name} with confidence {c.confidence}.")
        # DO SOMETHING

Safe arm example: The arm wrapper resource uses the vision service to check for people. If a person is detected, the resource raises an error, otherwise it passes the command to the arm.

    async def _is_safe_to_move(self):
        detections = await self.vision_service.get_detections_from_camera(self.camera_name)
        for d in detections:
            if d.confidence > 0.4 and d.class_name == "Person":
                self.logger.warn(f"Detected {d.class_name} with confidence {d.confidence}.")
                return False
        self.logger.warn("No person detected. Safe arm will move.")
        return True

    async def move_to_position(
        self,
        pose: Pose,
        *,
        extra: Optional[Dict[str, Any]] = None,
        timeout: Optional[float] = None,
        **kwargs
    ):
        if await self._is_safe_to_move():
            await self.arm.move_to_position(pose, extra=extra, timeout=timeout)
        else:
            raise ValueError("Person detected. Safe arm will not move.")

    async def move_to_joint_positions(
        self,
        positions: JointPositions,
        *,
        extra: Optional[Dict[str, Any]] = None,
        timeout: Optional[float] = None,
        **kwargs
    ):
        if await self._is_safe_to_move():
            await self.arm.move_to_joint_positions(positions, extra=extra, timeout=timeout)
        else:
            raise ValueError("Person detected. Safe arm will not move.")

All other methods pass the method call through to the self.arm resource. For example:

    async def do_command(
        self,
        command: Mapping[str, ValueTypes],
        *,
        timeout: Optional[float] = None,
        **kwargs
    ) -> Mapping[str, ValueTypes]:
        return await self.arm.do_command(command, timeout, **kwargs)

Enter the path to the run.sh file, for example, /home/naomi/vision-intelligence/run.sh on Linux or /Users/naomi/vision-intelligence/run.sh on macOS. Click Create.

Save your config.

Click +, click Local module, then click Local component or Local service depending on your resource type and fill in the fields as follows:

• model namespace triplet: <namespace>:vision-intelligence:intelligence-resource, you can see the full triplet in the module’s meta.json file
• type: <resource-type>
• name: resource-1

Configure the camera and vision service names by pasting the following in the configuration field and updating the names as needed:

{
  "camera_name": "camera-1",
  "vision_name": "person-detector"
}

Save the config.