Remove criterion combination

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-ignore = E203, E266, E501, W503, F403, F401, E402, C901, F405
+ignore = E203, E266, E501, W503, F403, F401, E402, C901, F405, E731
 max-line-length = 88
 max-complexity = 18
 select = B,C,E,F,W,T4,B9

.pre-commit-config.yaml

@@ -3,7 +3,7 @@ default_language_version:
 exclude: '^docs/'
 repos:
 - repo: https://github.com/pycqa/isort
-  rev: 5.13.2
+  rev: 6.0.1
   hooks:
     - id: isort
       name: isort (python)
@@ -17,13 +17,13 @@ repos:
     -   id: check-json
     -   id: pretty-format-json
         args: ['--autofix', '--no-sort-keys']
--   repo: https://github.com/ambv/black
-    rev: 24.10.0
+-   repo: https://github.com/psf/black
+    rev: 25.1.0
     hooks:
     - id: black
       language_version: python3.11
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v1.13.0'
+    rev: 'v1.15.0'
     hooks:
     -   id: mypy
         name: mypy
@@ -36,12 +36,12 @@ repos:
         exclude: tests/
         args: [--select, "D101,D102,D103,D105,D106"]
 -   repo: https://github.com/PyCQA/bandit
-    rev: '1.8.0'
+    rev: '1.8.3'
     hooks:
     - id: bandit
       args: [--skip, "B101,B303,B110,B311"]
 -   repo: https://github.com/PyCQA/flake8
-    rev: '7.1.1'
+    rev: '7.1.2'
     hooks:
     -   id: flake8
 -   repo: https://github.com/myint/autoflake

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(),

execution_engine/builder.py

@@ -28,7 +28,8 @@
 from execution_engine.converter.goal.ventilator_management import (
     VentilatorManagementGoal,
 )
-from execution_engine.converter.time_from_event.abstract import TemporalIndicator
+from execution_engine.converter.relative_time.abstract import RelativeTime
+from execution_engine.converter.time_from_event.abstract import TimeFromEvent
 
 if TYPE_CHECKING:
     from execution_engine.execution_engine import ExecutionEngine
@@ -42,7 +43,8 @@ class CriterionConverterType(TypedDict):
     characteristic: list[type[CriterionConverter]]
     action: list[type[CriterionConverter]]
     goal: list[type[CriterionConverter]]
-    time_from_event: list[type[TemporalIndicator]]
+    time_from_event: list[type[TimeFromEvent]]
+    relative_time: list[type[RelativeTime]]
 
 
 _default_converters: CriterionConverterType = {
@@ -63,6 +65,7 @@ class CriterionConverterType(TypedDict):
     ],
     "goal": [LaboratoryValueGoal, VentilatorManagementGoal, AssessmentScaleGoal],
     "time_from_event": [],
+    "relative_time": [],
 }
 
 
@@ -76,6 +79,7 @@ def default_execution_engine_builder() -> "ExecutionEngineBuilder":
     builder.set_action_converters(_default_converters["action"])
     builder.set_goal_converters(_default_converters["goal"])
     builder.set_time_from_event_converters(_default_converters["time_from_event"])
+    builder.set_relative_time_converters(_default_converters["relative_time"])
 
     return builder
 
@@ -92,7 +96,8 @@ def __init__(self) -> None:
         self.characteristic_converters: list[type[CriterionConverter]] = []
         self.action_converters: list[type[CriterionConverter]] = []
         self.goal_converters: list[type[CriterionConverter]] = []
