Skip to content

Commit b5d1dd4

Browse files
a-hartensa-hartens
committed
added non functioning tests
1 parent d5394a9 commit b5d1dd4

4 files changed

+537
-0
lines changed

‎phenex/test/phenotypes/test_categoric_phenotype.py

+89
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
import datetime, os
2+
import pandas as pd
3+
4+
from phenex.phenotypes.categorical_phenotype import CategoricalPhenotype
5+
from phenex.codelists import LocalCSVCodelistFactory
6+
from phenex.filters.date_range_filter import DateRangeFilter
7+
from phenex.filters.relative_time_range_filter import RelativeTimeRangeFilter
8+
9+
from phenex.test.phenotype_test_generator import PhenotypeTestGenerator
10+
from phenex.filters.value import *
11+
12+
13+
class CategoricalPhenotypeTestGenerator(PhenotypeTestGenerator):
14+
name_space = "cgpt"
15+
16+
def define_input_tables(self):
17+
def add_flag(df, flag_name, flag_values):
18+
dfs = []
19+
for flag in flag_values:
20+
_df = df.copy()
21+
_df[flag_name] = flag
22+
dfs.append(_df)
23+
return pd.concat(dfs)
24+
25+
df = pd.DataFrame()
26+
df["PERSON_ID"] = ["p1"]
27+
df["CODE"] = ["c1"]
28+
df["CODE_TYPE"] = ["ICD10CM"]
29+
df = add_flag(df, "x", ["x1", "x2"])
30+
df = add_flag(df, "y", ["y1", "y2"])
31+
df = add_flag(df, "z", ["z1", "z2"])
32+
df["PERSON_ID"] = [f"P{i}" for i in range(df.shape[0])]
33+
34+
return [{"condition_occurrence": "input", "df": df, "column_types": {}}]
35+
36+
def define_phenotype_tests(self):
37+
c1 = {
38+
"name": "single_flag",
39+
"persons": [f"P{i}" for i in range(4)],
40+
"phenotype": CategoricalPhenotype(
41+
name_space=self.name_space,
42+
domain="condition_occurrence",
43+
categorical_filter=CategoricalFilter(
44+
allowed_values=["z1"], columnname="z"
45+
),
46+
),
47+
}
48+
49+
c2 = {
50+
"name": "two_categorical_filter_or",
51+
"persons": [f"P{i}" for i in range(4)] + [f"P{i}" for i in range(6, 8)],
52+
"phenotype": CategoricalPhenotype(
53+
name_space=self.name_space,
54+
domain="condition_occurrence",
55+
categorical_filter=CategoricalFilter(
56+
allowed_values=["z1"], columnname="z"
57+
)
58+
| CategoricalFilter(allowed_values=["y2"], columnname="y"),
59+
),
60+
}
61+
62+
c3 = {
63+
"name": "two_categorical_filter_and",
64+
"persons": [f"P{i}" for i in range(2, 4)],
65+
"phenotype": CategoricalPhenotype(
66+
name_space=self.name_space,
67+
domain="condition_occurrence",
68+
categorical_filter=CategoricalFilter(
69+
allowed_values=["z1"], columnname="z"
70+
)
71+
& CategoricalFilter(allowed_values=["y2"], columnname="y"),
72+
),
73+
}
74+
75+
test_infos = [c1, c2, c3]
76+
for test_info in test_infos:
77+
test_info["refactor"] = True # TODO remove once refactored
78+
test_info["phenotype"].name_phenotype = test_info["name"]
79+
80+
return test_infos
81+
82+
83+
def test_categorical_phenotype():
84+
spg = CategoricalPhenotypeTestGenerator()
85+
spg.run_tests()
86+
87+
88+
if __name__ == "__main__":
89+
test_categorical_phenotype()

‎phenex/test/phenotypes/test_continuous_coverage_phenotype.py

