Source code for bhf_smarthealth.process.summary_process

"""This module utilises functions from the parse.summary_parser module to batch process raw Fitbit daily summary data"""

import logging
import pandas as pd

from pathlib import Path

from bhf_smarthealth.log import metadata_logs
from bhf_smarthealth.parse import summary_parser as bhf_sum
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) -> pd.DataFrame:
    # returns None if no file matches
    data_path = next(part_path.glob("Fitbit/*_fitbit.json"), None)

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

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

    data_dict = bhf_sum.load_json(data_path)

    parsed_data = bhf_sum.validate_data(data_dict)

    data_df = bhf_sum.create_dataframe(parsed_data)

    logger.info(
        f"Processed {part_path}. Final dataframe shape: {data_df.shape}. Dataframe columns: {data_df.columns} with index: {data_df.index.names}"
    )

    return data_df





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

    Parameters
    ----------
    in_dir : Path
        Path of the directory containing raw data to batch process
    out_path : Path
        Path to output processed data files to
    """
    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_df = process_file(part_path)

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

        except RawDataProcessingError:
            logger.exception(
                f"Error with raw Health App 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_df)
            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:
        full_df = pd.DataFrame()
        valid_IDs = []
        logger.warning("Writing empty file..")
    else:
        full_df = df_utils.merge_dataframes(df_list)
        full_df = full_df.set_index(["sheffield_id", "date"])

        valid_IDs = full_df.reset_index()["sheffield_id"].unique().tolist()

    data_io.save_dataframe_ouput(full_df, out_paths["interim"])

    if not full_df.empty:
        logger.info(
            f"Merged DataFrame saved to {out_paths['interim']} with columns: {full_df.columns}"
        )

    # output metadata log of all IDs seen 'today'
    all_IDs = [str(part_path.stem) for part_path in part_paths]

    # Both summary and intraday stream receipts will have "fitbit" as the source stream, but will be saved out to different path names to be consolidated at a later stage.
    metadata_logs.generate_daily_IDs_receipt(
        all_IDs=all_IDs,
        valid_IDs=valid_IDs,
        data_stream=DataStreamNames.FITBIT,
        date_str=date_str,
        out_path=out_paths["metadata"],
    )