Source code for bhf_smarthealth.enrich.lsoa_ahah_enrich

"""This module enriches LSOA data with AHAH environmental data"""

import pandas as pd
import logging

from pathlib import Path
from bhf_smarthealth.utils import data_io, df_utils

from bhf_smarthealth.config import (
    DEFAULT_AHAH_AGG_DICT as agg_dict,
    DEFAULT_AHAH_TIME_PERIOD as time_period,
    DEFAULT_REMOVE_AHAH_COLUMNS as remove_columns,
    DEFAULT_RENAME_AHAH_COLUMNS as rename_columns,
    DEFAULT_AHAH_FIELDS as ahah_field_subset,
    DEFAULT_AHAH_MERGE_COL as merge_column,
)

logger = logging.getLogger(__name__)




def enrich_LSOA_out(
    lsoa_data: pd.DataFrame, ahah_data: pd.DataFrame, out_path: Path | str
) -> None:
    """Enrich derived LSOA data with AHAH data

    Parameters
    ----------
    lsoa_data : pd.DataFrame
        DataFrame containing the intraday LSOA data to be processed
    ahah_data : pd.DataFrame
        DataFrame containing reference AHAH (Access to Health Assets and Hazards) data
    out_path : Path | str
        Path to write output file to. File format will be determined from filename extension

    Raises
    ------
    TypeError
        Raised if reference AHAH data is not a DataFrame
    TypeError
        Raised if LSOA data is not a DataFrame
    """
    if not isinstance(ahah_data, pd.DataFrame):
        raise TypeError("Reference AHAH data must be a pd.DataFrame instance")

    if not isinstance(lsoa_data, pd.DataFrame):
        raise TypeError("LSOA data must be a pd.DataFrame instance.")

    if lsoa_data.empty:
        logger.warning(
            "No valid LSOA data to enrich. Writing empty 'Environmental' file.."
        )
        data_io.save_dataframe_ouput(df=lsoa_data, file_out_path=out_path)
        return

    logger.info(
        f"Loaded intraday LSOA data with shape: {lsoa_data.shape}. Combining with reference AHAH data"
    )

    # calculate modal lsoa
    lsoa_modal_df = df_utils.apply_agg_over_interval(
        lsoa_data, agg_dict=agg_dict, time_period=time_period, ID_col="sheffield_id"
    )
    lsoa_modal_df = df_utils.rename_columns(df=lsoa_modal_df, col_names=rename_columns)

    # merge with reference dataset
    lsoa_ahah_df = (
        lsoa_modal_df.reset_index()
        .merge(
            ahah_data[ahah_field_subset],
            how="inner",
            left_on=merge_column,
            right_on=["LSOA21CD"],
        )
        .set_index(["sheffield_id", "timestamp"])
    )

    lsoa_ahah_df = df_utils.remove_columns(df=lsoa_ahah_df, col_names=remove_columns)

    data_io.save_dataframe_ouput(df=lsoa_ahah_df, file_out_path=out_path)

    logger.info(
        f"Processed DataFrame saved to {out_path} with columns: {lsoa_ahah_df.columns} and shape: {lsoa_ahah_df.shape}"
    )