Pet Photographer: Create a Data Filtering Module

If your smart machine captures a lot of data, you might want to filter captured data to selectively store only the data you are interested in. For example, you might want to use your smart machine’s camera to capture images based on specific criteria, such as the presence of a certain color, and omit captured images that don’t meet that criteria.

In this tutorial, you will use a custom module to function as a color filter, and use it with a camera to only capture images where your pet is in the frame in the following way:

1. Attach a colored object, such as a blue collar, to your pet.

2. Set up a camera in an area where your pet is likely to appear in the frame, and configure the data management service to capture and sync images from that camera.

3. Configure the colorfilter custom module to filter captured images from your camera, saving them only when your pet, along with their easily-identifiable colored object, is present in the frame.

   

The source code for this module is available on the modular-filter-examples GitHub repository . In addition to the colorfilter module used in this tutorial, the example repository also includes a sensor reading filter which you could use to control and filter the data recorded by a sensor component.

Hardware requirements

To create your own filtering pet photographer robot, you’ll need the following hardware:

• A computer
• A webcam or other type of camera
• A colored object, such as a blue collar for enhanced accuracy (optional)

Setup

Add a new machine in the Viam app. Then follow the setup instructions to install viam-server on the computer you’re using for your project and connect to the Viam app. Wait until your machine has successfully connected.

Then, install Go or Python on both your local development computer and on your machine’s board if they are not the same device.

Add the custom module

In this tutorial, you can choose to add custom data filtering to your machine in one of two ways:

1. Download the colorfilter module from Viam and get started quickly.
2. Code your own color filtering module, exploring the process of building a module from scratch.

Download the colorfilter module

Follow the instructions below to download the colorfilter module in your preferred programming language:

Code your own module

To code your own color filtering module, first create the necessary files and directories on your smart machine:

Code a filter resource model

Next, include all the methods that the corresponding Viam SDK requires in the API definition of its built-in subtype.

The filter function in your custom filter module must contain two critical elements:

1. A utility function that will check if the caller of the filter function is the data management service.
2. A safeguard that ensures if the data management service is not the caller, an error and the unfiltered data is returned.

Follow the steps below to include the utility function and check whether the data management service is the caller of the function responsible for data capture. If a service other than the data management service calls the function, it will return the original, unfiltered data.

To check the caller of the collector function using the utility function:

from viam.utils import from_dm_from_extra

if from_dm_from_extra(extra):
    detections = await self.vision_service.get_detections(img)

if ctx.Value(data.FromDMContextKey{}) != true {
   // If not data management collector, return underlying stream contents without filtering.
   return fs.cameraStream.Next(ctx)
}

// Only return captured image if it contains a certain color set by the vision service.
img, release, err := fs.cameraStream.Next(ctx)

detections, err := fs.visionService.Detections(ctx, img, map[string]interface{}{})

After implementing a check to identify the initiator of the filter function, you must include the safeguard that will return an error if the data management service is not the caller.

To do this, include the following in your filter module’s resource model:

from viam.errors import NoCaptureToStoreError

if from_dm_from_extra(extra):
    detections = await self.vision_service.get_detections(img)
    if len(detections) == 0:
        raise NoCaptureToStoreError()

if len(detections) == 0 {
   return nil, nil, data.ErrNoCaptureToStore
}

Now that you’ve included the required utility function and safeguard, your complete color filter function should look like the following:

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

// Next contains the filtering logic and returns select data from the underlying camera.
func (fs filterStream) Next(ctx context.Context) (image.Image, func(), error) {
    if ctx.Value(data.FromDMContextKey{}) != true {
    // If not data management collector, return underlying stream contents without filtering.
      return fs.cameraStream.Next(ctx)
     }

    // Only return captured image if it contains a certain color set by the vision service.
    img, release, err := fs.cameraStream.Next(ctx)
    if err != nil {
      return nil, nil, errors.New("could not get next source image")
    }
    detections, err := fs.visionService.Detections(ctx, img, map[string]interface{}{})
    if err != nil {
      return nil, nil, errors.New("could not get detections")
    }

    if len(detections) == 0 {
      return nil, nil, data.ErrNoCaptureToStore
    }

    return img, release, err
}

