edsteva.probes.condition.completeness_predictors.per_visit

compute_completeness_predictor_per_visit

compute_completeness_predictor_per_visit(
    self,
    data: Data,
    start_date: datetime,
    end_date: datetime,
    care_site_levels: Union[bool, str, List[str]],
    stay_types: Union[bool, str, Dict[str, str]],
    care_site_ids: List[int],
    extra_data: Data,
    care_site_short_names: List[str],
    care_site_specialties: Union[bool, List[str]],
    care_sites_sets: Union[str, Dict[str, str]],
    specialties_sets: Union[str, Dict[str, str]],
    diag_types: Union[bool, str, Dict[str, str]],
    condition_types: Union[bool, str, Dict[str, str]],
    condition_concept_codes: Union[bool, List[str]],
    source_systems: Union[bool, List[str]],
    length_of_stays: List[float],
    age_ranges: List[int],
    gender_source_values: Union[bool, str, Dict[str, str]],
    provenance_sources: Union[bool, str, Dict[str, str]],
    stay_sources: Union[bool, str, Dict[str, str]],
    drg_sources: Union[bool, str, Dict[str, str]],
    **kwargs
)

Script to be used by compute()

The per_visit algorithm computes $c_{(}t)$ the availability of claim data linked to patients' administrative stays:

$$c(t)=\frac{n_{withcondition}(t)}{n_{visit}(t)}$$

Where $n_{visit}(t)$ is the number of administrative stays, $n_{withcondition}$ the number of stays having at least one claim code (e.g. ICD-10) recorded and $t$ is the month.

Source code in edsteva/probes/condition/completeness_predictors/per_visit.py

def compute_completeness_predictor_per_visit(
    self,
    data: Data,
    start_date: datetime,
    end_date: datetime,
    care_site_levels: Union[bool, str, List[str]],
    stay_types: Union[bool, str, Dict[str, str]],
    care_site_ids: List[int],
    extra_data: Data,
    care_site_short_names: List[str],
    care_site_specialties: Union[bool, List[str]],
    care_sites_sets: Union[str, Dict[str, str]],
    specialties_sets: Union[str, Dict[str, str]],
    diag_types: Union[bool, str, Dict[str, str]],
    condition_types: Union[bool, str, Dict[str, str]],
    condition_concept_codes: Union[bool, List[str]],
    source_systems: Union[bool, List[str]],
    length_of_stays: List[float],
    age_ranges: List[int],
    gender_source_values: Union[bool, str, Dict[str, str]],
    provenance_sources: Union[bool, str, Dict[str, str]],
    stay_sources: Union[bool, str, Dict[str, str]],
    drg_sources: Union[bool, str, Dict[str, str]],
    **kwargs
):
    r"""Script to be used by [``compute()``][edsteva.probes.base.BaseProbe.compute]

    The ``per_visit`` algorithm computes $c_(t)$ the availability of claim data linked to patients' administrative stays:

    $$
    c(t) = \frac{n_{with\,condition}(t)}{n_{visit}(t)}
    $$

    Where $n_{visit}(t)$ is the number of administrative stays, $n_{with\,condition}$ the number of stays having at least one claim code (e.g. ICD-10) recorded and $t$ is the month.
    """
    self._condition_columns = list(
        set(
            ["diag_type", "condition_type", "condition_source_value", "source_system"]
        ).intersection(set(self._index))
    )
    self._metrics = ["c", "n_visit", "n_visit_with_condition"]
    check_tables(
        data=data,
        required_tables=[
            "condition_occurrence",
            "visit_occurrence",
            "care_site",
            "fact_relationship",
        ],
    )
    care_site_relationship = prepare_care_site_relationship(
        data=data,
    )
    self.care_site_relationship = care_site_relationship
    person = (
        prepare_person(data, gender_source_values)
        if (age_ranges or gender_source_values)
        else None
    )
    cost = prepare_cost(data, drg_sources) if drg_sources else None

    visit_occurrence = prepare_visit_occurrence(
        data=data,
        start_date=start_date,
        end_date=end_date,
        stay_types=stay_types,
        length_of_stays=length_of_stays,
        provenance_sources=provenance_sources,
        stay_sources=stay_sources,
        cost=cost,
        person=person,
        age_ranges=age_ranges,
    )

    condition_occurrence = prepare_condition_occurrence(
        data=data,
        extra_data=extra_data,
        visit_occurrence=visit_occurrence,
        source_systems=source_systems,
        diag_types=diag_types,
        condition_types=condition_types,
    ).drop(columns=["condition_occurrence_id", "date"])

    care_site = prepare_care_site(
        data=data,
        care_site_ids=care_site_ids,
        care_site_short_names=care_site_short_names,
        care_site_relationship=care_site_relationship,
        care_site_specialties=care_site_specialties,
        care_sites_sets=care_sites_sets,
        specialties_sets=specialties_sets,
    )

    hospital_visit = get_hospital_visit(
        self,
        condition_occurrence,
        visit_occurrence,
        care_site,
    )
    hospital_name = CARE_SITE_LEVEL_NAMES["Hospital"]
    condition_predictor_by_level = {hospital_name: hospital_visit}

    # UF selection
    if not hospital_only(care_site_levels=care_site_levels):
        visit_detail = prepare_visit_detail(data, start_date, end_date)

        uf_visit = get_uf_visit(
            self,
            condition_occurrence=condition_occurrence,
            visit_occurrence=visit_occurrence,
            visit_detail=visit_detail,
            care_site=care_site,
        )
        uf_name = CARE_SITE_LEVEL_NAMES["UF"]
        condition_predictor_by_level[uf_name] = uf_visit

        pole_visit = get_pole_visit(
            uf_visit=uf_visit,
            care_site=care_site,
            care_site_relationship=care_site_relationship,
        )
        pole_name = CARE_SITE_LEVEL_NAMES["Pole"]
        condition_predictor_by_level[pole_name] = pole_visit

    condition_predictor = concatenate_predictor_by_level(
        predictor_by_level=condition_predictor_by_level,
        care_site_levels=care_site_levels,
    )

    return compute_completeness(self, condition_predictor)