categorical phenotypes

Author: sprivite

phenex/filters/categorical_filter.py

@@ -0,0 +1,35 @@
+from phenex.filters.filter import Filter
+from typing import List
+from typing import Optional
+from ibis.expr.types.relations import Table
+
+class CategoricalFilter(Filter):
+    """
+    This class filters events in an EventTable based on specified categorical values
+
+    Attributes:
+        category (Optional[str]): The category to filter events by.
+
+    Methods:
+        _filter(table: MeasurementTable) -> MeasurementTable:
+            Filters the given MeasurementTable based on the specified category.
+            Parameters:
+                table (Measurement): The table containing events to be filtered.
+            Returns:
+                MeasurementTable: The filtered MeasurementTable with events matching the category.
+    """
+
+    def __init__(
+            self, 
+            column_name: str,
+            allowed_values: List[str, int],
+            domain: Optional[str] = None
+    ):
+        self.column_name = column_name
+        self.allowed_values = allowed_values
+        self.domain = domain
+        super(CategoricalFilter, self).__init__()
+
+    def _filter(self, table: Table):
+        table = table.filter(table[self.column_name].isin(self.allowed_values))
+        return table

phenex/phenotypes/death_phenotype.py

@@ -0,0 +1,32 @@
+from typing import Dict
+from ibis.expr.types.relations import Table
+from phenex.phenotypes.phenotype import Phenotype
+from phenex.tables import PhenotypeTable, is_phenex_person_table
+
+class DeathPhenotype(Phenotype):
+    """
+    DeathPhenotype is a class that represents a death-based phenotype. It filters individuals
+    who have died and returns their date of death.
+
+    Attributes:
+        name (str): Name of the phenotype, default is 'death'.
+        domain (str): Domain of the phenotype, default is 'PERSON'.
+        children (list): List of dependent phenotypes.
+
+    Methods:
+        _execute(tables: Dict[str, Table]) -> PhenotypeTable:
+            Executes the phenotype calculation and returns a table with the filtered individuals.
+    """
+
+    def __init__(self, name: str = "death", domain: str = "PERSON"):
+        self.name = name
+        self.domain = domain
+        self.children = []
+        super(DeathPhenotype, self).__init__()
+
+    def _execute(self, tables: Dict[str, Table]) -> PhenotypeTable:
+        person_table = tables[self.domain]
+        assert is_phenex_person_table(person_table)
+
+        death_table = person_table.filter(person_table.DEATH_DATE.notnull())
+        return death_table.mutate(EVENT_DATE=death_table.DEATH_DATE)

phenex/phenotypes/hospitalization_phenotype.py

