Merge pull request #130 from oda-hub/fix_type_key

Refactor class hierarchy of NB2WProduct

Author: burnout87

dispatcher_plugin_nb2workflow/products.py

@@ -5,8 +5,10 @@
 import json
 
 from cdci_data_analysis.analysis.products import LightCurveProduct, BaseQueryProduct, ImageProduct, SpectrumProduct
-from cdci_data_analysis.analysis.parameters import Parameter
+from cdci_data_analysis.analysis.parameters import Parameter, subclasses_recursive
+from cdci_data_analysis.analysis.exceptions import ProductProcessingError
 from oda_api.data_products import NumpyDataProduct, ODAAstropyTable, BinaryProduct, PictureProduct
+
 from .util import AstropyTableViewParser, with_hashable_dict
 from oda_api.ontology_helper import Ontology
 from io import StringIO
@@ -35,34 +37,18 @@ def write(self, file_name=None, overwrite=True, file_dir=None):
             file_path = self.file_path.path
 
         self.table_data.write(file_path, overwrite=overwrite, format='ascii.ecsv')
-        
-class NB2WProduct:
-    def __init__(self, 
-                 encoded_data, 
-                 data_product_type=BaseQueryProduct,
-                 out_dir='./', 
-                 name='nb2w', 
-                 extra_metadata={}):
 
-        # this constructor is only valid for NumpyDataProduct-based products
-        
-        self.name = name
-        self.extra_metadata = extra_metadata 
-        metadata = encoded_data.get('meta_data', {})
-        self.out_dir = out_dir
-        numpy_data_prod = NumpyDataProduct.decode(encoded_data) 
-        
-        if not numpy_data_prod.name:
-            numpy_data_prod.name = self.name
 
-        self.dispatcher_data_prod = data_product_type(
-            name=self.name, 
-            data=numpy_data_prod,
-            meta_data=metadata,
-            file_dir=out_dir,
-            file_name = f"{self.name}.fits")
-    
+class NB2WProduct:
 
+    def __init__(self, *args, **kwargs):
+        error_msg = "The output"
+        name = kwargs.get('name', None)
+        if name is not None:
+            error_msg += f" with name \"{name}\""
+        error_msg += " has been wrongly annotated."
+        raise ProductProcessingError(error_msg)
+
     def write(self):
         file_path = self.dispatcher_data_prod.file_path
         self.dispatcher_data_prod.write()
@@ -74,10 +60,10 @@ def get_html_draw(self):
     @classmethod 
     def _init_as_list(cls, encoded_data, *args, **kwargs):
         encoded_data = cls._dejsonify(encoded_data)
-        
+
         if isinstance(encoded_data, list):
             return [cls(elem, *args, **kwargs) for elem in encoded_data]
-        
+
         return [cls(encoded_data, *args, **kwargs)]
 
     @classmethod
