Merge branch 'v1.0-main' into v1.0-es-update

OHDSI · Aug 28, 2024 · 8e0e005 · 8e0e005
2 parents 9e80a3d + c1b365f
commit 8e0e005
Show file tree

Hide file tree

Showing 16 changed files with 61 additions and 43 deletions.
diff --git a/R/Module-Characterization.R b/R/Module-Characterization.R
@@ -3,7 +3,7 @@
 #' @export
 #' @description
 #' Computes cohort characterization information against the OMOP CDM
-#' package version 2.0.0
+#' package version 2.0.1
 CharacterizationModule <- R6::R6Class(
   classname = "CharacterizationModule",
   inherit = StrategusModule,
@@ -19,8 +19,8 @@ CharacterizationModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
       workFolder <- jobContext$moduleExecutionSettings$workSubFolder
@@ -39,7 +39,7 @@ CharacterizationModule <- R6::R6Class(
         csvFilePrefix = self$tablePrefix,
         minCellCount =  jobContext$moduleExecutionSettings$minCellCount,
         minCharacterizationMean = jobContext$moduleExecutionSettings$minCharacterizationMean,
-        incremental = T, # any Strartegus param for this?
+        incremental = jobContext$moduleExecutionSettings$incremental,
         threads = as.double(ifelse(Sys.getenv('CharacterizationThreads') == "", 1,Sys.getenv('CharacterizationThreads') ))
       )
 

diff --git a/R/Module-CohortDiagnostics.R b/R/Module-CohortDiagnostics.R
@@ -19,8 +19,8 @@ CohortDiagnosticsModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
       cohortDefinitionSet <- super$.createCohortDefinitionSetFromJobContext()
@@ -133,7 +133,6 @@ CohortDiagnosticsModule <- R6::R6Class(
     #'                                    will help reduce the file size of the characterization output, but will remove information
     #'                                    on covariates that have very low values. The default is 0.001 (i.e. 0.1 percent)
     #' @param irWashoutPeriod             Number of days washout to include in calculation of incidence rates - default is 0
-    #' @param incremental                 Create only cohort diagnostics that haven't been created before?
     createModuleSpecifications = function(cohortIds = NULL,
                                           runInclusionStatistics = TRUE,
                                           runIncludedSourceConcepts = TRUE,
@@ -146,8 +145,7 @@ CohortDiagnosticsModule <- R6::R6Class(
                                           runTemporalCohortCharacterization = TRUE,
                                           temporalCovariateSettings = private$.getDefaultCovariateSettings(),
                                           minCharacterizationMean = 0.01,
-                                          irWashoutPeriod = 0,
-                                          incremental = FALSE) {
+                                          irWashoutPeriod = 0) {
       analysis <- list()
       for (name in names(formals(self$createModuleSpecifications))) {
         analysis[[name]] <- get(name)

diff --git a/R/Module-CohortGenerator.R b/R/Module-CohortGenerator.R
@@ -24,8 +24,8 @@ CohortGeneratorModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
       cohortDefinitionSet <- super$.createCohortDefinitionSetFromJobContext()
@@ -46,7 +46,7 @@ CohortGeneratorModule <- R6::R6Class(
         detectOnDescendants = negativeControlOutcomeSettings$detectOnDescendants,
         outputFolder = resultsFolder,
         databaseId = jobContext$moduleExecutionSettings$cdmDatabaseMetaData$databaseId,
-        incremental = jobContext$settings$incremental,
+        incremental = jobContext$moduleExecutionSettings$incremental,
         incrementalFolder = jobContext$moduleExecutionSettings$workSubFolder
       )
 
@@ -79,13 +79,9 @@ CohortGeneratorModule <- R6::R6Class(
       )
     },
     #' @description Creates the CohortGenerator Module Specifications
-    #' @param incremental When TRUE, the module will keep track of the cohorts
-    #' generated so that subsequent runs will skip any previously generated
-    #' cohorts.
     #' @param generateStats When TRUE, the Circe cohort definition SQL will
     #' include steps to compute inclusion rule statistics.
-    createModuleSpecifications = function(incremental = TRUE,
-                                          generateStats = TRUE) {
+    createModuleSpecifications = function(generateStats = TRUE) {
       analysis <- list()
       for (name in names(formals(self$createModuleSpecifications))) {
         analysis[[name]] <- get(name)

diff --git a/R/Module-CohortIncidence.R b/R/Module-CohortIncidence.R
@@ -18,8 +18,8 @@ CohortIncidenceModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       refId <- 1 # this should be part of execution settings
       resultsFolder <- private$jobContext$moduleExecutionSettings$resultsSubFolder

diff --git a/R/Module-CohortMethod.R b/R/Module-CohortMethod.R
@@ -17,11 +17,11 @@ CohortMethodModule <- R6::R6Class(
     #' @param analysisSpecifications The analysis specifications for the study
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
-      multiThreadingSettings <- CohortMethod::createDefaultMultiThreadingSettings(parallel::detectCores())
+      multiThreadingSettings <- CohortMethod::createDefaultMultiThreadingSettings(jobContext$moduleExecutionSettings$maxCores)
 
       args <- jobContext$settings
       args$connectionDetails <- connectionDetails
@@ -41,7 +41,7 @@ CohortMethodModule <- R6::R6Class(
         exportFolder = exportFolder,
         databaseId = jobContext$moduleExecutionSettings$cdmDatabaseMetaData$databaseId,
         minCellCount = jobContext$moduleExecutionSettings$minCellCount,
-        maxCores = parallel::detectCores(),
+        maxCores = jobContext$moduleExecutionSettings$maxCores,
         cmDiagnosticThresholds = jobContext$settings$cmDiagnosticThresholds
       )
       # TODO: Removing this to make the upload easier

diff --git a/R/Module-EvidenceSynthesis.R b/R/Module-EvidenceSynthesis.R
@@ -20,8 +20,8 @@ EvidenceSynthesisModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateResultsExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "ResultsExecutionSettings")
       jobContext <- private$jobContext
 
       resultsFolder <- jobContext$moduleExecutionSettings$resultsSubFolder
@@ -357,7 +357,7 @@ EvidenceSynthesisModule <- R6::R6Class(
       fullKeys <- perDbEstimates$estimates[, c(perDbEstimates$key, "analysisId")] |>
         distinct()
 
-      cluster <- ParallelLogger::makeCluster(min(10, parallel::detectCores()))
+      cluster <- ParallelLogger::makeCluster(min(10, jobContext$moduleExecutionSettings$maxCores))
       ParallelLogger::clusterRequire(cluster, "dplyr")
       on.exit(ParallelLogger::stopCluster(cluster))
 

diff --git a/R/Module-PatientLevelPrediction.R b/R/Module-PatientLevelPrediction.R
@@ -19,8 +19,8 @@ PatientLevelPredictionModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
       cohortDefinitionSet <- super$.createCohortDefinitionSetFromJobContext()

diff --git a/R/Module-SelfControlledCaseSeries.R b/R/Module-SelfControlledCaseSeries.R
@@ -20,11 +20,11 @@ SelfControlledCaseSeriesModule <- R6::R6Class(
     #' @template analysisSpecifications
     #' @template executionSettings
     execute = function(connectionDetails, analysisSpecifications, executionSettings) {
+      super$.validateCdmExecutionSettings(executionSettings)
       super$execute(connectionDetails, analysisSpecifications, executionSettings)
-      checkmate::assertClass(executionSettings, "CdmExecutionSettings")
 
       jobContext <- private$jobContext
-      sccsMultiThreadingSettings <- SelfControlledCaseSeries::createDefaultSccsMultiThreadingSettings(parallel::detectCores())
+      sccsMultiThreadingSettings <- SelfControlledCaseSeries::createDefaultSccsMultiThreadingSettings(jobContext$moduleExecutionSettings$maxCores)
 
       args <- jobContext$settings
       args$connectionDetails <- connectionDetails

diff --git a/R/Module-StrategusModule.R b/R/Module-StrategusModule.R
@@ -272,6 +272,16 @@ StrategusModule <- R6::R6Class(
           detectOnDescendants = FALSE
         ))
       }
+    },
+    .validateCdmExecutionSettings = function(executionSettings) {
+      errorMessages <- checkmate::makeAssertCollection()
+      checkmate::assertClass(executionSettings, "CdmExecutionSettings", add = errorMessages)
+      checkmate::assertInt(executionSettings$maxCores, lower = 1, upper = parallel::detectCores())
+    },
+    .validateResultsExecutionSettings = function(executionSettings) {
+      errorMessages <- checkmate::makeAssertCollection()
+      checkmate::assertClass(executionSettings, "ResultsExecutionSettings", add = errorMessages)
+      checkmate::assertInt(executionSettings$maxCores, lower = 1, upper = parallel::detectCores())
     }
   )
 )
diff --git a/R/Settings.R b/R/Settings.R
@@ -253,6 +253,11 @@ createEmptyAnalysisSpecificiations <- function() {
 #' @param logFileName                Logging information from Strategus and all modules will be located in this file. Individual modules will continue to have their own module-specific logs. By default this will be written to the root of the `resultsFolder`
 #' @param minCellCount               The minimum number of subjects contributing to a count before it can be included
 #'                                   in results.
+#' @param incremental                This value will be passed to each module that supports execution in an incremental manner. Modules
+#'                                   and their underlying packages may use the `workFolder` contents to determine their state of execution
+#'                                   and attempt to pick up where they left off when this value is set to TRUE.
+#' @param maxCores                   The maximum number of processing cores to use for execution. The default is to
+#'                                   use all available cores on the machine.
 #'
 #' @return
 #' An object of type `ExecutionSettings`.
@@ -265,7 +270,9 @@ createCdmExecutionSettings <- function(workDatabaseSchema,
                                        workFolder,
                                        resultsFolder,
                                        logFileName = file.path(resultsFolder, "strategus-log.txt"),
-                                       minCellCount = 5) {
+                                       minCellCount = 5,
+                                       incremental = TRUE,
+                                       maxCores = parallel::detectCores()) {
   errorMessages <- checkmate::makeAssertCollection()
   checkmate::assertCharacter(workDatabaseSchema, len = 1, add = errorMessages)
   checkmate::assertCharacter(cdmDatabaseSchema, len = 1, add = errorMessages)
@@ -274,6 +281,8 @@ createCdmExecutionSettings <- function(workDatabaseSchema,
   checkmate::assertCharacter(resultsFolder, len = 1, add = errorMessages)
   checkmate::assertCharacter(logFileName, len = 1, add = errorMessages)
   checkmate::assertInt(minCellCount, add = errorMessages)
+  checkmate::assertLogical(incremental, add = errorMessages)
+  checkmate::assertInt(maxCores, add = errorMessages)
   checkmate::reportAssertions(collection = errorMessages)
 
   # Normalize paths to convert relative paths to absolute paths
@@ -297,7 +306,8 @@ createCdmExecutionSettings <- function(workDatabaseSchema,
 #' @param logFileName                Logging information from Strategus and all modules will be located in this file. Individual modules will continue to have their own module-specific logs. By default this will be written to the root of the `resultsFolder`
 #' @param minCellCount               The minimum number of subjects contributing to a count before it can be included
 #'                                   in results.
-#'
+#' @param maxCores                   The maximum number of processing cores to use for execution. The default is to
+#'                                   use all available cores on the machine.
 #' @return
 #' An object of type `ExecutionSettings`.
 #'
@@ -306,13 +316,15 @@ createResultsExecutionSettings <- function(resultsDatabaseSchema,
                                            workFolder,
                                            resultsFolder,
                                            logFileName = file.path(resultsFolder, "strategus-log.txt"),
-                                           minCellCount = 5) {
+                                           minCellCount = 5,
+                                           maxCores = parallel::detectCores()) {
   errorMessages <- checkmate::makeAssertCollection()
   checkmate::assertCharacter(resultsDatabaseSchema, len = 1, add = errorMessages)
   checkmate::assertCharacter(workFolder, len = 1, add = errorMessages)
   checkmate::assertCharacter(resultsFolder, len = 1, add = errorMessages)
   checkmate::assertCharacter(logFileName, len = 1, add = errorMessages)
   checkmate::assertInt(minCellCount, add = errorMessages)
+  checkmate::assertInt(maxCores, add = errorMessages)
   checkmate::reportAssertions(collection = errorMessages)
 
   # Normalize paths to convert relative paths to absolute paths

diff --git a/man/CharacterizationModule.Rd b/man/CharacterizationModule.Rd
diff --git a/man/CohortDiagnosticsModule.Rd b/man/CohortDiagnosticsModule.Rd
diff --git a/man/CohortGeneratorModule.Rd b/man/CohortGeneratorModule.Rd
diff --git a/man/createCdmExecutionSettings.Rd b/man/createCdmExecutionSettings.Rd
diff --git a/man/createResultsExecutionSettings.Rd b/man/createResultsExecutionSettings.Rd
diff --git a/tests/testthat/test-Settings.R b/tests/testthat/test-Settings.R
@@ -48,8 +48,7 @@ test_that("Test analysis specification creation", {
     runBreakdownIndexEvents = TRUE,
     runIncidenceRate = TRUE,
     runCohortRelationship = TRUE,
-    runTemporalCohortCharacterization = TRUE,
-    incremental = FALSE
+    runTemporalCohortCharacterization = TRUE
   )
 
   # Cohort Generator -----------------