completing mincellcount

- fixing bug where mincellcount wasnt being used by aggregate covs - adding tests for all mincellcount cells - turning off GHA mac spark tests due to odd java error on there (tested manually on my mac)
OHDSI · Aug 7, 2024 · 638d24e · 638d24e
1 parent 0da14c2
commit 638d24e
Show file tree

Hide file tree

Showing 4 changed files with 295 additions and 10 deletions.
diff --git a/R/AggregateCovariates.R b/R/AggregateCovariates.R
@@ -188,6 +188,7 @@ computeTargetAggregateCovariateAnalyses <- function(
     databaseId = "database 1",
     outputFolder = file.path(getwd(),'characterization_results'),
     minCharacterizationMean = 0,
+    minCellCount = 0,
     ...
 ) {
 
@@ -317,7 +318,8 @@ computeTargetAggregateCovariateAnalyses <- function(
     cohortDetails = cohortDetails,
     counts = counts,
     databaseId = databaseId,
-    minCharacterizationMean = minCharacterizationMean
+    minCharacterizationMean = minCharacterizationMean,
+    minCellCount = minCellCount
   )
 
   return(invisible(T))
@@ -337,6 +339,7 @@ computeCaseAggregateCovariateAnalyses <- function(
     databaseId = "database 1",
     outputFolder = file.path(getwd(),'characterization_results'),
     minCharacterizationMean = 0,
+    minCellCount = 0,
     ...
     ) {
   # check inputs
@@ -550,7 +553,8 @@ computeCaseAggregateCovariateAnalyses <- function(
     cohortDetails = cohortDetails,
     counts = counts,
     databaseId = databaseId,
-    minCharacterizationMean = minCharacterizationMean
+    minCharacterizationMean = minCharacterizationMean,
+    minCellCount = minCellCount
   )
   exportAndromedaToCsv(
     andromeda = result2,
@@ -559,7 +563,8 @@ computeCaseAggregateCovariateAnalyses <- function(
     counts = NULL, # previously added
     databaseId = databaseId,
     minCharacterizationMean = minCharacterizationMean,
-    includeSettings = F
+    includeSettings = F,
+    minCellCount = minCellCount
   )
 
   return(invisible(T))
@@ -654,6 +659,8 @@ exportAndromedaToCsv <- function(
           ParallelLogger::logInfo(paste0("Removing sum_value counts less than ", minCellCount))
           if (sum(removeInd) > 0) {
             data$sum_value[removeInd] <- -1*minCellCount
+            # adding other calculated columns
+            data$average_value[removeInd] <- NA
           }
         }
 
@@ -686,6 +693,16 @@ exportAndromedaToCsv <- function(
           ParallelLogger::logInfo(paste0("Removing count_value counts less than ", minCellCount))
           if (sum(removeInd) > 0) {
             data$count_value[removeInd] <- -1*minCellCount
+            # adding columns calculated from count
+            data$min_value[removeInd] <- NA
+            data$max_value[removeInd] <- NA
+            data$average_value[removeInd] <- NA
+            data$standard_deviation[removeInd] <- NA
+            data$median_value[removeInd] <- NA
+            data$p_10_value[removeInd] <- NA
+            data$p_25_value[removeInd] <- NA
+            data$p_75_value[removeInd] <- NA
+            data$p_90_value[removeInd] <- NA
           }
         }
 

diff --git a/R/SaveLoad.R b/R/SaveLoad.R
@@ -27,7 +27,8 @@
 exportTimeToEventToCsv <- function(
     result,
     saveDirectory,
-    minCellCount = 0) {
+    minCellCount = 0
+    ) {
   if (!dir.exists(saveDirectory)) {
     dir.create(
       path = saveDirectory,
@@ -120,53 +121,67 @@ exportDechallengeRechallengeToCsv <- function(
         string = colnames(dat)
       )
 
+      removeInd <- dat$num_exposure_eras < minCellCount
+      if (sum(removeInd) > 0) {
+        ParallelLogger::logInfo(paste0("Censoring num_exposure_eras counts less than ", minCellCount))
+        if (sum(removeInd) > 0) {
+          dat$num_exposure_eras[removeInd] <- -minCellCount
+        }
+      }
+
       removeInd <- dat$num_persons_exposed < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing num_persons_exposed counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring num_persons_exposed counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$num_persons_exposed[removeInd] <- -minCellCount
         }
       }
 
       removeInd <- dat$num_cases < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing num_cases counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring num_cases counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$num_cases[removeInd] <- -minCellCount
         }
       }
 
       removeInd <- dat$dechallenge_attempt < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing dechallenge_attempt counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring/removing dechallenge_attempt counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$dechallenge_attempt[removeInd] <- -minCellCount
+          dat$pct_dechallenge_attempt[removeInd] <- NA
         }
       }
 
       removeInd <- dat$dechallenge_fail < minCellCount | dat$dechallenge_success < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing DECHALLENGE FAIL or SUCCESS counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring/removing DECHALLENGE FAIL or SUCCESS counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$dechallenge_fail[removeInd] <- -minCellCount
           dat$dechallenge_success[removeInd] <- -minCellCount
