Circepy/circe/execution/lower/specimen.py at 7d5a9c30e752a1bd3baef0643c2968fca6970d8d · OHDSI/Circepy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
from __future__ import annotations

from circe.extensions import lowerer

from ...cohortdefinition.criteria import Specimen
from ..normalize.criteria import NormalizedCriterion
from ..plan.events import EventPlan
from .common import (
    append_concept_filters,
    append_numeric_filter,
    append_text_filter,
    build_standard_domain_plan,
    lower_common_steps,
)


@lowerer(Specimen)
def lower_specimen(
    criterion: NormalizedCriterion,
    *,
    criterion_index: int,
) -> EventPlan:
    raw = criterion.raw_criteria
    if not isinstance(raw, Specimen):
        raise TypeError("lower_specimen requires Specimen criteria")

    steps = lower_common_steps(criterion)

    append_concept_filters(
        steps,
        column="specimen_type_concept_id",
        concepts=raw.specimen_type,
        codeset_selection=raw.specimen_type_cs,
        exclude=bool(raw.specimen_type_exclude),
    )
    append_numeric_filter(steps, column="quantity", value=raw.quantity)
    append_concept_filters(
        steps,
        column="unit_concept_id",
        concepts=raw.unit,
        codeset_selection=raw.unit_cs,
    )
    append_concept_filters(
        steps,
        column="anatomic_site_concept_id",
        concepts=raw.anatomic_site,
        codeset_selection=raw.anatomic_site_cs,
    )
    append_concept_filters(
        steps,
        column="disease_status_concept_id",
        concepts=raw.disease_status,
        codeset_selection=raw.disease_status_cs,
    )
    append_text_filter(steps, column="specimen_source_id", value=raw.source_id)

    return build_standard_domain_plan(
        criterion,
        criterion_index=criterion_index,
        steps=steps,
    )