Merge pull request #245 from nanglo123/stock_flow_export

Import and Export Stock and Flow acsets into and from Mira template models

mira/dkg/model.py

@@ -31,14 +31,14 @@
 from mira.modeling import Model
 from mira.modeling.askenet.petrinet import AskeNetPetriNetModel, ModelSpecification
 from mira.modeling.bilayer import BilayerModel
-from mira.modeling.petri import PetriNetModel, PetriNetResponse
+from mira.modeling.acsets.petri import PetriNetModel, PetriNetResponse
 from mira.modeling.viz import GraphicalModel
 from mira.sources.askenet.flux_span import reproduce_ode_semantics, \
     test_file_path, docker_test_file_path
 from mira.sources.askenet.petrinet import template_model_from_askenet_json
 from mira.sources.bilayer import template_model_from_bilayer
 from mira.sources.biomodels import get_sbml_model
-from mira.sources.petri import template_model_from_petri_json
+from mira.sources.acsets.petri import template_model_from_petri_json
 from mira.sources.sbml import template_model_from_sbml_string
 
 __all__ = [

mira/metamodel/schema.json

@@ -83,6 +83,10 @@
         "name": {
           "title": "Name",
           "type": "string"
+        },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
         }
       }
     },
@@ -105,6 +109,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "ControlledConversion",
@@ -151,6 +159,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "GroupedControlledConversion",
@@ -201,6 +213,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "GroupedControlledProduction",
@@ -247,6 +263,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "ControlledProduction",
@@ -289,6 +309,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "NaturalConversion",
@@ -331,6 +355,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "NaturalProduction",
@@ -369,6 +397,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "NaturalDegradation",
@@ -407,6 +439,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "ControlledDegradation",
@@ -449,6 +485,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "GroupedControlledDegradation",
@@ -495,6 +535,10 @@
           "title": "Name",
           "type": "string"
         },