@@ -0,0 +1,96 @@
+from typing import Union, List
+from phenex.phenotypes.phenotype import Phenotype
+from phenex.filters.relative_time_range_filter import RelativeTimeRangeFilter
+from phenex.filters.date_range_filter import DateRangeFilter
+from phenex.filters.aggregator import First, Last
+from phenex.filters.categorical_filter import CategoricalFilter
+from phenex.tables import is_phenex_code_table, PHENOTYPE_TABLE_COLUMNS, PhenotypeTable
+from phenex.phenotypes.functions import select_phenotype_columns
+from ibis import _
+
+
+class HospitalizationPhenotype(Phenotype):
+    """
+    HospitalizationPhenotype filters an EncounterTable to identify inpatient events based on the encounter_type column.
+    It uses a CategoricalFilter to filter for inpatient events and can apply additional date and time range filters.
+
+    Attributes:
+        name (str): The name of the phenotype.
+        domain (str): The domain of the phenotype, default is 'ENCOUNTER'.
+        column_name (str): The name of the column to filter on, default is 'ENCOUNTER_TYPE'.
+        allowed_values (List[str]): List of allowed values for the encounter_type column, default is ['inpatient'].
+        date_range (DateRangeFilter, optional): A date range filter to apply.
+        relative_time_range (Union[RelativeTimeRangeFilter, List[RelativeTimeRangeFilter]], optional): A relative time range filter or a list of filters to apply.
+        return_date (str): Specifies whether to return the 'first', 'last', 'nearest', or 'all' event dates. Default is 'first'.
+        table (PhenotypeTable): The resulting phenotype table after filtering.
+        children (list): List of child phenotypes.
+
+    Methods:
+        _execute(tables: Dict[str, Table]) -> PhenotypeTable:
+            Executes the filtering process on the provided tables and returns the filtered phenotype table.
+    """
+    def __init__(
+        self,
+        domain,
+        column_name: str,
+        allowed_values: List[str],
+        name = None,
+        date_range: DateRangeFilter = None,
+        relative_time_range: Union[
+            RelativeTimeRangeFilter, List[RelativeTimeRangeFilter]
+        ] = None,
+        return_date="first",
+    ):
+        super(HospitalizationPhenotype, self).__init__()
+
+        self.categorical_filter = CategoricalFilter(column_name=column_name, allowed_values=allowed_values)
+        self.name = name
+        self.date_range = date_range
+        self.return_date = return_date
+        assert self.return_date in [
+            "first",
+            "last",
+            "nearest",
+            "all",
+        ], f"Unknown return_date: {return_date}"
+        self.table = None
+        self.domain = domain
+        if isinstance(relative_time_range, RelativeTimeRangeFilter):
+            relative_time_range = [relative_time_range]
+
+        self.relative_time_range = relative_time_range
+        if self.relative_time_range is not None:
+            for rtr in self.relative_time_range:
+                if rtr.anchor_phenotype is not None:
+                    self.children.append(rtr.anchor_phenotype)
+
+    def _execute(self, tables) -> PhenotypeTable:
+        code_table = tables[self.domain]
+        code_table = self._perform_categorical_filtering(code_table)
+        code_table = self._perform_time_filtering(code_table)
+        code_table = self._perform_date_selection(code_table)
+        return select_phenotype_columns(code_table)
+
+    def _perform_categorical_filtering(self, code_table):
+        assert is_phenex_code_table(code_table)
+        code_table = self.categorical_filter.filter(code_table)
+        return code_table
+
+    def _perform_time_filtering(self, code_table):
+        if self.date_range is not None:
+            code_table = self.date_range.filter(code_table)
+        if self.relative_time_range is not None:
+            for rtr in self.relative_time_range:
+                code_table = rtr.filter(code_table)
+        return code_table
+
+    def _perform_date_selection(self, code_table):
+        if self.return_date is None or self.return_date == "all":
+            return code_table
+        if self.return_date == "first":
+            aggregator = First()
+        elif self.return_date == "last":
+            aggregator = Last()
+        else:
+            raise ValueError(f"Unknown return_date: {self.return_date}")
+        return aggregator.aggregate(code_table)

phenex/phenotypes/sex_phenotype.py

@@ -0,0 +1,42 @@
+from typing import Dict, List, Optional
+from ibis.expr.types.relations import Table
+from phenex.phenotypes.phenotype import Phenotype
+from phenex.filters.categorical_filter import CategoricalFilter
+from phenex.tables import PhenotypeTable, is_phenex_person_table
+
+class SexPhenotype(Phenotype):
+    """
+    SexPhenotype is a class that represents a sex-based phenotype. It filters individuals
+    based on their sex (e.g., male, female) using the CategoricalFilter.
+
+    Attributes:
+        name (str): Name of the phenotype, default is 'sex'.
+        allowed_values (List[str]): List of allowed values for the sex column.
+        domain (str): Domain of the phenotype, default is 'PERSON'.
+        children (list): List of dependent phenotypes.
+
+    Methods:
+        _execute(tables: Dict[str, Table]) -> PhenotypeTable:
+            Executes the phenotype calculation and returns a table with the filtered individuals.
+    """
+
+    def __init__(
+        self,
+        name: str = "sex",
+        allowed_values: List[str] = ["male", "female"],
+        domain: str = "PERSON",
+    ):
+        self.name = name
+        self.allowed_values = allowed_values
+        self.domain = domain
+        self.children = []
+        super(SexPhenotype, self).__init__()
+
+    def _execute(self, tables: Dict[str, Table]) -> PhenotypeTable:
+        person_table = tables[self.domain]
+        assert is_phenex_person_table(person_table)
+
+        sex_filter = CategoricalFilter(column_name="SEX", allowed_values=self.allowed_values)
+        filtered_table = sex_filter._filter(person_table)
+
+        return filtered_table.mutate(VALUE=filtered_table.SEX)