refactor: remove criterion combination

Author: glichtner

.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

execution_engine/converter/action/abstract.py

@@ -8,11 +8,8 @@
 from execution_engine.fhir.recommendation import RecommendationPlan
 from execution_engine.fhir.util import get_coding
 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
 
@@ -144,27 +141,27 @@ def process_timing(cls, timing: FHIRTiming) -> Timing:
         )
 
     @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,10 +45,12 @@ 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(
-            concept=self._code,
-            timing=self._timing,
+        return logic.Symbol(
+            criterion=ProcedureOccurrence(
+                concept=self._code,
+                timing=self._timing,
+            )
         )

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,26 +273,30 @@ 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
-            return DrugExposure(
-                ingredient_concept=self._ingredient_concept,
-                dose=None,
-                route=None,
+            return logic.Symbol(
+                criterion=DrugExposure(
+                    ingredient_concept=self._ingredient_concept,
+                    dose=None,
+                    route=None,
+                )
             )
 
         for dosage in self._dosages:
-            drug_action = DrugExposure(
-                ingredient_concept=self._ingredient_concept,
-                dose=dosage["dose"],
-                route=dosage.get("route", None),
+            drug_action = logic.Symbol(
+                criterion=DrugExposure(
+                    ingredient_concept=self._ingredient_concept,
+                    dose=dosage["dose"],
+                    route=dosage.get("route", None),
+                )
             )
 
             extensions = dosage.get("extensions", None)
@@ -316,28 +315,29 @@ def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
                         f"Extension type {extension['type']} not supported yet"
                     )
 
-                ext_criterion = PointInTimeCriterion(
-                    concept=extension["code"],
-                    value=extension["value"],
+                ext_criterion = logic.Symbol(
+                    criterion=PointInTimeCriterion(
+                        concept=extension["code"],
+                        value=extension["value"],
+                    )
                 )
 
                 # A Conditional Filter returns `right` iff left is POSITIVE, otherwise it returns NEGATIVE
                 # 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

execution_engine/converter/action/procedure.py

@@ -3,13 +3,11 @@
 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.measurement import Measurement
 from execution_engine.omop.criterion.observation import Observation
 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
 
 
@@ -57,7 +55,7 @@ def from_fhir(cls, action_def: RecommendationPlan.Action) -> AbstractAction:
 
         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."""
 
         criterion: Criterion
@@ -89,4 +87,4 @@ def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
                     f"Concept domain {self._code.domain_id} is not supported for {self.__class__.__name__}]"
                 )
 
-        return criterion
+        return logic.Symbol(criterion=criterion)

execution_engine/converter/action/ventilator_management.py

@@ -2,10 +2,6 @@
 
 from execution_engine.converter.action.abstract import AbstractAction
 from execution_engine.fhir.recommendation import RecommendationPlan
-from execution_engine.omop.criterion.abstract import Criterion
-from execution_engine.omop.criterion.combination.logical import (
-    LogicalCriterionCombination,
-)
 from execution_engine.omop.vocabulary import SNOMEDCT
 
 
@@ -29,6 +25,6 @@ def from_fhir(cls, action_def: RecommendationPlan.Action) -> Self:
             exclude=False,
         )  # fixme: no way to exclude goals (e.g. "do not ventilate")
 
-    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
+    def _to_expression(self) -> None:
         """Converts this characteristic to a Criterion."""
         return None

execution_engine/converter/characteristic/abstract.py

@@ -8,6 +8,7 @@
 from execution_engine.omop.concepts import Concept
 from execution_engine.omop.criterion.abstract import Criterion
 from execution_engine.omop.vocabulary import standard_vocabulary
+from execution_engine.util import logic as logic
 from execution_engine.util.value import Value
 
 
@@ -30,7 +31,7 @@ class AbstractCharacteristic(CriterionConverter, ABC):
     Subclasses must define the following methods:
     - valid: returns True if the supplied characteristic falls within the scope of the subclass
     - from_fhir: creates a new instance of the subclass from a FHIR EvidenceVariable.characteristic element
-    - to_criterion(): converts the characteristic to a Criterion
+    - to_expression(): converts the characteristic to a Criterion
     """
 
     _criterion_class: Type[Criterion]
@@ -82,6 +83,13 @@ def valid(
         """Checks if the given FHIR EvidenceVariable is a valid characteristic."""
         raise NotImplementedError()
 
+    @abstractmethod
+    def to_positive_expression(self) -> logic.BaseExpr:
+        """
+        Converts this characteristic to a positive expression (i.e. neglecting the exlude flag).
+        """
+        raise NotImplementedError()
+
     @staticmethod
     def get_standard_concept(cc: Coding) -> Concept:
         """
@@ -95,13 +103,3 @@ def get_concept(cc: Coding, standard: bool = True) -> Concept:
         Get the OMOP Standard Vocabulary standard concept for the given code in the given vocabulary.
         """
         return standard_vocabulary.get_concept(cc.system, cc.code, standard=standard)
-
-    @abstractmethod
-    def to_positive_criterion(self) -> Criterion:
-        """
-        Converts this characteristic to a "Positive" Criterion.
-
-        Positive criterion means that a possible excluded flag is disregarded. Instead, the exclusion
-        is later introduced (in the to_criterion() method) via a LogicalCriterionCombination.Not).
-        """
-        raise NotImplementedError()

execution_engine/converter/characteristic/codeable_concept.py

@@ -6,9 +6,9 @@
 from execution_engine.converter.characteristic.abstract import AbstractCharacteristic
 from execution_engine.fhir.util import get_coding
 from execution_engine.omop.concepts import Concept
-from execution_engine.omop.criterion.abstract import Criterion
 from execution_engine.omop.criterion.concept import ConceptCriterion
 from execution_engine.omop.vocabulary import AbstractVocabulary
+from execution_engine.util import logic
 
 
 class AbstractCodeableConceptCharacteristic(AbstractCharacteristic):
@@ -78,10 +78,12 @@ def from_fhir(
 
         return c
 
-    def to_positive_criterion(self) -> Criterion:
+    def to_positive_expression(self) -> logic.BaseExpr:
         """Converts this characteristic to a Criterion."""
-        return self._criterion_class(
-            concept=self.value,
-            value=None,
-            static=self._concept_value_static,
+        return logic.Symbol(
+            criterion=self._criterion_class(
+                concept=self.value,
+                value=None,
+                static=self._concept_value_static,
+            )
         )

execution_engine/converter/characteristic/value.py

@@ -6,6 +6,7 @@
 from execution_engine.converter.characteristic.abstract import AbstractCharacteristic
 from execution_engine.converter.criterion import parse_code, parse_value
 from execution_engine.omop.criterion.concept import ConceptCriterion
+from execution_engine.util import logic
 
 
 class AbstractValueCharacteristic(AbstractCharacteristic, ABC):
@@ -32,10 +33,12 @@ def from_fhir(
 
         return c
 
-    def to_positive_criterion(self) -> ConceptCriterion:
+    def to_positive_expression(self) -> logic.Symbol:
         """Converts this characteristic to a Criterion."""
-        return self._criterion_class(
-            concept=self.type,
-            value=self.value,
-            static=self._concept_value_static,
+        return logic.Symbol(
+            criterion=self._criterion_class(
+                concept=self.type,
+                value=self.value,
+                static=self._concept_value_static,
+            )
         )

execution_engine/converter/criterion.py

@@ -1,6 +1,6 @@
 import json
 from abc import ABC, abstractmethod
-from typing import Tuple, Type
+from typing import Tuple, Type, final
 
 from fhir.resources.codeableconcept import CodeableConcept
 from fhir.resources.element import Element
@@ -11,11 +11,8 @@
 
 from execution_engine.fhir.util import get_coding
 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.vocabulary import standard_vocabulary
+from execution_engine.util import logic as logic
 from execution_engine.util.value import Value, ValueConcept, ValueNumber
 from execution_engine.util.value.value import ValueScalar
 
@@ -186,22 +183,23 @@ def valid(cls, fhir_definition: Element) -> bool:
         raise NotImplementedError()
 
     @abstractmethod
-    def to_positive_criterion(self) -> Criterion | LogicalCriterionCombination:
+    def to_positive_expression(self) -> logic.BaseExpr:
         """Converts this characteristic to a Criterion or a combination of criteria but no negation."""
         raise NotImplementedError()
 
-    def to_criterion(self) -> Criterion | LogicalCriterionCombination:
+    @final
+    def to_expression(self) -> logic.BaseExpr:
         """
-        Converts this characteristic to a Criterion or a
-        combination of criteria. The result may be a "negative"
-        criterion, that is the result of to_positive_criterion wrapped
-        in a LogicalCriterionCombination with operator NOT.
+        Converts this characteristic to an expression. The result may be a "negative"
+        criterion, that is the result of to_positive_expression wrapped
+        in a logic.Not.
         """
-        positive_criterion = self.to_positive_criterion()
+        positive_expression = self.to_positive_expression()
+
         if self._exclude:
-            return LogicalCriterionCombination.Not(positive_criterion)
+            return logic.Not(positive_expression)
         else:
-            return positive_criterion
+            return positive_expression
 
 
 class CriterionConverterFactory: