DATA-1813 Add example modular python component for filtering data capture

examples/filters/pycolorfilter/color_filter.py

@@ -0,0 +1,74 @@
+from typing import ClassVar, Mapping, Sequence, Optional, cast, Tuple, List, Any, Dict
+
+from typing_extensions import Self
+
+from viam.module.types import Reconfigurable
+from viam.proto.app.robot import ComponentConfig
+from viam.proto.common import ResourceName, ResponseMetadata, Geometry
+from viam.components.camera import Camera
+from viam.resource.types import Model, ModelFamily
+from viam.resource.base import ResourceBase
+from viam.media.video import NamedImage
+from PIL import Image
+from viam.errors import NoCaptureToStoreError
+from viam.services.vision import Vision
+from viam.utils import from_dm_from_extra
+
+class ColorFilterCam(Camera, Reconfigurable):
+    """A ColorFilterCam wraps the underlying camera `actual_cam` and only keeps the data captured on the actual camera if `vision_service`
+    detects a certain color in the captured image.
+    """
+    MODEL: ClassVar[Model] = Model(ModelFamily("example", "camera"), "colorfilter")
+
+    def __init__(self, name: str):
+        super().__init__(name)
+
+    @classmethod
+    def new_cam(cls, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]) -> Self:
+        cam = cls(config.name)
+        cam.reconfigure(config, dependencies)
+        return cam
+
+    @classmethod
+    def validate_config(cls, config: ComponentConfig) -> Sequence[str]:
+        """Validates JSON configuration"""
+        actual_cam = config.attributes.fields["actual_cam"].string_value
+        if actual_cam == "":
+            raise Exception("actual_cam attribute is required for a ColorFilterCam component")
+        vision_service = config.attributes.fields["vision_service"].string_value
+        if vision_service == "":
+            raise Exception("vision_service attribute is required for a ColorFilterCam component")
+        return [actual_cam, vision_service]
+
+    def reconfigure(self, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]):
+        """Handles attribute reconfiguration"""
+        actual_cam_name = config.attributes.fields["actual_cam"].string_value
+        actual_cam = dependencies[Camera.get_resource_name(actual_cam_name)]
+        self.actual_cam = cast(Camera, actual_cam)
+
+        vision_service_name = config.attributes.fields["vision_service"].string_value
+        vision_service = dependencies[Vision.get_resource_name(vision_service_name)]
+        self.vision_service = cast(Vision, vision_service)
+
+    async def get_properties(self, *, timeout: Optional[float] = None, **kwargs) -> Camera.Properties:
+        """Returns details about the camera"""
+        return await self.actual_cam.get_properties()
+
+    async def get_image(self, mime_type: str = "", *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Image.Image:
+        """Filters the output of the underlying camera"""
+        img = await self.actual_cam.get_image()
+        if from_dm_from_extra(extra):
+            detections = await self.vision_service.get_detections(img)
+            if len(detections) == 0:
+                raise NoCaptureToStoreError()
+
+        return img
+
+    async def get_images(self, *, timeout: Optional[float] = None, **kwargs) -> Tuple[List[NamedImage], ResponseMetadata]:
+        raise NotImplementedError
+
+    async def get_point_cloud(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Tuple[bytes, str]:
+        raise NotImplementedError
+
+    async def get_geometries(self) -> List[Geometry]:
+        raise NotImplementedError

examples/filters/pycolorfilter/main.py

@@ -0,0 +1,20 @@
+import asyncio
+
+from viam.components.camera import Camera
+from viam.module.module import Module
+from viam.resource.registry import Registry, ResourceCreatorRegistration
+
+import color_filter
+
+async def main():
+    """This function creates and starts a new module, after adding all desired resource models.
+    Resource creators must be registered to the resource registry before the module adds the resource model. 
+    """
+    Registry.register_resource_creator(Camera.SUBTYPE, color_filter.ColorFilterCam.MODEL, ResourceCreatorRegistration(color_filter.ColorFilterCam.new_cam))
+    module = Module.from_args()
+
+    module.add_model_from_registry(Camera.SUBTYPE, color_filter.ColorFilterCam.MODEL)
+    await module.start()
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/filters/pycolorfilter/requirements.txt

@@ -0,0 +1,10 @@
+googleapis-common-protos==1.60.0
+grpclib==0.4.5
+h2==4.1.0
+hpack==4.0.0
+hyperframe==6.0.1
+multidict==6.0.4
+Pillow==10.0.0
+protobuf==4.24.0
+typing_extensions==4.7.1
+/Users/katherinewu/viam/viam-python-sdk

examples/filters/pycolorfilter/run.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+cd `dirname $0`
+
+# Create a virtual environment to run our code
+VENV_NAME="venv"
+PYTHON="$VENV_NAME/bin/python"
+
+python3 -m venv $VENV_NAME
+$PYTHON -m pip install -r requirements.txt -U # remove -U if viam-sdk should not be upgraded whenever possible
+
+# Be sure to use `exec` so that termination signals reach the python process,
+# or handle forwarding termination signals manually
+exec $PYTHON main.py $@
+
+# To make this shell script executable, run the terminal command: sudo chmod +x run.sh