Merge remote-tracking branch 'upstream/master'

agent/demo/document_summarizer/__init__.py

@@ -12,26 +12,20 @@
 from sn_agent.ontology.service_descriptor import ServiceDescriptor
 from sn_agent import ontology
 import os
+import asyncio
 
 import logging
 
 log = logging.getLogger(__name__)
 
 
-
-
 class DocumentSummarizer(ModuleServiceAdapterABC):
     type_name = "DocumentSummarizer"
 
     def __init__(self, app, service_ontology_node, required_service_nodes, name: str):
         super().__init__(app, service_ontology_node, required_service_nodes, name)
         self.app = app
-
         self.settings = DocumentSummarizerSettings()
-        directory = self.settings.TEST_OUTPUT_DIRECTORY
-
-        if not os.path.exists(directory):
-            os.makedirs(directory)
 
     def post_load_initialize(self, service_manager: ServiceManager):
         self.word_sense_disambiguater = service_manager.get_service_adapter_for_id(ontology.WORD_SENSE_DISAMBIGUATER_ID)
@@ -40,27 +34,31 @@ def post_load_initialize(self, service_manager: ServiceManager):
         self.video_summarizer = service_manager.get_service_adapter_for_id(ontology.VIDEO_SUMMARIZER_ID)
         self.entity_extracter = service_manager.get_service_adapter_for_id(ontology.ENTITY_EXTRACTER_ID)
 
-    def transform_output_url(self, sub_adapter: str, item_count: int, output_url: str):
+    def transform_output_url(self, tag: str, item_count: int, output_url: str):
         last_part = output_url.split("/")[-1]
         if last_part == "":
-            output_url = self.settings.TEST_OUTPUT_DIRECTORY + sub_adapter + ".out"
+            sub_adapter_output = tag + ".out"
         else:
-            output_url = self.settings.TEST_OUTPUT_DIRECTORY + sub_adapter + "_" + last_part
-        return output_url
+            sub_adapter_output = tag + "_" + last_part
+        sub_adapter_url = os.path.join(self.settings.TEST_OUTPUT_DIRECTORY, sub_adapter_output)
+        return sub_adapter_url
 
-    def sub_adapter_job(self, sub_adapter: ModuleServiceAdapterABC,  tag: str, job: JobDescriptor):
+    def sub_adapter_job(self, tag: str, sub_adapter: ModuleServiceAdapterABC, job: JobDescriptor):
         new_service_descriptor = ServiceDescriptor(sub_adapter.service)
         new_job = JobDescriptor(new_service_descriptor)
         item_count = 0
         for job_item in job:
+
+            # Just pass the inputs on directly to the subtasks.
             new_job_item = {}
             new_job_item['input_type'] = job_item['input_type']
+            new_job_item['input_url'] = job_item['input_url']
+
             output_type = job_item['output_type']
             new_job_item['output_type'] = output_type
 
-            new_job_item['input_url'] = job_item['input_url']
-
-            # Transform the output so we can get separate outputs for each sub-adapter.
+            # Transform the output URL so we can get separate output files for each sub-adapter
+            # That way we can assemble the various parts at the end using these separate URLs.
             if output_type == 'file_url_put':
                 output_url = job_item['output_url']
                 sub_adapter_output_url = self.transform_output_url(tag, item_count, output_url)
@@ -86,21 +84,39 @@ def perform(self, job: JobDescriptor):
         if not os.path.exists(directory):
             os.mkdir(directory)
 
-        # Perform the sub-jobs...
-        word_job = self.sub_adapter_job(self.word_sense_disambiguater, 'word', job)
-        self.word_sense_disambiguater.perform(word_job)
+        # 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)
+
+        async def recognize_faces():
+            self.face_recognizer.perform(face_job)
+
+        async def summarize_text():
+            self.text_summarizer.perform(text_job)
 
-        face_job = self.sub_adapter_job(self.face_recognizer, 'face', job)
-        self.face_recognizer.perform(face_job)
+        async def summarize_video():
+            self.video_summarizer.perform(video_job)
 
-        text_job = self.sub_adapter_job(self.text_summarizer, 'text', job)
-        self.text_summarizer.perform(text_job)
+        async def extract_entities():
+            self.entity_extracter.perform(entity_job)
 
