Source code for bhf_smarthealth.validate.gps_schema

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

import logging

from datetime import datetime
from pydantic import BaseModel, RootModel, ConfigDict, model_validator
from typing_extensions import Self


logger = logging.getLogger(__name__)




class GPSDataPoint(BaseModel):
    """
    Represents a measurement for a single GPS timepoint.
    """

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

    # From raw data
    d: datetime
    long: float
    lat: float
    # Extracted from file metadata
    sheffield_id: str





class GPSDataList(RootModel):
    """
    Validation model for a full list of GPS timepoints.
    Will filter out timepoints with invlaid GPS coordinates after initial validation.
    """

    root: list[GPSDataPoint]

    # filter out invalid GPSDataPoints after initial validation
    # prevents entire .json being rejected for a few bad points.


    @model_validator(mode="after")
    def filter_invalid_gps_points(self) -> Self:
        valid_points = []

        for gps_point in self.root:
            is_lat_valid = -90.0 <= gps_point.lat <= 90.0
            is_long_valid = -180.0 <= gps_point.long <= 180.0

            if is_lat_valid and is_long_valid:
                valid_points.append(gps_point)

        if len(valid_points) != len(self.root):
            logger.warning(
                f"Removed {len(self.root) - len(valid_points)} invalid GPS points"
            )
        # Replace the original list with the filtered list
        self.root = valid_points
        return self