Merge pull request #328 from nanglo123/sbml_qual_support

Add support for sbml qual models

docs/source/sources.rst

@@ -41,6 +41,13 @@ SBML extraction (:py:mod:`mira.sources.sbml`)
     :members:
     :show-inheritance:
 
+SBML Qual extraction (:py:mod:`mira.sources.sbml.qual_api`)
+-----------------------------------------------------------
+.. automodule:: mira.sources.sbml.qual_api
+    :members:
+    :show-inheritance:
+
+
 BioModels client (:py:mod:`mira.sources.biomodels`)
 ---------------------------------------------------
 .. automodule:: mira.sources.biomodels

mira/modeling/amr/regnet.py

@@ -5,7 +5,6 @@
 __all__ = ["AMRRegNetModel", "ModelSpecification",
            "template_model_to_regnet_json"]
 
-
 import json
 import logging
 from copy import deepcopy
@@ -47,7 +46,7 @@ def __init__(self, model: Model):
             model.template_model.annotations.name else "Model"
         self.model_description = model.template_model.annotations.description \
             if model.template_model.annotations and \
-            model.template_model.annotations.description else self.model_name
+               model.template_model.annotations.description else self.model_name
 
         self.rates = []
         self.observables = []
@@ -84,7 +83,7 @@ def __init__(self, model: Model):
             self.states.append(state_data)
             self._states_by_id[name] = state_data
 
-        idx = 0
+        edge_id = 1
         # It's possible that something is naturally degraded/replicated
         # by multiple transitions so we need to collect and aggregate rates
         intrinsic_by_var = defaultdict(list)
@@ -132,47 +131,70 @@ def __init__(self, model: Model):
             # a regular transition in the regnet framework
             # Possibilities are:
             # - ControlledReplication / GroupedControlledProduction
-            # - ControlledDegradation
+            # - ControlledDegradation / GroupedControlledDegradation
+            # - ControlledProduction if the control and produced concept are not the same
             else:
-                tid = f"t{idx + 1}"
-                transition_dict = {'id': tid}
                 # If we have multiple controls then the thing that replicates
                 # is both a control and a produced variable.
                 if len(transition.control) > 1:
-                    indep_ctrl = {c.key for c in transition.control} - \
-                        {transition.produced[0].key}
-                    # There is one corner case where both controllers are also
-                    # the same as the produced variable, in which case.
-                    if not indep_ctrl:
-                        indep_ctrl = {transition.produced[0].key}
-                    transition_dict['source'] = vmap[sorted(indep_ctrl)[0]]
+                    # GroupedControlledProduction
+                    if is_production(transition.template) or is_replication(
+                        transition.template
+                    ):
+                        indep_ctrl = {c.key for c in transition.control} - {
+                            transition.produced[0].key
+                        }
+                        # There is one corner case where both controllers are also
+                        # the same as the produced variable, in which case.
+                        if not indep_ctrl:
+                            indep_ctrl = {transition.produced[0].key}
+
+                        for index, controller in enumerate(indep_ctrl):
+                            self.create_edge(
+                                transition,
+                                vmap[controller],
+                                vmap[transition.produced[0].key],
+                                edge_id,
+                                False if index == 0 else True
+                            )
+
+                            edge_id += 1
+                    else:
+                        # GroupedControlledDegradation
+                        indep_ctrl = {c.key for c in transition.control} - {
+                            transition.consumed[0].key
+                        }
+                        # There is one corner case where both controllers are also
+                        # the same as the consumed variable, in which case.
+                        if not indep_ctrl:
+                            indep_ctrl = {transition.consumed[0].key}
+
+                        for index, controller in enumerate(indep_ctrl):
+                            self.create_edge(
+                                transition,
+                                vmap[controller],
+                                vmap[transition.consumed[0].key],
+                                edge_id,
+                                False if index == 0 else True
+                            )
+
+                            edge_id += 1
                 else:
-                    transition_dict['source'] = vmap[transition.control[0].key]
-                transition_dict['target'] = \
-                    vmap[transition.consumed[0].key if
-                         transition.consumed else transition.produced[0].key]
-                transition_dict['sign'] = (is_production(transition.template) or
-                                           is_replication(transition.template))
-
-                # Include rate law
-                if transition.template.rate_law:
-                    rate_law = transition.template.rate_law.args[0]
-                    pnames = transition.template.get_parameter_names()
-                    # We just choose an arbitrary one deterministically
-                    rate_const = sorted(pnames)[0] if pnames else None
-
-                    transition_dict['properties'] = {
-                        'name': tid,
-                        'rate_constant': rate_const,
-                    }
-                    self.rates.append({
-                        'target': tid,
-                        'expression': str(rate_law),
-                        'expression_mathml': expression_to_mathml(rate_law)
-                    })
-
-                self.transitions.append(transition_dict)
-                idx += 1
+                    # ControlledProduction if produced and controller are not the same
+                    # ControlledDegradation
+                    target = vmap[
+                        transition.consumed[0].key
+                        if transition.consumed
+                        else transition.produced[0].key
+                    ]
+                    self.create_edge(
+                        transition,
+                        vmap[transition.control[0].key],
+                        target,
+                        edge_id,
+                        False
+                    )
+                    edge_id += 1
 
         for var, rates in intrinsic_by_var.items():
             rate_law = sum(rates)
@@ -224,6 +246,55 @@ def __init__(self, model: Model):
             self.time = None
         add_metadata_annotations(self.metadata, model)
 
+    def create_edge(self, transition, source, target, edge_id, duplicate):
+        """Create and append a transition dictionary to the list of transitions
+
+        Parameters
+        ----------
+        transition : Transition
+            The Transition object
+        source : str
+            The name of the source of the transition
+        target : str
+            The name of the target of the transition
+        edge_id : int
+            The id to assign to the transition
+        duplicate : bool
+            A boolean that tells us whether the transition we are processing has already been
+            processed at least once. This is for the purpose of not adding duplicate rate laws
+            to the output amr.
+        """
+        tid = f"t{edge_id}"
+        transition_dict = {"id": tid}
+        transition_dict["source"] = source
+        transition_dict["target"] = target
+        transition_dict["sign"] = is_production(
+            transition.template
+        ) or is_replication(transition.template)
+
+        #
+        if transition.template.rate_law:
+            # If we are processing a duplicate rate, set the rate to 0
+            rate_law = transition.template.rate_law.args[0] if not duplicate \
+                else safe_parse_expr('0')
+            pnames = transition.template.get_parameter_names()
+            # We just choose an arbitrary one deterministically
+            rate_const = sorted(pnames)[0] if pnames else None
+
+            transition_dict["properties"] = {
+                "name": tid,
+                "rate_constant": rate_const,
+            }
+            self.rates.append(
+                {
+                    "target": tid,
+                    "expression": str(rate_law),
+                    "expression_mathml": expression_to_mathml(rate_law),
+                }
+            )
+
+        self.transitions.append(transition_dict)
+
     def to_json(
         self,
         name: str = None,
@@ -461,4 +532,3 @@ class ModelSpecification(BaseModel):
     model: RegNetModel
     semantics: Optional[Ode]
     metadata: Optional[Dict]
-

mira/sources/sbml/__init__.py

@@ -1 +1,2 @@
 from .api import *
+from .qual_api import *