-        video_job = self.sub_adapter_job(self.video_summarizer, 'video', job)
-        self.video_summarizer.perform(video_job)
+        # Gather all the subservice tasks to process them asynchronously.
+        loop = self.app.loop
+        sub_services = [
+            asyncio.ensure_future(disambiguate_words(), loop=loop),
+            asyncio.ensure_future(recognize_faces(), loop=loop),
+            asyncio.ensure_future(summarize_text(), loop=loop),
+            asyncio.ensure_future(summarize_video(), loop=loop),
+            asyncio.ensure_future(extract_entities(), loop=loop)]
 
-        entity_job = self.sub_adapter_job(self.entity_extracter, 'entity', job)
-        self.entity_extracter.perform(entity_job)
+        # Wait until the sub-service tasks all complete.
+        loop.run_until_complete(asyncio.gather(*sub_services))
 
         # Now copy the outputs of each of the sub-jobs...
         item_count = 0

agent/demo/document_summarizer/settings.py

@@ -3,12 +3,12 @@
 
 from sn_agent import SettingsBase
 
-THIS_DIR = Path(__file__).parent
+DEMO_PARENT_DIR = Path(__file__).parent.parent.parent
 
 
 class DocumentSummarizerSettings(SettingsBase):
     def __init__(self, **custom_settings):
         self._ENV_PREFIX = 'SN_DS_'
-        self.TEST_OUTPUT_DIRECTORY = os.path.join(THIS_DIR, "tests", "output")
+        self.TEST_OUTPUT_DIRECTORY = os.path.join(DEMO_PARENT_DIR, "tests", "output")
 
         super().__init__(**custom_settings)

agent/demo/entity_extracter/__init__.py

@@ -25,11 +25,9 @@ def perform(self, job: JobDescriptor):
         item_count = 0
         for job_item in job:
             file_name = job[item_count]['output_url']
-            file = open(file_name, 'w')
-            file.write("entity:\n")
-            file.write("    pig\n")
-            file.write("    farmer\n")
-            file.write("    tractor\n")
-            file.write("    cornfield\n")
-            file.close()
-
+            with open(file_name, 'w') as file:
+                file.write("entity:\n")
+                file.write("    pig\n")
+                file.write("    farmer\n")
+                file.write("    tractor\n")
+                file.write("    cornfield\n")

agent/demo/face_recognizer/__init__.py

@@ -25,9 +25,8 @@ def perform(self, job: JobDescriptor):
         item_count = 0
         for job_item in job:
             file_name = job[item_count]['output_url']
-            file = open(file_name, 'w')
-            file.write("face:\n")
-            file.write("    Mary Jones\n")
-            file.write("    Henry Jones\n")
-            file.close()
+            with open(file_name, 'w') as file:
+                file.write("face:\n")
+                file.write("    Mary Jones\n")
+                file.write("    Henry Jones\n")
 

agent/demo/text_summarizer/__init__.py

@@ -25,8 +25,6 @@ def perform(self, job: JobDescriptor):
         item_count = 0
         for job_item in job:
             file_name = job[item_count]['output_url']
-            file = open(file_name, 'w')
-            file.write("text:\n")
-            file.write("    Farmer Jones and her husband, Henry, are standing in a field\n")
-            file.close()
-
+            with open(file_name, 'w') as file:
+                file.write("text:\n")
+                file.write("    Farmer Jones and her husband, Henry, are standing in a field\n")

agent/demo/video_summarizer/__init__.py

@@ -25,7 +25,6 @@ def perform(self, job: JobDescriptor):
         item_count = 0
         for job_item in job:
             file_name = job[item_count]['output_url']
-            file = open(file_name, 'w')
-            file.write("video:\n")
-            file.write("    A woman farmer is plowing a cornfield, a man watches.\n")
-            file.close()
+            with open(file_name, 'w') as file:
+                file.write("video:\n")
+                file.write("    A woman farmer is plowing a cornfield, a man watches.\n")

agent/demo/word_sense_disambiguater/__init__.py

@@ -25,7 +25,6 @@ def perform(self, job: JobDescriptor):
         item_count = 0
         for job_item in job:
             file_name = job[item_count]['output_url']
-            file = open(file_name, 'w')
-            file.write("word:\n")
-            file.write("    farm - to cultivate\n")
-            file.close()
+            with open(file_name, 'w') as file:
+                file.write("word:\n")
+                file.write("    farm - to cultivate\n")

