"""This module contains utility functions for common Pandas DataFrame operations"""
import functools
import logging
from collections.abc import Callable
from typing import Any
import geopandas as gpd
import numpy as np
import pandas as pd
logger = logging.getLogger(__name__)
[docs]
def apply_func_list_to_df(
df: gpd.GeoDataFrame | pd.DataFrame,
func_list: list[Callable],
) -> gpd.GeoDataFrame | pd.DataFrame:
"""Apply a list of data cleaning functions to a DataFrame in turn. Logs any changes to data shape.
Parameters
----------
df: gpd.GeoDataFrame | pd.DataFrame,
Dataframe containing data to be cleaned
func_list : list[Callable]
list of functions to apply to dataframe
Returns
-------
gpd.GeoDataFrame | pd.DataFrame
Cleaned Dataframe
Notes
-----
- Logs any changes to the shape of the DataFrame after processing
"""
original_size = df.shape
for func in func_list:
df = func(df)
logger.debug(f"Ran {func.__name__}")
filtered_size = df.shape
if original_size != filtered_size:
logger.debug(
f"Dataframe filtered. Original shape: {original_size}. New shape: {filtered_size}."
)
return df
[docs]
def remove_columns(
df: gpd.GeoDataFrame | pd.DataFrame,
col_names: list[str],
) -> gpd.GeoDataFrame | pd.DataFrame:
"""Remove specified columns from a dataframe
Parameters
----------
df: gpd.GeoDataFrame | pd.DataFrame,
DataFrame
col_names : list[str]
Names of columns to be removed
Returns
-------
df: gpd.GeoDataFrame | pd.DataFrame
DataFrame without specified columns
"""
logger.debug(f"Dropped {len(col_names)} columns: {col_names}.")
return df.drop(columns=col_names)
[docs]
def rename_columns(
df: gpd.GeoDataFrame | pd.DataFrame, col_names: list[str]
) -> gpd.GeoDataFrame | pd.DataFrame:
"""Rename dataframe columns from list
Parameters
----------
df: gpd.GeoDataFrame | pd.DataFrame,
DataFrame
col_names : list[str]
list of column names in order
Returns
-------
df: gpd.GeoDataFrame | pd.DataFrame
DataFrame with renamed columns
"""
logger.debug(f"Renaming columns to {col_names}")
return df.set_axis(labels=col_names, axis="columns")
[docs]
def apply_agg_over_interval(
df: gpd.GeoDataFrame | pd.DataFrame,
agg_dict: dict[str, list[str]],
time_period: str = "5min",
ID_col: str | None = None,
**kwargs: Any,
) -> gpd.GeoDataFrame | pd.DataFrame:
"""Apply dictionary of aggregation functions to a DataFrame resampled over a specified time interval.
Parameters
----------
df : gpd.GeoDataFrame | pd.DataFrame
DataFrame with a datetime index and fields to be aggregated
agg_dict : dict[str, list[str]]
Dictionary containing column names and aggregation functions to be applied
time_period : str, optional
Time period to resample data over, by default "5min"
ID_col : str, optional
ID column name to group by
Returns
-------
gpd.GeoDataFrame | pd.DataFrame
Resampled dataframe with aggregated fields
Raises
------
AttributeError
Raised if DataFrame has no datetime index or the specified ID_col does not exists in the DataFrame
Notes
-----
- Columns are renamed to join the original column name with the aggregate function
- Empty time windows are removed from the result
"""
defaults = {
"label": "left",
"closed": "left",
}
kwargs = {**defaults, **kwargs}
# Resample datetime index over time period
validate_datetime_index(df)
grouper = [pd.Grouper(level="timestamp", freq=time_period, **kwargs)]
if ID_col is not None:
if ID_col in df.columns:
grouper_params = {"key": ID_col}
elif ID_col in df.index.names:
grouper_params = {"level": ID_col}
else:
raise AttributeError(
f"Specified ID_col {ID_col} does not exist in the DataFrame"
)
grouper.insert(0, pd.Grouper(**grouper_params))
# Apply aggregation functions
# TypeChecker ignore as types confuse checker. .agg() correctly implemented at runtime
grouped_pseudo_df = df.groupby(grouper, observed=False).agg(agg_dict) # type: ignore
# Flatten multi-index column names (if needed) as columns are a multiindex (column name, aggregate function)
flat_col_list = [
"_".join(col).strip() if isinstance(col, tuple) else col
for col in grouped_pseudo_df.columns.values
]
grouped_pseudo_df.columns = flat_col_list
logger.debug(f"Flattened columns: {grouped_pseudo_df.columns}")
# Drop rows where all of the aggregated cols are NaN
grouped_pseudo_df = grouped_pseudo_df.replace([0, 0.0, None], np.nan).dropna(
how="all"
)
return grouped_pseudo_df
[docs]
def validate_dataframe_columns(
df: pd.DataFrame | gpd.GeoDataFrame, required_cols: list[str]
) -> bool:
"""Validate that dataframe contains necessary columns
Parameters
----------
df : pd.DataFrame | gpd.GeoDataFrame
DataFrame to validate
required_cols : list[str]
list of columns required
Raises
------
KeyError
Raised if DataFrame does not contain the required fields
"""
missing = list(set(required_cols) - set(df.columns))
if missing:
raise KeyError(
f"Dataframe missing required columns. Expected {required_cols}, missing {missing}"
)
return True
[docs]
def validate_datetime_index(
df: pd.DataFrame | gpd.GeoDataFrame,
) -> bool:
"""Validates DataFrame has DatetimeIndex
Parameters
----------
df : pd.DataFrame | gpd.GeoDataFrame
DataFrame to validate
Raises
------
AttributeError
Raised if DataFrame does not have a DatetimeIndex
"""
has_multiindex_datetime = isinstance(df.index, pd.MultiIndex) and any(
isinstance(level, pd.DatetimeIndex) for level in df.index.levels
)
has_datetime = isinstance(df.index, pd.DatetimeIndex)
if not has_multiindex_datetime and not has_datetime:
raise AttributeError("Dataframe missing DatetimeIndex")
return True
[docs]
def merge_dataframes(dataframe_list: list[pd.DataFrame], **kwargs) -> pd.DataFrame:
"""Merge a list of multiple dataframes into a single dataframe.
Effectively just a a wrapper around pandas.concat.
Parameters
----------
dataframe_list : list[pd.DataFrame]
list of dataframes, one per participant
Returns
-------
pd.DataFrame
A single, concatenated pandas DataFrame containing all the rows from the
input list. DataFrame will be empty if no non-empty DataFrames exist in dataframe_list.
Notes
-----
* Empty dataframes are filtered before merging
* If the filtered list of dataframes is empty, a single, empty DataFrame is returned
"""
valid_dataframe_list = [df for df in dataframe_list if not df.empty]
if not valid_dataframe_list:
logger.warning(
f"No non-empty dataframes were found out of {len(dataframe_list)} dataframes. Returning empty dataframe."
)
return pd.DataFrame()
else:
if len(valid_dataframe_list) != len(dataframe_list):
logger.warning(
f"{len(dataframe_list) - len(valid_dataframe_list)} empty dataframes were found and will not be merged"
)
merged_dfs = pd.concat(valid_dataframe_list, **kwargs)
logger.debug(
f"Merged {len(valid_dataframe_list)} DataFrames into one with columns: {merged_dfs.columns} and index: {merged_dfs.index}"
)
return merged_dfs
[docs]
def join_dataframes(
main_df: pd.DataFrame, other_dfs: pd.DataFrame | list[pd.DataFrame], **kwargs
):
"""Join one or more other data frames to a main dataframe on their common indices
Effectively just a wrapper around Dataframe.join, and will use the same defaults.
Parameters
----------
main_df : pd.DataFrame
primary dataframe to join other dataframes to
other_dfs : pd.DataFrame | list[pd.DataFrame]
One or more other dataframes to be joined to the main dataframe
Returns
-------
pd.DataFrame
A single, joined pandas DataFrame containing all columns from the input dataframes
"""
logger.debug(f"Joining {len(list(other_dfs)) + 1} DataFrames into one")
return main_df.join(other_dfs, **kwargs)
[docs]
def split_timestamp_to_date_and_time(
dataframe: pd.DataFrame, column_name: str = "timestamp"
) -> pd.DataFrame:
"""Splits timestamp column or index into separate time and date columns
Will look for a column first then in the index
Parameters
----------
dataframe : pd.DataFrame
Incoming dataframe
column_name : Optional[str], optional
column or index to split, by default "timestamp"
Returns
-------
pd.DataFrame
Modified dataframe with date and time columns, returns None if no matching column or index name found
Raises
------
ValueError
Raised if the specified column_name cannot be found within the DataFrame
"""
if column_name in dataframe.columns:
timestamp = dataframe[column_name].dt
elif column_name in (dataframe.index.names or []):
timestamp = dataframe.index.get_level_values(column_name)
else:
raise ValueError(
f"Could not find a column or index level named '{column_name}' "
f"in the DataFrame."
)
dataframe["date"] = timestamp.date
dataframe["time"] = timestamp.time
return dataframe
[docs]
def get_agg_column_name(
data_type: str, agg_type: str | functools.partial, overrides: dict | None = None
) -> str:
"""Returns the name of an aggregated column combined with the data type. Includes option to override specific column names using a dictionary of overrides.
Parameters
----------
data_type : str
The data type being aggregated, for example "heart" or "steps"
agg_type : str | functools.partial
The type of aggregation e.g. min, max, size, mean, or a functools.partial generated by bhf_smarthealth.utils.aggregators.config_filtered_aggs factory
overrides : Optional[dict]
A dictionary containing columns to be overridden (keys) and what to overwrite the names with (values)
Returns
-------
_type_
_description_
"""
# getattr is needed to extract the __name__ from any Partials. Defaults to the str(agg_type) if __name__ is not available, i.e. the agg_type is a standard string
name = getattr(agg_type, "__name__", str(agg_type))
if overrides:
# default to the above name if not in the overrides dict
name = overrides.get(name, name)
# e.g. heart_n_samples or steps_min
return f"{data_type}_{name}"
