Source code for bhf_smarthealth.parse.summary_parser

"""This module contains functions to process raw Fitbit daily summary data ingressed to the Smart Health pipeline"""

import pandas as pd
import logging
from typing import Dict, Any
from pydantic import ValidationError
from pathlib import Path
from bhf_smarthealth.validate.fitbit_schema import FitbitSummaryData
from bhf_smarthealth.utils import data_io
from bhf_smarthealth.exceptions import DataValidationError, EmptyFileError
from bhf_smarthealth.validate import metadata_validation

logger = logging.getLogger(__name__)




def load_json(json_path: str | Path) -> Dict[str, Any]:
    """
    Load the raw JSON file, reshape it into a flat dictionary, and validate
    data consistency against filename metadata.

    The function extracts the Sheffield ID and date from the filename and
    compares the filename date against all date portions of timestamps found
    within the data for consistency. Files with inconsistent dates (either
    externally or internally inconsistent) will raise a ValueError.

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

    Returns
    -------
    Dict[str, Any]
        A flat dictionary containing the raw, unvalidated Fitbit daily summary data, with sheffield_id and date appended

    Raises
    ------
    ValueError
        - If the filename date is known but data dates do not match it (External Consistency Failure).
        - If the filename date is unknown (Mandatory Error).
    """
    # Load raw JSON data
    json_dict = data_io.load_data_from_file(json_path)

    if not json_dict:
        raise EmptyFileError(f"No data to process in file {json_path}")

    # Convert original, nested json into something flatter
    data_dict = {key: value[0] for key, value in json_dict.items() if value}

    # Extract a list of dates from the datapoints
    dates = [
        value[0].get("dateTime")
        for value in json_dict.values()
        if isinstance(value, list) and len(value) > 0 and "dateTime" in value[0]
    ]

    # Extract and validate the metadata in the filename
    metadata = metadata_validation.process_metadata(filepath=json_path, dates=dates)

    # Add the validated metadata to the data dict
    data_dict = {
        "date": metadata.file_date,
        "sheffield_id": metadata.sheffield_id,
        **data_dict,
    }

    logger.info(f"Loaded Daily Summary data from {json_path}")
    logger.debug(f"Loaded Daily Summary data contains columns: {data_dict.keys()}")

    return data_dict





def validate_data(data_dict: Dict[str, Any]) -> FitbitSummaryData:
    """Uses Pydantic model to convert and validate the dictionary containig the raw data.
    This should raise an error if there is an issue.

    Parameters
    ----------
    data_dict : dict[str, str]
        Dictionary of the raw data

    Returns
    -------
    FitbitSummaryData
        Pydantic model class containing the converted data

    Raises
    ------
    DataValidationError
        Raised if the pydantic schema validation fails
    """
    try:
        return FitbitSummaryData(**data_dict)
    except (ValidationError, ValueError) 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_dataframe(summary_data: FitbitSummaryData) -> pd.DataFrame:
    """Creates a dataframe from the Pydantic model with nested FitbitSummaryDatapoints under the daily activity fields. From these nested fields, the "value" is extracted.

    Parameters
    ----------
    summary_data : FitbitSummaryData
        Pydantic model class containing the converted data

    Returns
    -------
    pd.DataFrame
        DataFrame containing data
    """
    validated_dict = summary_data.model_dump()

    # Need to extract value from the nested metric fields
    extracted_dict = {
        key: (val.get("value") if isinstance(val, dict) else val)
        for key, val in validated_dict.items()
    }

    df = pd.DataFrame([extracted_dict])

    logger.debug(f"DataFrame created with shape {df.shape}")

    return df