bhf_smarthealth.utils.df_utils

bhf_smarthealth.utils.df_utils#

This module contains utility functions for common Pandas DataFrame operations

Attributes#

logger

Functions#

apply_func_list_to_df(...)

Apply a list of data cleaning functions to a DataFrame in turn. Logs any changes to data shape.

remove_columns(→ geopandas.GeoDataFrame | pandas.DataFrame)

Remove specified columns from a dataframe

rename_columns(→ geopandas.GeoDataFrame | pandas.DataFrame)

Rename dataframe columns from list

apply_agg_over_interval(...)

Apply dictionary of aggregation functions to a DataFrame resampled over a specified time interval.

validate_dataframe_columns(→ bool)

Validate that dataframe contains necessary columns

validate_datetime_index(→ bool)

Validates DataFrame has DatetimeIndex

merge_dataframes(→ pandas.DataFrame)

Merge a list of multiple dataframes into a single dataframe.

join_dataframes(main_df, other_dfs, **kwargs)

Join one or more other data frames to a main dataframe on their common indices

split_timestamp_to_date_and_time(→ pandas.DataFrame)

Splits timestamp column or index into separate time and date columns

get_agg_column_name(→ 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.

Module Contents#

bhf_smarthealth.utils.df_utils.logger#
bhf_smarthealth.utils.df_utils.apply_func_list_to_df(df: geopandas.GeoDataFrame | pandas.DataFrame, func_list: list[collections.abc.Callable]) → geopandas.GeoDataFrame | pandas.DataFrame[source]#

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:

Cleaned Dataframe

Return type:

gpd.GeoDataFrame | pd.DataFrame

Notes

  • Logs any changes to the shape of the DataFrame after processing

bhf_smarthealth.utils.df_utils.remove_columns(df: geopandas.GeoDataFrame | pandas.DataFrame, col_names: list[str]) → geopandas.GeoDataFrame | pandas.DataFrame[source]#

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 – DataFrame without specified columns

Return type:

gpd.GeoDataFrame | pd.DataFrame

bhf_smarthealth.utils.df_utils.rename_columns(df: geopandas.GeoDataFrame | pandas.DataFrame, col_names: list[str]) → geopandas.GeoDataFrame | pandas.DataFrame[source]#

Rename dataframe columns from list

Parameters:

  • df (gpd.GeoDataFrame | pd.DataFrame,) – DataFrame

  • col_names (list[str]) – list of column names in order

Returns:

df – DataFrame with renamed columns

Return type:

gpd.GeoDataFrame | pd.DataFrame

bhf_smarthealth.utils.df_utils.apply_agg_over_interval(df: geopandas.GeoDataFrame | pandas.DataFrame, agg_dict: dict[str, list[str]], time_period: str = '5min', ID_col: str | None = None, **kwargs: Any) → geopandas.GeoDataFrame | pandas.DataFrame[source]#

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:

Resampled dataframe with aggregated fields

Return type:

gpd.GeoDataFrame | pd.DataFrame

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

bhf_smarthealth.utils.df_utils.validate_dataframe_columns(df: pandas.DataFrame | geopandas.GeoDataFrame, required_cols: list[str]) → bool[source]#

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

bhf_smarthealth.utils.df_utils.validate_datetime_index(df: pandas.DataFrame | geopandas.GeoDataFrame) → bool[source]#

Validates DataFrame has DatetimeIndex

Parameters:

df (pd.DataFrame | gpd.GeoDataFrame) – DataFrame to validate

Raises:

AttributeError – Raised if DataFrame does not have a DatetimeIndex

bhf_smarthealth.utils.df_utils.merge_dataframes(dataframe_list: list[pandas.DataFrame], **kwargs) → pandas.DataFrame[source]#

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:

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.

Return type:

pd.DataFrame

Notes

  • Empty dataframes are filtered before merging

  • If the filtered list of dataframes is empty, a single, empty DataFrame is returned

bhf_smarthealth.utils.df_utils.join_dataframes(main_df: pandas.DataFrame, other_dfs: pandas.DataFrame | list[pandas.DataFrame], **kwargs)[source]#

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:

A single, joined pandas DataFrame containing all columns from the input dataframes

Return type:

pd.DataFrame

bhf_smarthealth.utils.df_utils.split_timestamp_to_date_and_time(dataframe: pandas.DataFrame, column_name: str = 'timestamp') → pandas.DataFrame[source]#

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:

Modified dataframe with date and time columns, returns None if no matching column or index name found

Return type:

pd.DataFrame

Raises:

ValueError – Raised if the specified column_name cannot be found within the DataFrame

bhf_smarthealth.utils.df_utils.get_agg_column_name(data_type: str, agg_type: str | functools.partial, overrides: dict | None = None) → str[source]#

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:

_description_

Return type:

_type_

On this page