Source code for bhf_smarthealth.validate.healthapp_schema

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

import logging

from datetime import datetime, date
from pydantic import (
    BaseModel,
    ConfigDict,
    model_validator,
    Field,
    AfterValidator,
)
from typing import Optional, Union, List, Literal, Annotated

logger = logging.getLogger(__name__)




def validate_offset(v: int) -> int:
    """Validate that the timezone minutes offset (from UTC+00:00) is a valid value

    Parameters
    ----------
    v : int
        Value (in minutes) of timezone offset to validate

    Returns
    -------
    int
        Validated offset value (in minutes)

    Raises
    ------
    ValueError
        Raised if the offset value is out of bounds or not in 15-minute increments
    """
    # check range between westernmost (-12:00) and easternmost (+14:00) UTC offset
    if not (-720 <= v <= 840):
        raise ValueError(f"Offset {v} is out of bounds (-720 to 840)")

    # check divisible by 15
    if v % 15 != 0:
        raise ValueError(f"Offset {v} is not aligned with 15-minute intervals")

    return v





class StandardField(BaseModel):
    model_config = ConfigDict(extra="forbid")

    source: str
    unit: str

    # 'whitelist' of expected Health App fields
    type: Literal[
        "HealthDataType.ACTIVE_ENERGY_BURNED",
        "HealthDataType.BASAL_ENERGY_BURNED",
        "HealthDataType.HEART_RATE",
        "HealthDataType.RESTING_HEART_RATE",
        "HealthDataType.STEPS",
        "HealthDataType.FLIGHTS_CLIMBED",
        "HealthDataType.SLEEP_ASLEEP",
        "HealthDataType.SLEEP_AWAKE",
        "HealthDataType.SLEEP_DEEP",
        "HealthDataType.SLEEP_LIGHT",
        "HealthDataType.SLEEP_REM",
        "HealthDataType.WALKING_SPEED",
        "HealthDataType.DISTANCE_DELTA",
    ]

    value: Optional[float] = None
    min_value: Optional[float] = None
    max_value: Optional[float] = None



    @model_validator(mode="after")
    def check_value_or_range(self) -> "StandardField":
        has_val = self.value is not None
        has_min = self.min_value is not None
        has_max = self.max_value is not None

        # must have min AND max, or neither
        if has_min != has_max:
            raise ValueError(
                "Incomplete range of values. Both 'min_value' and 'max_value' must be present."
            )

        # check for either "value" or "min_value" AND "max_value", not both cases
        if has_val == has_min:
            raise ValueError(
                "Provide either 'value' OR both 'min_value' and 'max_value'"
            )

        return self






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

    energy: int
    startTime: datetime
    type: str
    distance: Optional[float]  # some workouts do not track a distance
    endTime: datetime



# inherits all fields from the BaseHealtAppField model


class WorkoutField(BaseModel):
    type: Literal["WORKOUT"]

    count: int
    source: str
    workouts: List[IndividualWorkout]
    unit: str
    totalDistance: Optional[float]
    totalEnergy: int





class HealthAppSummaryData(BaseModel):
    """
    Validation model for the Health App data received from Daiser.
    Compulsory fields: sheffield_id, date, date of synchronisation, the UTC timezone offset in mins, and specific activity metrics.

    """

    # 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
    sync_date: datetime
    timezone_offset_mins: Annotated[int, AfterValidator(validate_offset)]

    # the discriminator distinguishes between standard Health App fields and the nested "WORKOUT" fields
    health_data: List[
        Annotated[Union[WorkoutField, StandardField], Field(discriminator="type")]
    ]