Data cleaning with multiple platforms

Snakefile

@@ -397,9 +397,12 @@ 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/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_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:

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
     ```

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

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"]:

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}",

rules/preprocessing.smk

@@ -19,10 +19,46 @@ 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 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_with_datetime.csv", pid=config["PIDS"])
+    output:
+        "data/interim/platforms/all_participants/all_sensor_platforms_with_datetime.csv"
+    script:
+        "../src/data/streams/merge_phone_platforms_for_all_participants.py"
+
 rule process_time_segments:
     input: 
         segments_file = config["TIME_SEGMENTS"]["FILE"],

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)

src/data/streams/merge_phone_platforms_for_individual_participants.py

@@ -0,0 +1,22 @@
+import pandas as pd
+import numpy as np
+
+pid = snakemake.params["pid"]
+
+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)
+    # 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)

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,56 @@ 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=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)
-    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), 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)
   }
-  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()

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