From 99e667a157bfeb7a2bda1110253579d6e559057f Mon Sep 17 00:00:00 2001 From: Meng Li <34143965+Meng6@users.noreply.github.com> Date: Thu, 28 Apr 2022 01:16:51 +0000 Subject: [PATCH 1/2] Add platforms output for pull_phone_data rule --- Snakefile | 2 + rules/common.smk | 11 +++ rules/preprocessing.smk | 21 ++++- ...ge_phone_platforms_for_all_participants.py | 10 +++ ...e_platforms_for_individual_participants.py | 17 ++++ src/data/streams/pull_phone_data.R | 87 +++++++++++-------- 6 files changed, 109 insertions(+), 39 deletions(-) create mode 100644 src/data/streams/merge_phone_platforms_for_all_participants.py create mode 100644 src/data/streams/merge_phone_platforms_for_individual_participants.py diff --git a/Snakefile b/Snakefile index 7dc5ad3a3..52d0c9caf 100644 --- a/Snakefile +++ b/Snakefile @@ -397,9 +397,11 @@ if config["HEATMAP_FEATURE_CORRELATION_MATRIX"]["PLOT"]: # Data Cleaning for provider in config["ALL_CLEANING_INDIVIDUAL"]["PROVIDERS"].keys(): if config["ALL_CLEANING_INDIVIDUAL"]["PROVIDERS"][provider]["COMPUTE"]: + files_to_compute.extend(expand("data/interim/platforms/{pid}/all_sensor_platforms.csv", pid=config["PIDS"])) files_to_compute.extend(expand("data/processed/features/{pid}/all_sensor_features_cleaned_" + provider.lower() +".csv", pid=config["PIDS"])) for provider in config["ALL_CLEANING_OVERALL"]["PROVIDERS"].keys(): if config["ALL_CLEANING_OVERALL"]["PROVIDERS"][provider]["COMPUTE"]: + files_to_compute.extend(expand("data/interim/platforms/all_participants/all_sensor_platforms.csv", pid=config["PIDS"])) files_to_compute.extend(expand("data/processed/features/all_participants/all_sensor_features_cleaned_" + provider.lower() +".csv")) rule all: diff --git a/rules/common.smk b/rules/common.smk index 017330c1f..93e5d19d7 100644 --- a/rules/common.smk +++ b/rules/common.smk @@ -10,6 +10,17 @@ def get_script_language(script_path): return "r" +# Preprocessing.smk ######################################################################################################### +def input_merge_phone_platforms_for_individual_participants(wildcards): + platforms_files = [] + for config_key in config.keys(): + if config_key.startswith("PHONE") and "PROVIDERS" in config[config_key] and isinstance(config[config_key]["PROVIDERS"], dict) and config_key != "PHONE_DATA_YIELD": + for provider_key, provider in config[config_key]["PROVIDERS"].items(): + if "COMPUTE" in provider.keys() and provider["COMPUTE"]: + platforms_files.append("data/interim/platforms/{pid}/" + config_key.lower() + "_platforms.csv") + break + return platforms_files + # Features.smk ######################################################################################################### def optional_phone_yield_input_for_locations(wildcards): if config["PHONE_LOCATIONS"]["LOCATIONS_TO_USE"] in ["ALL_RESAMPLED","FUSED_RESAMPLED"]: diff --git a/rules/preprocessing.smk b/rules/preprocessing.smk index 70f8ea632..ebce3d42a 100644 --- a/rules/preprocessing.smk +++ b/rules/preprocessing.smk @@ -19,10 +19,29 @@ rule pull_phone_data: sensor = "phone_" + "{sensor}", tables = lambda wildcards: config["PHONE_" + str(wildcards.sensor).upper()]["CONTAINER"], output: - "data/raw/{pid}/phone_{sensor}_raw.csv" + sensor_raw_file = "data/raw/{pid}/phone_{sensor}_raw.csv", + platforms_file = "data/interim/platforms/{pid}/phone_{sensor}_platforms.csv" script: "../src/data/streams/pull_phone_data.R" +rule merge_phone_platforms_for_individual_participants: + input: + platforms_files = input_merge_phone_platforms_for_individual_participants + params: + pid = "{pid}" + output: + "data/interim/platforms/{pid}/all_sensor_platforms.csv" + script: + "../src/data/streams/merge_phone_platforms_for_individual_participants.py" + +rule merge_phone_platforms_for_all_participants: + input: + platforms_files = expand("data/interim/platforms/{pid}/all_sensor_platforms.csv", pid=config["PIDS"]) + output: + "data/interim/platforms/all_participants/all_sensor_platforms.csv" + script: + "../src/data/streams/merge_phone_platforms_for_all_participants.py" + rule process_time_segments: input: segments_file = config["TIME_SEGMENTS"]["FILE"], diff --git a/src/data/streams/merge_phone_platforms_for_all_participants.py b/src/data/streams/merge_phone_platforms_for_all_participants.py new file mode 100644 index 000000000..12c979a5c --- /dev/null +++ b/src/data/streams/merge_phone_platforms_for_all_participants.py @@ -0,0 +1,10 @@ +import pandas as pd + + +platforms_of_all_participants = pd.DataFrame() +for platforms_file_path in snakemake.input["platforms_files"]: + platforms_per_participant = pd.read_csv(platforms_file_path) + platforms_per_participant.insert(0, "pid", platforms_file_path.split("/")[3]) + platforms_of_all_participants = pd.concat([platforms_of_all_participants, platforms_per_participant], axis=0) + +platforms_of_all_participants.to_csv(snakemake.output[0], index=False) diff --git a/src/data/streams/merge_phone_platforms_for_individual_participants.py b/src/data/streams/merge_phone_platforms_for_individual_participants.py new file mode 100644 index 000000000..c202b9d4c --- /dev/null +++ b/src/data/streams/merge_phone_platforms_for_individual_participants.py @@ -0,0 +1,17 @@ +import pandas as pd +import numpy as np + +pid = snakemake.params["pid"] + +merged_platforms = pd.DataFrame() +for platforms_file_path in snakemake.input["platforms_files"]: + platforms_per_sensor = pd.read_csv(platforms_file_path) + if merged_platforms.shape[0] < platforms_per_sensor.shape[0]: + merged_platforms = platforms_per_sensor + elif merged_platforms.shape[0] == platforms_per_sensor.shape[0]: + if (merged_platforms["os"] == platforms_per_sensor["os"]).all(): + merged_platforms["timestamp"] = np.minimum(merged_platforms["timestamp"], platforms_per_sensor["timestamp"]) + else: + raise ValueError("Phone platforms of " + pid + " inferred from different sensors are not consistent. Please check it manually!") + +merged_platforms.to_csv(snakemake.output[0], index=False) diff --git a/src/data/streams/pull_phone_data.R b/src/data/streams/pull_phone_data.R index 6ef589cfe..407657270 100644 --- a/src/data/streams/pull_phone_data.R +++ b/src/data/streams/pull_phone_data.R @@ -125,7 +125,8 @@ pull_phone_data <- function(){ tables <- snakemake@params[["tables"]] sensor <- toupper(snakemake@params[["sensor"]]) device_type <- "phone" - output_data_file <- snakemake@output[[1]] + output_sensor_raw_file <- snakemake@output[["sensor_raw_file"]] + output_platforms_file <- snakemake@output[["platforms_file"]] validate_participant_file_without_device_ids(participant_file) participant_data <- read_yaml(participant_file) @@ -144,49 +145,59 @@ pull_phone_data <- function(){ expected_columns <- tolower(rapids_schema[[sensor]]) participant_data <- setNames(data.frame(matrix(ncol = length(expected_columns), nrow = 0)), expected_columns) + platforms_data <- setNames(data.frame(matrix(ncol=2, nrow=0)), c("timestamp", "os")) + if(length(devices) == 0){ warning("There were no PHONE device ids in this participant file:", participant_file) - write_csv(participant_data, output_data_file) - return() - } + }else{ + container_functions <- load_container_script(stream_container) + infer_device_os_container <- container_functions$infer_device_os + pull_data_container <- container_functions$pull_data - container_functions <- load_container_script(stream_container) - infer_device_os_container <- container_functions$infer_device_os - pull_data_container <- container_functions$pull_data - - for(idx in seq_along(devices)){ - - device <- devices[idx] - message(paste0("\nProcessing ", sensor, " for ", device)) - device_os <- ifelse(device_oss[idx] == "infer", infer_device_os_container(data_configuration, device), device_oss[idx]) - validate_inferred_os(basename(stream_container), participant_file, device, device_os) - - if(!toupper(device_os) %in% names(stream_schema[[sensor]])){ # the current sensor is only available in a single OS (like PHONE_MESSAGES) - warning(sensor, " data is not available for ", device_os, ". No data to download for ", device) - next - } + for(idx in seq_along(devices)){ + + device <- devices[idx] + message(paste0("\nProcessing ", sensor, " for ", device)) + device_os <- ifelse(device_oss[idx] == "infer", infer_device_os_container(data_configuration, device), device_oss[idx]) + validate_inferred_os(basename(stream_container), participant_file, device, device_os) + + if(!toupper(device_os) %in% names(stream_schema[[sensor]])){ # the current sensor is only available in a single OS (like PHONE_MESSAGES) + warning(sensor, " data is not available for ", device_os, ". No data to download for ", device) + next + } + + os_table <- ifelse(length(tables) > 1, tables[[toupper(device_os)]], tables) # some sensor tables have a different name for android and ios - os_table <- ifelse(length(tables) > 1, tables[[toupper(device_os)]], tables) # some sensor tables have a different name for android and ios - - columns_to_download <- c(stream_schema[[sensor]][[toupper(device_os)]][["RAPIDS_COLUMN_MAPPINGS"]], stream_schema[[sensor]][[toupper(device_os)]][["MUTATION"]][["COLUMN_MAPPINGS"]]) - columns_to_download <- columns_to_download[(columns_to_download != "FLAG_TO_MUTATE")] - data <- pull_data_container(data_configuration, device, sensor, os_table, columns_to_download) - - if(!setequal(columns_to_download, colnames(data))) - stop(paste0("The pulled data for ", device, " does not have the expected columns (including [RAPIDS_COLUMN_MAPPINGS] and [MUTATE][COLUMN_MAPPINGS]). The container script returned [", paste(colnames(data), collapse=","),"] but the format mappings expected [",paste(columns_to_download, collapse=","), "]. The conainer script is: ", stream_container)) - - renamed_data <- rename_columns(columns_to_download, data) - - mutation_scripts <- stream_schema[[sensor]][[toupper(device_os)]][["MUTATION"]][["SCRIPTS"]] - mutated_data <- mutate_data(mutation_scripts, renamed_data, data_configuration) - - if(!setequal(expected_columns, colnames(mutated_data))) - stop(paste0("The mutated data for ", device, " does not have the columns RAPIDS expects. The mutation script returned [", paste(colnames(mutated_data), collapse=","),"] but RAPIDS expected [",paste(expected_columns, collapse=","), "]. One ore more mutation scripts in [", sensor,"][MUTATION][SCRIPTS] are adding extra columns or removing or not adding the ones expected")) - participant_data <- rbind(participant_data, mutated_data %>% distinct()) + columns_to_download <- c(stream_schema[[sensor]][[toupper(device_os)]][["RAPIDS_COLUMN_MAPPINGS"]], stream_schema[[sensor]][[toupper(device_os)]][["MUTATION"]][["COLUMN_MAPPINGS"]]) + columns_to_download <- columns_to_download[(columns_to_download != "FLAG_TO_MUTATE")] + data <- pull_data_container(data_configuration, device, sensor, os_table, columns_to_download) + if(!setequal(columns_to_download, colnames(data))) + stop(paste0("The pulled data for ", device, " does not have the expected columns (including [RAPIDS_COLUMN_MAPPINGS] and [MUTATE][COLUMN_MAPPINGS]). The container script returned [", paste(colnames(data), collapse=","),"] but the format mappings expected [",paste(columns_to_download, collapse=","), "]. The conainer script is: ", stream_container)) + + renamed_data <- rename_columns(columns_to_download, data) + + mutation_scripts <- stream_schema[[sensor]][[toupper(device_os)]][["MUTATION"]][["SCRIPTS"]] + mutated_data <- mutate_data(mutation_scripts, renamed_data, data_configuration) + + if(!setequal(expected_columns, colnames(mutated_data))) + stop(paste0("The mutated data for ", device, " does not have the columns RAPIDS expects. The mutation script returned [", paste(colnames(mutated_data), collapse=","),"] but RAPIDS expected [",paste(expected_columns, collapse=","), "]. One ore more mutation scripts in [", sensor,"][MUTATION][SCRIPTS] are adding extra columns or removing or not adding the ones expected")) + if(nrow(mutated_data) > 0){ + platforms_data <- rbind(platforms_data, list(timestamp=min(mutated_data$timestamp), os=device_os)) + participant_data <- rbind(participant_data, mutated_data %>% distinct()) + } + } + + participant_data <- participant_data %>% arrange(timestamp) + platforms_data <- platforms_data %>% arrange(timestamp) + if(lengths(unique(platforms_data$os)) == 1 && nrow(platforms_data) > 1){ + platforms_data <- head(platforms_data, 1) + } } - participant_data <- participant_data %>% arrange(timestamp) - write_csv(participant_data, output_data_file) + + write_csv(participant_data, output_sensor_raw_file) + write_csv(platforms_data, output_platforms_file) + return() } pull_phone_data() \ No newline at end of file From b9e71da20cb4f17b7534631c5c1c1d4f9e5e6959 Mon Sep 17 00:00:00 2001 From: Meng Li <34143965+Meng6@users.noreply.github.com> Date: Fri, 29 Apr 2022 01:00:05 +0000 Subject: [PATCH 2/2] Fix the bug of data cleaning module to support multiple platforms --- Snakefile | 3 +- docs/analysis/data-cleaning.md | 3 ++ docs/change-log.md | 1 + rules/features.smk | 6 ++- rules/preprocessing.smk | 21 ++++++++- ...e_platforms_for_individual_participants.py | 21 +++++---- src/data/streams/pull_phone_data.R | 7 +-- .../all_cleaning_individual/rapids/main.R | 33 +++++++++++--- .../all_cleaning_overall/rapids/main.R | 43 ++++++++++++++++--- 9 files changed, 109 insertions(+), 29 deletions(-) diff --git a/Snakefile b/Snakefile index 52d0c9caf..45f06738f 100644 --- a/Snakefile +++ b/Snakefile @@ -398,10 +398,11 @@ if config["HEATMAP_FEATURE_CORRELATION_MATRIX"]["PLOT"]: for provider in config["ALL_CLEANING_INDIVIDUAL"]["PROVIDERS"].keys(): if config["ALL_CLEANING_INDIVIDUAL"]["PROVIDERS"][provider]["COMPUTE"]: files_to_compute.extend(expand("data/interim/platforms/{pid}/all_sensor_platforms.csv", pid=config["PIDS"])) + files_to_compute.extend(expand("data/interim/platforms/{pid}/all_sensor_platforms_with_datetime.csv", pid=config["PIDS"])) files_to_compute.extend(expand("data/processed/features/{pid}/all_sensor_features_cleaned_" + provider.lower() +".csv", pid=config["PIDS"])) for provider in config["ALL_CLEANING_OVERALL"]["PROVIDERS"].keys(): if config["ALL_CLEANING_OVERALL"]["PROVIDERS"][provider]["COMPUTE"]: - files_to_compute.extend(expand("data/interim/platforms/all_participants/all_sensor_platforms.csv", pid=config["PIDS"])) + files_to_compute.extend(expand("data/interim/platforms/all_participants/all_sensor_platforms_with_datetime.csv", pid=config["PIDS"])) files_to_compute.extend(expand("data/processed/features/all_participants/all_sensor_features_cleaned_" + provider.lower() +".csv")) rule all: diff --git a/docs/analysis/data-cleaning.md b/docs/analysis/data-cleaning.md index dede1e9eb..e6a8a3527 100644 --- a/docs/analysis/data-cleaning.md +++ b/docs/analysis/data-cleaning.md @@ -7,6 +7,8 @@ The goal of this module is to perform basic clean tasks on the behavioral featur !!! info "File Sequence" ```bash + - data/interim/platforms/{pid}/all_sensor_platforms.csv + - data/interim/platforms/{pid}/all_sensor_platforms_with_datetime.csv - data/processed/features/{pid}/all_sensor_features.csv - data/processed/features/{pid}/all_sensor_features_cleaned_{provider_key}.csv ``` @@ -80,6 +82,7 @@ Steps to clean sensor features for individual participants. It only considers th !!! info "File Sequence" ```bash + - data/interim/platforms/all_participants/all_sensor_platforms_with_datetime.csv - data/processed/features/all_participants/all_sensor_features.csv - data/processed/features/all_participants/all_sensor_features_cleaned_{provider_key}.csv ``` diff --git a/docs/change-log.md b/docs/change-log.md index 558ba47c9..07c963164 100644 --- a/docs/change-log.md +++ b/docs/change-log.md @@ -2,6 +2,7 @@ ## v1.9.0 - Upgrade generics package - Optimize memory usage in readable_datetime.R script +- Fix the bug of data cleaning module to support multiple platforms ## v1.8.0 - Add data stream for AWARE Micro server - Fix the NA bug in PHONE_LOCATIONS BARNETT provider diff --git a/rules/features.smk b/rules/features.smk index f0fea9459..675a5bbcc 100644 --- a/rules/features.smk +++ b/rules/features.smk @@ -961,7 +961,8 @@ rule merge_sensor_features_for_all_participants: rule clean_sensor_features_for_individual_participants: input: - sensor_data = rules.merge_sensor_features_for_individual_participants.output + sensor_data = rules.merge_sensor_features_for_individual_participants.output, + platforms = "data/interim/platforms/{pid}/all_sensor_platforms_with_datetime.csv" wildcard_constraints: pid = "("+"|".join(config["PIDS"])+")" params: @@ -975,7 +976,8 @@ rule clean_sensor_features_for_individual_participants: rule clean_sensor_features_for_all_participants: input: - sensor_data = rules.merge_sensor_features_for_all_participants.output + sensor_data = rules.merge_sensor_features_for_all_participants.output, + platforms = "data/interim/platforms/all_participants/all_sensor_platforms_with_datetime.csv" params: provider = lambda wildcards: config["ALL_CLEANING_OVERALL"]["PROVIDERS"][wildcards.provider_key.upper()], provider_key = "{provider_key}", diff --git a/rules/preprocessing.smk b/rules/preprocessing.smk index ebce3d42a..826f7b6ed 100644 --- a/rules/preprocessing.smk +++ b/rules/preprocessing.smk @@ -34,11 +34,28 @@ rule merge_phone_platforms_for_individual_participants: script: "../src/data/streams/merge_phone_platforms_for_individual_participants.py" +rule phone_platforms_with_datetime: + input: + sensor_input = "data/interim/platforms/{pid}/all_sensor_platforms.csv", + time_segments = "data/interim/time_segments/{pid}_time_segments.csv", + pid_file = "data/external/participant_files/{pid}.yaml", + tzcodes_file = input_tzcodes_file, + params: + device_type = "phone_platforms", + timezone_parameters = config["TIMEZONE"], + pid = "{pid}", + time_segments_type = config["TIME_SEGMENTS"]["TYPE"], + include_past_periodic_segments = config["TIME_SEGMENTS"]["INCLUDE_PAST_PERIODIC_SEGMENTS"] + output: + "data/interim/platforms/{pid}/all_sensor_platforms_with_datetime.csv" + script: + "../src/data/datetime/readable_datetime.R" + rule merge_phone_platforms_for_all_participants: input: - platforms_files = expand("data/interim/platforms/{pid}/all_sensor_platforms.csv", pid=config["PIDS"]) + platforms_files = expand("data/interim/platforms/{pid}/all_sensor_platforms_with_datetime.csv", pid=config["PIDS"]) output: - "data/interim/platforms/all_participants/all_sensor_platforms.csv" + "data/interim/platforms/all_participants/all_sensor_platforms_with_datetime.csv" script: "../src/data/streams/merge_phone_platforms_for_all_participants.py" diff --git a/src/data/streams/merge_phone_platforms_for_individual_participants.py b/src/data/streams/merge_phone_platforms_for_individual_participants.py index c202b9d4c..b9d2c0af9 100644 --- a/src/data/streams/merge_phone_platforms_for_individual_participants.py +++ b/src/data/streams/merge_phone_platforms_for_individual_participants.py @@ -3,15 +3,20 @@ pid = snakemake.params["pid"] -merged_platforms = pd.DataFrame() +merged_platforms = pd.DataFrame(columns=["timestamp", "device_id", "os"]) for platforms_file_path in snakemake.input["platforms_files"]: platforms_per_sensor = pd.read_csv(platforms_file_path) - if merged_platforms.shape[0] < platforms_per_sensor.shape[0]: - merged_platforms = platforms_per_sensor - elif merged_platforms.shape[0] == platforms_per_sensor.shape[0]: - if (merged_platforms["os"] == platforms_per_sensor["os"]).all(): - merged_platforms["timestamp"] = np.minimum(merged_platforms["timestamp"], platforms_per_sensor["timestamp"]) - else: - raise ValueError("Phone platforms of " + pid + " inferred from different sensors are not consistent. Please check it manually!") + # Unable to merge automatically if a device_id was used multiple times (e.g. device_id#1, device_id#2, device_id#1) + if platforms_per_sensor["device_id"].nunique() != platforms_per_sensor.shape[0]: + raise ValueError("Due to the complexity of " + pid + "'s devices, please merge platforms across all sensors manually!") + # Merge platforms across all sensors based on device_id + merged_platforms = merged_platforms.merge(platforms_per_sensor, on=["device_id", "os"], how="outer", sort=False) + # Keep the smaller timestamp per device_id + merged_platforms.insert(0, "timestamp", merged_platforms[["timestamp_x", "timestamp_y"]].min(axis=1).astype(int)) + merged_platforms.drop(["timestamp_x", "timestamp_y"], axis=1, inplace=True) + +# Sort by timestamp, and drop consecutive duplicates of os column +merged_platforms.sort_values(by=["timestamp"], inplace=True) +merged_platforms = merged_platforms[merged_platforms["os"] != merged_platforms["os"].shift()] merged_platforms.to_csv(snakemake.output[0], index=False) diff --git a/src/data/streams/pull_phone_data.R b/src/data/streams/pull_phone_data.R index 407657270..403fdeb05 100644 --- a/src/data/streams/pull_phone_data.R +++ b/src/data/streams/pull_phone_data.R @@ -145,7 +145,7 @@ pull_phone_data <- function(){ expected_columns <- tolower(rapids_schema[[sensor]]) participant_data <- setNames(data.frame(matrix(ncol = length(expected_columns), nrow = 0)), expected_columns) - platforms_data <- setNames(data.frame(matrix(ncol=2, nrow=0)), c("timestamp", "os")) + platforms_data <- setNames(data.frame(matrix(ncol=3, nrow=0)), c("timestamp", "device_id", "os")) if(length(devices) == 0){ warning("There were no PHONE device ids in this participant file:", participant_file) @@ -183,16 +183,13 @@ pull_phone_data <- function(){ if(!setequal(expected_columns, colnames(mutated_data))) stop(paste0("The mutated data for ", device, " does not have the columns RAPIDS expects. The mutation script returned [", paste(colnames(mutated_data), collapse=","),"] but RAPIDS expected [",paste(expected_columns, collapse=","), "]. One ore more mutation scripts in [", sensor,"][MUTATION][SCRIPTS] are adding extra columns or removing or not adding the ones expected")) if(nrow(mutated_data) > 0){ - platforms_data <- rbind(platforms_data, list(timestamp=min(mutated_data$timestamp), os=device_os)) + platforms_data <- rbind(platforms_data, list(timestamp=min(mutated_data$timestamp), device_id=device, os=device_os)) participant_data <- rbind(participant_data, mutated_data %>% distinct()) } } participant_data <- participant_data %>% arrange(timestamp) platforms_data <- platforms_data %>% arrange(timestamp) - if(lengths(unique(platforms_data$os)) == 1 && nrow(platforms_data) > 1){ - platforms_data <- head(platforms_data, 1) - } } write_csv(participant_data, output_sensor_raw_file) diff --git a/src/features/all_cleaning_individual/rapids/main.R b/src/features/all_cleaning_individual/rapids/main.R index 6221c428c..f251878f4 100644 --- a/src/features/all_cleaning_individual/rapids/main.R +++ b/src/features/all_cleaning_individual/rapids/main.R @@ -8,6 +8,7 @@ library(corrr) rapids_cleaning <- function(sensor_data_files, provider){ clean_features <- read.csv(sensor_data_files[["sensor_data"]], stringsAsFactors = FALSE) + platforms <- read.csv(sensor_data_files[["platforms"]], stringsAsFactors = FALSE) impute_selected_event_features <- provider[["IMPUTE_SELECTED_EVENT_FEATURES"]] cols_nan_threshold <- as.numeric(provider[["COLS_NAN_THRESHOLD"]]) drop_zero_variance_columns <- as.logical(provider[["COLS_VAR_THRESHOLD"]]) @@ -23,16 +24,36 @@ rapids_cleaning <- function(sensor_data_files, provider){ stop("Error: RAPIDS provider needs to impute the selected event features based on phone_data_yield_rapids_ratiovalidyieldedminutes column, please set config[PHONE_DATA_YIELD][PROVIDERS][RAPIDS][COMPUTE] to True and include 'ratiovalidyieldedminutes' in [FEATURES].") } column_names <- colnames(clean_features) - selected_apps_features <- column_names[grepl("^phone_applications_foreground_rapids_(countevent|countepisode|minduration|maxduration|meanduration|sumduration)", column_names)] + # Features which can be extracted from both Android and iOS devices selected_battery_features <- column_names[grepl("^phone_battery_rapids_", column_names)] selected_calls_features <- column_names[grepl("^phone_calls_rapids_.*_(count|distinctcontacts|sumduration|minduration|maxduration|meanduration|modeduration)", column_names)] + selected_screen_features <- column_names[grepl("^phone_screen_rapids_(sumduration|maxduration|minduration|avgduration|countepisode)", column_names)] + selected_wifi_connected_features <- column_names[grepl("^phone_wifi_connected_rapids_", column_names)] + selected_androidios_columns <- c(selected_battery_features, selected_calls_features, selected_screen_features, selected_wifi_connected_features) + clean_features[selected_androidios_columns][is.na(clean_features[selected_androidios_columns]) & (clean_features$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + + # Features which can only be extracted from Android devices + selected_apps_features <- column_names[grepl("^phone_applications_foreground_rapids_(countevent|countepisode|minduration|maxduration|meanduration|sumduration)", column_names)] selected_keyboard_features <- column_names[grepl("^phone_keyboard_rapids_(sessioncount|averagesessionlength|changeintextlengthlessthanminusone|changeintextlengthequaltominusone|changeintextlengthequaltoone|changeintextlengthmorethanone|maxtextlength|totalkeyboardtouches)", column_names)] selected_messages_features <- column_names[grepl("^phone_messages_rapids_.*_(count|distinctcontacts)", column_names)] - selected_screen_features <- column_names[grepl("^phone_screen_rapids_(sumduration|maxduration|minduration|avgduration|countepisode)", column_names)] - selected_wifi_features <- column_names[grepl("^phone_wifi_(connected|visible)_rapids_", column_names)] - - selected_columns <- c(selected_apps_features, selected_battery_features, selected_calls_features, selected_keyboard_features, selected_messages_features, selected_screen_features, selected_wifi_features) - clean_features[selected_columns][is.na(clean_features[selected_columns]) & (clean_features$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + selected_wifi_visible_features <- column_names[grepl("^phone_wifi_visible_rapids_", column_names)] + selected_android_columns <- c(selected_apps_features, selected_keyboard_features, selected_messages_features, selected_wifi_visible_features) + if(nrow(platforms) == 1){ # Single platform + if(platforms$os == "android"){ + clean_features[selected_android_columns][is.na(clean_features[selected_android_columns]) & (clean_features$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + } + }else{ # Multiple platforms + for(idx in 1:nrow(platforms)){ + if(platforms[idx, "os"] == "android"){ + if(idx < nrow(platforms)){ + selected_android_rows <- (clean_features$local_segment_start_datetime >= platforms[idx, "local_date_time"]) & (clean_features$local_segment_end_datetime <= platforms[idx + 1, "local_date_time"]) + }else{ + selected_android_rows <- clean_features$local_segment_start_datetime >= platforms[idx, "local_date_time"] + } + clean_features[selected_android_rows, selected_android_columns][is.na(clean_features[selected_android_rows, selected_android_columns]) & (clean_features[selected_android_rows, ]$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + } + } + } } # Drop rows with the value of data_yield_column less than data_yield_ratio_threshold diff --git a/src/features/all_cleaning_overall/rapids/main.R b/src/features/all_cleaning_overall/rapids/main.R index 6221c428c..175b8cb20 100644 --- a/src/features/all_cleaning_overall/rapids/main.R +++ b/src/features/all_cleaning_overall/rapids/main.R @@ -8,6 +8,7 @@ library(corrr) rapids_cleaning <- function(sensor_data_files, provider){ clean_features <- read.csv(sensor_data_files[["sensor_data"]], stringsAsFactors = FALSE) + platforms <- read.csv(sensor_data_files[["platforms"]], stringsAsFactors = FALSE) impute_selected_event_features <- provider[["IMPUTE_SELECTED_EVENT_FEATURES"]] cols_nan_threshold <- as.numeric(provider[["COLS_NAN_THRESHOLD"]]) drop_zero_variance_columns <- as.logical(provider[["COLS_VAR_THRESHOLD"]]) @@ -23,16 +24,48 @@ rapids_cleaning <- function(sensor_data_files, provider){ stop("Error: RAPIDS provider needs to impute the selected event features based on phone_data_yield_rapids_ratiovalidyieldedminutes column, please set config[PHONE_DATA_YIELD][PROVIDERS][RAPIDS][COMPUTE] to True and include 'ratiovalidyieldedminutes' in [FEATURES].") } column_names <- colnames(clean_features) - selected_apps_features <- column_names[grepl("^phone_applications_foreground_rapids_(countevent|countepisode|minduration|maxduration|meanduration|sumduration)", column_names)] + # Features which can be extracted from both Android and iOS devices selected_battery_features <- column_names[grepl("^phone_battery_rapids_", column_names)] selected_calls_features <- column_names[grepl("^phone_calls_rapids_.*_(count|distinctcontacts|sumduration|minduration|maxduration|meanduration|modeduration)", column_names)] + selected_screen_features <- column_names[grepl("^phone_screen_rapids_(sumduration|maxduration|minduration|avgduration|countepisode)", column_names)] + selected_wifi_connected_features <- column_names[grepl("^phone_wifi_connected_rapids_", column_names)] + selected_androidios_columns <- c(selected_battery_features, selected_calls_features, selected_screen_features, selected_wifi_connected_features) + + # Features which can only be extracted from Android devices + selected_apps_features <- column_names[grepl("^phone_applications_foreground_rapids_(countevent|countepisode|minduration|maxduration|meanduration|sumduration)", column_names)] selected_keyboard_features <- column_names[grepl("^phone_keyboard_rapids_(sessioncount|averagesessionlength|changeintextlengthlessthanminusone|changeintextlengthequaltominusone|changeintextlengthequaltoone|changeintextlengthmorethanone|maxtextlength|totalkeyboardtouches)", column_names)] selected_messages_features <- column_names[grepl("^phone_messages_rapids_.*_(count|distinctcontacts)", column_names)] - selected_screen_features <- column_names[grepl("^phone_screen_rapids_(sumduration|maxduration|minduration|avgduration|countepisode)", column_names)] - selected_wifi_features <- column_names[grepl("^phone_wifi_(connected|visible)_rapids_", column_names)] + selected_wifi_visible_features <- column_names[grepl("^phone_wifi_visible_rapids_", column_names)] + selected_android_columns <- c(selected_apps_features, selected_keyboard_features, selected_messages_features, selected_wifi_visible_features) - selected_columns <- c(selected_apps_features, selected_battery_features, selected_calls_features, selected_keyboard_features, selected_messages_features, selected_screen_features, selected_wifi_features) - clean_features[selected_columns][is.na(clean_features[selected_columns]) & (clean_features$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + merged_features <- data.frame() + for(pid in unique(clean_features$pid)){ + + clean_features_per_participant <- clean_features[clean_features$pid == pid, ] + platforms_per_participant <- platforms[platforms$pid == pid, ] + + # For selected_androidios_columns + clean_features_per_participant[selected_androidios_columns][is.na(clean_features_per_participant[selected_androidios_columns]) & (clean_features_per_participant$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + # For selected_android_columns + if(nrow(platforms_per_participant) == 1){ # Single platform + if(platforms_per_participant$os == "android"){ + clean_features_per_participant[selected_android_columns][is.na(clean_features_per_participant[selected_android_columns]) & (clean_features_per_participant$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + } + }else{ # Multiple platforms + for(idx in 1:nrow(platforms_per_participant)){ + if(platforms_per_participant[idx, "os"] == "android"){ + if(idx < nrow(platforms_per_participant)){ + selected_android_rows <- (clean_features_per_participant$local_segment_start_datetime >= platforms_per_participant[idx, "local_date_time"]) & (clean_features_per_participant$local_segment_end_datetime <= platforms_per_participant[idx + 1, "local_date_time"]) + }else{ + selected_android_rows <- clean_features_per_participant$local_segment_start_datetime >= platforms_per_participant[idx, "local_date_time"] + } + clean_features_per_participant[selected_android_rows, selected_android_columns][is.na(clean_features_per_participant[selected_android_rows, selected_android_columns]) & (clean_features_per_participant[selected_android_rows, ]$phone_data_yield_rapids_ratiovalidyieldedminutes > impute_selected_event_features$MIN_DATA_YIELDED_MINUTES_TO_IMPUTE)] <- 0 + } + } + } + merged_features <- rbind(merged_features, clean_features_per_participant) + } + clean_features <- merged_features } # Drop rows with the value of data_yield_column less than data_yield_ratio_threshold