After you have implemented your resource subtype’s required methods and written your filter function, your final code should look like this:

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

// Package colorfilter implements a modular camera that filters the output of an underlying camera and only keeps
// captured data if the vision service detects a certain color in the captured image.
package colorfilter

import (
"context"
"fmt"
"image"

    "go.viam.com/rdk/logging"
    "github.com/pkg/errors"
    "github.com/viamrobotics/gostream"

    "go.viam.com/rdk/components/camera"
    "go.viam.com/rdk/data"
    "go.viam.com/rdk/pointcloud"
    "go.viam.com/rdk/resource"
    "go.viam.com/rdk/rimage/transform"
    "go.viam.com/rdk/services/vision"

)

var (
    // Model is the full model definition.
    Model = resource.NewModel("example", "camera", "colorfilter")
    errUnimplemented = errors.New("unimplemented")
)

func init() {
    resource.RegisterComponent(camera.API, Model, resource.Registration[camera.Camera, *Config]{
    Constructor: newCamera,
    })
}

func newCamera(ctx context.Context, deps resource.Dependencies, conf resource.Config, logger logging.Logger) (camera.Camera, error) {
    c := &colorFilterCam{
        Named: conf.ResourceName().AsNamed(),
        logger: logger,
    }
    if err := c.Reconfigure(ctx, deps, conf); err != nil {
        return nil, err
    }
    return c, nil
}

// Config contains the name to the underlying camera and the name of the vision service to be used.
type Config struct {
    ActualCam string `json:"actual_cam"`
    VisionService string `json:"vision_service"`
}

// Validate validates the config and returns implicit dependencies.
func (cfg \*Config) Validate(path string) ([]string, error) {
    if cfg.ActualCam == "" {
        return nil, fmt.Errorf(`expected "actual_cam" attribute in %q`, path)
    }
    if cfg.VisionService == "" {
        return nil, fmt.Errorf(`expected "vision_service" attribute in %q`, path)
    }

    return []string{cfg.ActualCam, cfg.VisionService}, nil

}

// A colorFilterCam wraps the underlying camera `actualCam` and only keeps the data captured on the actual camera if `visionService`
// detects a certain color in the captured image.
type colorFilterCam struct {
    resource.Named
    actualCam camera.Camera
    visionService vision.Service
    logger loggingg.Logger
}

// Reconfigure reconfigures the modular component with new settings.
func (c *colorFilterCam) Reconfigure(ctx context.Context, deps resource.Dependencies, conf resource.Config) error {
    camConfig, err := resource.NativeConfig[*Config](conf)
    if err != nil {
        return err
    }

    c.actualCam, err = camera.FromDependencies(deps, camConfig.ActualCam)
    if err != nil {
        return errors.Wrapf(err, "unable to get camera %v for colorfilter", camConfig.ActualCam)
    }

    c.visionService, err = vision.FromDependencies(deps, camConfig.VisionService)
    if err != nil {
        return errors.Wrapf(err, "unable to get vision service %v for colorfilter", camConfig.VisionService)
    }

    return nil

}

// DoCommand simply echoes whatever was sent.
func (c \*colorFilterCam) DoCommand(ctx context.Context, cmd map[string]interface{}) (map[string]interface{}, error) {
    return cmd, nil
}

// Close closes the underlying camera.
func (c \*colorFilterCam) Close(ctx context.Context) error {
    return c.actualCam.Close(ctx)
}

// Images does nothing.
func (c \*colorFilterCam) Images(ctx context.Context) ([]camera.NamedImage, resource.ResponseMetadata, error) {
    return nil, resource.ResponseMetadata{}, errUnimplemented
}

