"""This module contains functions to process raw Health App daily summary data ingressed to the Smart Health pipeline"""
import pandas as pd
import logging
from datetime import datetime, timedelta
from typing import Any
from pathlib import Path
from pydantic import ValidationError
from bhf_smarthealth.validate.healthapp_schema import HealthAppSummaryData
from bhf_smarthealth.utils import data_io, identifiers
from bhf_smarthealth.validate import metadata_validation
from bhf_smarthealth.exceptions import (
DataValidationError,
JSONStructureError,
EmptyFileError,
)
from bhf_smarthealth.config import DEFAULT_HA_WORKOUT_UNITS
logger = logging.getLogger(__name__)
[docs]
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 validates the metadata.
Parameters
----------
json_path : str | Path
Path to the data file to be processed.
Returns
-------
dict[str, Any]
A dictionary containing the raw, unvalidated Health App daily summary data, with the extracted file metadata (sheffield_id and date) appended
Raises
------
JSONStructureError
Raised if the JSON file does not contain the expected keys
"""
# 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}")
try:
data_list = json_dict[0]["health_data"]
metadata_date = str(
datetime.fromisoformat(json_dict[0]["when"]) + timedelta(days=1)
) # needed due to offset bug
timezone_offset_mins = json_dict[0]["timezone_offset"]
sync_date = json_dict[0]["synced"]
except KeyError as e:
raise JSONStructureError(
"Health app JSON file has an invalid structure. Missing key."
) from e
# Extract and validate the metadata in the filename
metadata = metadata_validation.process_metadata(
filepath=json_path,
dates=metadata_date,
)
# Add the validated metadata to the data dict
data_dict = {
"date": metadata.file_date,
"sheffield_id": metadata.sheffield_id,
"sync_date": sync_date,
"timezone_offset_mins": timezone_offset_mins,
"health_data": data_list,
}
logger.info(f"Loaded Health App Daily Summary data from {json_path}")
logger.debug(
f"Loaded Health App Daily Summary data contains {len(data_list)} fields."
)
return data_dict
[docs]
def validate_data(data_dict: dict[str, Any]) -> HealthAppSummaryData:
"""Uses Pydantic model to convert and validate the dictionary containig the raw data.
Parameters
----------
data_dict : dict[str, Any]
Dictionary of the raw data
Returns
-------
HealthAppSummaryData
Pydantic model class containing the validated data
Raises
------
DataValidationError
Raised if the pydantic schema validation fails
"""
try:
return HealthAppSummaryData.model_validate(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
[docs]
def create_dataframes(validated_data: HealthAppSummaryData) -> dict[str, pd.DataFrame]:
"""Creates the metrics and workouts dataframes from the Pydantic model.
Parameters
----------
validated_data : HealthAppSummaryData
Pydantic model class containing the validated data
Returns
-------
dict[str, pd.DataFrame]
A dictionary containing the "metrics" and "workouts" DataFrames
"""
validated_dict = validated_data.model_dump()
metrics_rows = []
workout_rows = []
sheffield_id = validated_dict["sheffield_id"]
date = validated_dict["date"]
timezone_offset_mins = validated_dict["timezone_offset_mins"]
sync_date = validated_dict["sync_date"]
# loop through each health_data array
for entry in validated_dict["health_data"]:
entry_type = entry.get("type")
source = entry.get("source")
# common metadata shared across all rows in the metrics table
common_meta = {
"sheffield_id": sheffield_id,
"date": date,
"sync_date": sync_date,
"timezone_offset_mins": timezone_offset_mins,
"source": source,
}
# if workout field, extract the workout summary metrics
if entry_type != "WORKOUT":
metrics_rows.append(
{
**common_meta,
# the key is a hash of fields that WILL NOT CHANGE with late arriving data
"metric_id": identifiers.generate_deterministic_uuid_from_args(
sheffield_id, date, entry_type, source
),
"type": entry_type,
"unit": entry.get("unit"),
"value": entry.get("value"),
"min_value": entry.get("min_value"),
"max_value": entry.get("max_value"),
"workout_summary_id": identifiers.generate_nil_uuid(), # No workout link, use 'sentinel value' of nil UUID rather than NULL
}
)
continue
# this is a referential identifier that links parent workout summary metrics to the child workouts
# the key is a hash of fields that WILL NOT CHANGE with late arriving data
workout_summary_id = identifiers.generate_deterministic_uuid_from_args(
sheffield_id, date, source, "WORKOUT"
)
# This might need updating once we know the units of the workout summary metric fields
for name, unit in DEFAULT_HA_WORKOUT_UNITS.items():
metrics_rows.append(
{
**common_meta,
# the key is a hash of fields that WILL NOT CHANGE with late arriving data
"metric_id": identifiers.generate_deterministic_uuid_from_args(
sheffield_id, date, name, source
),
"workout_summary_id": workout_summary_id, # referential link to the child workouts
"type": name,
"value": entry.get(name),
"unit": unit,
"min_value": entry.get("min_value"),
"max_value": entry.get("max_value"),
}
)
for exercise in entry.get("workouts", []):
workout_rows.append(
{
# the key is a hash of fields that WILL NOT CHANGE with late arriving data
"workout_id": identifiers.generate_deterministic_uuid_from_args(
sheffield_id,
date,
exercise.get("type"),
exercise.get("startTime"),
),
"workout_summary_id": workout_summary_id, # referential link to the child parent summary metrics
"sheffield_id": sheffield_id,
"date": date,
"workout_type": exercise.get("type"),
"workout_energy": exercise.get("energy"),
"workout_distance": exercise.get("distance"),
"start_time": exercise.get("startTime"),
"end_time": exercise.get("endTime"),
}
)
df_dict = {
"metrics": pd.DataFrame(metrics_rows),
"workouts": pd.DataFrame(workout_rows),
}
# Set index only if data exists else will throw and error
if metrics_rows:
df_dict["metrics"].set_index("metric_id", inplace=True)
if workout_rows:
df_dict["workouts"].set_index("workout_id", inplace=True)
logger.debug(
f"Metrics DataFrame created with shape {df_dict['metrics'].shape} and columns {df_dict['metrics'].columns}"
)
logger.debug(
f"Workouts DataFrame created with shape {df_dict['workouts'].shape} and columns {df_dict['workouts'].columns}"
)
return df_dict
