functioning sex and continuous coverage phenotypes and tests

mkdocs.yml

@@ -20,6 +20,7 @@ nav:
           - AgePhenotype: api/phenotypes/age_phenotype.md
           - SexPhenotype: api/phenotypes/sec_phenotype.md
           - DeathPhenotype: api/phenotypes/death_phenotype.md
+          - ContinuousCoveragePhenotype: api/phenotypes/continuous_coverage_phenotype.md
           - AgePhenotype: api/phenotypes/age_phenotype.md
           - ArithmeticPhenotype: api/phenotypes/arithmetic_phenotype.md
           - LogicPhenotype: api/phenotypes/logic_phenotype.md

phenex/codelists/codelists.py

@@ -156,11 +156,10 @@ def to_pandas(self) -> pd.DataFrame:
         """
 
         _df = pd.DataFrame(self.to_tuples(), columns=["code_type", "code"])
-        _df['codelist'] = self.name
+        _df["codelist"] = self.name
         return _df
 
 
-
 class LocalCSVCodelistFactory:
     """ """
 

phenex/filters/aggregator.py

@@ -8,7 +8,7 @@ def __init__(
         aggregation_index=["PERSON_ID"],
         aggregation_function="sum",
         event_date_column="EVENT_DATE",
-        reduce=False
+        reduce=False,
     ):
         self.aggregation_index = aggregation_index
         self.aggregation_function = aggregation_function
@@ -40,7 +40,9 @@ def aggregate(self, input_table: Table):
         input_table = input_table.mutate(aggregated_date=aggregated_date)
 
         # Filter rows where the original date matches the aggregated date
-        input_table = input_table.filter(input_table[self.event_date_column] == input_table.aggregated_date)
+        input_table = input_table.filter(
+            input_table[self.event_date_column] == input_table.aggregated_date
+        )
 
         # Select the necessary columns
 
@@ -52,6 +54,7 @@ def aggregate(self, input_table: Table):
 
         return input_table
 
+
 class Nearest(VerticalDateAggregator):
     def __init__(self, **kwargs):
         super().__init__(aggregation_function="max", **kwargs)

phenex/filters/categorical_filter.py

@@ -2,6 +2,7 @@
 from typing import List, Optional, Union
 from ibis.expr.types.relations import Table
 
+
 class CategoricalFilter(Filter):
     """
     This class filters events in an EventTable based on specified categorical values
