Source code for bhf_smarthealth.process.gps_process
"""This module utilises functions from the parse.GPS_parser module to batch process raw GPS data"""
import logging
from pathlib import Path
import pandas as pd
from geopandas import GeoDataFrame
from pandas import DataFrame
from bhf_smarthealth.config import (
DEFAULT_GPS_AGG_DICT,
DEFAULT_GPS_OUTPUT_COLUMNS,
DEFAULT_REMOVE_GPS_COLUMNS,
DEFAULT_RENAME_GPS_COLUMNS,
DEFAULT_TIME_PERIOD,
DEFAULT_REPLACE_NON_MATCHED_LSOA_VALUES,
)
from bhf_smarthealth.exceptions import (
EmptyFileError,
RawDataProcessingError,
)
from bhf_smarthealth.log import metadata_logs
from bhf_smarthealth.models import DataStreamNames
from bhf_smarthealth.parse import gps_parser as bhf_gps
from bhf_smarthealth.utils import data_io, df_utils
logger = logging.getLogger(__name__)
[docs]
def process_gps_file(
part_path: Path, lsoa_data: GeoDataFrame
) -> DataFrame | GeoDataFrame:
"""Process an individual raw GPS .json file
Parameters
----------
part_path : Path
Path to subdirectory containing raw GPS .json file to be processed
lsoa_data : GeoDataFrame
DataFrame containing reference LSOA information
Returns
-------
DataFrame | GeoDataFrame
DataFrame containing the aggregated, pseudonymised LSOA data and mobility measures
Raises
------
FileNotFoundError
Raised if a *_gps.json file cannot be located in this subdirectory
"""
# returns None if no file matches
data_path = next(part_path.glob("*_gps.json"), None)
if not data_path:
raise FileNotFoundError(
f"No file matching the glob pattern '*_gps.json' could be found in the directory {part_path}"
)
logger.info(f"Processing {data_path}")
gps_json = bhf_gps.load_gps_json(data_path)
gps_val = bhf_gps.validate_gps_data(gps_json)
gps_gdf = bhf_gps.gps_to_dataframe(gps_val)
gps_gdf = bhf_gps.add_gps_as_geometry(gps_gdf)
gps_gdf["distance"] = bhf_gps.calculate_distances(gps_gdf)
gps_gdf["speed"] = bhf_gps.calculate_speeds(gps_gdf)
gps_gdf_clean = df_utils.apply_func_list_to_df(
df=gps_gdf,
func_list=[
bhf_gps.filter_speeds,
bhf_gps.filter_time_jumps,
],
)
gps_lsoa_gdf = bhf_gps.join_spatial_data(
gps_gdf_clean,
lsoa_data,
nan_replace_values=DEFAULT_REPLACE_NON_MATCHED_LSOA_VALUES,
)
lsoa_gdf = df_utils.remove_columns(
gps_lsoa_gdf, col_names=DEFAULT_REMOVE_GPS_COLUMNS
)
lsoa_gdf_clean = bhf_gps.nullify_lsoa_at_invalid_timepoints(
lsoa_gdf, drop_null_rows=True
)
lsoa_agg = df_utils.apply_agg_over_interval(
lsoa_gdf_clean,
agg_dict=DEFAULT_GPS_AGG_DICT,
time_period=DEFAULT_TIME_PERIOD,
ID_col="sheffield_id",
)
lsoa_agg = df_utils.rename_columns(
df=lsoa_agg,
col_names=DEFAULT_RENAME_GPS_COLUMNS,
)
logger.info(
f"Processed {part_path}. Final dataframe shape: {lsoa_agg.shape}. Dataframe columns: {lsoa_agg.columns} with index: {lsoa_agg.index.names}"
)
return lsoa_agg
[docs]
def process_day(
in_dir: Path, out_paths: dict[str, Path], lsoa_data: GeoDataFrame, date_str: str
):
"""Process a daily batch of raw GPS .json files
Parameters
----------
in_dir : Path
Path of the subdirectory of raw GPS data to batch process
out_paths : dict[str, Path]
Dictionary of Paths (values) for the required output tables (keys)
lsoa_data : GeoDataFrame
DataFrame containing reference LSOA information
"""
df_list = []
quality_counter = metadata_logs.DataQualityCounter()
part_paths = [
part_path for part_path in sorted(in_dir.iterdir()) if part_path.is_dir()
]
for part_path in part_paths:
try:
processed_df = process_gps_file(part_path, lsoa_data)
except EmptyFileError:
logger.error(f"File {part_path} is empty. Skipping file..")
quality_counter["empty"] += 1
continue
except RawDataProcessingError:
logger.exception(
f"Error with raw Health App data file at {part_path}. Skipping file.."
)
quality_counter["processing_error"] += 1
continue
except FileNotFoundError:
logger.error(f"File missing in {part_path}. Skipping file..")
quality_counter["missing"] += 1
continue
else:
df_list.append(processed_df)
quality_counter["valid"] += 1
# check whether enough valid data was processed
metadata_logs.evaluate_daily_processing_success(counter=quality_counter)
# back stop for DEFAULT_VALID_DATA_THRESHOLD being set to 0
if quality_counter["valid"] == 0:
full_df = pd.DataFrame()
valid_IDs = []
else:
full_df = df_utils.merge_dataframes(df_list)
valid_IDs = full_df.reset_index()["sheffield_id"].unique().tolist()
# Subset the dataframe and save the result to corresponding parquet file
for file, cols in DEFAULT_GPS_OUTPUT_COLUMNS.items():
if full_df.empty:
logger.warning("Writing empty file..")
data_io.save_dataframe_ouput(
full_df, out_paths[file]
) # save empty .parquet if insufficient data
else:
full_df_subset = full_df[cols]
full_df_subset = full_df_subset.rename(
# prepend the "n_samples" fields with the table name
columns={"n_samples": f"{file.split('_')[0]}_n_samples"}
)
data_io.save_dataframe_ouput(full_df_subset, out_paths[file])
logger.info(
f"Merged DataFrame saved to {out_paths[file]} with columns: {cols} and index: {full_df_subset.index.names}"
)
# output metadata log of all IDs seen 'today'
all_IDs = [str(part_path.stem) for part_path in part_paths]
metadata_logs.generate_daily_IDs_receipt(
all_IDs=all_IDs,
valid_IDs=valid_IDs,
data_stream=DataStreamNames.GPS,
date_str=date_str,
out_path=out_paths["metadata"],
)