@@ -96,7 +82,7 @@ def _prod_list_description_analyser(
             onto = None
             par_prod_class_dict = {}
 
-        mapping = {getattr(x, 'type_key'): x for x in cls.__subclasses__() if hasattr(x, 'type_key')}
+        mapping = {getattr(x, 'type_key'): x for x in subclasses_recursive(cls) if hasattr(x, 'type_key')}
         mapping.update(par_prod_class_dict)
 
         prod_classes_dict = {}
@@ -135,7 +121,7 @@ def _prod_list_description_analyser(
                     extra_kw.update({'extra_metadata': extra_metadata})
 
             prod_classes_dict[key] = (mapping.get(cls_owl_type, cls), name, extra_kw)
-        
+
         return prod_classes_dict
 
 
@@ -152,7 +138,8 @@ def prod_list_factory(cls, output_description_dict, output, out_dir = './', onto
                                                       name=val[1],
                                                       **val[2]))
             except Exception as e:
-                logger.error('unable to construct %s product: %s from this: %s ', key, e, output[key])
+                logger.error('unable to construct %s product: %s from %s', key, e, val[0])
+                raise
 
         return prod_list
 
@@ -165,8 +152,44 @@ def _dejsonify(encoded_data):
                 pass
         return encoded_data
 
+
+class _CommentProduct(NB2WProduct):
+    type_key = 'http://odahub.io/ontology#WorkflowResultComment'
+
+    def __init__(self, *args, **kwargs): ...
+
+    @classmethod
+    def _init_as_list(cls, encoded_data, *args, **kwargs):
+        return []
+
+class NB2WNumpyDataProduct(NB2WProduct):
+    type_key = 'http://odahub.io/ontology#NumpyDataProduct'
+
+    def __init__(self,
+                 encoded_data,
+                 data_product_type=BaseQueryProduct,
+                 out_dir='./',
+                 name='nb2w',
+                 extra_metadata={}):
+
+        self.name = name
+        self.extra_metadata = extra_metadata
+        metadata = encoded_data.get('meta_data', {})
+        self.out_dir = out_dir
+        numpy_data_prod = NumpyDataProduct.decode(encoded_data)
+        if not numpy_data_prod.name:
+            numpy_data_prod.name = self.name
+
+        self.dispatcher_data_prod = data_product_type(
+            name=self.name,
+            data=numpy_data_prod,
+            meta_data=metadata,
+            file_dir=out_dir,
+            file_name=f"{self.name}.fits")
+
+
 class NB2WParameterProduct(NB2WProduct):
-    type_key = 'oda:WorkflowParameter'
+    type_key = 'http://odahub.io/ontology#WorkflowParameter'
 
     ontology_path = None
 
@@ -248,7 +271,6 @@ def get_html_draw(self):
 
 
 class NB2WProgressProduct(NB2WProduct):
-
     def __init__(self, progress_html_data, out_dir=None, name='progress'):
         self.out_dir = out_dir
         self.name = name
@@ -315,7 +337,7 @@ def get_html_draw(self):
         return {'image': {'div': '<br><br>'+parser.tabcode,
                           'script': f"<script>{script_text}</script>"} }
 
-class NB2WLightCurveProduct(NB2WProduct): 
+class NB2WLightCurveProduct(NB2WNumpyDataProduct):
     type_key = 'http://odahub.io/ontology#LightCurve'
 
     def __init__(self, 
@@ -366,7 +388,7 @@ def get_html_draw(self, unit_id=None):
                                                          )
         return im_dic
 
-class NB2WSpectrumProduct(NB2WProduct):
+class NB2WSpectrumProduct(NB2WNumpyDataProduct):
     type_key = 'http://odahub.io/ontology#Spectrum'
 
     def __init__(self, 
@@ -379,15 +401,16 @@ def __init__(self,
                          out_dir=out_dir, name=name, 
                          extra_metadata=extra_metadata)
 
-class NB2WImageProduct(NB2WProduct):
+class NB2WImageProduct(NB2WNumpyDataProduct):
     type_key = 'http://odahub.io/ontology#Image'
 
     def __init__(self, 
                  encoded_data, 
                  out_dir='./', 
                  name='image', 
                  extra_metadata={}):
-        super().__init__(encoded_data, 
+
+        super().__init__(encoded_data,
                          data_product_type=ImageProduct, 
                          out_dir=out_dir, 
                          name=name, 

dispatcher_plugin_nb2workflow/queries.py

@@ -6,7 +6,9 @@
                        NB2WPictureProduct,
                        NB2WTextProduct,
                        NB2WParameterProduct,
-                       NB2WProgressProduct)
+                       NB2WProgressProduct,
+                       NB2WNumpyDataProduct,
+                       NB2WImageProduct)
 from oda_api.ontology_helper import Ontology
 import os
 from functools import lru_cache
@@ -186,7 +188,7 @@ def process_product_method(self, instrument, prod_list, api=False):
                 elif isinstance(product, NB2WProgressProduct):
                     progress_dp_list.append({'name': product.name,
                                              'value': product.progress_data})
-                else: # NB2WProduct contains NumpyDataProd by default
+                else:
                     np_dp_list.append(product.dispatcher_data_prod.data)
 
                 extra_meta[product.name] = getattr(product, 'extra_metadata', {})

test-requirements.txt

@@ -1,3 +1,3 @@
 git+https://github.com/oda-hub/nb2workflow.git@master#egg=nb2workflow[service,rdf]
 git+https://github.com/oda-hub/oda_api.git@master#egg=oda_api
-git+https://github.com/oda-hub/dispatcher-app.git@master#egg=cdci_data_analysis
+git+https://github.com/oda-hub/dispatcher-app.git@dedicated-exception-product-processing#egg=cdci_data_analysis

tests/conftest.py

@@ -119,6 +119,8 @@ def mock_backend(httpserver):
         bin_json = json.loads(fd.read())
     with open(os.path.join(responses_path, 'image.json'), 'r') as fd:
         image_json = json.loads(fd.read())
+    with open(os.path.join(responses_path, 'data_product.json'), 'r') as fd:
+        data_product_json = json.loads(fd.read())
     # with open(os.path.join(responses_path, 'test_output.html'), 'r') as fd:
     #     test_output_html = fd.read()
 
@@ -128,6 +130,8 @@ def mock_backend(httpserver):
     httpserver.expect_request(f'/api/v1.0/get/table').respond_with_json(table_json)
     httpserver.expect_request(f'/api/v1.0/get/ascii_binary').respond_with_json(bin_json)
     httpserver.expect_request(f'/api/v1.0/get/image').respond_with_json(image_json)
+    httpserver.expect_request(f'/api/v1.0/get/data_product').respond_with_json(data_product_json)
+    httpserver.expect_request(f'/api/v1.0/get/data_product_no_annotations').respond_with_json(data_product_json)
     # httpserver.expect_request(f'/trace/nb2w-ylp5ovnm/lightcurve').respond_with_data(test_output_html)
     httpserver.expect_request(f'/trace/nb2w-ylp5ovnm/lightcurve').respond_with_handler(trace_get_func_handler)
     httpserver.expect_request(f'/api/v1.0/get/dummy_echo').respond_with_handler(return_request_query_dict)

tests/responses/ascii_binary.json

@@ -0,0 +1,7 @@
+{
+    "exceptions": [],
+    "jobdir": "/tmp/nb2w-aqkvl461",
+    "output": {
+        "result": {}
+    }
+}

tests/responses/data_product.json

@@ -38,6 +38,34 @@
         },
         "parameters": {}
     },
