14
14
from execution_engine .omop .criterion .combination .temporal import (
15
15
FixedWindowTemporalIndicatorCombination ,
16
16
PersonalWindowTemporalIndicatorCombination ,
17
- TimeIntervalType , TemporalIndicatorCombination ,
17
+ TimeIntervalType ,
18
18
)
19
19
from execution_engine .omop .criterion .condition_occurrence import ConditionOccurrence
20
20
from execution_engine .omop .criterion .drug_exposure import DrugExposure
27
27
from execution_engine .util .value import ValueNumber
28
28
from tests ._fixtures .concept import (
29
29
concept_artificial_respiration ,
30
- concept_delir_screening ,
31
30
concept_body_weight ,
32
31
concept_covid19 ,
32
+ concept_delir_screening ,
33
33
concept_heparin_ingredient ,
34
34
concept_surgical_procedure ,
35
35
concept_unit_kg ,
@@ -208,10 +208,16 @@ def test_operator_str(self, operator, threshold):
208
208
op = FixedWindowTemporalIndicatorCombination .Operator (operator , threshold )
209
209
210
210
if operator in ["AT_LEAST" , "AT_MOST" , "EXACTLY" ]:
211
- assert repr (op ) == f'Operator(operator="{ operator } ", threshold={ threshold } )'
211
+ assert (
212
+ repr (op )
213
+ == f'TemporalIndicatorCombination.Operator(operator="{ operator } ", threshold={ threshold } )'
214
+ )
212
215
assert str (op ) == f"{ operator } (threshold={ threshold } )"
213
216
else :
214
- assert repr (op ) == f'Operator(operator="{ operator } ")'
217
+ assert (
218
+ repr (op )
219
+ == f'TemporalIndicatorCombination.Operator(operator="{ operator } ")'
220
+ )
215
221
assert str (op ) == f"{ operator } "
216
222
217
223
def test_repr (self ):
@@ -228,9 +234,7 @@ def test_repr(self):
228
234
" interval_type=TimeIntervalType.MORNING_SHIFT,\n "
229
235
" start_time=None,\n "
230
236
" end_time=None,\n "
231
- ' operator=Operator(operator="AT_LEAST", threshold=1),\n '
232
- " criteria=[\n "
233
- " ],\n "
237
+ ' operator=TemporalIndicatorCombination.Operator(operator="AT_LEAST", threshold=1),\n '
234
238
")"
235
239
)
236
240
@@ -245,9 +249,7 @@ def test_repr(self):
245
249
" interval_type=None,\n "
246
250
" start_time=datetime.time(8, 0),\n "
247
251
" end_time=datetime.time(16, 0),\n "
248
- ' operator=Operator(operator="AT_LEAST", threshold=1),\n '
249
- " criteria=[\n "
250
- " ],\n "
252
+ ' operator=TemporalIndicatorCombination.Operator(operator="AT_LEAST", threshold=1),\n '
251
253
")"
252
254
)
253
255
@@ -309,6 +311,7 @@ def test_add_all(self):
309
311
concept = concept_delir_screening ,
310
312
)
311
313
314
+
312
315
class TestCriterionCombinationDatabase (TestCriterion ):
313
316
"""
314
317
Test class for testing criterion combinations on the database.
@@ -323,13 +326,13 @@ def observation_window(self) -> TimeRange:
323
326
@pytest .fixture
324
327
def criteria (self , db_session ):
325
328
criteria = [
326
- c1 ,
327
- c2 ,
328
- c3 ,
329
- bodyweight_measurement_without_forward_fill ,
330
- bodyweight_measurement_with_forward_fill ,
331
- delir_screening ,
332
- ]
329
+ c1 ,
330
+ c2 ,
331
+ c3 ,
332
+ bodyweight_measurement_without_forward_fill ,
333
+ bodyweight_measurement_with_forward_fill ,
334
+ delir_screening ,
335
+ ]
333
336
for i , c in enumerate (criteria ):
334
337
if not c .is_persisted ():
335
338
c .set_id (i + 1 )
@@ -1668,6 +1671,7 @@ def test_combination_on_database(
1668
1671
persons ,
1669
1672
)
1670
1673
1674
+
1671
1675
class TestTemporalCountNearObservationWindowEnd (TestCriterionCombinationDatabase ):
1672
1676
"""This test ensures that counting criteria with minimum count
1673
1677
thresholds adapt to the temporal interval of the population
@@ -1683,13 +1687,17 @@ class TestTemporalCountNearObservationWindowEnd(TestCriterionCombinationDatabase
1683
1687
1684
1688
@pytest .fixture
1685
1689
def observation_window (self ) -> TimeRange :
1686
- return TimeRange (name = "observation" , start = "2025-02-18 13:55:00Z" , end = "2025-02-22 11:00:00Z" )
1690
+ return TimeRange (
1691
+ name = "observation" , start = "2025-02-18 13:55:00Z" , end = "2025-02-22 11:00:00Z"
1692
+ )
1687
1693
1688
1694
def patient_events (self , db_session , visit_occurrence ):
1689
- c1 = create_condition (vo = visit_occurrence ,
1690
- condition_concept_id = concept_covid19 .concept_id ,
1691
- condition_start_datetime = pendulum .parse ("2025-02-19 22:00:00+01:00" ),
1692
- condition_end_datetime = pendulum .parse ("2025-02-22 02:00:00+01:00" ))
1695
+ c1 = create_condition (
1696
+ vo = visit_occurrence ,
1697
+ condition_concept_id = concept_covid19 .concept_id ,
1698
+ condition_start_datetime = pendulum .parse ("2025-02-19 22:00:00+01:00" ),
1699
+ condition_end_datetime = pendulum .parse ("2025-02-22 02:00:00+01:00" ),
1700
+ )
1693
1701
# One screen on the 19th
1694
1702
e1 = create_procedure (
1695
1703
vo = visit_occurrence ,
@@ -1725,7 +1733,7 @@ def patient_events(self, db_session, visit_occurrence):
1725
1733
start_datetime = pendulum .parse ("2025-02-22 04:00:00+01:00" ),
1726
1734
end_datetime = pendulum .parse ("2025-02-22 04:01:00+01:00" ),
1727
1735
)
1728
- db_session .add_all ([c1 ,e1 ,e2 ,e3 ,e4 ,e5 ])
1736
+ db_session .add_all ([c1 , e1 , e2 , e3 , e4 , e5 ])
1729
1737
db_session .commit ()
1730
1738
1731
1739
@pytest .mark .parametrize (
@@ -1777,13 +1785,14 @@ def test_at_least_combination_on_database(
1777
1785
observation_window ,
1778
1786
criteria ,
1779
1787
):
1780
- persons = [person [0 ]] # only one person
1788
+ persons = [person [0 ]] # only one person
1781
1789
vos = [
1782
1790
create_visit (
1783
1791
person_id = person .person_id ,
1784
1792
visit_start_datetime = observation_window .start
1785
1793
+ datetime .timedelta (hours = 3 ),
1786
- visit_end_datetime = observation_window .end - datetime .timedelta (hours = 6.5 ),
1794
+ visit_end_datetime = observation_window .end
1795
+ - datetime .timedelta (hours = 6.5 ),
1787
1796
visit_concept_id = concepts .INTENSIVE_CARE ,
1788
1797
)
1789
1798
for person in persons
0 commit comments