+          dat$pct_dechallenge_fail[removeInd] <- NA
+          dat$pct_dechallenge_success[removeInd] <- NA
         }
       }
 
       removeInd <- dat$rechallenge_attempt < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing rechallenge_attempt counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring/removing rechallenge_attempt counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$rechallenge_attempt[removeInd] <- -minCellCount
+          dat$pct_rechallenge_attempt[removeInd] <- NA
         }
       }
 
       removeInd <- dat$rechallenge_fail < minCellCount | dat$rechallenge_success < minCellCount
       if (sum(removeInd) > 0) {
-        ParallelLogger::logInfo(paste0("Removing rechallenge_fail or rechallenge_success counts less than ", minCellCount))
+        ParallelLogger::logInfo(paste0("Censoring/removing rechallenge_fail or rechallenge_success counts less than ", minCellCount))
         if (sum(removeInd) > 0) {
           dat$rechallenge_fail[removeInd] <- -minCellCount
           dat$rechallenge_success[removeInd] <- -minCellCount
+          dat$pct_rechallenge_fail[removeInd] <- NA
+          dat$pct_rechallenge_success[removeInd] <- NA
         }
       }
 

diff --git a/tests/testthat/test-dbs.R b/tests/testthat/test-dbs.R
@@ -102,6 +102,7 @@ getPlatformConnectionDetails <- function(dbmsPlatform) {
       cohortDatabaseSchema <- Sys.getenv("CDM_SNOWFLAKE_OHDSI_SCHEMA")
       options(sqlRenderTempEmulationSchema = Sys.getenv("CDM_SNOWFLAKE_OHDSI_SCHEMA"))
     } else if (dbmsPlatform == "spark") {
+      if (.Platform$OS.type == "windows") { # skipping Mac for GHA due to JAVA issue
       connectionDetails <- DatabaseConnector::createConnectionDetails(
         dbms = dbmsPlatform,
         user = Sys.getenv("CDM5_SPARK_USER"),
@@ -113,6 +114,10 @@ getPlatformConnectionDetails <- function(dbmsPlatform) {
       vocabularyDatabaseSchema <- Sys.getenv("CDM5_SPARK_CDM_SCHEMA")
       cohortDatabaseSchema <- Sys.getenv("CDM5_SPARK_OHDSI_SCHEMA")
       options(sqlRenderTempEmulationSchema = Sys.getenv("CDM5_SPARK_OHDSI_SCHEMA"))
+      }
+      else{
+        return(NULL)
+      }
     } else if (dbmsPlatform == "sql server") {
       connectionDetails <- DatabaseConnector::createConnectionDetails(
         dbms = dbmsPlatform,