Source code for bhf_smarthealth.validate.fitbit_schema

"""This module contains pydantic validation models for the raw Fitbit data ingested into the Smart Health pipeline"""

import logging

from datetime import datetime, date
from pydantic import BaseModel, ConfigDict, model_validator, Field, RootModel
from typing_extensions import Self
from typing import Optional


logger = logging.getLogger(__name__)




class FitbitSummaryDatapoint(BaseModel):
    model_config = ConfigDict(validate_by_alias=True, extra="forbid")

    # Maps JSON key 'd' (datetime) to Python field 'timestamp'
    timestamp: datetime = Field(alias="dateTime")

    # Checks if value is a non-negative integer
    value: int = Field(ge=0)
    units: str





class FitbitSummaryData(BaseModel):
    """
    Validation model for the Fitbit summary data received from Daiser.
    Compulsory fields: sheffield_id, timestamp, and specific activity metrics.
    All other fields are optional.

    Each of these fields should contain a nested JSON object containing a dateTime, value, and units field.
    """

    # Raise ValidationError if any unexpected additional data fields are present
    model_config = ConfigDict(extra="forbid")

    # Compulsary fields
    # Derived from filename metadata
    sheffield_id: str
    date: date
    # Raw data fields
    calories: FitbitSummaryDatapoint
    minutesSedentary: FitbitSummaryDatapoint

    # Optional fields
    steps: Optional[FitbitSummaryDatapoint] = None
    minutesLightlyActive: Optional[FitbitSummaryDatapoint] = None
    minutesFairlyActive: Optional[FitbitSummaryDatapoint] = None
    minutesVeryActive: Optional[FitbitSummaryDatapoint] = None
    HeartRatePeakMinutes: Optional[FitbitSummaryDatapoint] = None
    HeartRateCardioMinutes: Optional[FitbitSummaryDatapoint] = None
    HeartRateFatBurnMinutes: Optional[FitbitSummaryDatapoint] = None
    HeartRateResting: Optional[FitbitSummaryDatapoint] = None
    minutesAsleep: Optional[FitbitSummaryDatapoint] = None
    minutesAwake: Optional[FitbitSummaryDatapoint] = None
    minutesToFallAsleep: Optional[FitbitSummaryDatapoint] = None
    timeInBed: Optional[FitbitSummaryDatapoint] = None
    efficiency: Optional[FitbitSummaryDatapoint] = None



    @model_validator(mode="before")
    @classmethod
    def log_optional_fields_present(cls, data: dict) -> dict:
        required_fields = {
            key for key, value in cls.model_fields.items() if value.is_required()
        }

        # Any key in the dataset that is not a required key must be optional
        optional_fields = set(
            key
            for key in data.keys()
            if key in cls.model_fields.keys() and key not in required_fields
        )

        if optional_fields:
            logger.info(
                f"Additional optional fields present: {sorted(list(optional_fields))}"
            )

        return data


    # Checks if the values of activity fields sum to 1440 minutes, which is the total minutes in a day


    @model_validator(mode="after")
    def check_sum_of_minutes(self) -> Self:
        activity_fields = [
            self.minutesSedentary,
            self.minutesLightlyActive,
            self.minutesFairlyActive,
            self.minutesVeryActive,
        ]

        available_activity_fields = [field.value for field in activity_fields if field]

        total_mins = sum(available_activity_fields)

        if total_mins != 24 * 60:
            raise ValueError(
                f"{total_mins} total minutes in activity fields is different than total minutes in a day"
            )

        return self




# 1. Model for a single data point (inside "dataset")


class IntradayDataPoint(BaseModel):
    """
    Represents a measurement for a single fitbit intraday timepoint.
    """

    model_config = ConfigDict(validate_by_alias=True)

    # Maps JSON key 'd' (datetime) to Python field 'timestamp'
    timestamp: datetime = Field(alias="d")

    # Maps JSON key 'v' (value) to Python field 'value'
    value: int = Field(alias="v")

    # Extracted from file metadata
    sheffield_id: str



# 2. Model for the Intraday data container


class IntradayDataList(RootModel):
    """
    Represents the full list of intraday data points captured for a single Intraday metric.
    It manages the Intraday time-series data, e.g. Steps, Heart Rate, Calories, etc collected by Fitbit.
    Please note, that while it should be compatible with other intraday measures from fitbit,
    it has only been tested with Steps and Heart Rate
    """

    root: list[IntradayDataPoint]