Adding a New Service

from cbsr.factory import CBSRfactory
from my_service import MyService


class MyServiceFactory(CBSRfactory):
    def __init__(self):
        """ MyServiceFactory constructor
            inherit from the CBSRfactory constructor to create the new service as a service
        """
        super(MyServiceFactory, self).__init__()

    def get_connection_channel(self):
        """ initialise the name of the service Redis channel
        """
        return 'my_service'

    def create_service(self, connect, identifier, disconnect):
        return MyService(connect, identifier, disconnect)


if __name__ == '__main__':
    my_factory = MyServiceFactory()
    my_factory.run()

""" This file shows an example of a mock service that uses the text_transcript channel result
    from the Dialogflow speech-to-text and publishes it into the new my_service channel.
    The file can be extended to as many services as necessary
"""

from cbsr.service import CBSRservice


class MyService(CBSRservice):
    def __init__(self, connect, identifier, disconnect):
        super(MyService, self).__init__(connect, identifier, disconnect)

    def get_device_types(self):
        """ :return list of devices the service uses
            This method returns a list of all the devices my_service uses; in this case, only the microphone
        """
        return ['mic']

    def get_channel_action_mapping(self):
        """ :return key-value dictionary pairing, with listened channel as key and linked method as value;
                    multiple channels can be listened to, linked to a method when separated by a comma in the output dictionary
            Use an already-existing channel and links it to a method to be called when there is a message in the channel
            The channel should correspond to one of the topics in abstract_connector.py; 
            in this case, my_service uses the text_transcript channel and links it to the method self.execute
        """
        return {self.get_full_channel('text_transcript'): self.execute}

    def execute(self, message):
        """ :param message: the message published in the channel the method was linked to
            This method decodes the data received in the channel it was linked to and further publishes
            a message in the new service channel
            Extra functionalities can be added to the method
        """
        sentence = message['data'].decode()
        self.publish('my_service', sentence)

...
echo Deploying my_service...
cd ../my_service
cp -f {cert.pem,  *.py} ../../docker/cbsr/my_service    # extra files can be copied if used
...

# ------------------------------------------------------------
# My service
# ------------------------------------------------------------
my_service:
  image: python3
  build:
    context: .
    dockerfile: Dockerfile.python3
  hostname: my_service
  user: "${NEW_UID}:${NEW_GID}"
  env_file:
    - ./.env

  working_dir: /my_service
  command: python3 my_service_factory.py
  volumes:
    - ./cbsr/mock:/my_service:rw${MOUNT_OPTIONS}

  tty: true
  stdin_open: false

  networks:
    app_net:
      ipv4_address: 172.16.238.x   # address has to differ from those of the already existing services' 

  depends_on:
    - redis
    - dialogflow
  # - any other services my_service depends on

###########################
# Management              #
###########################

def enable_service(self, name: str) -> None:
    ...
    elif ... and name == 'my_service':
        for mic in self.devices[self.device_types['mic']]:
            pipe.publish(name, mic)
    ...

...
elif channel = 'my_service':
    self.on_my_service(message=data.decode('utf-8'))
...

###########################
# Event handlers          #
###########################
...
def on_my_service(self, message: str) -> None:
    pass
...

###########################
# Event handlers          #
###########################
...
def on_my_service(self, message: str) -> None:
    """ :param message: the message published on the my_service channel
        This method notifies the listeners that a new message has been posted on the my_service channel;
        This method can be further inherited and overridden
    """
    self.__notify_listeners('onMyService', message)
...

from social_interaction_cloud.action import ActionRunner
from social_interaction_cloud.basic_connector import BasicSICConnector

class MyConnector(BasicSICConnector):
    def __init__(self, server_ip: str):
        """ :param server_ip:
            This method inherits the BasicSICConnector and enables the new my_service service;
            my_service needs to be manually enabled in this way
        """
        super(MyConnector, self).__init__(server_ip)
        self.enable_service('my_service')

    def on_my_service(self, message: str) -> None:
        """ :param message: message published in the my_service channel
            This method overrides the event function on_my_service inherited from the BasicSICConnector
        """
        print(message)
        self.stop()