Source code for bhf_smarthealth.parse.qualtrics_parser

"""This module contains functions to process raw Qualtrics data ingressed to the Smart Health pipeline"""

import numpy as np
import pandas as pd
import logging

from typing import Any
from pydantic import ValidationError
from pathlib import Path

# from bhf_smarthealth.validate.qualtrics_schema import FitbitSummaryData
from bhf_smarthealth.utils import data_io
from bhf_smarthealth.utils import identifiers
from bhf_smarthealth.validate.qualtrics_schema import QualtricsStreamSchema
from bhf_smarthealth.exceptions import EmptyFileError, DataValidationError

logger = logging.getLogger(__name__)




def load_qualtrics_csv(file_path: Path | str) -> list[dict[str, Any]]:
    """
    Load the raw CSV file, and convert it to a flat dictionary. This function assumes row  is the header, and skips rows 1 and 2 that are multi-index rows output from the Qualtrics platform.

    Parameters
    ----------
    file_path : str | Path
        Path to the data file to be processed.

    Returns
    -------
    dict[str, Any]
        A dictionary containing the raw, unvalidated Qualtrics data, with the extracted file metadata (sheffield_id and date) appended

    """
    # Load raw file and raise a custom exception if the file is completely empty
    try:
        data = data_io.load_data_from_file(file_path, header=0, skiprows=[1, 2])
    except pd.errors.EmptyDataError:
        raise EmptyFileError(f"No usable data in file {file_path}")

    # Also want to raise an "empty file error" if only the headers exist
    if len(data) == 0:
        raise EmptyFileError(f"No usable data in file {file_path}")

    data = data.replace({np.nan: None})

    # convert dataframe to dictionary of rows
    data = data.to_dict(orient="records")

    return data





def validate_data(base_questions_data, bridge_data) -> QualtricsStreamSchema:
    """Uses Pydantic model to convert and validate the raw Qualtrics data. Stream is validated together to ensure all recieved IDs exist within each of the component files.

    Parameters
    ----------
    base_questions_data : list[dict[str, Any]]
        List of dictionaries containing one row of the raw base questions data

    bridge_data : list[dict[str, Any]]
        List of dictionaries containing one row of the raw bridge file data

    Returns
    -------
    QualtricsStreamSchema
        Pydantic model class containing the validated data

    Raises
    ------
    DataValidationError
        Raised if the pydantic schema validation fails
    """

    try:
        return QualtricsStreamSchema(
            base_questions=base_questions_data, bridge_file=bridge_data
        )
    except ValidationError as e:
        # raised ValidationError is re-raised as a custom error that inherits from the RawDataProcessingError exception class
        raise DataValidationError(f"Schema validation failed: {e}") from e





def create_dataframes(validated_data: QualtricsStreamSchema) -> dict[str, pd.DataFrame]:
    """Creates the "base_questions" and "bridge_file" dataframes from the validated Pydantic model, output as a dictionary. Includes derivation of the deterministic global_id for the bridge_file.

    Parameters
    ----------
    validated_data : QualtricsStreamSchema
        Pydantic model class containing the validated data


    Returns
    -------
    dict[str, pd.DataFrame]
        Dictionary containing the "base_questions" and "bridge_file" dataframe
    """
    validated_dict = validated_data.model_dump()

    base_df = pd.DataFrame(validated_dict["base_questions"]).set_index("sheffield_id")

    bridge_df = pd.DataFrame(validated_dict["bridge_file"]).set_index("sheffield_id")

    # deterministically generate the global_id using the sheffield_id index to ensure idempotency
    bridge_df["global_id"] = bridge_df.index.map(
        lambda x: identifiers.generate_deterministic_uuid_from_args(x)
    )

    return {"base": base_df, "bridge": bridge_df}