Source code for bhf_smarthealth.process.qualtrics_process

"""This module utilises functions from the parse.qualtrics_parser module to process raw daily Qualtrics data"""

import logging
from datetime import date
from pathlib import Path

import pandas as pd

from bhf_smarthealth.log import metadata_logs
from bhf_smarthealth.models import DataStreamNames
from bhf_smarthealth.parse import qualtrics_parser as bhf_qualtrics
from bhf_smarthealth.utils import data_io

logger = logging.getLogger(__name__)




def process_day(in_paths: dict[str, Path], out_paths: dict[str, Path], date_str: str):
    """Process daily Qualtrics base and bridge .csv files

    Parameters
    ----------
    in_dir : Path
        Path of the directory containing raw data to process
    out_paths : dict[str, Path]
        Dictionary of paths to save each of the output DataFrames to
    date : date
        datetime.date object containing date to process
    """
    # ensure date in correct format for Qualtrics filename (YYYY-MM-DD)
    date_str = date.fromisoformat(date_str).strftime("%Y-%m-%d")

    logger.info(f"Processing Qualtrics data stream for {date_str}")

    bridge_path = in_paths.get("bridge")
    base_path = in_paths.get("base")

    # if no file paths are passed, i.e. there is no Qualtrics data, save out empty files
    if (bridge_path is None) and (base_path is None):
        for table in ["base", "bridge"]:
            logger.warning(
                f"No Qualtrics data recieved for date {date_str}. Saving empty parquet files..."
            )

            data_io.save_dataframe_ouput(
                df=pd.DataFrame(), file_out_path=out_paths[table]
            )

            # No IDs seen for this stream today
            all_IDs = []

    # else both paths are available
    elif (bridge_path is not None) and (base_path is not None):
        # no errors handled here as an error with this stream should be handled at a higher level to either stop the pipeline, or continue without this stream but flag for re-processing
        bridge_data = bhf_qualtrics.load_qualtrics_csv(bridge_path)
        base_data = bhf_qualtrics.load_qualtrics_csv(base_path)
        validated_data = bhf_qualtrics.validate_data(
            base_questions_data=base_data, bridge_data=bridge_data
        )
        out_dict: dict[str, pd.DataFrame] = bhf_qualtrics.create_dataframes(
            validated_data
        )

        # Save out the base and bridge DataFrames
        for table in ["base", "bridge"]:
            data_io.save_dataframe_ouput(out_dict[table], out_paths[table])

            logger.info(
                f"Saved {table} DataFrame to {out_paths[table]} with shape: {out_dict[table].shape} and columns: {out_dict[table].columns}"
            )

        # Generate receipt of IDs "seen" today
        # validate_data() checks all IDs are unique and identical between bridge and base files
        all_IDs = out_dict["base"].reset_index()["sheffield_id"].tolist()

    metadata_logs.generate_daily_IDs_receipt(
        all_IDs=all_IDs,
        valid_IDs=all_IDs,  # all processed IDs will be valid
        data_stream=DataStreamNames.QUALTRICS,
        date_str=date_str,
        out_path=out_paths["metadata"],
    )