refactor: serializable handling

Author: glichtner

apps/graph/static/script.js

@@ -129,15 +129,15 @@ async function loadGraph(recommendationId) {
                         return nodeColors[ele.data('category')] || '#666'; // Assign color based on 'type', with a default
                     },
                     'shape': function(ele) {
-                        return nodeShapes[ele.data('type')] || 'star'; // Assign color based on 'type', with a default
+                        return nodeShapes[ele.data("is_atom") ? "Symbol" : ele.data('type')] || 'star'; // Assign color based on 'type', with a default
                     },
                     'text-valign': 'center',
                     'color': '#000000',
                     'width': function(ele) {
-                        return ele.data('type') === 'Symbol' ? '120px': '40px';
+                        return ele.data('is_atom') ? '120px': '40px';
                     },
                     'height': function(ele) {
-                        return ele.data('type') === 'Symbol' ? '80px': '40px';
+                        return ele.data('is_atom') ? '80px': '40px';
                     },
                     'font-size': '10px',
                     'text-wrap': 'wrap',

apps/rest_api/app/routers/recommendation.py

@@ -46,7 +46,7 @@ async def recommendation_criteria(
 
     data = []
 
-    for c in recommendation.flatten():
+    for c in recommendation.atoms():
         data.append(
             {
                 "description": c.description(),

apps/viz-backend/app/main.py

@@ -89,6 +89,8 @@ def get_execution_graph(recommendation_id: int, db: Session = Depends(get_db)) -
     if not result:
         raise HTTPException(status_code=404, detail="Recommendation not found")
 
+    print(result)
+
     # Decode the bytes to a string and parse it as JSON
     execution_graph = json.loads(result.recommendation_execution_graph.decode("utf-8"))
 

execution_engine/converter/action/procedure.py

@@ -71,15 +71,15 @@ def _to_expression(self) -> logic.Symbol:
                 # as Observation and Measurement normally expect a value.
                 criterion = Measurement(
                     concept=self._code,
-                    override_value_required=False,
+                    value_required=False,
                     timing=self._timing,
                 )
             case "Observation":
                 # we need to explicitly set the VALUE_REQUIRED flag to false, otherwise creating the query will raise an error
                 # as Observation and Measurement normally expect a value.
                 criterion = Observation(
                     concept=self._code,
-                    override_value_required=False,
+                    value_required=False,
                     timing=self._timing,
                 )
             case _:

execution_engine/converter/recommendation_factory.py

@@ -63,7 +63,7 @@ def parse_recommendation_from_url(
         pi_pairs: list[PopulationInterventionPairExpr] = []
 
         base_criterion = PatientsActiveDuringPeriod()
-
+        base_criterion.dict()
         for rec_plan in rec.plans():
 
             # parse population and create criteria
@@ -72,6 +72,9 @@ def parse_recommendation_from_url(
             # parse intervention and create criteria
             actions = parser.parse_actions(rec_plan.actions, rec_plan)
 
+            # population_expr is assigned a NoDataPreservingAnd to ensure creation of negative intervals
+            # todo: not sure we really need this - we can just always create negative intervals when store_results=True
+            # in the graph
             pi_pair = PopulationInterventionPairExpr(
                 population_expr=population_criteria,
                 intervention_expr=actions,

execution_engine/converter/time_from_event/abstract.py

@@ -55,9 +55,9 @@ def valid(cls, fhir: Element) -> bool:
         raise NotImplementedError("must be implemented by class")
 
     @abstractmethod
-    def to_temporal_combination(self, combo: logic.BaseExpr) -> logic.TemporalCount:
+    def to_temporal_combination(self, expr: logic.BaseExpr) -> logic.Expr:
         """
-        Wraps Criterion/CriterionCombinaion with a TemporalIndicatorCombination
+        Wraps Criterion/CriterionCombination with a TemporalIndicatorCombination
         """
         raise NotImplementedError("must be implemented by class")
 
@@ -122,8 +122,8 @@ def valid(cls, fhir: Element) -> bool:
         return cls._event_vocabulary.is_system(cc.system) and cc.code == cls._event_code
 
     @abstractmethod
-    def to_temporal_combination(self, combo: logic.BaseExpr) -> logic.TemporalCount:
+    def to_temporal_combination(self, expr: logic.BaseExpr) -> logic.Expr:
         """
-        Wraps Criterion/CriterionCombinaion with a TemporalIndicatorCombination
+        Wraps expression with a TemporalIndicatorCombination
         """
         raise NotImplementedError("must be implemented by class")

execution_engine/execution_engine.py

@@ -17,8 +17,8 @@
 from execution_engine.omop.cohort import PopulationInterventionPairExpr
 from execution_engine.omop.criterion.abstract import Criterion
 from execution_engine.omop.db.celida import tables as result_db
-from execution_engine.omop.serializable import Serializable
 from execution_engine.task import runner
+from execution_engine.util.serializable import Serializable
 
 
 class ExecutionEngine:
@@ -208,7 +208,7 @@ def load_recommendation_from_database(
                     pi_pair, rec_db.recommendation_id
                 )
 
-            for criterion in recommendation.flatten():
+            for criterion in recommendation.atoms():
                 self.register_criterion(criterion)
 
             # All objects in the deserialized object graph must have an id.
@@ -219,7 +219,7 @@ def load_recommendation_from_database(
             for pi_pair in recommendation.population_intervention_pairs():
                 assert pi_pair.id is not None
 
-            for criterion in recommendation.flatten():
+            for criterion in recommendation.atoms():
                 assert criterion.id is not None
 
             return recommendation
@@ -294,7 +294,7 @@ def register_recommendation(self, recommendation: cohort.Recommendation) -> None
                 pi_pair, recommendation_id=recommendation.id
             )
 
-        for criterion in recommendation.flatten():
+        for criterion in recommendation.atoms():
             self.register_criterion(criterion)
 
         assert recommendation.id is not None
@@ -304,7 +304,7 @@ def register_recommendation(self, recommendation: cohort.Recommendation) -> None
         for pi_pair in recommendation.population_intervention_pairs():
             assert pi_pair.id is not None
 
-        for criterion in recommendation.flatten():
+        for criterion in recommendation.atoms():
             assert criterion.id is not None
 
         # Update the recommendation in the database with the final

execution_engine/execution_graph/graph.py

@@ -1,5 +1,4 @@
-import copy
-from typing import Any
+from typing import Any, cast
 
 import networkx as nx
 
@@ -40,27 +39,6 @@ def is_sink(self, expr: logic.Expr) -> bool:
         """
         return self.out_degree(expr) == 0
 
-    @classmethod
-    def filter_symbols(cls, node: logic.Expr, filter_: logic.Expr) -> logic.Expr:
-        """
-        Filter (=AND-combine) all symbols by the applied filter function
-
-        Used to filter all intervention criteria (symbols) by the population output in order to exclude
-        all intervention events outside the population intervals, which may otherwise interfere with corrected
-        determination of temporal combination, i.e. the presence of an intervention event during some time window.
-        """
-
-        if isinstance(node, logic.Symbol):
-            return logic.LeftDependentToggle(left=filter_, right=node)
-
-        if hasattr(node, "args") and isinstance(node.args, tuple):
-            converted_args = [cls.filter_symbols(a, filter_) for a in node.args]
-
-            if any(a is not b for a, b in zip(node.args, converted_args)):
-                node.args = tuple(converted_args)
-
-        return node
-
     @classmethod
     def from_expression(
         cls, expr: logic.Expr, base_criterion: Criterion, category: CohortCategory
@@ -71,9 +49,7 @@ def from_expression(
 
         from execution_engine.omop.cohort import PopulationInterventionPairExpr
 
-        # we might make changes to the expression (e.g. filtering), so we must preserve
-        # the original expression from the caller
-        expr = copy.deepcopy(expr)
+        expr_hash = hash(expr)
 
         graph = cls()
         base_node = base_criterion
@@ -89,42 +65,33 @@ def traverse(
             parent: logic.Expr | None = None,
             category: CohortCategory = category,
         ) -> None:
+
+            graph.add_node(expr, category=category, store_result=False)
+
+            if parent is not None:
+                assert expr in graph.nodes
+                assert parent in graph.nodes
+                graph.add_edge(expr, parent)
+
             if isinstance(expr, PopulationInterventionPairExpr):
                 # special case for PopulationInterventionPairExpr:
                 # we need explicitly set the category of the population and intervention nodes
 
                 p, i = expr.left, expr.right
 
-                # filter all intervention criteria by the output of the population - this is performed to filter out
-                # intervention events that outside of the population intervals (i.e. the time windows during which
-                # patients are part of the population) as otherwise events outside of the population time may be picked up
-                # by Temporal criteria that determine the presence of some event or condition during a specific time window.
-
-                # the following command changes expr, i.e. we must not add expr before this command to the graph
-
-                i = cls.filter_symbols(i, filter_=p)
-
                 traverse(i, parent=expr, category=CohortCategory.INTERVENTION)
                 traverse(p, parent=expr, category=CohortCategory.POPULATION)
 
-                graph.add_node(expr, category=category, store_result=False)
-
-                if parent is not None:
-                    graph.add_edge(expr, parent)
-
-                subgraph = graph.subgraph(nx.ancestors(graph, expr) | {expr})
-
-                subgraph.set_sink_nodes_store(bind_params=dict(pi_pair_id=expr._id))
-
-                # children are already traversed
-                return
-
-            graph.add_node(expr, category=category, store_result=False)
+                # create a subgraph for the pair in order to determine the sink nodes (i.e. the nodes that have no
+                # outgoing edges) for POPULATION and POPULATION_INTERVENTION and mark them for storing their result
+                # in the database
+                subgraph = cast(
+                    ExecutionGraph, graph.subgraph(nx.ancestors(graph, expr) | {expr})
+                )
+                subgraph.set_sink_nodes_store(bind_params=dict(pi_pair_id=expr.id))
 
-            if parent is not None:
-                graph.add_edge(expr, parent)
-
-            if expr.is_Atom:
+            elif expr.is_Atom:
+                assert expr in graph.nodes
                 graph.nodes[expr]["store_result"] = True
                 graph.add_edge(base_node, expr)
             else:
@@ -133,6 +100,9 @@ def traverse(
 
         traverse(expr, category=category)
 
+        if hash(expr) != expr_hash:
+            raise ValueError("Expression has been modified during traversal")
+
         return graph
 
     def add_node(
@@ -219,6 +189,7 @@ def to_cytoscape_dict(self) -> dict:
                         self.nodes[node]["store_result"]
                     ),  # Convert to string if necessary
                     "is_sink": self.is_sink(node),
+                    "is_atom": node.is_Atom,
                     "bind_params": self.nodes[node]["bind_params"],
                 }
             }

execution_engine/omop/cohort/graph_builder.py

@@ -0,0 +1,129 @@
+import copy
+
+import execution_engine.util.logic as logic
+from execution_engine.constants import CohortCategory
+from execution_engine.execution_graph import ExecutionGraph
+from execution_engine.omop.cohort.population_intervention_pair import (
+    PopulationInterventionPairExpr,
+)
+from execution_engine.omop.criterion.abstract import Criterion
+
+
+class RecommendationGraphBuilder:
+    """
+    A builder class for constructing ExecutionGraph objects based on
+    population/intervention expressions. It provides utility methods to filter
+    intervention criteria by population constraints, and then converts the
+    filtered expression into an ExecutionGraph ready for execution and storage.
+    """
+
+    @classmethod
+    def filter_symbols(cls, node: logic.Expr, filter_: logic.Expr) -> logic.Expr:
+        """
+        Filter (=AND-combine) all symbols by the applied filter function
+
+        Used to filter all intervention criteria (symbols) by the population output in order to exclude
+        all intervention events outside the population intervals, which may otherwise interfere with corrected
+        determination of temporal combination, i.e. the presence of an intervention event during some time window.
+
+        :param node: The expression node to be filtered.
+        :type node: logic.Expr
+        :param filter_: The filter expression to AND-combine with symbols in the node.
+        :type filter_: logic.Expr
+        :return: A new expression in which all symbols are constrained by the filter expression.
+        :rtype: logic.Expr
+        """
+
+        if isinstance(node, logic.Symbol):
+            return logic.LeftDependentToggle(left=filter_, right=node)
+        elif isinstance(node, logic.Expr):
+            converted_args = [cls.filter_symbols(a, filter_) for a in node.args]
+
+            if any(a is not b for a, b in zip(node.args, converted_args)):
+                node.update_args(*converted_args)
+
+        return node
+
+    @classmethod
+    def filter_intervention_criteria_by_population(cls, expr: logic.Expr) -> logic.Expr:
+        """
+        Filter all intervention criteria in a given expression by the population part of the expression.
+
+        :param expr: The expression that may contain population and intervention parts.
+        :type expr: logic.Expr
+        :return: A new expression where all intervention symbols are constrained by the population intervals.
+        :rtype: logic.Expr
+        """
+
+        from execution_engine.omop.cohort import PopulationInterventionPairExpr
+
+        # we might make changes to the expression (e.g. filtering), so we must preserve
+        # the original expression from the caller
+        expr = copy.deepcopy(expr)
+
+        def traverse(
+            expr: logic.Expr,
+        ) -> None:
+            if isinstance(expr, PopulationInterventionPairExpr):
+                p, i = expr.left, expr.right
+
+                # filter all intervention criteria by the output of the population - this is performed to filter out
+                # intervention events that outside of the population intervals (i.e. the time windows during which
+                # patients are part of the population) as otherwise events outside of the population time may be picked up
+                # by Temporal criteria that determine the presence of some event or condition during a specific time window.
+                i = cls.filter_symbols(i, filter_=p)
+
+                expr.update_args(p, i)
+
+                traverse(i)
+                traverse(p)
+
+            elif not expr.is_Atom:
+                for child in expr.args:
+                    traverse(child)
+
+        traverse(expr)
+
+        return expr
+
+    @classmethod
+    def build(cls, expr: logic.Expr, base_criterion: Criterion) -> ExecutionGraph:
+        """
+        Build an ExecutionGraph for a population/intervention expression.
+
+        If the expression is a PopulationInterventionPairExpr, it is wrapped in a
+        NonSimplifiableAnd to ensure a top-level result entry is generated in the database.
+        Then the expression is filtered and converted into an ExecutionGraph with the
+        appropriate sink nodes and edges.
+
+        :param expr: The population/intervention expression to build the graph from.
+        :type expr: logic.Expr
+        :param base_criterion: The base criterion used to label the execution graph.
+        :type base_criterion: Criterion
+        :return: The constructed ExecutionGraph for the given expression.
+        :rtype: ExecutionGraph
+        """
+        if isinstance(expr, PopulationInterventionPairExpr):
+            expr = logic.NonSimplifiableAnd(expr)
+
+        # Make sure the expr is filtered
+        expr_filtered = cls.filter_intervention_criteria_by_population(expr)
+
+        graph = ExecutionGraph.from_expression(
+            expr_filtered,
+            base_criterion=base_criterion,
+            category=CohortCategory.POPULATION_INTERVENTION,
+        )
+
+        p_sink_nodes = graph.sink_nodes(CohortCategory.POPULATION)
+        graph.set_sink_nodes_store(
+            bind_params={}, desired_category=CohortCategory.POPULATION_INTERVENTION
+        )
+
+        p_combination_node = logic.NoDataPreservingOr(*p_sink_nodes)
+        graph.add_node(
+            p_combination_node, store_result=True, category=CohortCategory.POPULATION
+        )
+        graph.add_edges_from((src, p_combination_node) for src in p_sink_nodes)
+
+        return graph