Page Comparison

from os import getcwd

from cbsrsic.factory import CBSRfactorySICfactory
from nltk import download
from nltk.data import path

from mysentiment_service import MyServiceSentimentAnalysisService


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

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

    def create_service(self, connect, identifier, disconnect):
        return MyServiceSentimentAnalysisService(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)
     mydownload('wordnet', download_dir=cwd)
    path.append(cwd)

    sentiment_analysis_factory = MyServiceFactorySentimentAnalysisFactory()
    mysentiment_analysis_factory.run()

""" This file shows an example of a mocksentiment 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 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 MyServiceSentimentAnalysisService(CBSRserviceSICservice):
    def __init__(self, connect, identifier, disconnect):
        super(MyServiceSentimentAnalysisService, self).__init__(connect, identifier, disconnect)
     def get_device_types(self):  with open('classifier.pickle', 'rb')     """ :return list of devices the service usesas pickle:
            self.classifier = load(pickle)
   This method returns a list ofself.lemmatizer all= theWordNetLemmatizer()
devices
my_service uses; in this case, only the microphone
    def get_device_types(self):
   """         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 publishessentence = message['data'].decode()
        tokens = self.remove_noise(word_tokenize(sentence))
        sentiment = self.classifier.classify(dict([token, True] afor messagetoken in thetokens))
new service channel             Extra functionalities can be added to the method
        """
        sentence = message['data'].decode(print(sentiment)
        self.publish('mytext_servicesentiment', sentencesentiment)
...

...
echo Deploying mysentiment_serviceanalysis...
cd ../myaudio_servicesentiment
cp -f {cert.pem,  classifier.pickle,*.py} ../../docker/cbsrsic/my_servicesentiment    # extra files should canalso be copiedinclduded if usedhere
...

...
  # ------------------------------------------------------------
  # MySentiment Analysis service
  # ------------------------------------------------------------
my  sentiment_serviceanalysis:
    image: sic_python3
    build:
      context: .
      dockerfile: Dockerfile.python3

 hostname: my_service   user: "${NEW_UID}:${NEW_GID}"
    env_file:
      - ./.env

    working_dir: /my_servicesentiment
    command: python3 my_servicesentiment_factory.py
    
    volumes:
      - ./cbsrsic/mocksentiment:/my_service:rw${MOUNT_OPTIONS}sentiment:rw,delegated

    tty: true
    stdin_open: false

  networks:     app_netdepends_on:
      ipv4_address: 172.16.238.x- redis
  # address has to differ- fromdialogflow
those
of the already# existing- services'any other services the service depends_on:
    - redis
    - dialogflow
  # - any other services my_service depends on

 on
...

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

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

...
elif channel = 'mytext_servicesentiment':
    self.on_mytext_servicesentiment(message=data.decode('utf-8'))
...

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

...
###########################
# Event handlers          #
###########################
...
def on_mytext_servicesentiment(self, message: str) -> None:
    """ :param message: the message published on the mytext_servicesentiment channel
        This method notifies the listeners that a new message has been posted on the mytext_servicesentiment channel;
        This method can be further inherited and overridden
    """
    self.__notify_listeners('onMyServiceonTextSentiment', 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 MyConnectorSentimentConnector(BasicSICConnector):
    def __init__(self, server_ip: str, dialogflow_key_file: str, dialogflow_agent_id: str):
        """ :param super(SentimentConnector, self).__init__(server_ip:
  , 'en-US', dialogflow_key_file, dialogflow_agent_id)
         This method inherits the BasicSICConnector and enables the new my_service service;self.enable_service('sentiment_analysis')
        self.sentiment = None
    
       my_service needs to be manually enabled in this waydef on_text_sentiment(self, sentiment: str) -> None:
         print(sentiment)
 """         super(MyConnector, self).__init__(server_ip)
        self.enable_service('my_service').sentiment = sentiment

class Example:
     def on__init_my_service(self, messageserver_ip: str) -> None, dialogflow_key_file:         """ :param message: message published in the my_service channel
       str, dialogflow_agent_id: str):
        self.sic = SentimentConnector(server_ip, dialogflow_key_file, dialogflow_agent_id)
    This method overrides the event function on_my_service inherited from the BasicSICConnector self.action_runner = ActionRunner(self.sic)

        """
     self.recognition_manager = RecognitionManager(2)
  print(message)      self.user_model = {}
self...stop()