-        self.time_from_event_converters: list[type[TemporalIndicator]] = []
+        self.time_from_event_converters: list[type[TimeFromEvent]] = []
+        self.relative_time_converters: list[type[RelativeTime]] = []
 
     def set_characteristic_converters(
         self, converters: list[type[CriterionConverter]]
@@ -128,7 +133,7 @@ def set_goal_converters(
         return self
 
     def set_time_from_event_converters(
-        self, converters: list[type[TemporalIndicator]]
+        self, converters: list[type[TimeFromEvent]]
     ) -> "ExecutionEngineBuilder":
         """
         Sets (overwrites) the time from event converters for this builder.
@@ -140,6 +145,19 @@ def set_time_from_event_converters(
 
         return self
 
+    def set_relative_time_converters(
+        self, converters: list[type[RelativeTime]]
+    ) -> "ExecutionEngineBuilder":
+        """
+        Sets (overwrites) the time from event converters for this builder.
+        """
+        self.relative_time_converters.clear()
+
+        for converter_type in converters:
+            self.append_relative_time_converter(converter_type)
+
+        return self
+
     def append_characteristic_converter(
         self, converter_type: type[CriterionConverter]
     ) -> "ExecutionEngineBuilder":
@@ -207,27 +225,49 @@ def prepend_goal_converter(
         return self
 
     def append_time_from_event_converter(
-        self, converter_type: type[TemporalIndicator]
+        self, converter_type: type[TimeFromEvent]
     ) -> "ExecutionEngineBuilder":
         """
         Appends a single time_from_event converter at the end of the list.
         """
-        if not issubclass(converter_type, TemporalIndicator):
+        if not issubclass(converter_type, TimeFromEvent):
             raise ValueError(f"Invalid TimeFromEvent converter type: {converter_type}")
         self.time_from_event_converters.append(converter_type)
         return self
 
     def prepend_time_from_event_converter(
-        self, converter_type: type[TemporalIndicator]
+        self, converter_type: type[TimeFromEvent]
     ) -> "ExecutionEngineBuilder":
         """
         Inserts a single time_from_event converter at the front of the list.
         """
-        if not issubclass(converter_type, TemporalIndicator):
+        if not issubclass(converter_type, TimeFromEvent):
             raise ValueError(f"Invalid TimeFromEvent converter type: {converter_type}")
         self.time_from_event_converters.insert(0, converter_type)
         return self
 
+    def append_relative_time_converter(
+        self, converter_type: type[RelativeTime]
+    ) -> "ExecutionEngineBuilder":
+        """
+        Appends a single relative_time converter at the end of the list.
+        """
+        if not issubclass(converter_type, RelativeTime):
+            raise ValueError(f"Invalid TimeFromEvent converter type: {converter_type}")
+        self.relative_time_converters.append(converter_type)
+        return self
+
+    def prepend_relative_time_converter(
+        self, converter_type: type[RelativeTime]
+    ) -> "ExecutionEngineBuilder":
+        """
+        Inserts a single relative_time converter at the front of the list.
+        """
+        if not issubclass(converter_type, RelativeTime):
+            raise ValueError(f"Invalid TimeFromEvent converter type: {converter_type}")
+        self.relative_time_converters.insert(0, converter_type)
+        return self
+
     def build(self, verbose: bool = False) -> "ExecutionEngine":
         """
         Builds an ExecutionEngine with the specified converters.

execution_engine/constants.py

@@ -26,6 +26,7 @@
 EXT_DOSAGE_CONDITION = "https://www.netzwerk-universitaetsmedizin.de/fhir/cpg-on-ebm-on-fhir/StructureDefinition/ext-dosage-condition"
 EXT_ACTION_COMBINATION_METHOD = "https://www.netzwerk-universitaetsmedizin.de/fhir/cpg-on-ebm-on-fhir/StructureDefinition/ext-action-combination-method"
 EXT_CPG_PARTOF = "http://hl7.org/fhir/uv/cpg/StructureDefinition/cpg-partOf"
+EXT_RELATIVE_TIME = "https://www.netzwerk-universitaetsmedizin.de/fhir/cpg-on-ebm-on-fhir/StructureDefinition/relative-time"
 
 CS_ACTION_COMBINATION_METHOD = "https://www.netzwerk-universitaetsmedizin.de/fhir/cpg-on-ebm-on-fhir/CodeSystem/cs-action-combination-method"
 

execution_engine/converter/action/abstract.py

@@ -3,16 +3,14 @@
 
 from fhir.resources.timing import Timing as FHIRTiming
 
+from execution_engine import constants
 from execution_engine.converter.criterion import CriterionConverter, parse_code
 from execution_engine.converter.goal.abstract import Goal
 from execution_engine.fhir.recommendation import RecommendationPlan
-from execution_engine.fhir.util import get_coding
+from execution_engine.fhir.util import get_coding, get_extensions
 from execution_engine.omop.criterion.abstract import Criterion
-from execution_engine.omop.criterion.combination.logical import (
-    LogicalCriterionCombination,
-)
 from execution_engine.omop.vocabulary import AbstractVocabulary
-from execution_engine.util import AbstractPrivateMethods
+from execution_engine.util import AbstractPrivateMethods, logic
 from execution_engine.util.types import Timing
 from execution_engine.util.value.time import ValueCount, ValueDuration, ValuePeriod
 
@@ -139,32 +137,40 @@ def process_timing(cls, timing: FHIRTiming) -> Timing:
         if rep.offset is not None:
             raise NotImplementedError("offset has not been implemented")
 
+        relative_time = get_extensions(timing, constants.EXT_RELATIVE_TIME)
+
+        if relative_time:
+            raise NotImplementedError(
+                "RelativeTime processing within AbstractAction not implemented - "
+                "should be performed in the parser"
+            )
+
         return Timing(
             count=count, duration=duration, frequency=frequency, interval=interval
         )
 
     @abstractmethod
-    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
+    def _to_expression(self) -> logic.BaseExpr | None:
         """Converts this action to a Criterion."""
         raise NotImplementedError()
 
     @final
-    def to_positive_criterion(self) -> Criterion | LogicalCriterionCombination:
+    def to_positive_expression(self) -> logic.BaseExpr:
         """
         Converts this action to a criterion.
         """
-        action = self._to_criterion()
+        action = self._to_expression()
 
         if action is None:
             assert (
                 self.goals
             ), "Action without explicit criterion must have at least one goal"
 
         if self.goals:
-            criteria = [goal.to_criterion() for goal in self.goals]
+            criteria = [goal.to_expression() for goal in self.goals]
             if action is not None:
                 criteria.append(action)
-            return LogicalCriterionCombination.And(*criteria)
+            return logic.And(*criteria)
         else:
             return action  # type: ignore
 

execution_engine/converter/action/body_positioning.py

@@ -4,12 +4,9 @@
 from execution_engine.converter.criterion import parse_code
 from execution_engine.fhir.recommendation import RecommendationPlan
 from execution_engine.omop.concepts import Concept
-from execution_engine.omop.criterion.abstract import Criterion
-from execution_engine.omop.criterion.combination.logical import (
-    LogicalCriterionCombination,
-)
 from execution_engine.omop.criterion.procedure_occurrence import ProcedureOccurrence
 from execution_engine.omop.vocabulary import SNOMEDCT
+from execution_engine.util import logic
 from execution_engine.util.types import Timing
 
 
@@ -48,7 +45,7 @@ def from_fhir(cls, action_def: RecommendationPlan.Action) -> Self:
 
         return cls(exclude=exclude, code=code, timing=timing)
 
-    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
+    def _to_expression(self) -> logic.Symbol:
         """Converts this characteristic to a Criterion."""
 
         return ProcedureOccurrence(

execution_engine/converter/action/drug_administration.py

@@ -15,19 +15,14 @@
 )
 from execution_engine.fhir.recommendation import RecommendationPlan
 from execution_engine.omop.concepts import Concept
-from execution_engine.omop.criterion.abstract import Criterion
-from execution_engine.omop.criterion.combination.combination import CriterionCombination
-from execution_engine.omop.criterion.combination.logical import (
-    LogicalCriterionCombination,
-    NonCommutativeLogicalCriterionCombination,
-)
 from execution_engine.omop.criterion.drug_exposure import DrugExposure
 from execution_engine.omop.criterion.point_in_time import PointInTimeCriterion
 from execution_engine.omop.vocabulary import (
     SNOMEDCT,
     VocabularyFactory,
     standard_vocabulary,
 )
+from execution_engine.util import logic
 from execution_engine.util.types import Dosage
 from execution_engine.util.value import Value, ValueNumber
 
@@ -278,12 +273,12 @@ def filter_same_unit(cls, df: pd.DataFrame, unit: Concept) -> pd.DataFrame:
 
         return df_filtered
 
-    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
+    def _to_expression(self) -> logic.BaseExpr:
         """
         Returns a criterion that represents this action.
         """
 
-        drug_actions: list[Criterion | LogicalCriterionCombination] = []
+        drug_actions: list[logic.BaseExpr] = []
 
         if not self._dosages:
             # no dosages, just return the drug exposure
@@ -325,19 +320,18 @@ def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
                 # rational: "conditional" extensions are some conditions for dosage, such as body weight ranges.
                 # Thus, the actual drug administration (drug_action, "right") must only be fulfilled if the
                 # condition (ext_criterion, "left") is fulfilled. Thus, we here add this conditional filter.
-                comb = NonCommutativeLogicalCriterionCombination.ConditionalFilter(
+                comb = logic.ConditionalFilter(
                     left=ext_criterion,
                     right=drug_action,
                 )
 
                 drug_actions.append(comb)
 
-        result: Criterion | CriterionCombination
+        result: logic.BaseExpr
+
         if len(drug_actions) == 1:
             result = drug_actions[0]
         else:
-            result = LogicalCriterionCombination(
-                operator=LogicalCriterionCombination.Operator("OR"),
-            )
-            result.add_all(drug_actions)
+            result = logic.Or(*drug_actions)
+
         return result