// Stream returns a stream that filters the output of the underlying camera stream in the stream.Next method.
func (c \*colorFilterCam) Stream(ctx context.Context, errHandlers ...gostream.ErrorHandler) (gostream.VideoStream, error) {
camStream, err := c.actualCam.Stream(ctx, errHandlers...)
    if err != nil {
        return nil, err
    }

    return filterStream{camStream, c.visionService}, nil

}

// NextPointCloud does nothing.
func (c \*colorFilterCam)NextPointCloud(ctx context.Context)(pointcloud.PointCloud, error)
{
    return nil, errUnimplemented
}

// Properties returns details about the camera.
func (c \*colorFilterCam) Properties(ctx context.Context)(camera.Properties, error)
{
    return c.actualCam.Properties(ctx)
}

// Projector does nothing.
func (c \*colorFilterCam) Projector(ctx context.Context) (transform.Projector, error) {
    return nil, errUnimplemented
}

type filterStream struct {
    cameraStream gostream.VideoStream
    visionService vision.Service
}

// Next contains the filtering logic and returns select data from the underlying camera.
func (fs filterStream) Next(ctx context.Context) (image.Image, func(), error) {
    if ctx.Value(data.FromDMContextKey{}) != true {
        // If not data management collector, return underlying stream contents without filtering.
        return fs.cameraStream.Next(ctx)
    }

    // Only return captured image if it contains a certain color set by the vision service.
    img, release, err := fs.cameraStream.Next(ctx)
    if err != nil {
        return nil, nil, errors.New("could not get next source image")
    }
    detections, err := fs.visionService.Detections(ctx, img, map[string]interface{}{})
    if err != nil {
        return nil, nil, errors.New("could not get detections")
    }

    if len(detections) == 0 {
        return nil, nil, data.ErrNoCaptureToStore
    }

    return img, release, err
}

// Close closes the stream.
func (fs filterStream) Close(ctx context.Context) error {
    return fs.cameraStream.Close(ctx)
}

For more information, see Define a new resource model.

Code an entry point file

Next, code your module entry point file which viam-server will use to initialize and start the filter module.

To code an entry point file yourself, locate the subtype API as defined in the relevant <resource-name>/<resource-name>.go file in the RDK source code.

• In this example, the camera’s API is defined in the camera.go file in the RDK source code. When developing your main.go or main.py file, reference this file.

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,
            color_filter.ColorFilterCam.validate_config
        )
    )
    module = Module.from_args()
    module.add_model_from_registry(
        Camera.SUBTYPE,
        color_filter.ColorFilterCam.MODEL
    )
    await module.start()

if __name__ == "__main__":
    asyncio.run(main())

// Package main is a module which serves the colorfilter custom module.
package main

import (
  "context"

  "go.viam.com/rdk/logging"
  "go.viam.com/utils"

  "github.com/viam-labs/modular-filter-examples/colorfilter"
  "go.viam.com/rdk/components/camera"
  "go.viam.com/rdk/module"
)

func main() {
  utils.ContextualMain(mainWithArgs, module.NewLoggerFromArgs("colorfilter_module"))
}

func mainWithArgs(ctx context.Context, args []string, logger logging.Logger) (err error) {
  myMod, err := module.NewModuleFromArgs(ctx, logger)
  if err != nil {
    return err
  }

  err = myMod.AddModelFromRegistry(ctx, camera.API, colorfilter.Model)
  if err != nil {
    return err
  }

  err = myMod.Start(ctx)
  defer myMod.Close(ctx)
  if err != nil {
    return err
  }
  <-ctx.Done()
  return nil
}

For more information, see Create a new module.

Once you’ve written your filter module, compile your module into a single executable that runs your module when executed.

Note the absolute path to your module’s executable for use in the next section.

Add as a local module

Whether you’ve downloaded the colorfilter module, or written your own color filtering module, the next step is to add the module to your smart machine as a local module:

1. Navigate to the CONFIGURE tab of your machine’s page in the Viam app.

2. Click the + (Create) button next to your main part in the left-hand menu and select Local module, then Local module.

3. Enter a name or use the suggested name for your local module, enter the module’s executable path, then click Create.

   • The name must use only lowercase characters.

