Source code for bhf_smarthealth.process.intraday_process
"""This module utilises functions from the parse.intraday_parser module to batch process raw Fitbit intraday activity data"""
import pandas as pd
import logging
from pathlib import Path
from bhf_smarthealth.log import metadata_logs
from bhf_smarthealth.parse import intraday_parser as bhf_intraday
from bhf_smarthealth.utils import df_utils, data_io
from bhf_smarthealth.exceptions import (
RawDataProcessingError,
EmptyFileError,
)
from bhf_smarthealth.models import DataStreamNames
from bhf_smarthealth.config import (
DEFAULT_FB_INTRADAY_AGG_DICT,
DEFAULT_FB_INTRADAY_AGG_SETTINGS,
DEFAULT_FB_INTRADAY_COLUMN_OVERRIDES,
)
logger = logging.getLogger(__name__)
[docs]
def process_file(
part_path: Path, data_type: str, rename_columns: list[str]
) -> dict[str, pd.DataFrame]:
# returns None if no file matches
data_path = next(part_path.glob(f"Fitbit/*_{data_type}.json"), None)
if not data_path:
raise FileNotFoundError(
f"No file matching the glob pattern 'Fitbit/*_{data_type}.json' could be found in the directory {part_path}"
)
logger.info(f"Processing {data_path}")
data_dict = bhf_intraday.load_json(data_path, data_type)
parsed_data = bhf_intraday.validate_data(data_dict)
data_df = bhf_intraday.create_dataframe(parsed_data)
data_df = data_df.astype({"value": "int16"})
out_dict = {}
# Perform both a daily and intraday aggregation of the Fitbit data
for name, settings in DEFAULT_FB_INTRADAY_AGG_SETTINGS.items():
# get optional kwargs
kwargs = settings.get("kwargs", {})
agg_df = df_utils.apply_agg_over_interval(
df=data_df,
agg_dict=DEFAULT_FB_INTRADAY_AGG_DICT[data_type],
time_period=DEFAULT_FB_INTRADAY_AGG_SETTINGS[name]["time_period"],
ID_col="sheffield_id",
**kwargs,
)
agg_df = df_utils.rename_columns(
df=agg_df,
col_names=rename_columns,
)
# convert timestamp to date and set index for daily aggregate resolution
if name == "summary":
agg_df = (
agg_df.reset_index()
.assign(date=lambda df_: df_["timestamp"].dt.date)
.drop(columns=["timestamp"])
.set_index(DEFAULT_FB_INTRADAY_AGG_SETTINGS[name]["index"])
)
else:
# ensure correct indices
agg_df = agg_df.reset_index().set_index(
[idx for idx in DEFAULT_FB_INTRADAY_AGG_SETTINGS[name]["index"]]
)
out_dict[name] = agg_df
logger.info(
f"Processed {part_path}. Final {name} data length: {len(out_dict[name].values)}"
)
return out_dict
[docs]
def process_day(
in_dir: Path, out_paths: dict[str, Path], data_type: str, date_str: str
):
"""Process a daily batch of raw Fitbit intraday activity .json files
Parameters
----------
in_dir : Path
Path of the directory containing raw data to batch process
out_paths : dict[str, Path]
Dictionary of Paths (values) to save each of the aggregation periods (keys) to
data_type : str
Intraday activity data type, e.g. heart, steps
"""
# Defines aggregation names we want to override
overrides = DEFAULT_FB_INTRADAY_COLUMN_OVERRIDES
rename_columns = [
# get string of column name from functools.partial filtered Series objects
df_utils.get_agg_column_name(
agg_type=agg_type, data_type=data_type, overrides=overrides
)
for agg_type in DEFAULT_FB_INTRADAY_AGG_DICT[data_type]["value"]
]
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, data_type, rename_columns)
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:
df_list.append(
{key: pd.DataFrame() for key in DEFAULT_FB_INTRADAY_AGG_SETTINGS.keys()}
)
valid_IDs_set = set()
# Combine the DataFrames for both the Daily summary and intraday data
for agg_resolution, settings in DEFAULT_FB_INTRADAY_AGG_SETTINGS.items():
unwrapped_list = [processed_file[agg_resolution] for processed_file in df_list]
full_df = df_utils.merge_dataframes(unwrapped_list)
if full_df.empty:
logger.warning("Writing empty file..")
data_io.save_dataframe_ouput(full_df, out_paths[agg_resolution])
continue
data_io.save_dataframe_ouput(full_df, out_paths[agg_resolution])
logger.info(
f"Merged DataFrame saved to {out_paths[agg_resolution]} with columns: {full_df.columns}"
)
# add the valid IDs seen to the set of valid IDs.
valid_IDs_set.update(full_df.reset_index()["sheffield_id"].unique().tolist())
# output metadata log of all IDs seen 'today'
all_IDs = [str(part_path.stem) for part_path in part_paths]
valid_IDs = list(valid_IDs_set)
# 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"],
)
