Update to get everything working

agent/agent.py

@@ -6,6 +6,8 @@
 
 import logging
 
+from sn_agent.network import NetworkSettings
+
 logger = logging.getLogger(__name__)
 
 app = create_app()
@@ -16,7 +18,7 @@
 # TODO Make the port configurable from the ENV
 # web.run_app(app, port=8000, ssl_context=sslcontext)
 
-settings = AgentSettings()
+settings = NetworkSettings()
 
 logger.info('Host setting: %s', settings.WEB_HOST)
 

agent/alice_config.yml

@@ -0,0 +1,12 @@
+---
+
+modules:
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000001
+  name: demo.document_summarizer.DocumentSummarizer
+  required_ontology_node_ids:
+    - deadbeef-aaaa-bbbb-cccc-000000000002
+    - deadbeef-aaaa-bbbb-cccc-000000000003
+    - deadbeef-aaaa-bbbb-cccc-000000000004
+    - deadbeef-aaaa-bbbb-cccc-000000000005
+    - deadbeef-aaaa-bbbb-cccc-000000000006

agent/bob_config.yml

@@ -0,0 +1,23 @@
+---
+
+modules:
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000002
+  name: demo.word_sense_disambiguater.WordSenseDisambiguater
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000003
+  name: demo.face_recognizer.FaceRecognizer
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000004
+  name: demo.text_summarizer.TextSummarizer
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000005
+  name: demo.video_summarizer.VideoSummarizer
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000006
+  name: demo.entity_extracter.EntityExtracter

agent/demo/document_summarizer/__init__.py

@@ -7,7 +7,7 @@
 #
 from demo.document_summarizer.settings import DocumentSummarizerSettings
 from sn_agent.job.job_descriptor import JobDescriptor
-from sn_agent.service_adapter.base import ModuleServiceAdapterABC
+from sn_agent.service_adapter.base import ModuleServiceAdapterABC, ServiceAdapterABC
 from sn_agent.service_adapter.manager import ServiceManager
 from sn_agent.ontology.service_descriptor import ServiceDescriptor
 from sn_agent import ontology
@@ -27,7 +27,14 @@ def __init__(self, app, service_ontology_node, required_service_nodes, name: str
         self.app = app
         self.settings = DocumentSummarizerSettings()
 
+        self.word_sense_disambiguater = None
+        self.face_recognizer = None
+        self.text_summarizer = None
+        self.video_summarizer = None
+        self.entity_extracter = None
+
     def post_load_initialize(self, service_manager: ServiceManager):
+        super().post_load_initialize(service_manager)
         self.word_sense_disambiguater = service_manager.get_service_adapter_for_id(ontology.WORD_SENSE_DISAMBIGUATER_ID)
         self.face_recognizer = service_manager.get_service_adapter_for_id(ontology.FACE_RECOGNIZER_ID)
         self.text_summarizer = service_manager.get_service_adapter_for_id(ontology.TEXT_SUMMARIZER_ID)
@@ -43,8 +50,8 @@ def transform_output_url(self, tag: str, item_count: int, output_url: str):
         sub_adapter_url = os.path.join(self.settings.TEST_OUTPUT_DIRECTORY, sub_adapter_output)
         return sub_adapter_url
 
-    def sub_adapter_job(self, tag: str, sub_adapter: ModuleServiceAdapterABC, job: JobDescriptor):
-        new_service_descriptor = ServiceDescriptor(sub_adapter.service)
+    def sub_adapter_job(self, tag: str, sub_adapter: ServiceAdapterABC, job: JobDescriptor):
+        new_service_descriptor = ServiceDescriptor(sub_adapter.service.node_id)
         new_job = JobDescriptor(new_service_descriptor)
         item_count = 0
         for job_item in job:
@@ -91,6 +98,41 @@ def perform(self, job: JobDescriptor):
         video_job = self.sub_adapter_job('video', self.video_summarizer, job)
         entity_job = self.sub_adapter_job('entity', self.entity_extracter, job)
 
+        self.word_sense_disambiguater.perform(word_job)
+        self.face_recognizer.perform(face_job)
+        self.text_summarizer.perform(text_job)
+        self.video_summarizer.perform(video_job)
+        self.entity_extracter.perform(entity_job)
+
+        # Now copy the outputs of each of the sub-jobs...
+        item_count = 0
+        for job_item in job:
+            output_file_name = self.transform_output_url('document', item_count, job_item['output_url'])
+
+            with open(output_file_name, "w") as output_file:
+                self.copy_adapter_output(output_file, word_job[item_count]['output_url'], 'word')
+                self.copy_adapter_output(output_file, face_job[item_count]['output_url'], 'face')
+                self.copy_adapter_output(output_file, text_job[item_count]['output_url'], 'text')
+                self.copy_adapter_output(output_file, video_job[item_count]['output_url'], 'video')
+                self.copy_adapter_output(output_file, entity_job[item_count]['output_url'], 'entity')
+
+            item_count += 1
+
+    def perform_async(self, job: JobDescriptor):
+        log.debug("      summarizing document")
+
+        # Make sure we have a directory.
+        directory = self.settings.TEST_OUTPUT_DIRECTORY
+        if not os.path.exists(directory):
+            os.mkdir(directory)
+
+        # Create new job descriptors for the sub-services...
+        word_job = self.sub_adapter_job('word', self.word_sense_disambiguater, job)
+        face_job = self.sub_adapter_job('face', self.face_recognizer, job)
+        text_job = self.sub_adapter_job('text', self.text_summarizer, job)
+        video_job = self.sub_adapter_job('video', self.video_summarizer, job)
+        entity_job = self.sub_adapter_job('entity', self.entity_extracter, job)
+
         async def disambiguate_words():
             self.word_sense_disambiguater.perform(word_job)
 
@@ -116,7 +158,7 @@ async def extract_entities():
             asyncio.ensure_future(extract_entities(), loop=loop)]
 
         # Wait until the sub-service tasks all complete.
