"""This module contains functions to process raw GPS data ingressed to the Smart Health pipeline"""
import geopandas as gpd
import pandas as pd
import logging
import numpy as np
from typing import Any, List, Dict
from pydantic import ValidationError
from pathlib import Path
from bhf_smarthealth.exceptions import (
DataValidationError,
JSONStructureError,
EmptyFileError,
)
from bhf_smarthealth.utils import data_io, df_utils
from bhf_smarthealth.validate import metadata_validation
from bhf_smarthealth.validate.gps_schema import GPSDataList
from bhf_smarthealth.config import (
DEFAULT_GPS_CRS,
DEFAULT_PLANAR_CRS,
DEFAULT_DECIMAL_PLACES,
DEFAULT_MAX_SPEED_MS,
DEFAULT_MAX_TIME_JUMP_MINS,
)
logger = logging.getLogger(__name__)
[docs]
def load_gps_json(
file_path: str | Path,
) -> List[Dict[str, Any]]:
"""Load GPS from JSON file provided by DAISER
Parameters
----------
file_path : str | Path
Path where JSON file is located
Returns
-------
List[Dict[str, Any]]
List of dictionaries containing unvalidated, raw GPS datapoints with sheffield_id and date appended
Raises
------
KeyError
Raised if the JSON file has an unexpected structure
FileNotFoundError
Raised if file not found
ValueError
Raised if file metadata cannot be validated
"""
data = data_io.load_data_from_file(file_path)
try:
time_series = data["gps-coordinates"]["dataset"]
except KeyError as e:
raise JSONStructureError(
"GPS JSON file has an invalid structure. Missing key."
) from e
if not time_series:
raise EmptyFileError(f"No data to process in file {file_path}")
# Extract a list of dates from the datapoints
datetimes = [datapoint["d"] for datapoint in time_series if datapoint.get("d")]
# Extract and validate the metadate in the filename
metadata = metadata_validation.process_metadata(file_path, dates=datetimes)
# Append the validated metadata to the raw data
time_series_with_metadata = [
{
**datapoint,
"sheffield_id": metadata.sheffield_id,
}
for datapoint in time_series
]
logger.info(
f"Loaded {len(time_series_with_metadata)} GPS data points from {file_path}"
)
return time_series_with_metadata
[docs]
def validate_gps_data(
gps_dict_list: List[Dict[str, Any]],
) -> GPSDataList:
"""Validate the dictionaries containing the raw GPS data using Pydantic
Parameters
----------
gps_dict_list : List[GPSDataPoint]
List of dictionaries containing raw GPS datapoints
Returns
-------
GPSDataList
Pydantic model class containing the validated GPS data
Raises
------
ValidationError
Raised when input data is invalid
"""
try:
return GPSDataList.model_validate(gps_dict_list)
except ValidationError as e:
# raised ValidationError is re-raised as a custom error that inherits from the RawDataProcessingError exception class
raise DataValidationError(f"Schema validation failed: {e}") from e
[docs]
def gps_to_dataframe(
time_series: GPSDataList,
) -> pd.DataFrame:
"""Convert raw GPS pydantic model to pandas DataFrame
Parameters
----------
time_series : GPSDataList
Pydantic model class containing raw GPS data
Returns
-------
pd.DataFrame
DataFrame containing raw GPS data
"""
gps_df = pd.DataFrame(time_series.model_dump())
gps_df = gps_df.rename(columns={"d": "timestamp"})
gps_df["timestamp"] = pd.to_datetime(
gps_df["timestamp"], format="%Y-%m-%dT%H:%M:%S", errors="coerce"
) # Invalid datetimes are coerced to NaTs
invalid_dates = gps_df["timestamp"].isna()
if invalid_dates.any():
logger.warning(f"Found {invalid_dates.sum()} invalid datetime values")
gps_df = gps_df.set_index("timestamp")
return gps_df
[docs]
def add_gps_as_geometry(
gps_df: pd.DataFrame,
crs: str = DEFAULT_GPS_CRS,
) -> gpd.GeoDataFrame:
"""Converts the raw GPS latitude and longitude to a geometry column
Parameters
----------
gps_gdf : pd.DataFrame
DataFrame containing raw GPS data
crs : str, optional
Coordinate Reference System to use for GPS data, by default "EPSG:4326"
Returns
-------
gpd.GeoDataFrame
DataFrame containing raw GPS data, with geometry column
Raises
------
KeyError
Raised if DataFrame does not contain necessary latitude and/or longitude columns
"""
df_utils.validate_dataframe_columns(gps_df, ["lat", "long"])
gps_geom_series = gpd.GeoSeries.from_xy(
gps_df["long"],
gps_df["lat"],
crs=crs,
)
gps_gdf = gpd.GeoDataFrame(gps_df)
gps_gdf = gps_gdf.set_geometry(gps_geom_series)
return gps_gdf
[docs]
def calculate_distances(
gps_gdf: gpd.GeoDataFrame,
crs: str = DEFAULT_PLANAR_CRS,
decimal_places: int = DEFAULT_DECIMAL_PLACES,
) -> gpd.pd.Series:
"""Calculate distance between adjacent GeoDataFrame timepoints
Parameters
----------
gps_gdf : gpd.GeoDataFrame
DataFrame containing raw GPS data, with geometry column
crs : str, optional
Planar Coordinate Reference System to use for GPS data. Defaults to value set in config.py.
decimal_places : int, optional
Number of decimal places to round calculated distance to. Defaults to value set in config.py. If invalid value in config.py, will be set to 3.
Returns
-------
gpd.pd.Series
Series containing calculated distances. Units are determined by the CRS (metres if using "EPSG:27700", British National Grid)
Raises
------
KeyError
Raised if DataFrame does not contain necessary geometry column
ValueError
Raised if decimal_places in not a non-negative integer
Notes
-----
- The data is reprojected to a planar coordiante reference system for accurate distance calculations.
- The first value will be NaN as there is no previous point for comparison
"""
if decimal_places < 0:
decimal_places = 3
logger.warning(
"decimal_places must be a non-negative integer. Defaulting to 3."
)
df_utils.validate_dataframe_columns(gps_gdf, ["geometry"])
gps_gdf_projected = gps_gdf.to_crs(crs)
distances = gps_gdf_projected["geometry"].distance(
gps_gdf_projected["geometry"].shift()
) # distance between adjacent timepoints
return distances.round(decimal_places)
[docs]
def calculate_speeds(
gps_gdf: gpd.GeoDataFrame,
decimal_places: int = DEFAULT_DECIMAL_PLACES,
) -> gpd.pd.Series:
"""Calculate speed of travel between adjacent GeoDataframe timepoints
Parameters
----------
gps_gdf : gpd.GeoDataFrame
Dataframe containing raw GPS data, with a datetime index and a distance column
decimal_places : int, optional
Number of decimal places to round calculated distance to. Defaults to value set in config.py. If invalid value in config.py, will be set to 3.
Returns
-------
gpd.pd.Series
Series containing calculated speeds. Units are XX per second, where XX is the unit of distance in the distance column
Raises
------
KeyError
Raised if distance column is missing.
AttributeError
Raised if DataFrame does not have a datetime index
ValueError
Raised if decimal_places in not a non-negative integer
Notes
-----
- The first value will be NaN as there is no previous point for comparison
"""
if decimal_places < 0:
decimal_places = 3
logger.warning(
"decimal_places must be a non-negative integer. Defaulting to 3."
)
df_utils.validate_dataframe_columns(gps_gdf, ["distance"])
df_utils.validate_datetime_index(gps_gdf)
time_diffs = gps_gdf.index.to_series().diff().dt.total_seconds()
# Avoid ZeroDivisionError by replacing zeroes (where time difference is duplicated) with NaN
time_diffs = time_diffs.replace(0.0, np.nan)
speed = gps_gdf["distance"] / time_diffs
return speed.round(decimal_places)
[docs]
def filter_speeds(
gps_gdf: gpd.GeoDataFrame,
max_speed_ms: float = DEFAULT_MAX_SPEED_MS,
) -> gpd.GeoDataFrame:
"""Filter rows from a GeoDataframe where the speed of travel is unrealistic, i.e. less than zero or greater than the max_speed_ms
Parameters
----------
gps_gdf : gpd.GeoDataFrame
Dataframe containing raw GPS data, with a speed column
max_speed_ms : float, optional
Maximum accepted speed in metres per second. Defaults to value set in config.py.
Returns
-------
gpd.GeoDataFrame
Cleaned Dataframe
Raises
------
KeyError
Raised if speed column not in GeoDataframe
Notes
-----
- NaN values are not filtered out
"""
df_utils.validate_dataframe_columns(gps_gdf, ["speed"])
speed_mask = (gps_gdf["speed"] <= max_speed_ms) & (gps_gdf["speed"] >= 0) | gps_gdf[
"speed"
].isna()
logger.debug(f"Removed {gps_gdf.shape[0] - sum(speed_mask)} invalid speed values")
return gps_gdf[speed_mask]
[docs]
def filter_time_jumps(
gps_gdf: gpd.GeoDataFrame,
max_jump_mins: int = DEFAULT_MAX_TIME_JUMP_MINS,
) -> gpd.GeoDataFrame:
"""Nullify distance and speed for rows following a large time jump.
Parameters
----------
gps_gdf : gpd.GeoDataFrame
Dataframe containing raw GPS data, with datetime index
max_jump_mins : int, optional
Maximum acceptable time jump in whole minutes. Defaults to value set in config.py.
Returns
-------
gpd.GeoDataFrame
Cleaned Dataframe
Raises
------
AttributeError
Raised if index is not a datetime index
Notes
-----
This function sets distance and speed to NaN rather than removing rows as the location (LSOA) at that timepoint is still valid data
"""
df_utils.validate_datetime_index(gps_gdf)
max_jump_seconds = max_jump_mins * 60 # to convert mins to seconds
time_diffs = gps_gdf.index.to_series().diff().dt.total_seconds()
jump_mask = time_diffs >= max_jump_seconds
gps_gdf = gps_gdf.copy() # avoid inplace mutation
gps_gdf.loc[jump_mask, ["distance", "speed"]] = np.nan
logger.debug(f"Nullified {sum(jump_mask)} distance and speed values")
return gps_gdf
[docs]
def join_spatial_data(
gps_gdf: gpd.GeoDataFrame | pd.DataFrame,
ref_data: Any,
nan_replace_values: dict[str, str] | None = None,
**kwargs: Any,
) -> gpd.GeoDataFrame:
"""Spatial join for GPS and reference data, by default a left join with a within predicate
Parameters
----------
gps_gdf : gpd.GeoDataFrame
GPS GeoDataFrame
ref_data : Any
Reference data to spatial join to GPS data (most likely LSOA data)
Returns
-------
gpd.GeoDataFrame
GPS GeoDataFrame with reference data joined
Raises
-------
KeyError
Raised if the DataFrame doesn't have a geometry column
"""
if isinstance(gps_gdf, pd.DataFrame):
df_utils.validate_dataframe_columns(gps_gdf, ["geometry"])
gps_gdf = gpd.GeoDataFrame(gps_gdf, geometry="geometry")
defaults = {
"how": "left",
"predicate": "within",
}
kwargs = {
**defaults,
**kwargs,
}
gps_gdf = gps_gdf.sjoin(
ref_data,
**kwargs,
)
logger.debug(
f"Merged spatial data with DataFrame using the following conditions:\nHow: {kwargs['how']}, Predicate: {kwargs['predicate']}"
)
if nan_replace_values:
gps_gdf = gps_gdf.fillna(value=nan_replace_values)
return gps_gdf
[docs]
def nullify_lsoa_at_invalid_timepoints(
df: gpd.GeoDataFrame | pd.DataFrame,
lsoa_cols: list[str] = ["LSOA21CD", "LSOA21NM"],
drop_null_rows: bool = True,
) -> gpd.GeoDataFrame | pd.DataFrame:
"""Set LSOA values to NULL for bad timepoints (i.e. distance is NULL)
Parameters
----------
df : gpd.GeoDataFrame | pd.DataFrame
GPS GeoDataFrame or DataFrame
lsoa_cols: List[str], optional
List of column names containing LSOA data
drop_null_rows: bool , optional
Optionally drop the rows where the distance and LSOA values are NULL
Returns
-------
gpd.GeoDataFrame | pd.DataFrame
GPS GeoDataFrame or DataFrame
Raises
------
KeyError
Raised if distance or LSOA columns are not present
"""
df_utils.validate_dataframe_columns(df, ["distance", *lsoa_cols])
# Set LSOA to be NULL at invalid timepoints where distance is NULL (e.g. after a time jump)
nan_mask = df["distance"].isna()
df_copy = df.copy() # avoid inplace mutation
df_copy.loc[nan_mask, lsoa_cols] = np.nan
logger.debug(f"Nullified {sum(nan_mask)} LSOA values")
if drop_null_rows:
df_copy = df_copy.dropna(how="all", axis=0, subset=["distance", *lsoa_cols])
return df_copy
