Source code for bhf_smarthealth.parse.intraday_parser

"""This module contains functions to process raw Fitbit intraday activity data ingressed to the Smart Health pipeline"""

import pandas as pd
import logging

from pathlib import Path
from pydantic import ValidationError
from typing import List, Dict, Any

from bhf_smarthealth.validate.fitbit_schema import IntradayDataList
from bhf_smarthealth.utils import data_io
from bhf_smarthealth.validate import metadata_validation
from bhf_smarthealth.exceptions import DataValidationError, EmptyFileError

logger = logging.getLogger(__name__)


# Load raw json


def load_json(json_path: str | Path, activity_source: str) -> List[Dict[str, Any]]:
    """Load the raw JSON file and return list of dictionaries for validation.

    Parameters
    ----------
    json_path : str | Path
        JSON file Path
    activity_source : str
        The source of intraday activity data to load (e.g., heart rate or steps).
        The key is constructed as "activities-{activity_source}-intraday".

    Returns
    -------
    List[Dict[str, Any]]
        List of dictionaries containing raw, unvalidated Fitbit activity datapoints with sheffield_id and date appended

    Raises
    ------
    FileNotFoundError
        Raised if file not found
    ValueError
        Raised if file metadata cannot be validated

    """
    # Load raw JSON data
    data = data_io.load_data_from_file(json_path)

    # The next line should be conditioned on the Activity source (HR or steps, extracted from file name)
    # Construct the full name for the time series data
    activity_source_full = f"activities-{activity_source}-intraday"
    time_series = data[activity_source_full]["dataset"]

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

    # Extract a list of dates from the datapoints
    datetimes = [datapoint["d"] for datapoint in time_series if datapoint.get("d")]

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

    # Append the validated metadata to the raw data
    time_series_with_metadata = [
        {**datapoint, "sheffield_id": metadata.sheffield_id}
        for datapoint in time_series
    ]

    logger.info(
        f"Loaded {len(time_series_with_metadata)} intraday {activity_source} data points from {json_path}"
    )

    return time_series_with_metadata





def validate_data(intraday_data_dict: List[Dict[str, Any]]) -> IntradayDataList:
    """Uses Pydantic model to convert and validate the dictionaries
    containing the raw data.

    Parameters
    ----------
    intraday_data_dict : List of intraday_datapoint
        List of dictionaries containing raw activity datapoints
    Returns
    -------
    IntradayDataList
        Pydantic model class containing the converted data
    Raises
    ------
    DataValidationError
        Raised if the pydantic schema validation fails
    """
    try:
        # Use model_validate for best practice when loading a dictionary
        return IntradayDataList.model_validate(intraday_data_dict)
    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_dataframe(time_series: IntradayDataList) -> pd.DataFrame:
    """Creates a pandas DataFrame from a Pydantic model instance.

    Parameters
    ----------
    time_series : IntradayDataList
        The Pydantic model instance containing the time-series data,
        which is expected to have a 'timestamp' field.

    Returns
    -------
    pd.DataFrame
        A DataFrame containing the data from the Pydantic model, with the
        'timestamp' field converted to datetime objects and set as the index.
    """
    intraday_df = pd.DataFrame(time_series.model_dump())
    intraday_df["timestamp"] = pd.to_datetime(
        intraday_df["timestamp"], format="%Y-%m-%dT%H:%M:%S", errors="coerce"
    )  # Invalid datetimes are coerced to NaTs

    invalid_dates = intraday_df["timestamp"].isna()
    if invalid_dates.any():
        logger.warning(f"Found {invalid_dates.sum()} invalid datetime values")

    intraday_df.set_index(
        "timestamp",
        inplace=True,
    )

    return intraday_df