-        loop.run_until_complete(asyncio.gather(*sub_services))
+        results = asyncio.gather(*sub_services)
 
         # Now copy the outputs of each of the sub-jobs...
         item_count = 0

agent/demo/text_summarizer/__init__.py

@@ -6,15 +6,14 @@
 # Distributed under the MIT software license, see LICENSE file.
 #
 
-from typing import List
+import logging
 
 from sn_agent.job.job_descriptor import JobDescriptor
 from sn_agent.service_adapter.base import ModuleServiceAdapterABC
 
-import logging
-
 log = logging.getLogger(__name__)
 
+
 class TextSummarizer(ModuleServiceAdapterABC):
     type_name = "TextSummarizer"
 

agent/sn_agent/agent/base.py

@@ -7,14 +7,10 @@
 #
 # Distributed under the MIT software license, see LICENSE file.
 #
-
-from sn_agent.network.enum import NetworkStatus
-from sn_agent.ontology.service_descriptor import ServiceDescriptor
-from enum import Enum
-
 import logging
 from abc import abstractmethod, ABC
 
+from sn_agent.job.job_descriptor import JobDescriptor
 from sn_agent.ontology.service_descriptor import ServiceDescriptor
 
 logger = logging.getLogger(__name__)
@@ -34,18 +30,8 @@ def can_perform(self, service: ServiceDescriptor) -> bool:
         pass
 
     @abstractmethod
-    def perform(self, agent_id, service: ServiceDescriptor):
-        """
-        :param service: the service to perform
-        """
-        pass
-
-    @abstractmethod
-    def list_providers(self, service: ServiceDescriptor) -> list:
+    def perform(self, job: JobDescriptor):
         """
-        This is used for creating the tree of subprovider services behind a given service
-
-        :param service: the service for which to list sub-providers.
-        :return:
+        :param job: the service to perform
         """
         pass

agent/sn_agent/agent/settings.py

@@ -8,6 +8,4 @@ def __init__(self, **custom_settings):
         self._ENV_PREFIX = 'SN_AGENT_'
         self.CLASS = 'sn_agent.agent.test.TestAgent'
         self.ID = Required(uuid.UUID)
-        self.WEB_HOST = "0.0.0.0"
-        self.WEB_PORT = 8000
         super().__init__(**custom_settings)

agent/sn_agent/agent/test.py

@@ -31,6 +31,3 @@ def can_perform(self, service: ServiceDescriptor) -> bool:
 
     def perform(self, service: ServiceDescriptor) -> bool:
         pass
-
-    def list_providers(self, service: ServiceDescriptor) -> list:
-         pass

agent/sn_agent/api/__init__.py

@@ -1,5 +1,6 @@
 import logging
 import os
+
 from aiohttp import web, WSMsgType
 from aiohttp.web_response import Response
 from jsonrpcserver.aio import methods