4. Then, click the Save button in the top right corner of the page to save your changes.

Add services

Next, add the following services to your smart machine to support the color filter module:

• The data management service enables your smart machine to capture data and sync it to the cloud.
• The vision service enables your smart machine to perform color detection on objects in a camera stream.

Add the data management service

To enable data capture on your machine, add and configure the data management service to capture and store data on your machine’s computer:

{
  "name": "dm",
  "type": "data_manager",
  "namespace": "rdk",
  "attributes": {
    "sync_interval_mins": 0.1,
    "capture_dir": "",
    "tags": [],
    "additional_sync_paths": []
  }
}

Add the vision service

To enable your smart machine to detect a specific color in its camera stream, add a color_detector vision service. For this tutorial, we will configure the vision service to recognize a blue dog collar using #43A1D0 or rgb(67, 161, 208). If you have a different item you want to use, or want to match to a color that matches your pet closely, you can use a different color.

{
  "name": "my_color_detector",
  "type": "vision",
  "model": "color_detector",
  "attributes": {
    "segment_size_px": 100,
    "detect_color": "#43a1d0",
    "hue_tolerance_pct": 0.06
  }
}

Enable filtering by color

With the vision and data management services configured, you can now configure your camera to filter by color and sync photos to Viam’s cloud.

Configure your camera

If you haven’t already, add a camera component to your smart machine:

1. On the CONFIGURE tab, click the + (Create) button next to your main part in the left-hand menu and select Component. Start typing “webcam” and select camera / webcam. Enter a name or use the suggested name for your camera. This tutorial uses the name ‘cam’ in all example code. Click Create.

2. Click the video path dropdown and select the webcam you’d like to use for this project from the list of suggestions.

3. Click Save in the top right corner of the screen to save your changes.

Add the color filter component

1. Click the + icon next to your machine part in the left-hand menu and select Local module, then Local component.

2. On the Create menu:

   1. Select the camera type from the dropdown menu.

   2. Enter example:camera:colorfilter, the model namespace triplet of your modular resource’s model.

   3. Provide a name for this instance of your modular resource. This name must be different from the module name.

      

3. Click Create to create the modular resource.

4. In the resulting module configuration pane, copy the following JSON configuration into the attributes field:

   {
     "vision_service": "my_color_detector",
     "actual_cam": "cam"
   }
   

   

Configure data capture

To add data capture for the color filter camera, click Add method in the Data capture section of your color filter camera component. Toggle the Method dropdown menu, select ReadImage, and set the Frequency of the capture to 0.1, which will configure the data management service to capture images from your camera about once every 10 seconds. Then, click Save.

Test your color filter camera

To test that your color filter camera is capturing and filtering images properly, navigate to the CONTROL tab on your machine’s page.

On the colorfiltercam panel, toggle View colorfiltercam to view your camera’s live feed. Test the filter by positioning your smart machine so that it captures an image of your pet wearing its collar. Then examine the DATA tab to confirm that only pictures containing your pet wearing their collar are stored.

For example, the following is the result of several dozen pictures of the same dog, but only those pictures where he is wearing the blue collar were captured and synced to the cloud:

Next steps

Your pet photographer is now set up. Place it in an area your pet frequently visits and don’t forget to attach the colored object to your pet.

Now you can follow similar steps and customize the code you’ve written to configure a sensor for detecting specific thresholds or filter out blurry images from your camera’s captures.

Try these other tutorials for more on working with the data management and vision services:

Detect Color

Use the vision service in the Viam app to detect a color with the Viam Rover.

Smart Pet Treat Dispenser

Use a Raspberry Pi, a motor, and machine learning to build a smart pet feeder.

Pet Guardian

Make a functional guardian with a servo motor, some LEDs, a camera, and the ML Model and vision service to detect people and pets.

Detect a Person and Send a Photo

Use the vision service and the Python SDK to send yourself a text message when your webcam detects a person.

Capture Data and Train a Model

Configure data capture and cloud sync, filter and tag captured data, and train an ML model.