Source code for bhf_smarthealth.process.healthapp_process

"""This module utilises functions from the parse.healthapp_parser module to batch process raw Health App metrics data"""

import logging
import pandas as pd

from pathlib import Path
from pandas import DataFrame

from bhf_smarthealth.log import metadata_logs
from bhf_smarthealth.parse import healthapp_parser as bhf_ha
from bhf_smarthealth.utils import df_utils, data_io
from bhf_smarthealth.exceptions import (
    RawDataProcessingError,
    EmptyFileError,
)
from bhf_smarthealth.models import DataStreamNames


logger = logging.getLogger(__name__)




def process_file(part_path: Path) -> dict[str, DataFrame]:
    """Process an individual raw Health App .json file

    Parameters
    ----------
    part_path : Path
       Path to subdirectory containing raw Health App .json file to be processed

    Returns
    -------
    dict[str, DataFrame]
        Dictionary containing "workouts" and "metrics" DataFrames

    Raises
    ------
    FileNotFoundError
        Raised if a healthapp/*_healthdata.json file cannot be located in this subdirectory
    """
    # returns None if no file matches
    data_path = next(part_path.glob("healthapp/*_healthdata.json"), None)

    if not data_path:
        raise FileNotFoundError(
            f"No file matching the glob pattern 'healthapp/*_healthdata.json' could be found in the directory {part_path}"
        )

    logger.info(f"Processing {data_path}")

    data_dict = bhf_ha.load_json(data_path)

    parsed_data = bhf_ha.validate_data(data_dict)

    data_dicts = bhf_ha.create_dataframes(parsed_data)

    logger.info(f"Successfully processed {part_path}")

    for name in ["workouts", "metrics"]:
        logger.debug(
            f"Processed {part_path}. Final metrics dataframe shape: {data_dicts[name].shape}. Dataframe columns: {data_dicts[name].columns} with index: {data_dicts[name].index.names}"
        )

    return data_dicts





def process_day(in_dir: Path, out_paths: dict[str, Path], date_str: str):
    """Process a daily batch of raw healthapp .json files

    Parameters
    ----------
    in_dir : Path
        Path of the directory containing raw data to batch process
    out_paths : Path
        Dictionary containing Paths to output the merged "workouts" and "metrics" DataFrames to
    date_str:
        String of date being processed
    """

    df_list = []
    quality_counter = metadata_logs.DataQualityCounter()

    part_paths = [
        part_path for part_path in sorted(in_dir.iterdir()) if part_path.is_dir()
    ]

    for part_path in part_paths:
        try:
            processed_dfs = process_file(part_path)
        # base exception which catches all raw data validation issues

        except EmptyFileError:
            logger.warning(f"File {part_path} is empty. Skipping file..")
            quality_counter["empty"] += 1
            continue

        except RawDataProcessingError:
            logger.exception(
                f"Error with raw data file at {part_path}. Skipping file.."
            )
            quality_counter["processing_error"] += 1
            continue

        except FileNotFoundError:
            logger.error(f"File missing in {part_path}. Skipping file..")
            quality_counter["missing"] += 1
            continue

        else:
            df_list.append(processed_dfs)
            quality_counter["valid"] += 1

    # check whether enough valid data was processed
    metadata_logs.evaluate_daily_processing_success(counter=quality_counter)

    # back stop for DEFAULT_VALID_DATA_THRESHOLD being set to 0
    if quality_counter["valid"] == 0:
        df_list.append({"metrics": pd.DataFrame(), "workouts": pd.DataFrame()})

    valid_IDs_set = set()

    # Combine the DataFrames from each participant for both the Health App Metrics and Workouts
    for table in ["workouts", "metrics"]:
        # collect the DataFrames comprising one of the output tables
        frames_per_table = [processed_file[table] for processed_file in df_list]

        # merge the DataFrames from each participant into a single, daily DataFrame
        full_df = df_utils.merge_dataframes(frames_per_table)

        # It is possible that no data was recorded for one of the tables so we should continue to process the next table.
        if full_df.empty:
            logger.warning(
                f"No valid {table} data processed from {in_dir}. Writing empty file.."
            )
            data_io.save_dataframe_ouput(full_df, out_paths[table])
            continue

        data_io.save_dataframe_ouput(full_df, out_paths[table])

        logger.info(
            f"Merged DataFrame saved to {out_paths[table]} with shape: {full_df.shape}, columns: {full_df.columns}, and index: {full_df.index.names}"
        )

        # add the valid IDs seen to the set of valid IDs
        valid_IDs_set.update(full_df.reset_index()["sheffield_id"].unique().tolist())

    all_IDs = [str(part_path.stem) for part_path in part_paths]
    valid_IDs = list(valid_IDs_set)

    metadata_logs.generate_daily_IDs_receipt(
        all_IDs=all_IDs,
        valid_IDs=valid_IDs,
        data_stream=DataStreamNames.HEALTHAPP,
        date_str=date_str,
        out_path=out_paths["metadata"],
    )