Merge pull request #23 from OHDSI/issue_22

issue 22 fix
OHDSI · Mar 15, 2023 · fac4b1b · fac4b1b
2 parents 4b9a1c1 + 9d97421
commit fac4b1b
Show file tree

Hide file tree

Showing 9 changed files with 390 additions and 103 deletions.
diff --git a/R/AggregateCovariates.R b/R/AggregateCovariates.R
@@ -18,6 +18,7 @@
 #'
 #' @param targetIds   A list of cohortIds for the target cohorts
 #' @param outcomeIds  A list of cohortIds for the outcome cohorts
+#' @param minPriorObservation The minimum time in the database a patient in the target cohorts must be observed prior to index
 #' @template timeAtRisk
 #' @param covariateSettings   An object created using \code{FeatureExtraction::createCovariateSettings}
 #'
@@ -28,6 +29,7 @@
 createAggregateCovariateSettings <- function(
   targetIds,
   outcomeIds,
+  minPriorObservation = 0,
   riskWindowStart = 1,
   startAnchor = 'cohort start',
   riskWindowEnd = 365,
@@ -57,19 +59,26 @@ createAggregateCovariateSettings <- function(
     endAnchor = endAnchor,
     errorMessages = errorMessages
   )
-  
+
   # check covariateSettings
   .checkCovariateSettings(
     covariateSettings = covariateSettings,
     errorMessages = errorMessages
   )
 
+  # check minPriorObservation
+  .checkMinPriorObservation(
+    minPriorObservation = minPriorObservation,
+    errorMessages = errorMessages
+  )
+
   checkmate::reportAssertions(errorMessages)
 
   # create list
   result <- list(
     targetIds = targetIds,
     outcomeIds = outcomeIds,
+    minPriorObservation = minPriorObservation,
     riskWindowStart = riskWindowStart,
     startAnchor = startAnchor,
     riskWindowEnd = riskWindowEnd ,
@@ -126,6 +135,7 @@ computeAggregateCovariateAnalyses <- function(
   createCohortsOfInterest(
     connection = connection,
     dbms = connectionDetails$dbms,
+    cdmDatabaseSchema = cdmDatabaseSchema,
     aggregateCovariateSettings,
     targetDatabaseSchema,
     targetTable,
@@ -135,16 +145,16 @@ computeAggregateCovariateAnalyses <- function(
   )
 
   ## get counts
-  sql <- 'select cohort_definition_id, count(*) N from #agg_cohorts group by cohort_definition_id;'
+  sql <- 'select cohort_definition_id, count(*) row_count, count(distinct subject_id) person_count from #agg_cohorts group by cohort_definition_id;'
   sql <- SqlRender::translate(
     sql = sql,
     targetDialect = connectionDetails$dbms
     )
   counts <- DatabaseConnector::querySql(
     connection = connection,
-    sql = sql
+    sql = sql,
+    snakeCaseToCamelCase = T,
   )
-  #print(counts) # testing
 
   message("Computing aggregate covariate results")
 
@@ -159,6 +169,8 @@ computeAggregateCovariateAnalyses <- function(
     cdmVersion = cdmVersion,
     aggregated = T
   )
+  # adding counts as a new table
+  result$cohortCounts <- counts
 
   # add databaseId and runId to each table in results
   # could add settings table with this and just have setting id
@@ -234,6 +246,7 @@ computeAggregateCovariateAnalyses <- function(
 
 createCohortsOfInterest <- function(
   connection,
+  cdmDatabaseSchema,
   dbms,
   aggregateCovariateSettings,
   targetDatabaseSchema,
@@ -247,13 +260,15 @@ createCohortsOfInterest <- function(
     sqlFilename = "createTargetOutcomeCombinations.sql",
     packageName = "Characterization",
     dbms = dbms,
+    cdm_database_schema = cdmDatabaseSchema,
     tempEmulationSchema = tempEmulationSchema,
     target_database_schema = targetDatabaseSchema,
     target_table = targetTable,
     outcome_database_schema = outcomeDatabaseSchema,
     outcome_table = outcomeTable,
     target_ids = paste(aggregateCovariateSettings$targetIds, collapse = ',', sep = ','),
     outcome_ids = paste(aggregateCovariateSettings$outcomeIds, collapse = ',', sep = ','),
+    min_prior_observation = aggregateCovariateSettings$minPriorObservation,
     tar_start = aggregateCovariateSettings$riskWindowStart,
     tar_start_anchor = ifelse(
       aggregateCovariateSettings$startAnchor == 'cohort start',

diff --git a/R/HelperFunctions.R b/R/HelperFunctions.R
@@ -246,3 +246,15 @@
   }
 
 }
+
+.checkMinPriorObservation <- function(
+    minPriorObservation,
+  errorMessages
+) {
+  checkmate::assertCount(
+    x = minPriorObservation,
+    null.ok = F,
+    .var.name = 'minPriorObservation',
+    add = errorMessages
+  )
+}
diff --git a/R/SaveLoad.R b/R/SaveLoad.R
@@ -404,6 +404,38 @@ exportAggregateCovariateToCsv <- function(
 
     }
   )
+  # cohort details
+  Andromeda::batchApply(
+    tbl = result$cohortCounts,
+    fun = function(x) {
+
+      append <- file.exists(
+        file.path(
+          saveDirectory,
+          "cohort_counts.csv"
+        )
+      )
+
+      dat <- as.data.frame(
+        x %>%
+          dplyr::collect()
+      )
+
+      colnames(dat) <- SqlRender::camelCaseToSnakeCase(
+        string = colnames(dat)
+      )
+
+      readr::write_csv(
+        x = dat,
+        file = file.path(
+          saveDirectory,
+          "cohort_counts.csv"
+        ),
+        append = append
+      )
+
+    }
+  )
 
   # cohort details
   Andromeda::batchApply(
@@ -577,7 +609,8 @@ exportAggregateCovariateToCsv <- function(
         "analysis_ref.csv",
         "covariate_ref.csv",
         "covariates.csv",
-        "covariates_continuous.csv"
+        "covariates_continuous.csv",
+        "cohort_counts.csv"
       )
     )
   )