+        "display_name": {
+          "title": "Display Name",
+          "type": "string"
+        },
         "type": {
           "title": "Type",
           "default": "StaticConcept",

mira/metamodel/templates.py

@@ -294,6 +294,7 @@ class Config:
 
     rate_law: Optional[SympyExprStr] = Field(default=None)
     name: Optional[str] = Field(default=None)
+    display_name: Optional[str] = Field(default=None)
 
     @classmethod
     def from_json(cls, data, rate_symbols=None) -> "Template":

mira/modeling/petri.py → mira/modeling/acsets/petri.py

@@ -14,7 +14,7 @@
 from pydantic import BaseModel, Field
 import sympy
 
-from . import Model
+from mira.modeling import Model
 from mira.metamodel import expression_to_mathml
 from mira.metamodel.utils import safe_parse_expr
 

mira/modeling/acsets/stockflow.py

@@ -0,0 +1,84 @@
+from mira.modeling import Model
+from mira.metamodel import *
+
+
+class AskeNetStockFlowModel:
+
+    def __init__(self, model: Model):
+        self.properties = {}
+        self.stocks = []
+        self.flows = []
+        self.links = []
+        self.model_name = 'Model'
+
+        for idx, flow in enumerate(model.transitions.values()):
+            fid = flow.template.name
+            fname = flow.template.display_name
+
+            input = flow.consumed[0].key
+            output = flow.produced[0].key
+
+            rate_law_str = str(flow.template.rate_law) if flow.template.rate_law else None
+            if rate_law_str:
+                for param_key, param_obj in model.parameters.items():
+                    if param_obj.placeholder:
+                        continue
+                    index = rate_law_str.find(param_key)
+                    if index < 0:
+                        continue
+                    rate_law_str = rate_law_str[:index] + 'p.' + rate_law_str[index:]
+
+                for var_obj in model.variables.values():
+                    index = rate_law_str.find(var_obj.concept.display_name)
+                    if index < 0:
+                        continue
+                    rate_law_str = rate_law_str[:index] + 'u.' + rate_law_str[index:]
+
+            flow_dict = {'_id': fid,
+                         'u': input,
+                         'd': output,
+                         'fname': fname,
+                         'ϕf': rate_law_str}
+
+            self.flows.append(flow_dict)
+
+        vmap = {}
+        for var_key, var in model.variables.items():
+            vmap[var_key] = name = var.concept.name or str(var_key)
+            display_name = var.concept.display_name or name
+
+            stocks_dict = {
+                '_id': name,
+                'sname': display_name,
+            }
+            self.stocks.append(stocks_dict)
+
+            # Declare 's' and 't' field of a link before assignment, this is because if a stock is found to be
+            # a target for a flow before it is a source, then the 't' field of the link associated with a stock
+            # will be displayed first
+            links_dict = {'_id': name}
+            links_dict['s'] = None
+            links_dict['t'] = None
+            for flow in model.transitions.values():
+                if flow.consumed[0].concept.name == name:
+                    links_dict['s'] = flow.template.name
+                if flow.produced[0].concept.name == name:
+                    links_dict['t'] = flow.template.name
+
+            if not links_dict.get('s') and links_dict.get('t'):
+                links_dict['s'] = links_dict.get('t')
+            elif links_dict.get('s') and not links_dict.get('t'):
+                links_dict['t'] = links_dict.get('s')
+
+            self.links.append(links_dict)
+
+    def to_json(self):
+        return {
+            'Flow': self.flows,
+            'Stock': self.stocks,
+            'Link': self.links
+        }
+
+
+def template_model_to_stockflow_ascet_json(tm: TemplateModel):
+    return AskeNetStockFlowModel(Model(tm)).to_json()

mira/modeling/askenet/ops.py

@@ -60,7 +60,7 @@ def replace_transition_id(tm, old_id, new_id):
 
 @amr_to_mira
 def replace_observable_id(tm, old_id, new_id, name=None):
-    """Replace the ID of an observable."""
+    """Replace the ID of an observable"""
     for obs, observable in copy.deepcopy(tm.observables).items():
         if obs == old_id:
             observable.name = new_id
@@ -72,6 +72,7 @@ def replace_observable_id(tm, old_id, new_id, name=None):
 
 @amr_to_mira
 def remove_observable(tm, removed_id):
+    """Remove an observable from the template model"""
     for obs, observable in copy.deepcopy(tm.observables).items():
         if obs == removed_id:
             tm.observables.pop(obs)
@@ -80,6 +81,11 @@ def remove_observable(tm, removed_id):
 
 @amr_to_mira
 def remove_parameter(tm, removed_id, replacement_value=None):
+    """
+    If a replacement_value is supplied, substitute every instance of the parameter
+    in all expressions with the given replacement_value. If replacement_value is none,
+    substitute the parameter with 0.
+    """
     if replacement_value:
         tm.substitute_parameter(removed_id, replacement_value)
     else:
@@ -95,6 +101,7 @@ def remove_parameter(tm, removed_id, replacement_value=None):
 
 @amr_to_mira
 def add_observable(tm, new_id, new_name, new_expression):
+    """Add a new observable object to the template model"""
     # Note that if an observable already exists with the given
     # key, it will be replaced
     rate_law_sympy = mathml_to_expression(new_expression)
@@ -106,7 +113,7 @@ def add_observable(tm, new_id, new_name, new_expression):
 
 @amr_to_mira
 def replace_parameter_id(tm, old_id, new_id):
-    """Replace the ID of a parameter."""
+    """Replace the ID of a parameter"""
     if old_id not in tm.parameters:
         raise ValueError(f"Parameter with ID {old_id} not found in model.")
     for template in tm.templates:
@@ -134,10 +141,11 @@ def replace_parameter_id(tm, old_id, new_id):
 @amr_to_mira
 def add_parameter(tm, parameter_id: str,
                   name: str = None,
-                  description:str = None,
+                  description: str = None,
                   value: float = None,
                   distribution=None,
                   units_mathml: str = None):
+    """Add a new parameter to the template model"""
     tm.add_parameter(parameter_id, name, description, value, distribution, units_mathml)
     return tm
 
@@ -154,6 +162,7 @@ def replace_initial_id(tm, old_id, new_id):
 # Remove state
 @amr_to_mira
 def remove_state(tm, state_id):
+    """Remove a state from the template model"""
     new_templates = []
     for template in tm.templates:
         to_remove = False
@@ -174,6 +183,7 @@ def remove_state(tm, state_id):
 def add_state(tm, state_id: str, name: str = None,
               units_mathml: str = None, grounding: Mapping[str, str] = None,
               context: Mapping[str, str] = None):
+    """Add a new state to the template model"""
     if units_mathml:
         units = Unit(expression=SympyExprStr(mathml_to_expression(units_mathml)))
     else:
@@ -192,13 +202,36 @@ def add_state(tm, state_id: str, name: str = None,
 
 @amr_to_mira
 def remove_transition(tm, transition_id):
+    """Remove a transition object from the template model"""
     tm.templates = [t for t in tm.templates if t.name != transition_id]
     return tm
 
 
 @amr_to_mira
 def add_transition(tm, new_transition_id, src_id=None, tgt_id=None,
                    rate_law_mathml=None, params_dict: Mapping = None):
+    """Add a new transition to the template model
+
+       Parameters
+       ----------
+       tm:
+           The template model
+       new_transition_id:
+           The ID of the new transition to add
+       src_id:
+           The ID of the subject of the newly created transition (default None)
+       tgt_id:
+           The ID of the outcome of the newly created transition (default None)
+        rate_law_math_ml:
+            The rate law associated with the newly created transition
+        params_dict:
+            A mapping of parameter attributes to their respective values if the user
+            decides to explicitly create parameters
+       Returns
+       -------
+       :
+            The updated template model
+        """
     if src_id is None and tgt_id is None:
         ValueError("You must pass in at least one of source and target id")
     if src_id not in tm.get_concepts_name_map() and tgt_id not in tm.get_concepts_name_map():
@@ -219,6 +252,7 @@ def add_transition(tm, new_transition_id, src_id=None, tgt_id=None,
 
 @amr_to_mira
 def replace_rate_law_sympy(tm, transition_id, new_rate_law: sympy.Expr):
+    """Replace the rate law of transition. The new rate law passed in will be a sympy.Expr object"""
     # NOTE: this assumes that a sympy expression object is given
     # though it might make sense to take a string instead
     for template in tm.templates:
@@ -230,13 +264,15 @@ def replace_rate_law_sympy(tm, transition_id, new_rate_law: sympy.Expr):
 # This function isn't wrapped because it calls a wrapped function and just
 # passes the AMR through
 def replace_rate_law_mathml(amr, transition_id, new_rate_law):
+    """Replace the rate law of a transition. THe new rate law passed in will be a MathML str object"""
     new_rate_law_sympy = mathml_to_expression(new_rate_law)
     return replace_rate_law_sympy(amr, transition_id, new_rate_law_sympy)
 
 
 @amr_to_mira
 def replace_observable_expression_sympy(tm, obs_id,
                                         new_expression_sympy: sympy.Expr):
+    """Replace the expression of an observable. The new rate law passed in will be a sympy.Expr object"""
     for obs, observable in tm.observables.items():
         if obs == obs_id:
             observable.expression = SympyExprStr(new_expression_sympy)
@@ -246,6 +282,7 @@ def replace_observable_expression_sympy(tm, obs_id,
 @amr_to_mira
 def replace_initial_expression_sympy(tm, initial_id,
                                      new_expression_sympy: sympy.Expr):
+    """Replace the expression of an initial. THe new rate law passed in will be a sympy.Expr object"""
     for init, initial in tm.initials.items():
         if init == initial_id:
             initial.expression = SympyExprStr(new_expression_sympy)
@@ -255,6 +292,7 @@ def replace_initial_expression_sympy(tm, initial_id,
 # This function isn't wrapped because it calls a wrapped function and just
 # passes the AMR through
 def replace_observable_expression_mathml(amr, obs_id, new_expression_mathml):
+    """Replace the expression of an observable. The new rate law passed in will be MathML str object"""
     new_expression_sympy = mathml_to_expression(new_expression_mathml)
     return replace_observable_expression_sympy(amr, obs_id,
                                                new_expression_sympy)
@@ -263,6 +301,7 @@ def replace_observable_expression_mathml(amr, obs_id, new_expression_mathml):
 # This function isn't wrapped because it calls a wrapped function and just
 # passes the AMR through
 def replace_initial_expression_mathml(amr, initial_id, new_expression_mathml):
+    """Replace the expression of an initial. THe new rate law passed in will be a MathML str object"""
     new_expression_sympy = mathml_to_expression(new_expression_mathml)
     return replace_initial_expression_sympy(amr, initial_id,
                                             new_expression_sympy)