feat: add procedure action

Author: glichtner

execution_engine/builder.py

@@ -6,6 +6,7 @@
 from execution_engine.converter.action.drug_administration import (
     DrugAdministrationAction,
 )
+from execution_engine.converter.action.procedure import ProcedureAction
 from execution_engine.converter.action.ventilator_management import (
     VentilatorManagementAction,
 )
@@ -58,6 +59,7 @@ class CriterionConverterType(TypedDict):
         VentilatorManagementAction,
         BodyPositioningAction,
         AssessmentAction,
+        ProcedureAction,
     ],
     "goal": [LaboratoryValueGoal, VentilatorManagementGoal, AssessmentScaleGoal],
     "time_from_event": [],

execution_engine/converter/action/assessment.py

@@ -1,22 +1,8 @@
-from typing import Self, Type
-
-from execution_engine.converter.action.abstract import AbstractAction
-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.concept import ConceptCriterion
-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.converter.action.procedure import ProcedureAction
 from execution_engine.omop.vocabulary import SNOMEDCT
-from execution_engine.util.types import Timing
 
 
-class AssessmentAction(AbstractAction):
+class AssessmentAction(ProcedureAction):
     """
     An AssessmentAction is an action that is used to assess a patient's condition.
 
@@ -26,65 +12,3 @@ class AssessmentAction(AbstractAction):
 
     _concept_code = "386053000"  # Evaluation procedure (procedure)
     _concept_vocabulary = SNOMEDCT
-
-    def __init__(
-        self,
-        exclude: bool,
-        code: Concept,
-        timing: Timing | None = None,
-    ) -> None:
-        """
-        Initialize the assessment action.
-        """
-        super().__init__(exclude=exclude)
-        self._code = code
-        self._timing = timing
-
-    @classmethod
-    def from_fhir(cls, action_def: RecommendationPlan.Action) -> Self:
-        """Creates a new action from a FHIR PlanDefinition."""
-        assert (
-            action_def.activity_definition_fhir is not None
-        ), "ActivityDefinition is required"
-        assert (
-            action_def.activity_definition_fhir.code is not None
-        ), "Code is required for AssessmentAction"
-        assert (
-            action_def.activity_definition_fhir.timingTiming is not None
-        ), "Timing is required for AssessmentAction"
-
-        code = parse_code(action_def.activity_definition_fhir.code)
-        timing = cls.process_timing(action_def.activity_definition_fhir.timingTiming)
-
-        exclude = action_def.activity_definition_fhir.doNotPerform
-
-        return cls(exclude=exclude, code=code, timing=timing)
-
-    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
-        """Converts this characteristic to a Criterion."""
-
-        cls: Type[ConceptCriterion]
-
-        match self._code.domain_id:
-            case "Procedure":
-                cls = ProcedureOccurrence
-            case "Measurement":
-                cls = Measurement
-            case "Observation":
-                cls = Observation
-            case _:
-                raise ValueError(
-                    f"Concept domain {self._code.domain_id} is not supported for AssessmentAction"
-                )
-
-        criterion = cls(
-            concept=self._code,
-            timing=self._timing,
-        )
-
-        # we need to explicitly set the VALUE_REQUIRED flag to false after creation of the object,
-        # otherwise creating the query will raise an error as Observation and Measurement normally expect
-        # a value.
-        criterion._OMOP_VALUE_REQUIRED = False
-
-        return criterion

execution_engine/converter/action/procedure.py

@@ -0,0 +1,90 @@
+from typing import Self, Type
+
+from execution_engine.converter.action.abstract import AbstractAction
+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.concept import ConceptCriterion
+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.types import Timing
+
+
+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
+    is present in the respective OMOP CDM table.
+    """
+
+    _concept_code = "71388002"  # Procedure (procedure)
+    _concept_vocabulary = SNOMEDCT
+
+    def __init__(
+        self,
+        exclude: bool,
+        code: Concept,
+        timing: Timing | None = None,
+    ) -> None:
+        """
+        Initialize the procedure action.
+        """
+        super().__init__(exclude=exclude)
+        self._code = code
+        self._timing = timing
+
+    @classmethod
+    def from_fhir(cls, action_def: RecommendationPlan.Action) -> Self:
+        """Creates a new action from a FHIR PlanDefinition."""
+        assert (
+            action_def.activity_definition_fhir is not None
+        ), f"ActivityDefinition is required for {cls.__class__.__name__}"
+        assert (
+            action_def.activity_definition_fhir.code is not None
+        ), f"Code is required for {cls.__class__.__name__}"
+        assert (
+            action_def.activity_definition_fhir.timingTiming is not None
+        ), f"Timing is required for {cls.__class__.__name__}"
+
+        code = parse_code(action_def.activity_definition_fhir.code)
+        timing = cls.process_timing(action_def.activity_definition_fhir.timingTiming)
+
+        exclude = action_def.activity_definition_fhir.doNotPerform
+
+        return cls(exclude=exclude, code=code, timing=timing)
+
+    def _to_criterion(self) -> Criterion | LogicalCriterionCombination | None:
+        """Converts this characteristic to a Criterion."""
+
+        cls: Type[ConceptCriterion]
+
+        match self._code.domain_id:
+            case "Procedure":
+                cls = ProcedureOccurrence
+            case "Measurement":
+                cls = Measurement
+            case "Observation":
+                cls = Observation
+            case _:
+                raise ValueError(
+                    f"Concept domain {self._code.domain_id} is not supported for {self.__class__.__name__}]"
+                )
+
+        criterion = cls(
+            concept=self._code,
+            timing=self._timing,
+        )
+
+        # we need to explicitly set the VALUE_REQUIRED flag to false after creation of the object,
+        # otherwise creating the query will raise an error as Observation and Measurement normally expect
+        # a value.
+        criterion._OMOP_VALUE_REQUIRED = False
+
+        return criterion

execution_engine/fhir/recommendation.py

@@ -182,9 +182,9 @@ def __init__(
             self._action: fhir.resources.plandefinition.PlanDefinitionAction = (
                 action_def
             )
-            self._activity: fhir.resources.activitydefinition.ActivityDefinition | None = (
-                None
-            )
+            self._activity: (
+                fhir.resources.activitydefinition.ActivityDefinition | None
+            ) = None
             self._package_version = package_version
 
             # an action must not necessarily contain an activity definition (e.g. ventilator management)
@@ -234,6 +234,12 @@ def goals_fhir(
             """Get the goals for this action."""
             return self._goals
 
+        def model_dump(self) -> dict:
+            """
+            Return the FHIR PlanDefinitionAction model dump
+            """
+            return self.fhir().model_dump()
+
     def __init__(
         self, canonical_url: str, package_version: str, fhir_connector: FHIRClient
     ):

execution_engine/omop/sqlclient.py

@@ -328,6 +328,9 @@ def get_concept(
         """
         logging.info(f"Requesting standard concept: {vocabulary} #{code}")
 
+        if code is None:
+            raise ValueError("Code must be set")
+
         concept = omop.Concept.__table__.alias("c")
         query = concept.select().where(
             and_(

execution_engine/omop/vocabulary.py

@@ -232,7 +232,7 @@ def get(self, system_uri: str) -> AbstractVocabulary:
         if system_uri in self._vocabulary:
             return self._vocabulary[system_uri]
         else:
-            raise VocabularyNotFoundError(f"Vocabulary {system_uri} not found")
+            raise VocabularyNotFoundError(f'Vocabulary "{system_uri}" not found')
 
 
 class StandardVocabulary: