Adding a New Service

from os import getcwd

from sic.factory import SICfactory
from nltk import download
from nltk.data import path

from sentiment_service import SentimentAnalysisService


class SentimentAnalysisFactory(SICfactory):
    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 sic.service import SICservice
from nltk.stem.wordnet import WordNetLemmatizer
from nltk.tag import pos_tag
from nltk.tokenize import word_tokenize


class SentimentAnalysisService(SICservice):
    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 sentiment_analysis...
cd ../audio_sentiment
cp -f {cert.pem,classifier.pickle,*.py} ../../docker/sic/sentiment    # extra files should also be inclduded here
...

...
  # ------------------------------------------------------------
  # Sentiment Analysis service
  # ------------------------------------------------------------
  sentiment_analysis:
    image: sic_python3
    build:
      context: .
      dockerfile: Dockerfile.python3
    user: "${UID}:${GID}"
    env_file:
      - ./.env

    working_dir: /sentiment
    command: python3 sentiment_factory.py
    
    volumes:
      - ./sic/sentiment:/sentiment:rw,delegated

    tty: true
    stdin_open: false

    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 = {}
...