+141
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
import datetime, os
2+
import pandas as pd
3+
4+
from phenex.phenotypes.continuous_coverage_phenotype import ContinuousCoveragePhenotype
5+
from phenex.codelists import LocalCSVCodelistFactory
6+
from phenex.filters.date_range_filter import DateRangeFilter
7+
from phenex.filters.relative_time_range_filter import RelativeTimeRangeFilter
8+
9+
from phenex.test.phenotype_test_generator import PhenotypeTestGenerator
10+
from phenex.filters.value import *
11+
12+
13+
14+
class ContinuousCoveragePhenotypeTestGenerator(PhenotypeTestGenerator):
15+
name_space = "continuouscoverage"
16+
17+
def define_input_tables(self):
18+
oneday = datetime.timedelta(days=1)
19+
index_date = datetime.datetime.strptime("01-01-2022", "%m-%d-%Y")
20+
21+
observation_period_min = 90 * oneday
22+
possible_start_dates = [
23+
index_date - 4 * observation_period_min,
24+
index_date - 2 * observation_period_min,
25+
index_date - observation_period_min - oneday,
26+
index_date - observation_period_min,
27+
index_date - observation_period_min + oneday,
28+
index_date,
29+
index_date + oneday,
30+
]
31+
32+
intervals = [
33+
observation_period_min,
34+
observation_period_min - oneday,
35+
observation_period_min + oneday,
36+
2 * observation_period_min,
37+
]
38+
39+
start_dates = []
40+
end_dates = []
41+
for s in possible_start_dates:
42+
for i in intervals:
43+
start_dates.append(s)
44+
end_dates.append(s + i)
45+
46+
N = len(end_dates)
47+
df_observation_period = pd.DataFrame()
48+
df_observation_period["PERSON_ID"] = [
49+
f"P{x}" for x in list(range(N))
50+
]
51+
df_observation_period["INDEX_DATE"] = index_date
52+
df_observation_period["observation_period_start_date"] = start_dates
53+
df_observation_period["observation_period_end_date"] = end_dates
54+
55+
56+
self.df_input = df_observation_period
57+
input_info_observation_period = {
58+
"name": "observation_period",
59+
"df": df_observation_period,
60+
}
61+
62+
return [input_info_observation_period]
63+
64+
def define_phenotype_tests(self):
65+
t1 = {
66+
"name": "coverage_min_geq_90",
67+
"coverage_period_min": Value(value=90, operator=">="),
68+
"persons": ["P7", "P10", "P11", "P12", "P14", "P15"],
69+
}
70+
t2 = {
71+
"name": "coverage_min_gt_90",
72+
"coverage_period_min": Value(value=90, operator=">"),
73+
"persons": ["P7", "P10", "P11"],
74+
}
75+
test_infos = [t1, t2]
76+
77+
for test_info in test_infos:
78+
test_info["phenotype"] = ContinuousCoveragePhenotype(
79+
name=test_info["name"],
80+
domain="observation_period",
81+
coverage_period_min=test_info.get("coverage_period_min"),
82+
)
83+
test_info["refactor"] = True # TODO remove once refactored
84+
85+
return test_infos
86+
87+
88+
class ContinuousCoverageReturnLastPhenotypeTestGenerator(
89+
ContinuousCoveragePhenotypeTestGenerator
90+
):
91+
name_space = "ccpt_returnlast"
92+
93+
def define_phenotype_tests(self):
94+
persons = ["P7", "P10", "P11", "P12", "P14", "P15"]
95+
96+
t1 = {
97+
"name": "coverage_min_geq_90",
98+
"coverage_period_min": Value(value=90, operator=">="),
99+
"persons": persons,
100+
"dates": list(
101+
self.df_input[self.df_input["PERSON_ID"].isin(persons)][
102+
"observation_period_end_date"
103+
].values
104+
),
105+
}
106+
107+
persons = ["P7", "P10", "P11"]
108+
t2 = {
109+
"name": "coverage_min_gt_90",
110+
"coverage_period_min": Value(value=90, operator=">"),
111+
"persons": ["P7", "P10", "P11"],
112+
"dates": list(
113+
self.df_input[self.df_input["PERSON_ID"].isin(persons)][
114+
"observation_period_end_date"
115+
].values
116+
),
117+
}
118+
test_infos = [t1, t2]
119+
120+
for test_info in test_infos:
121+
test_info["phenotype"] = ContinuousCoveragePhenotype(
122+
name=test_info["name"],
123+
domain="observation_period",
124+
return_date="last",
125+
coverage_period_min=test_info.get("coverage_period_min"),
126+
)
127+
test_info["column_types"] = {f"{test_info['name']}_date": "date"}
128+
129+
return test_infos
130+
131+
132+
def test_continuous_coverage_phenotypes():
133+
spg = ContinuousCoveragePhenotypeTestGenerator()
134+
spg.run_tests()
135+
136+
spg = ContinuousCoverageReturnLastPhenotypeTestGenerator()
137+
spg.run_tests()
138+
139+
140+
if __name__ == "__main__":
141+
test_continuous_coverage_phenotypes()

0 commit comments

Comments
 (0)