Adding a New Service

from os import getcwd

from cbsr.factory import CBSRfactory
from nltk import download
from nltk.data import path

from sentiment_service import SentimentAnalysisService


class SentimentAnalysisFactory(CBSRfactory):
    def __init__(self):
        super(SentimentAnalysisFactory, self).__init__()

    def get_connection_channel(self):
        return 'sentiment_analysis'

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


if __name__ == '__main__':
    cwd = getcwd()
    download('punkt', download_dir=cwd)
    download('omw-1.4', download_dir=cwd)
    download('averaged_perceptron_tagger', download_dir=cwd)
    download('wordnet', download_dir=cwd)
    path.append(cwd)

    sentiment_analysis_factory = SentimentAnalysisFactory()
    sentiment_analysis_factory.run()

""" This file shows an example of a sentiment service that uses the text_transcript channel result
    from the Dialogflow speech-to-text to get the the type of sentiment
"""

from pickle import load
from re import sub
from string import punctuation

from cbsr.service import CBSRservice
from nltk.stem.wordnet import WordNetLemmatizer
from nltk.tag import pos_tag
from nltk.tokenize import word_tokenize


class SentimentAnalysisService(CBSRservice):
    def __init__(self, connect, identifier, disconnect):
        super(SentimentAnalysisService, self).__init__(connect, identifier, disconnect)
        with open('classifier.pickle', 'rb') as pickle:
            self.classifier = load(pickle)
        self.lemmatizer = WordNetLemmatizer()

    def get_device_types(self):
        return ['mic']

    def get_channel_action_mapping(self):
        return {self.get_full_channel('text_transcript'): self.execute}

    def execute(self, message):
        sentence = message['data'].decode()
        tokens = self.remove_noise(word_tokenize(sentence))
        sentiment = self.classifier.classify(dict([token, True] for token in tokens))
        print(sentiment)
        self.publish('text_sentiment', sentiment)
...

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

...
# ------------------------------------------------------------
# Sentiment Analysis
# ------------------------------------------------------------
sentiment_analysis:
  image: python3
  build:
    context: .
    dockerfile: Dockerfile.python3
  hostname: sentiment_analysis
  user: "${NEW_UID}:${NEW_GID}"
  env_file:
    - ./.env

  working_dir: /my_service
  command: python3 sentiment_factory.py
  volumes:
    - ./cbsr/mock:/sentiment: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 the service depends on
...

RUN pip3 install --no-cache-dir --upgrade --prefer-binary \
	redis~=4.1 \
	hiredis~=2.0 \
	simplejson~=3.17 \
	Pillow~=9.0 \
	numpy~=1.22 \
	imutils~=0.5 \
	[any other dependendencies] \
	...

...
###########################
# Management              #
###########################

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

...
elif channel = 'text_sentiment':
    self.on_text_sentiment(message=data.decode('utf-8'))
...

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

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

from enum import Enum
from functools import partial
from social_interaction_cloud.action import ActionRunner
from social_interaction_cloud.basic_connector import BasicSICConnector

class SentimentConnector(BasicSICConnector):
    def __init__(self, server_ip: str, dialogflow_key_file: str, dialogflow_agent_id: str):
        super(SentimentConnector, self).__init__(server_ip, 'en-US', dialogflow_key_file, dialogflow_agent_id)
        self.enable_service('sentiment_analysis')
        self.sentiment = None
    
    def on_text_sentiment(self, sentiment: str) -> None:
        print(sentiment)
        self.sentiment = sentiment

class Example:
    def __init__(self, server_ip: str, dialogflow_key_file: str, dialogflow_agent_id: str):
        self.sic = SentimentConnector(server_ip, dialogflow_key_file, dialogflow_agent_id)
        self.action_runner = ActionRunner(self.sic)

        self.recognition_manager = RecognitionManager(2)
        self.user_model = {}
...