Process raw Fitbit Intraday activity data

Process raw Fitbit Intraday activity data#

This notebook demonstrates the use of Smart Health modules to process synthetic raw Fitbit Intraday activities data ready for ingress to the “Intermediate” bucket.

# Import required libraries and modules
import logging
from pathlib import Path
from bhf_smarthealth.parse import intraday_parser as bhf_intraday
from bhf_smarthealth.utils import df_utils

# Initiate logging
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Path to the raw example data
hr_data_path = Path("../../../tests/data/raw/UOS001_20251204_heart.json")
steps_data_path = Path("../../../tests/data/raw/UOS001_20250714_steps.json")

  1. Load in the raw data

heart_data = bhf_intraday.load_json(hr_data_path, "heart")
steps_data = bhf_intraday.load_json(steps_data_path, "steps")

---------------------------------------------------------------------------
FileNotFoundError                         Traceback (most recent call last)
Cell In[2], line 1
----> 1 heart_data = bhf_intraday.load_json(hr_data_path, "heart")
      2 steps_data = bhf_intraday.load_json(steps_data_path, "steps")

File ~/work/BHF-SmartHealth/BHF-SmartHealth/src/bhf_smarthealth/parse/intraday_parser.py:44, in load_json(json_path, activity_source)
     20 """Load the raw JSON file and return list of dictionaries for validation.
     21 
     22 Parameters
   (...)     41 
     42 """
     43 # Load raw JSON data
---> 44 data = data_io.load_data_from_file(json_path)
     46 # The next line should be conditioned on the Activity source (HR or steps, extracted from file name)
     47 # Construct the full name for the time series data
     48 activity_source_full = f"activities-{activity_source}-intraday"

File ~/work/BHF-SmartHealth/BHF-SmartHealth/src/bhf_smarthealth/utils/data_io.py:48, in load_data_from_file(file_path, **kwargs)
     45     case _:
     46         raise ValueError("File extension not recognised.")
---> 48 with open(
     49     file_path,
     50     mode,
     51 ) as file:
     52     data = load_func(file, **kwargs)
     54 logger.debug(f"Data successfully loaded from {file_path}")

FileNotFoundError: [Errno 2] No such file or directory: '../../../tests/data/raw/UOS001_20251204_heart.json'

  1. The raw data is validated using the Pydantic library

This step checks that each datapoint contains an ID, timestamp, and a value field of type str, datetime and int, respectively

validated_heart_data = bhf_intraday.validate_data(heart_data)
validated_steps_data = bhf_intraday.validate_data(steps_data)

  1. A Pandas DataFrame is created from the validated data

validated_heart_df = bhf_intraday.create_dataframe(validated_heart_data)
validated_steps_df = bhf_intraday.create_dataframe(validated_steps_data)

# Show the first five rows of the heart data
validated_heart_df.head()

  1. Aggregate the data into reduced intervals and calculate aggregate measures

The intraday data is aggregated into two time resolutions:

  1. Short intervals, by default 5 minutes. This resolution is selected to balance granularity of data with storage costs

  2. 24hr period, to derive Daily Summary metrics such as daily average (mean) heart rate

INTRADAY_TIME_PERIOD = "5min"
DAILY_TIME_PERIOD = "24h"

# This dictionary dictates which aggregate calculations are performed for each column in the each of the intraday activity data types
DEFAULT_FB_INTRADAY_AGG_DICT = {
    "heart": {"value": ["mean", "min", "max", "size"]},
    "steps": {"value": ["sum", "size"]},
}

# Aggregate the heart data into 5 minutes intervals
intraday_aggregated_heart_df = df_utils.apply_agg_over_interval(
    df=validated_heart_df,
    time_period=INTRADAY_TIME_PERIOD,
    agg_dict=DEFAULT_FB_INTRADAY_AGG_DICT["heart"],
    ID_col="sheffield_id",
)

# Aggregate the heart data into a daily summary
daily_aggregated_heart_df = df_utils.apply_agg_over_interval(
    df=validated_heart_df,
    time_period=DAILY_TIME_PERIOD,
    agg_dict=DEFAULT_FB_INTRADAY_AGG_DICT["heart"],
    ID_col="sheffield_id",
)

# Show the first five rows
intraday_aggregated_heart_df.head()

# Show the aggregated daily metrics
daily_aggregated_heart_df.head()

  1. We rename the column names for clarity

In particular, the “size” field name is replaced with “n_samples” as this is more descriptive

rename_heart_columns = [
    "heart_n_samples" if agg_type == "size" else f"heart_{agg_type}"
    for agg_type in DEFAULT_FB_INTRADAY_AGG_DICT["heart"]["value"]
]

# Rename the columns in the intraday aggregated dataframe
intraday_aggregated_heart_df = df_utils.rename_columns(
    df=intraday_aggregated_heart_df, col_names=rename_heart_columns
)

# Rename the columns in the daily aggregated dataframe
daily_aggregated_heart_df = df_utils.rename_columns(
    df=daily_aggregated_heart_df, col_names=rename_heart_columns
)

# Show the first five rows
intraday_aggregated_heart_df.head()

4b and 5b: Repeat the same procedure for the steps data

# Aggregate the steps data into 5 minutes intervals
intraday_aggregated_steps_df = df_utils.apply_agg_over_interval(
    df=validated_steps_df,
    time_period=INTRADAY_TIME_PERIOD,
    agg_dict=DEFAULT_FB_INTRADAY_AGG_DICT["steps"],
    ID_col="sheffield_id",
)

# Aggregate the steps data into a daily summary
daily_aggregated_steps_df = df_utils.apply_agg_over_interval(
    df=validated_steps_df,
    time_period=DAILY_TIME_PERIOD,
    agg_dict=DEFAULT_FB_INTRADAY_AGG_DICT["steps"],
    ID_col="sheffield_id",
)

rename_steps_columns = [
    "steps_n_samples" if agg_type == "size" else f"steps_{agg_type}"
    for agg_type in DEFAULT_FB_INTRADAY_AGG_DICT["steps"]["value"]
]

# Rename the columns in the intraday aggregated dataframe
intraday_aggregated_steps_df = df_utils.rename_columns(
    df=intraday_aggregated_steps_df, col_names=rename_steps_columns
)

# Rename the columns in the daily aggregated dataframe
daily_aggregated_steps_df = df_utils.rename_columns(
    df=daily_aggregated_steps_df, col_names=rename_steps_columns
)

# Show the first five rows
intraday_aggregated_steps_df.head()

# Show the steps daily summary
daily_aggregated_steps_df

  1. These daily aggregate measures will be combined with the Daily Summary data for this user, before the combined data is stored in the “Intermediate” bucket