+    "data_product": {
+        "output": {
+            "result": {
+                "comment": " http://odahub.io/ontology#DataProduct",
+                "name": "result",
+                "owl_type": "http://odahub.io/ontology#DataProduct",
+                "python_type": {
+                    "type_object": "<class 'str'>"
+                },
+                "value": ""
+            }
+        },
+        "parameters": {}
+    },
+    "data_product_no_annotations": {
+        "output": {
+            "result": {
+                "comment": "",
+                "name": "result",
+                "owl_type": "",
+                "python_type": {
+                    "type_object": "<class 'str'>"
+                },
+                "value": ""
+            }
+        },
+        "parameters": {}
+    },
     "image": {
         "output": {
             "result": {

tests/responses/options.json

@@ -151,6 +151,8 @@ def test_instrument_products(dispatcher_live_fixture, mock_backend):
         if isinstance(elem, dict) and 'prod_dict' in elem.keys():
             prod_dict = elem['prod_dict']
     assert prod_dict == {'ascii_binary': 'ascii_binary_query',
+                         'data_product': 'data_product_query',
+                         'data_product_no_annotations': 'data_product_no_annotations_query',
                          'dummy_echo': 'dummy_echo_query',
                          'image': 'image_query',
                          'file_download': 'file_download_query',
@@ -289,6 +291,28 @@ def test_image_product(dispatcher_live_fixture, mock_backend):
     imdata = jdata['products']['numpy_data_product_list'][0]
     oda_ndp = ImageDataProduct.decode(imdata)
 
+@pytest.mark.parametrize("product_type", ["data_product", "data_product_no_annotations"])
+def test_data_product_product(dispatcher_live_fixture, mock_backend, product_type):
+    server = dispatcher_live_fixture
+    logger.info("constructed server: %s", server)
+
+    c = requests.get(server + "/run_analysis",
+                    params = {'instrument': 'example0',
+                              'query_status': 'new',
+                              'query_type': 'Real',
+                              'product_type': product_type,
+                              'api': 'True',
+                              'run_asynch': 'False'})
+    logger.info("content: %s", c.text)
+    jdata = c.json()
+    logger.info(json.dumps(jdata, indent=4, sort_keys=True))
+    logger.info(jdata)
+    assert c.status_code == 200
+    assert jdata['query_status'] == 'failed'
+    assert jdata['job_status'] == 'failed'
+    assert jdata['exit_status']['error_message'] == 'The output with name "result" has been wrongly annotated.'
+    assert jdata['exit_status']['message'] == 'Error during the products post processing'
+
 def test_get_config_dict_from_kg():
     from dispatcher_plugin_nb2workflow.exposer import get_config_dict_from_kg