@@ -19,10 +20,10 @@ class CategoricalFilter(Filter):
     """
 
     def __init__(
-            self,
-            column_name: str,
-            allowed_values: List[Union[str, int]],
-            domain: Optional[str] = None
+        self,
+        column_name: str,
+        allowed_values: List[Union[str, int]],
+        domain: Optional[str] = None,
     ):
         self.column_name = column_name
         self.allowed_values = allowed_values

phenex/filters/relative_time_range_filter.py

@@ -56,24 +56,7 @@ def __init__(
         when: Optional[str] = "before",
         anchor_phenotype: "Phenotype" = None,
     ):
-        if min_days is not None:
-            assert min_days.operator in [
-                ">",
-                ">=",
-            ], f"min_days operator must be > or >=, not {min_days.operator}"
-        if max_days is not None:
-            assert max_days.operator in [
-                "<",
-                "<=",
-            ], f"max_days operator must be > or >=, not {max_days.operator}"
-        if max_days is not None and min_days is not None:
-            assert (
-                min_days.value <= max_days.value
-            ), f"min_days must be less than or equal to max_days"
-        assert when in [
-            "before",
-            "after",
-        ], f"when must be 'before' or 'after', not {when}"
+        verify_relative_time_range_filter_input(min_days, max_days, when)
 
         self.min_days = min_days
         self.max_days = max_days
@@ -125,3 +108,24 @@ def _filter(self, table: EventTable):
             table = table.filter(conditions)
 
         return table
+
+
+def verify_relative_time_range_filter_input(min_days, max_days, when):
+    if min_days is not None:
+        assert min_days.operator in [
+            ">",
+            ">=",
+        ], f"min_days operator must be > or >=, not {min_days.operator}"
+    if max_days is not None:
+        assert max_days.operator in [
+            "<",
+            "<=",
+        ], f"max_days operator must be > or >=, not {max_days.operator}"
+    if max_days is not None and min_days is not None:
+        assert (
+            min_days.value <= max_days.value
+        ), f"min_days must be less than or equal to max_days"
+    assert when in [
+        "before",
+        "after",
+    ], f"when must be 'before' or 'after', not {when}"

phenex/mappers.py

@@ -58,7 +58,13 @@ def rename(self, table: Table) -> Table:
         mapping = copy.deepcopy(asdict(self))
         mapping.pop("NAME_TABLE")
         # delete optional params from mapping
-        for key in ["DATE_OF_BIRTH", "DATE_OF_DEATH", "YEAR_OF_BIRTH", "SEX", "ETHNICITY"]:
+        for key in [
+            "DATE_OF_BIRTH",
+            "DATE_OF_DEATH",
+            "YEAR_OF_BIRTH",
+            "SEX",
+            "ETHNICITY",
+        ]:
             if getattr(self, key) is None:
                 del mapping[key]
         return table.rename(**mapping)
@@ -118,6 +124,7 @@ class MeasurementTableColumnMapper(CodeTableColumnMapper):
 
     VALUE: str = "VALUE"
 
+
 @dataclass
 class ObservationPeriodTableMapper:
     NAME_TABLE: str = "OBSERVATION_PERIOD"
@@ -139,26 +146,30 @@ def rename(self, table: Table) -> Table:
         mapping.pop("NAME_TABLE")
         return table.rename(**mapping)
 
+
 #
 # OMOP Column Mappers
 #
 OMOPPersonTableColumnMapper = PersonTableColumnMapper(
-    NAME_TABLE="PERSON", PERSON_ID="PERSON_ID", 
+    NAME_TABLE="PERSON",
+    PERSON_ID="PERSON_ID",
     DATE_OF_BIRTH="BIRTH_DATETIME",
     YEAR_OF_BIRTH="YEAR_OF_BIRTH",
-    SEX="GENDER_CONCEPT_ID", ETHNICITY="ETHNICITY_CONCEPT_ID"
+    SEX="GENDER_CONCEPT_ID",
+    ETHNICITY="ETHNICITY_CONCEPT_ID",
 )
 
 OMOPDeathTableColumnMapper = PersonTableColumnMapper(
-    NAME_TABLE="DEATH", PERSON_ID="PERSON_ID",
-    DATE_OF_DEATH="DEATH_DATE"
+    NAME_TABLE="DEATH", PERSON_ID="PERSON_ID", DATE_OF_DEATH="DEATH_DATE"
 )
 
 OMOPPersonTableSourceColumnMapper = PersonTableColumnMapper(
-    NAME_TABLE="PERSON", PERSON_ID="PERSON_ID", 
+    NAME_TABLE="PERSON",
+    PERSON_ID="PERSON_ID",
     DATE_OF_BIRTH="BIRTH_DATETIME",
     YEAR_OF_BIRTH="YEAR_OF_BIRTH",
-    SEX="GENDER_SOURCE_VALUE", ETHNICITY="ETHNICITY_SOURCE_VALUE"
+    SEX="GENDER_SOURCE_VALUE",
+    ETHNICITY="ETHNICITY_SOURCE_VALUE",
 )
 
 OMOPConditionOccurrenceColumnMapper = CodeTableColumnMapper(
@@ -223,12 +234,14 @@ def rename(self, table: Table) -> Table:
 OMOPDomains = DomainsDictionary(**OMOPColumnMappers)
 
 
-
 #
 # Vera Column Mappers
 #
 VeraPersonTableColumnMapper = PersonTableColumnMapper(
-    NAME_TABLE="PERSON", PERSON_ID="PERSON_ID", DATE_OF_BIRTH="BIRTH_DATETIME", DATE_OF_DEATH="DEATH_DATETIME"
+    NAME_TABLE="PERSON",
+    PERSON_ID="PERSON_ID",
+    DATE_OF_BIRTH="BIRTH_DATETIME",
+    DATE_OF_DEATH="DEATH_DATETIME",
 )
 
 VeraConditionOccurrenceColumnMapper = CodeTableColumnMapper(
@@ -268,4 +281,4 @@ def rename(self, table: Table) -> Table:
 #
 # Domains
 #
-VeraDomains = DomainsDictionary(**VeraColumnMappers)
+VeraDomains = DomainsDictionary(**VeraColumnMappers)

phenex/phenotypes/categorical_phenotype.py

@@ -29,12 +29,13 @@ class HospitalizationPhenotype(Phenotype):
         _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,
+        name=None,
         date_range: DateRangeFilter = None,
         relative_time_range: Union[
             RelativeTimeRangeFilter, List[RelativeTimeRangeFilter]
@@ -43,7 +44,9 @@ def __init__(
     ):
         super(HospitalizationPhenotype, self).__init__()
 
-        self.categorical_filter = CategoricalFilter(column_name=column_name, allowed_values=allowed_values)
+        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
@@ -93,4 +96,4 @@ def _perform_date_selection(self, code_table):
             aggregator = Last()
         else:
             raise ValueError(f"Unknown return_date: {self.return_date}")
-        return aggregator.aggregate(code_table)
+        return aggregator.aggregate(code_table)