edsteva.probes.utils.filter_df

filter_table_by_care_site

filter_table_by_care_site(
    table_to_filter: DataFrame,
    care_site_relationship: pd.DataFrame,
    care_site_ids: Union[int, List[int]] = None,
    care_site_short_names: Union[str, List[str]] = None,
    care_site_specialties: Union[str, List[str]] = None,
)

Extend all care_site_ids and care_site_short_names by computing recursively their parents and child. For each extended lists of care sites, keep those with specialties in care_site_specialties.

Arguments

table_to_filter : DataFrame care_site_relationship : DataFrame care_site_ids: Union[int, List[int]] care_site_short_names: Union[str, List[str]] care_site_specialties: Union[str, List[str]]

Source code in edsteva/probes/utils/filter_df.py

def filter_table_by_care_site(
    table_to_filter: DataFrame,
    care_site_relationship: pd.DataFrame,
    care_site_ids: Union[int, List[int]] = None,
    care_site_short_names: Union[str, List[str]] = None,
    care_site_specialties: Union[str, List[str]] = None,
):
    """
    Extend all care_site_ids and care_site_short_names by computing recursively their parents and child.
    For each extended lists of care sites, keep those with specialties in care_site_specialties.

    Arguments
    ---------
    table_to_filter : DataFrame
    care_site_relationship : DataFrame
    care_site_ids: Union[int, List[int]]
    care_site_short_names: Union[str, List[str]]
    care_site_specialties: Union[str, List[str]]
    """
    care_site = care_site_relationship[
        [
            "care_site_id",
            "care_site_short_name",
            "care_site_level",
            "care_site_specialty",
        ]
    ]
    care_site_filter = []
    if care_site_ids:
        if not isinstance(care_site_ids, list):
            care_site_ids = [care_site_ids]
        care_site_filter.append(
            care_site[care_site["care_site_id"].isin(care_site_ids)]
        )
    if care_site_short_names:
        if not isinstance(care_site_short_names, list):
            care_site_short_names = [care_site_short_names]
        care_site_filter.append(
            care_site[care_site["care_site_short_name"].isin(care_site_short_names)]
        )

    if care_site_filter:
        care_site_filter = pd.concat(
            care_site_filter, ignore_index=True
        ).drop_duplicates()
        extended_care_site_id_to_filter = get_child_and_parent_cs(
            care_site_sample=care_site_filter,
            care_site_relationship=care_site_relationship,
        )
    else:
        extended_care_site_id_to_filter = care_site.copy()

    if care_site_specialties:
        if not isinstance(care_site_specialties, list):
            care_site_specialties = [care_site_specialties]
        extended_care_site_id_to_filter = extended_care_site_id_to_filter[
            extended_care_site_id_to_filter["care_site_specialty"].isin(
                care_site_specialties
            )
        ]
        extended_care_site_id_to_filter = get_child_and_parent_cs(
            care_site_sample=extended_care_site_id_to_filter,
            care_site_relationship=care_site_relationship,
        )

    extended_care_site_id_to_filter = list(
        extended_care_site_id_to_filter[
            (
                extended_care_site_id_to_filter.care_site_level.isin(
                    CARE_SITE_LEVEL_NAMES.values()
                )
            )
        ].care_site_id.unique()
    )

    return table_to_filter[
        table_to_filter.care_site_id.isin(extended_care_site_id_to_filter)
    ]