@@ -13,24 +14,32 @@
 
 WS_FILE = os.path.join(os.path.dirname(__file__), 'websocket.html')
 
+
 @methods.add
-async def can_perform(meaning_of_life=None, context=None):
+async def can_perform(service_node_id=None, context=None):
     # figure out what we are being asked to perform and answer
-    service = ServiceDescriptor(ontology.DOCUMENT_SUMMARIZER_ID)
+    service = ServiceDescriptor(service_node_id)
     app = context
     return await can_perform_service(app, service)
 
 
 @methods.add
-async def perform(request, context):
-    job = JobDescriptor()
+async def perform(service_node_id=None, job_params=None, context=None):
+    service_descriptor = ServiceDescriptor(service_node_id)
+
+    job = JobDescriptor(service_descriptor, job_params)
     app = context
-    return await perform_job(app, job)
+
+    result = await perform_job(app, job)
+    logging.debug('Result of perform was %s', result)
+    return result
 
 
 async def http_handler(request):
-    request = await request.text()
-    response = await methods.dispatch(request)
+    app = request.app
+    request_text = await request.text()
+
+    response = await methods.dispatch(request_text, app)
     if response.is_notification:
         return web.Response()
     else:

agent/sn_agent/api/job.py

@@ -31,3 +31,4 @@ async def perform_job(app, job_descriptor: JobDescriptor):
         raise Exception('Service not available')
 
     return service_adapter.perform(job_descriptor)
+

agent/sn_agent/job/job_descriptor.py

@@ -13,11 +13,13 @@
 
 
 class JobDescriptor(object):
-    def __init__(self, service: ServiceDescriptor, job_parameters: dict = None):
+    def __init__(self, service: ServiceDescriptor, job_parameters: list = None):
         self.service = service
-        self.job_parameters = []
-        if not job_parameters is None:
-            self.job_parameters.append(job_parameters)
+
+        if job_parameters is None:
+            job_parameters = []
+
+        self.job_parameters = job_parameters
 
     def __eq__(self, other):
         return self.__dict__ == other.__dict__
@@ -60,15 +62,25 @@ def init_test_jobs():
     test_jobs[ontology.VIDEO_SUMMARIZER_ID] = []
     test_jobs[ontology.ENTITY_EXTRACTER_ID] = []
 
-    job_parameters = {'input_type': 'file',
+    job_parameters = [
+        {
+            'input_type': 'file',
                       'input_url': 'http://test.com/inputs/test_input.txt',
                       'output_type': 'file_url_put',
-                      'output_url': 'test_output.txt'}
-    job_parameters_2 = {'input_type': 'file',
+                      'output_url': 'test_output.txt'
+        }
+                      ]
+
+    job_parameters_2 = [
+        {
+            'input_type': 'file',
                         'input_url': 'http://test.com/inputs/test_input_2.txt',
                         'output_type': 'file_url_put',
-                        'output_url': 'test_output_2.txt'}
+                        'output_url': 'test_output_2.txt'
+        }
+    ]
 
+    #TODO: These need to be fixed so that we can test for single and multiple jobs
     service_id = ontology.DOCUMENT_SUMMARIZER_ID
     job = JobDescriptor(ServiceDescriptor(service_id), job_parameters)
     test_jobs[service_id].append(job)

agent/sn_agent/network/dht.py

@@ -44,8 +44,8 @@ def get(self, key):
 
         # Deserialize from BSON - another serializer could be used but BSON works well here
         for d in data_bin:
-            data = bson.loads(d.data)
-            data.append(data)
+            loaded_data = bson.loads(d.data)
+            data.append(loaded_data)
 
         logger.debug('Key Value info: %s', data)
         return data

agent/sn_agent/network/settings.py

@@ -15,8 +15,10 @@ def __init__(self, **custom_settings):
 
         self.BOOT_HOST = 'bootstrap.ring.cx'
         self.BOOT_PORT = "4222"
+        self.WEB_HOST = "0.0.0.0"
+        self.WEB_PORT = 8000
 
         super().__init__(**custom_settings)
 
         # Must place after the init so as to pick up the proper gateway value
-        self.WEB_URL = Url(scheme='http', host=self.GATEWAY, port=8000, path='/api').url
+        self.WEB_URL = Url(scheme='http', host=self.GATEWAY, port=self.WEB_PORT, path='/api').url