agent/sn_agent/job/job_descriptor.py

@@ -22,16 +22,14 @@ def __eq__(self, other):
         return self.__dict__ == other.__dict__
 
     def __str__(self):
-        if not self.service:
+        if self.service is None:
             description = ""
         else:
             description = self.service.name()
         return '<Job: service %s>' % (description)
 
     def __iter__(self):
         return self.job_parameters.__iter__()
-    def __next__(self):
-        return self.job_parameters.__next__()
 
     def __delitem__(self, key):
         self.job_parameters.__delitem__(key)
@@ -40,7 +38,8 @@ def __getitem__(self, key):
     def __setitem__(self, key, value):
         self.job_parameters.__setitem__(key, value)
 
-
+    def __len__(self):
+        return len(self.job_parameters)
 
     def append_job_item(self, job_item: dict):
         self.job_parameters.append(job_item)
@@ -61,11 +60,11 @@ def init_test_jobs():
     job_parameters = {'input_type': 'file',
                         'input_url': 'http://test.com/inputs/test_input.txt',
                         'output_type': 'file_url_put',
-                        'output_url': 'tests/output/test_output.txt'}
+                        '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': 'tests/output/test_output_2.txt'}
+                        'output_url': 'test_output_2.txt'}
 
     service_id = ontology.DOCUMENT_SUMMARIZER_ID
     job = JobDescriptor(ServiceDescriptor(service_id), job_parameters)

agent/sn_agent/service_adapter/__init__.py

@@ -6,11 +6,17 @@
 from sn_agent.service_adapter.settings import ServiceAdapterSettings
 from sn_agent.service_adapter.manager import ServiceManager
 
+import logging
+
+log = logging.getLogger(__name__)
+
 def setup_service_manager(app):
     settings = ServiceAdapterSettings()
     config_file = settings.CONFIG_FILE
     ontology = app['ontology']
 
+    log.debug("reading configuration file {0}".format(config_file))
+
     with open(config_file, 'r') as ymlfile:
         cfg = yaml.load(ymlfile)
 

agent/sn_agent/test/mocks.py

@@ -1,12 +1,11 @@
 import logging
+import asyncio
 
 
 class MockApp(dict):
 
     def __init__(self):
-        self.loop = self.wait_loop
+        self.loop = asyncio.get_event_loop()
         pass
 
-    def wait_loop(self):
-        pass
 

agent/tests/service_adapter_test.yml

@@ -0,0 +1,17 @@
+---
+
+opencogs:
+- opencog:
+  required_ontology_node_ids:
+    - 86717381-0b4c-440f-9ed6-41206fbba0bc
+    - b95fb615-cb03-4d19-bf71-2a892c3d45a8
+  host: 127.0.0.1
+  port: 15010
+
+- opencog:
+  ontology_node_id: 413cfd03-14a2-45dd-a94c-3c7798aca992
+  required_ontology_node_ids:
+    - 86717381-0b4c-440f-9ed6-41206fbba0bc
+    - b95fb615-cb03-4d19-bf71-2a892c3d45a8
+  host: 127.0.0.1
+  port: 15010

agent/tests/service_adapter_test_2.yml

@@ -0,0 +1,12 @@
+---
+
+jsonrpcs:
+- jsonrpc:
+  required_ontology_node_ids:
+    - b95fb615-cb03-4d19-bf71-2a892c3d45a8
+  url: http://localhost:8091
+
+- jsonrpc:
+  comment: A worker that needs no supporting workers
+  ontology_node_id: b95fb615-cb03-4d19-bf71-2a892c3d45a8
+  url: http://localhost:8092

agent/tests/service_adapter_test_3.yml

@@ -0,0 +1,21 @@
+---
+
+modules:
+- module:
+  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
+
+- module:
+  ontology_node_id: deadbeef-aaaa-bbbb-cccc-000000000002
+  name: demo.word_sense_disambiguater.WordSenseDisambiguater
+
+
+bogusii:
+- bogus:
+  name: demo.document_summarizer.DocumentSummarizer
+  required_ontology_node_ids:

agent/tests/service_adapter_test_4.yml

@@ -0,0 +1,6 @@
+---
+
+bogusii:
+- bogus:
+  name: demo.document_summarizer.DocumentSummarizer
+  required_ontology_node_ids: