Merge pull request #93 from aidh-ms/issue/76-rework-mimic-dosage-defi…

…nition Rework MIMIC
aidh-ms · Oct 4, 2024 · 0e33f19 · 0e33f19
2 parents 1090754 + 1f8a7d2
commit 0e33f19
Show file tree

Hide file tree

Showing 6 changed files with 210 additions and 23 deletions.
diff --git a/conceptbase b/conceptbase
diff --git a/icu_pipeline/unit/frequency.py b/icu_pipeline/unit/frequency.py
@@ -6,10 +6,15 @@
 
 class FrequencyConverter(BaseConverter):
     SI_UNIT = "Hz"
-    AVAILABLE_UNITS = ["Hz", "bpm"]
+    AVAILABLE_UNITS = ["Hz", "bpm", "1/min"]
     # REQUIRED_CONCEPTS = ["SystolicBloodPressure"]
 
-    def _convertToSI(self, source_unit: str, data: Series[Quantity], dependencies: dict[str,DataFrame]):
+    def _convertToSI(
+        self,
+        source_unit: str,
+        data: Series[Quantity],
+        dependencies: dict[str, DataFrame],
+    ):
         convert: Callable[[float], float] = lambda v: v
         # Data can use any unit and will be transformed to Hz
         match source_unit:
@@ -19,17 +24,22 @@ def _convertToSI(self, source_unit: str, data: Series[Quantity], dependencies: d
 
             # Actual Conversions
             case "bpm":
-                convert = lambda v: v / 60 # bpm = 60 * Hz
+                convert = lambda v: v / 60  # bpm = 60 * Hz
+
+            case "1/min":
+                convert = lambda v: v / 60
 
             # Not Implemented
             case _:
                 raise NotImplementedError
 
-        return data.apply(lambda q: Quantity(
-            value=convert(q["value"]),
-            unit=self.SI_UNIT))
+        return data.apply(
+            lambda q: Quantity(value=convert(q["value"]), unit=self.SI_UNIT)
+        )
 
-    def _convertToTarget(self, sink_unit: str, data: Series[Quantity], dependencies: dict[str,DataFrame]):
+    def _convertToTarget(
+        self, sink_unit: str, data: Series[Quantity], dependencies: dict[str, DataFrame]
+    ):
         convert: Callable[[float], float] = lambda v: v
         # Data contains Hz values and can be transformed into any Unit
         match sink_unit:
@@ -39,12 +49,13 @@ def _convertToTarget(self, sink_unit: str, data: Series[Quantity], dependencies:
 
             # Actual Conversions
             case "bpm":
-                convert = lambda v: v * 60 # 60 Seconds in 1 Minute
+                convert = lambda v: v * 60  # 60 Seconds in 1 Minute
+
+            case "1/min":
+                convert = lambda v: v * 60
 
             # Not Implemented
             case _:
                 raise NotImplementedError
 
-        return data.apply(lambda q: Quantity(
-            value=convert(q["value"]),
-            unit=sink_unit))
+        return data.apply(lambda q: Quantity(value=convert(q["value"]), unit=sink_unit))
diff --git a/icu_pipeline/unit/length.py b/icu_pipeline/unit/length.py
@@ -0,0 +1,54 @@
+from typing import Callable
+from pandera.typing import Series, DataFrame
+from icu_pipeline.schema.fhir import Quantity
+from icu_pipeline.unit.converter import BaseConverter
+
+
+class LengthConverter(BaseConverter):
+    SI_UNIT = "m"
+    AVAILABLE_UNITS = ["cm", "m"]
+
+    def _convertToSI(
+        self,
+        source_unit: str,
+        data: Series[Quantity],
+        dependencies: dict[str, DataFrame],
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data can have any Unit and will be transformed to m
+        match source_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions from case to cm
+            case "cm":
+                convert = lambda v: v * 100
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(
+            lambda q: Quantity(value=convert(q["value"]), unit=self.SI_UNIT)
+        )
+
+    def _convertToTarget(
+        self, sink_unit: str, data: Series[Quantity], dependencies: dict[str, DataFrame]
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data uses m and can be transformed in to any Unit
+        match sink_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions
+            case "cm":
+                convert = lambda v: v / 100
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(lambda q: Quantity(value=convert(q["value"]), unit=sink_unit))
diff --git a/icu_pipeline/unit/none.py b/icu_pipeline/unit/none.py
@@ -0,0 +1,57 @@
+from typing import Callable
+from pandera.typing import Series, DataFrame
+from icu_pipeline.schema.fhir import Quantity
+from icu_pipeline.unit.converter import BaseConverter
+
+
+class NoUnitConverter(BaseConverter):
+    SI_UNIT = ""
+    AVAILABLE_UNITS = [
+        ""
+    ]  # year is amiguous (365, 366 days). TODO - Ignore since error is small and mimic does it anyway?
+    # REQUIRED_CONCEPTS = ["SystolicBloodPressure"]
+
+    def _convertToSI(
+        self,
+        source_unit: str,
+        data: Series[Quantity],
+        dependencies: dict[str, DataFrame],
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data can use any unit and will be transformed to Hz
+        match source_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions
+            case "":
+                convert = lambda v: v
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(
+            lambda q: Quantity(value=convert(q["value"]), unit=self.SI_UNIT)
+        )
+
+    def _convertToTarget(
+        self, sink_unit: str, data: Series[Quantity], dependencies: dict[str, DataFrame]
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data contains Hz values and can be transformed into any Unit
+        match sink_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions
+            case "":
+                convert = lambda v: v
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(lambda q: Quantity(value=convert(q["value"]), unit=sink_unit))
diff --git a/icu_pipeline/unit/quantity.py b/icu_pipeline/unit/quantity.py
@@ -5,10 +5,15 @@
 
 
 class UnitConverter(BaseConverter):
-    SI_UNIT = "unit" # Abstract Unit. Usually used for unspecific medications (eg. Vasopressine)
-    AVAILABLE_UNITS = ["unit", "milliunit"]
-
-    def _convertToSI(self, source_unit: str, data: Series[Quantity], dependencies: dict[str,DataFrame]):
+    SI_UNIT = "unit"  # Abstract Unit. Usually used for unspecific medications (eg. Vasopressine)
+    AVAILABLE_UNITS = ["unit", "milliunit", "%"]
+
+    def _convertToSI(
+        self,
+        source_unit: str,
+        data: Series[Quantity],
+        dependencies: dict[str, DataFrame],
+    ):
         convert: Callable[[float], float] = lambda v: v
         # Data can have any Unit and will be transformed to °C
         match source_unit:
@@ -20,15 +25,20 @@ def _convertToSI(self, source_unit: str, data: Series[Quantity], dependencies: d
             case "milliunit":
                 convert = lambda v: v / 10e3
 
+            case "%":
+                convert = lambda v: v / 100
+
             # Not Implemented
             case _:
                 raise NotImplementedError
 
-        return data.apply(lambda q: Quantity(
-            value=convert(q["value"]),
-            unit=self.SI_UNIT))
+        return data.apply(
+            lambda q: Quantity(value=convert(q["value"]), unit=self.SI_UNIT)
+        )
 
-    def _convertToTarget(self, sink_unit: str, data: Series[Quantity], dependencies: dict[str,DataFrame]):
+    def _convertToTarget(
+        self, sink_unit: str, data: Series[Quantity], dependencies: dict[str, DataFrame]
+    ):
         convert: Callable[[float], float] = lambda v: v
         # Data uses °C and can be transformed in to any Unit
         match sink_unit:
@@ -40,10 +50,11 @@ def _convertToTarget(self, sink_unit: str, data: Series[Quantity], dependencies:
             case "milliunit":
                 convert = lambda v: v * 10e3
 
+            case "%":
+                convert = lambda v: v * 100
+
             # Not Implemented
             case _:
                 raise NotImplementedError
 
-        return data.apply(lambda q: Quantity(
-            value=convert(q["value"]),
-            unit=sink_unit))
+        return data.apply(lambda q: Quantity(value=convert(q["value"]), unit=sink_unit))
diff --git a/icu_pipeline/unit/volume.py b/icu_pipeline/unit/volume.py
@@ -0,0 +1,54 @@
+from typing import Callable
+from pandera.typing import Series, DataFrame
+from icu_pipeline.schema.fhir import Quantity
+from icu_pipeline.unit.converter import BaseConverter
+
+
+class LengthConverter(BaseConverter):
+    SI_UNIT = "l"
+    AVAILABLE_UNITS = ["l", "ml"]
+
+    def _convertToSI(
+        self,
+        source_unit: str,
+        data: Series[Quantity],
+        dependencies: dict[str, DataFrame],
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data can have any Unit and will be transformed to m
+        match source_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions from case to cm
+            case "ml":
+                convert = lambda v: v * 100
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(
+            lambda q: Quantity(value=convert(q["value"]), unit=self.SI_UNIT)
+        )
+
+    def _convertToTarget(
+        self, sink_unit: str, data: Series[Quantity], dependencies: dict[str, DataFrame]
+    ):
+        convert: Callable[[float], float] = lambda v: v
+        # Data uses m and can be transformed in to any Unit
+        match sink_unit:
+            # Already SI-Unit
+            case self.SI_UNIT:
+                return data
+
+            # Actual Conversions
+            case "ml":
+                convert = lambda v: v / 100
+
+            # Not Implemented
+            case _:
+                raise NotImplementedError
+
+        return data.apply(lambda q: Quantity(value=convert(q["value"]), unit=sink_unit))
+8 −0		concepts/Age.yml
+8 −0		concepts/BodyHeight.yml
+8 −0		concepts/BodyTemperature.yml
+8 −0		concepts/DiastolicBloodPressue_Arterial_Invasive.yml
+8 −0		concepts/DiastolicBloodPressue_Arterial_noninvasive.yml
+8 −0		concepts/Gender.yml
+8 −0		concepts/HeartRate.yml
+8 −0		concepts/MeanBloodPressue_Arterial_Invasive.yml
+8 −0		concepts/MeanBloodPressue_Arterial_Noninvasive.yml
+8 −0		concepts/OxygenSaturation_Peripheral.yml
+8 −0		concepts/SystolicBloodPressure_Arterial_Invasive.yml
+8 −0		concepts/SystolicBloodPressure_Arterial_noninvasive.yml