Page Comparison

from os import getcwd

from cbsr.factory import CBSRfactory
from my_servicenltk import MyServicedownload
from nltk.data class MyServiceFactory(CBSRfactory):import path

   from sentiment_service import SentimentAnalysisService


class SentimentAnalysisFactory(CBSRfactory):
    def __init__(self):
        """ MyServiceFactory constructorsuper(SentimentAnalysisFactory, self).__init__()

      def get_connection_channel(self):
     inherit from the CBSRfactory constructor to create the new service as a service
  return 'sentiment_analysis'

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


super(MyServiceFactory, self).if __initname__() 
    def get_connection_channel(self)== '__main__':
    cwd = getcwd()
 """ initialise the name of the service Redis channel
  download('punkt', download_dir=cwd)
    download('omw-1.4', download_dir=cwd)
      """download('averaged_perceptron_tagger', download_dir=cwd)
        return 'my_service'
download('wordnet', download_dir=cwd)
    def create_service(self, connect, identifier, disconnect):path.append(cwd)

       return MyService(connect, identifier, disconnect)


if __name__ == '__main__':
    my_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 andto publishesget it intothe the newtype my_service channel.
of sentiment
"""

from pickle Theimport fileload
canfrom bere extendedimport tosub
asfrom manystring servicesimport aspunctuation
necessary
"""

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


class MyServiceSentimentAnalysisService(CBSRservice):
    def __init__(self, connect, identifier, disconnect):
        super(MyServiceSentimentAnalysisService, self).__init__(connect, identifier, disconnect)
     def get_device_types(self)   with open('classifier.pickle', 'rb') as pickle:
        """ :return list of devicesself.classifier the service uses= load(pickle)
        self.lemmatizer = WordNetLemmatizer()

 This method returns a list of all the 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;{self.get_full_channel('text_transcript'): self.execute}

    def execute(self, message):
        sentence = message['data'].decode()
        tokens    multiple channels can be listened to, linked to a method when separated by a comma in the output dictionary= self.remove_noise(word_tokenize(sentence))
        sentiment = self.classifier.classify(dict([token, True] for token in tokens))
        print(sentiment)
     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('mytext_servicesentiment', sentencesentiment)
...

...
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
...

...
# ------------------------------------------------------------
# MySentiment serviceAnalysis
# ------------------------------------------------------------
mysentiment_serviceanalysis:
  image: python3
  build:
    context: .
    dockerfile: Dockerfile.python3
  hostname: mysentiment_serviceanalysis
  user: "${NEW_UID}:${NEW_GID}"
  env_file:
    - ./.env

  working_dir: /my_service
  command: python3 my_servicesentiment_factory.py
  volumes:
    - ./cbsr/mock:/my_servicesentiment: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

existing services' 

  depends_on:
    - redis
    - dialogflow
  # - any other services the service depends 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 my_service channel;
        This method can be further inherited and overridden
    """
    self.__notify_listeners('onMyService', message)
...