bhf_smarthealth.parse.healthapp_parser#
This module contains functions to process raw Health App daily summary data ingressed to the Smart Health pipeline
Attributes#
Functions#
|
Load the raw JSON file, reshape it into a flat dictionary, and validate |
|
Uses Pydantic model to convert and validate the dictionary containig the raw data. |
|
Creates the metrics and workouts dataframes from the Pydantic model. |
Module Contents#
- bhf_smarthealth.parse.healthapp_parser.logger#
- bhf_smarthealth.parse.healthapp_parser.load_json(json_path: str | pathlib.Path) dict[str, Any][source]#
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:
A dictionary containing the raw, unvalidated Health App daily summary data, with the extracted file metadata (sheffield_id and date) appended
- Return type:
dict[str, Any]
- Raises:
JSONStructureError – Raised if the JSON file does not contain the expected keys
- bhf_smarthealth.parse.healthapp_parser.validate_data(data_dict: dict[str, Any]) bhf_smarthealth.validate.healthapp_schema.HealthAppSummaryData[source]#
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:
Pydantic model class containing the validated data
- Return type:
- Raises:
DataValidationError – Raised if the pydantic schema validation fails
- bhf_smarthealth.parse.healthapp_parser.create_dataframes(validated_data: bhf_smarthealth.validate.healthapp_schema.HealthAppSummaryData) dict[str, pandas.DataFrame][source]#
Creates the metrics and workouts dataframes from the Pydantic model.
- Parameters:
validated_data (HealthAppSummaryData) – Pydantic model class containing the validated data
- Returns:
A dictionary containing the “metrics” and “workouts” DataFrames
- Return type:
dict[str, pd.DataFrame]