feat: implement relativeTime extension

Author: glichtner

.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

converter/parser/util.py

@@ -0,0 +1,84 @@
+import logging
+from typing import Callable, cast
+
+from execution_engine.omop.criterion.abstract import Criterion
+from execution_engine.omop.criterion.procedure_occurrence import ProcedureOccurrence
+from execution_engine.util import logic
+from execution_engine.util.temporal_logic_util import Presence
+
+
+def _wrap_criteria_with_factory(
+    expr: logic.BaseExpr,
+    factory: Callable[[logic.BaseExpr], logic.TemporalCount],
+) -> logic.Expr:
+    """
+    Recursively wraps all Criterion instances within a combination using the specified factory.
+
+    :param expr: A single Criterion or an expression to be processed.
+    :param factory: A callable that takes a Criterion or expression and returns a TemporalCount.
+    :return: A new TemporalCount where all Criterion instances have been wrapped using the factory.
+    :raises ValueError: If an unexpected element type is encountered.
+    """
+
+    from digipod.concepts import OMOP_SURGICAL_PROCEDURE
+
+    new_expr: logic.Expr
+
+    if isinstance(expr, Criterion):
+        new_expr = factory(expr)
+    elif isinstance(expr, logic.Expr):
+
+        # Create a new combination of the same type with the same operator
+        args = []
+
+        # Loop through all elements
+        for element in expr.args:
+            if isinstance(element, logic.Expr):
+                # Recursively wrap nested combinations
+                args.append(_wrap_criteria_with_factory(element, factory))
+            elif isinstance(element, Criterion):
+                # Wrap individual criteria with the factory
+
+                if (
+                    isinstance(element, ProcedureOccurrence)
+                    and element.concept.concept_id == OMOP_SURGICAL_PROCEDURE
+                    and element.concept.vocabulary_id == "SNOMED"
+                ):
+                    logging.warning(
+                        "Removing Surgical Procedure Criterion in TimeFromEvent-SurgicalOperationDate"
+                    )
+                    continue
+
+                args.append(factory(element))
+
+            else:
+                raise ValueError(f"Unexpected element type: {type(element)}")
+
+        new_expr = expr.__class__(*args)
+    else:
+        raise ValueError(f"Unexpected element type: {type(expr)}")
+
+    return new_expr
+
+
+def wrap_criteria_with_temporal_indicator(
+    expr: logic.BaseExpr,
+    interval_criterion: logic.BaseExpr,
+) -> logic.TemporalMinCount:
+    """
+    Wraps all Criterion instances in a combination with a TemporalCount (with interval_criterion).
+
+    :param expr: A single Criterion or an expression to be wrapped.
+    :param interval_criterion: A Criterion or expression that defines the temporal interval.
+    :return: A new expression where all Criterion instances are wrapped with a TemporalCount (with interval_criterion).
+    """
+    temporal_combo_factory = lambda criterion: Presence(
+        criterion=criterion, interval_criterion=interval_criterion
+    )
+
+    new_combo = cast(
+        logic.TemporalMinCount,
+        _wrap_criteria_with_factory(expr, temporal_combo_factory),
+    )
+
+    return new_combo

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,10 +3,11 @@
 
 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.vocabulary import AbstractVocabulary
 from execution_engine.util import AbstractPrivateMethods, logic
@@ -136,6 +137,14 @@ 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
         )

execution_engine/converter/action/procedure.py

@@ -3,6 +3,8 @@
 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.condition_occurrence import ConditionOccurrence
+from execution_engine.omop.criterion.device_exposure import DeviceExposure
 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
@@ -15,7 +17,7 @@ class ProcedureAction(AbstractAction):
     """
     An ProcedureAction is an action that describes a procedure to be performed
 
-    This action tests whether the procedure has been performed by determining whether it is
+    This action tests whether the procedure has been performed by determining whether it
     is present in the respective OMOP CDM table.
     """
 
@@ -82,6 +84,18 @@ def _to_expression(self) -> logic.Symbol:
                     value_required=False,
                     timing=self._timing,
                 )
+            case "Device":
+                criterion = DeviceExposure(
+                    concept=self._code,
+                    value_required=False,
+                    timing=self._timing,
+                )
+            case "Condition":
+                criterion = ConditionOccurrence(
+                    concept=self._code,
+                    value_required=False,
+                    timing=self._timing,
+                )
             case _:
                 raise ValueError(
                     f"Concept domain {self._code.domain_id} is not supported for {self.__class__.__name__}]"

execution_engine/converter/parser/base.py

@@ -1,7 +1,12 @@
 from abc import ABC, abstractmethod
 from typing import Callable, Type
 
-from fhir.resources.evidencevariable import EvidenceVariable
+from fhir.resources.evidencevariable import (
+    EvidenceVariable,
+    EvidenceVariableCharacteristic,
+    EvidenceVariableCharacteristicTimeFromEvent,
+)
+from fhir.resources.extension import Extension
 from fhir.resources.plandefinition import PlanDefinition, PlanDefinitionAction
 
 from execution_engine import fhir
@@ -19,11 +24,13 @@ def __init__(
         action_converters: CriterionConverterFactory,
         goal_converters: CriterionConverterFactory,
         time_from_event_converters: TemporalIndicatorConverterFactory,
+        relative_time_converters: TemporalIndicatorConverterFactory,
     ):
         self.characteristics_converters = characteristic_converters
         self.action_converters = action_converters
         self.goal_converters = goal_converters
         self.time_from_event_converters = time_from_event_converters
+        self.relative_time_converters = relative_time_converters
 
     @abstractmethod
     def parse_characteristics(self, ev: EvidenceVariable) -> logic.BooleanFunction:
@@ -53,3 +60,31 @@ def parse_action_combination_method(
         combination method in form of a logical expression.
         """
         raise NotImplementedError()
+
+    def parse_time_from_event(
+        self,
+        tfes: list[EvidenceVariableCharacteristicTimeFromEvent],
+    ) -> list[logic.BaseExpr]:
+        """
+        Parses `timeFromEvent` elements and converts them into interval-based logical criteria.
+        """
+        raise NotImplementedError()
+
+    @abstractmethod
+    def parse_relative_time(
+        self,
+        relative_time: list[Extension],
+    ) -> list[logic.BaseExpr]:
+        """
+        Parses `extension[relativeTime]` elements and converts them into interval-based logical criteria.
+        """
+        raise NotImplementedError()
+
+    def parse_timing(
+        self, characteristic: EvidenceVariableCharacteristic, expr: logic.BaseExpr
+    ) -> logic.BaseExpr:
+        """
+        Applies temporal constraints to a given criterion expression based on `timeFromEvent` and
+        the relativeTime extension elements.
+        """
+        raise NotImplementedError()