Merge pull request #340 from nanglo123/fix_serialization_for_datetime

Fix serialization for datetime

mira/metamodel/template_model.py

@@ -8,6 +8,7 @@
     "Time",
     "model_has_grounding",
     "Concept",
+    "Author"
 ]
 
 import datetime

mira/modeling/amr/stockflow.py

@@ -4,10 +4,14 @@
 
 __all__ = ["AMRStockFlowModel", "template_model_to_stockflow_json"]
 
+import logging
+
 import sympy
+
 from mira.modeling import Model
 from mira.metamodel import *
-import logging
+from .utils import add_metadata_annotations
+
 
 logger = logging.getLogger(__name__)
 
@@ -193,6 +197,8 @@ def __init__(self, model: Model):
                     link_id += 1
                     self.links.append(link_dict)
 
+        add_metadata_annotations(self.metadata, model)
+
     def to_json(self):
         """Return a JSON dict structure of the Stock and Flow model."""
         return {

mira/modeling/amr/utils.py

@@ -3,7 +3,7 @@ def add_metadata_annotations(metadata, model):
         metadata['annotations'] = {}
         return
     annotations_subset = {
-        k: v
+        k: (str(v) if k in ["time_start", "time_end"] and v is not None else v)
         for k, v in model.template_model.annotations.dict().items()
         if k not in ["name", "description"]
         # name and description already have a privileged place

mira/sources/amr/petrinet.py

@@ -216,7 +216,16 @@ def template_model_from_amr_json(model_json) -> TemplateModel:
     # Finally, we gather some model-level annotations
     name = model_json.get('header', {}).get('name')
     description = model_json.get('header', {}).get('description')
-    anns = Annotations(name=name, description=description)
+
+    annotations = model_json.get('metadata', {}).get('annotations', {})
+    annotation_attributes = {"name": name, "description": description}
+    for key, val in annotations.items():
+        # convert list of author names to list of author objects
+        if key == "authors":
+            val = [Author(name=author_dict["name"]) for author_dict in val]
+        annotation_attributes[key] = val
+
+    anns = Annotations(**annotation_attributes)
     return TemplateModel(templates=templates,
                          parameters=mira_parameters,
                          initials=initials,

mira/sources/amr/regnet.py

@@ -173,7 +173,16 @@ def template_model_from_amr_json(model_json) -> TemplateModel:
     # Finally, we gather some model-level annotations
     name = model_json.get('header', {}).get('name')
     description = model_json.get('header', {}).get('description')
-    anns = Annotations(name=name, description=description)
+
+    annotations = model_json.get('metadata', {}).get('annotations', {})
+    annotation_attributes = {"name": name, "description": description}
+    for key, val in annotations.items():
+        # convert list of author names to list of author objects
+        if key == "authors":
+            val = [Author(name=author_dict["name"]) for author_dict in val]
+        annotation_attributes[key] = val
+
+    anns = Annotations(**annotation_attributes)
     return TemplateModel(templates=templates,
                          parameters=mira_parameters,
                          initials=initials,

mira/sources/amr/stockflow.py

@@ -135,8 +135,16 @@ def template_model_from_amr_json(model_json) -> TemplateModel:
         # Finally, we gather some model-level annotations
     name = model_json.get('header', {}).get('name')
     description = model_json.get('header', {}).get('description')
-    anns = Annotations(name=name, description=description)
 
+    annotations = model_json.get('metadata', {}).get('annotations', {})
+    annotation_attributes = {"name": name, "description": description}
+    for key, val in annotations.items():
+        # convert list of author names to list of author objects
+        if key == "authors":
+            val = [Author(name=author_dict["name"]) for author_dict in val]
+        annotation_attributes[key] = val
+
+    anns = Annotations(**annotation_attributes)
     return TemplateModel(templates=templates,
                          parameters=mira_parameters,
                          initials=initials,

tests/test_amr_source.py

@@ -1,18 +1,62 @@
 import requests
+import json
+
 import sympy
+
 from mira.metamodel import *
 from mira.sources.amr import model_from_url
 from mira.sources.amr import petrinet
 from mira.sources.amr import regnet
 from mira.sources.amr import stockflow
 from mira.modeling.amr.regnet import template_model_to_regnet_json
+from mira.modeling.amr.petrinet import template_model_to_petrinet_json
+from mira.modeling.amr.stockflow import template_model_to_stockflow_json
 
 petrinet_example = 'https://raw.githubusercontent.com/DARPA-ASKEM/' \
-    'Model-Representations/main/petrinet/examples/sir.json'
+                   'Model-Representations/main/petrinet/examples/sir.json'
 regnet_example = 'https://raw.githubusercontent.com/DARPA-ASKEM/' \
-    'Model-Representations/main/regnet/examples/lotka_volterra.json'
+                 'Model-Representations/main/regnet/examples/lotka_volterra.json'
 stockflow_example = 'https://raw.githubusercontent.com/DARPA-ASKEM/' \
-    'Model-Representations/7f5e377225675259baa6486c64102f559edfd79f/stockflow/examples/sir.json'
+                    'Model-Representations/7f5e377225675259baa6486c64102f559edfd79f/stockflow/examples/sir.json'
+
+template_model = TemplateModel(
+    templates=[
+        ControlledReplication(
+            name='replication',
+            controller=Concept(name='A'),
+            subject=Concept(name='B'),
+            rate_law=SympyExprStr(sympy.sympify('k * A * B / (1 + B)'))
+        ),
+        NaturalDegradation(
+            name='degradation',
+            subject=Concept(name='B'),
+            rate_law=SympyExprStr(sympy.sympify('k * B'))
+        )
+    ],
+    observables={
+        'obs1': Observable(
+            name='obs1',
+            expression=SympyExprStr(sympy.sympify('A + B')),
+            display_name='obs1'
+        )
+    },
+    annotations=Annotations(
+        name="test_name",
+        description="test_description",
+        diseases=["test_disease"],
+        hosts=["test_host"],
+        license="test_license",
+        locations=["test_location"],
+        model_types=["test_model_type"],
+        pathogens=["test_pathogen"],
+        references=["test_reference"],
+        authors=[Author(name="test_author")],
+        time_start="2020-03-01T00:00:00",
+        time_end="2020-08-01T00:00:00",
+        time_scale="days"
+    ),
+    time=Time(name='timexx'),
+)
 
 
 def stockflow_set_up_file():
@@ -76,21 +120,25 @@ def test_stockflow_flow_to_template():
 def test_stockflow_parameter_to_mira():
     sfamr = stockflow_set_up_file()
     tm = stockflow.model_from_url(stockflow_example)
-    for amr_param, tm_param in zip(sfamr['semantics']['ode']['parameters'], tm.parameters.values()):
+    for amr_param, tm_param in zip(sfamr['semantics']['ode']['parameters'],
+                                   tm.parameters.values()):
         assert amr_param['id'] == tm_param.name
         assert amr_param['name'] == tm_param.display_name
         assert amr_param['description'] == tm_param.description
         assert amr_param['value'] == tm_param.value
-        assert amr_param.get('units', {}) == (tm_param.units if tm_param.units else {})
+        assert amr_param.get('units', {}) == (
+            tm_param.units if tm_param.units else {})
 
 
 def test_stockflow_initial_to_mira():
     sfamr = stockflow_set_up_file()
     tm = stockflow.model_from_url(stockflow_example)
-    for amr_initial, tm_initial in zip(sfamr['semantics']['ode']['initials'], tm.initials.values()):
+    for amr_initial, tm_initial in zip(sfamr['semantics']['ode']['initials'],
+                                       tm.initials.values()):
         assert amr_initial['target'] == tm_initial.concept.name
         assert amr_initial['expression'] == str(tm_initial.expression)
-        assert amr_initial['expression_mathml'] == expression_to_mathml(tm_initial.expression)
+        assert amr_initial['expression_mathml'] == expression_to_mathml(
+            tm_initial.expression)
 
 
 def test_stockflow_stock_to_concept():
@@ -118,33 +166,41 @@ def test_regnet_rate_laws():
     # Make a simple template model with rate laws, then export
     # into AMR, ingest, then make sure we get proper rate laws back
     # out.
-    template_model = TemplateModel(
-        templates=[
-            ControlledReplication(
-                name='replication',
-                controller=Concept(name='A'),
-                subject=Concept(name='B'),
-                rate_law=SympyExprStr(sympy.sympify('k * A * B / (1 + B)'))
-            ),
-            NaturalDegradation(
-                name='degradation',
-                subject=Concept(name='B'),
-                rate_law=SympyExprStr(sympy.sympify('k * B'))
-            )
-        ],
-        observables={
-            'obs1': Observable(
-                name='obs1',
-                expression=SympyExprStr(sympy.sympify('A + B')),
-                display_name='obs1'
-            )
-        },
-        time=Time(name='timexx')
-    )
     amr_json = template_model_to_regnet_json(template_model)
     tm = regnet.template_model_from_amr_json(amr_json)
     assert isinstance(tm.templates[0].rate_law, SympyExprStr)
     assert isinstance(tm.templates[1].rate_law, SympyExprStr)
-    assert tm.templates[1].rate_law.args[0].equals(sympy.sympify('k * A * B / (1 + B)'))
+    assert tm.templates[1].rate_law.args[0].equals(
+        sympy.sympify('k * A * B / (1 + B)'))
     assert tm.time.name == 'timexx'
-    assert isinstance(tm.observables['obs1'].expression, SympyExprStr)
+    assert isinstance(tm.observables['obs1'].expression, SympyExprStr)
+
+
+def test_annotation_serialization_ingestion():
+    """Test to see if we can serialize template models that contain
+    datetime in their annotations into different frameworks.
+
+    Also test to see if we can extract all annotation related attributes
+    when we ingest an amr
+    """
+    amrs = [template_model_to_regnet_json(template_model),
+            template_model_to_petrinet_json(template_model),
+            template_model_to_stockflow_json(template_model)
+            ]
+    for amr in amrs:
+        json.dumps(amr)
+
+    # test to see if we can extract all annotation attributes during ingestion
+    # for each framework
+    regnet_tm = regnet.template_model_from_amr_json(amrs[0])
+    petrinet_tm = petrinet.template_model_from_amr_json(amrs[1])
+    stockflow_tm = stockflow.template_model_from_amr_json(amrs[2])
+
+    zipped_annotations = zip(regnet_tm.annotations.dict().values(),
+                              petrinet_tm.annotations.dict().values(),
+                              stockflow_tm.annotations.dict().values())
+
+    for annotation_attribute_tuple in zipped_annotations:
+        assert annotation_attribute_tuple[0]
+        assert annotation_attribute_tuple[1]
+        assert annotation_attribute_tuple[2]