Creating a Simple Service

This tutorial will walk through creating a simple service that detects faces in images from the robot. We’ll walk through:

Creating the service
Defining required methods
Creating the main function of the service
Making the service available to SICApplications

We start by creating a new folder face_detection in the sic/sic_framework/services/ folder. This folder should contain all of the files related to the new service, such as model weights and so on. Lets start by creating the service itself in a new python file face_detection/face_detection_service.py.

In the constructor, we set up the service by calling the super class constructor, which sets up input buffers and registers handlers. All other things related to the service, such as loading model parameters or defining/handling parameters should also be done here.

class FaceDetectionService(SICService):
    def __init__(self, *args, **kwargs):
        super(FaceDetectionService, self).__init__(*args, **kwargs)
        
        cascadePath = "haarcascade_frontalface_default.xml"
        self.faceCascade = cv2.CascadeClassifier(cascadePath)
  
        # Define min window size to be recognized as a face_img
        self.minW = 150
        self.minH = 150

We are required to define the inputs and outputs of the service. These make it easier for anyone using your service to know what they can expect from your service, and which inputs to provide to make it work.

    @staticmethod
    def get_inputs():
        return [CompressedImageMessage]
    
    @staticmethod
    def get_output():
        return BoundingBoxMessage

Finally, we get to the execute function. This function is the main body of the service. For a SICService, the execute will be called once all input buffers contain a timestamp-aligned message. For more information on what exactly this means, and how to adjust it, please read SICService subtypes description | SICService

The execute function has a single parameter: an inputs dict, which has the message types as key and the message itself as values. In the example below you can see how the image message is accessed, and the property image is accessed which contains the actual numpy array.

To read more about the SICMessage.id function, see https://socialrobotics.atlassian.net/wiki/spaces/CBSR/pages/2164391937/Architecture+v2#The-ID-function

The function then uses opencv to detect faces, and converts tuples to objects to make them easier to use. Finally the output message is created.

def execute(self, inputs):
    image = inputs[CompressedImageMessage.id()].image

    img = array(image).astype(np.uint8)

    gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)

    faces = self.faceCascade.detectMultiScale(
        gray,        
        scaleFactor=1.2,        
        minNeighbors=5,        
        minSize=(int(self.minW), 
        int(self.minH)))

    faces = [BoundingBox(x, y, w, h) for (x, y, w, h) in faces]

    return BoundingBoxMessage(faces)

The BoundingBox and BoundingBoxMessage used here are base classes (see sic/sic_framework/core/message_python2.py) provided to allow services to be compatible with one another. As we have no special information other than the bounding box to transmit, we can use these classes. The message inherits form SICMessage to make it transferable over the network, and we provide the bounding objects as a list.

Finally, we need to have a way to make the service available (that is, to run the code we have just written). This can be done by passing the service as an argument to a service manager that is started when we run this file by itself. Now simply run this file to start the manager, which makes your service available to use in a SICApplication (with device id local).

You can also add the service to the predefined devices, if the service should be part of a device (see sic/sic_framework/devices/nao.py for an example).

if __name__ == '__main__':
    SICServiceManager([FaceDetectionService], "local")

There you have it! A service that can perform sensor/data fusion, process the data and send it back to the user in a custom object format. For the full code, take a look at the sic/sic_framework/services/face_detection folder.

You can call this service in a SICApplication using, for example:

# [in SICApplication]
face_service = self.connect(FaceDetectionService, device_id='local', 
                            inputs_to_service=[nao_camera])