CNAM-154 Refactored MLPPMain class and added MLPPConfig

CNAM-154 Refactored global testing and added MLPPMainSuite

CNAM-154 Small fixes

Author: danielpes

src/main/resources/config/filtering-cmap.conf

@@ -0,0 +1,12 @@
+env_name = "cmap"
+
+dates = {
+  age_reference = [2010, 12, 31, 23, 59, 59]
+  study_start = [2010, 1, 1]
+  study_end = [2014, 12, 31, 23, 59, 59]
+}
+
+mlpp_parameters = {
+  min_timestamp = ${dates.study_start}
+  max_timestamp = ${dates.study_end}
+}

src/main/resources/config/filtering-cnam.conf

@@ -0,0 +1 @@
+env_name = "cnam"

src/main/resources/filtering-default.conf …/resources/config/filtering-default.confsrc/main/resources/filtering-default.conf renamed to src/main/resources/config/filtering-default.conf src/main/resources/filtering-default.conf renamed to src/main/resources/config/filtering-default.conf

@@ -16,6 +16,8 @@ default = {
   }
   dates = {
     age_reference = [2006, 12, 31, 23, 59, 59]
+    study_start = [2006, 1, 1]
+    study_end = [2009, 12, 31, 23, 59, 59]
   }
   paths = {
     input = {
@@ -32,38 +34,41 @@ default = {
       root = "/shared/filtered_data"
       patients = "/shared/filtered_data/patients"
       flat_events = "/shared/filtered_data/flat_events"
+      mlpp_features = "/shared/features/mlpp"
+    }
+  }
+  cox = {
+
+  }
+  mlpp_parameters = {
+    bucket_size = 30  # in days
+    lag_count = 10
+    min_timestamp = ${default.dates.study_start}
+    max_timestamp = ${default.dates.study_end}
+
+    exposures = {
+      min_purchases = 1
+      start_delay = 0
+      purchases_window = 0
+      only_first = false
+      filter_diagnosed_patients = true
+      filter_delayed_entries = true
+      delayed_entry_threshold = 12
     }
   }
 }
 
-# For the CNAM environment, we use the default values
 cnam = ${default}
-cnam.env_name = "cnam"
+cnam = {
+  include "filtering-cnam.conf"
+}
 
-# Overriding only the dates for cmap environment:
 cmap = ${default}
-cmap.env_name = "cmap"
-cmap.dates = {
-  age_reference = [2010, 12, 31, 23, 59, 59]
+cmap = {
+  include "filtering-cmap.conf"
 }
 
-# Overriding only the paths for test environment:
 test = ${default}
-test.env_name = "test"
-test.paths = {
-  input = {
-    dcir = "src/test/resources/test-input/DCIR.parquet"
-    pmsi_mco = "src/test/resources/test-input/MCO.parquet"
-    pmsi_had = "src/test/resources/test-input/HAD.parquet"
-    pmsi_ssr = "src/test/resources/test-input/SSR.parquet"
-    ir_ben = "src/test/resources/test-input/IR_BEN_R.parquet"
-    ir_imb = "src/test/resources/test-input/IR_IMB_R.parquet"
-    ir_pha = "src/test/resources/test-input/IR_PHA_R.parquet"
-    dosages = "src/test/resources/test-input/DOSE_PER_MOLECULE.CSV"
-  }
-  output = {
-    root = "target/test/output"
-    patients = "target/test/output/patients"
-    flat_events = "target/test/output/flat_events"
-  }
+test = {
+  include "filtering-test.conf"
 }

src/main/resources/config/filtering-test.conf

@@ -0,0 +1,20 @@
+env_name = "test"
+
+paths = {
+  input = {
+    dcir = "src/test/resources/test-input/DCIR.parquet"
+    pmsi_mco = "src/test/resources/test-input/MCO.parquet"
+    pmsi_had = "src/test/resources/test-input/HAD.parquet"
+    pmsi_ssr = "src/test/resources/test-input/SSR.parquet"
+    ir_ben = "src/test/resources/test-input/IR_BEN_R.parquet"
+    ir_imb = "src/test/resources/test-input/IR_IMB_R.parquet"
+    ir_pha = "src/test/resources/test-input/IR_PHA_R.parquet"
+    dosages = "src/test/resources/test-input/DOSE_PER_MOLECULE.CSV"
+  }
+  output = {
+    root = "target/test/output"
+    patients = "target/test/output/patients"
+    flat_events = "target/test/output/flat_events"
+    mlpp_features = "target/test/output/mlpp_features"
+  }
+}

src/main/scala/fr/polytechnique/cmap/cnam/filtering/ExposuresTransformer.scala

@@ -5,9 +5,7 @@ import fr.polytechnique.cmap.cnam.utilities.functions._
 
 trait ExposuresTransformer extends DatasetTransformer[FlatEvent, FlatEvent] {
 
-  // Constant definitions. Should be verified before compiling.
-  // In the future, we may want to export them to an external file.
-  val StudyStart = makeTS(2006, 1, 1)
+  lazy val StudyStart = FilteringConfig.dates.studyStart
 
   def transform(input: Dataset[FlatEvent]): Dataset[FlatEvent]
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/FilteringConfig.scala

@@ -43,8 +43,8 @@ object FilteringConfig {
     val configPath: String = sqlContext.getConf("conf", "")
     val environment: String = sqlContext.getConf("env", "test")
 
-    val defaultConfig = ConfigFactory.parseResources("filtering-default.conf").resolve().getConfig(environment)
-    val newConfig = ConfigFactory.parseFile(new java.io.File(configPath))
+    val defaultConfig = ConfigFactory.parseResources("config/filtering-default.conf").resolve().getConfig(environment)
+    val newConfig = ConfigFactory.parseFile(new java.io.File(configPath)).resolve()
 
     newConfig.withFallback(defaultConfig).resolve()
   }
@@ -60,7 +60,12 @@ object FilteringConfig {
     dosages: String
   )
 
-  case class OutputPaths(root: String, patients: String, flatEvents: String)
+  case class OutputPaths(
+    root: String,
+    patients: String,
+    flatEvents: String,
+    mlppFeatures: String
+  )
 
   case class Limits(
     minYear: Int,
@@ -73,7 +78,16 @@ object FilteringConfig {
     maxAge: Int
   )
 
-  case class Dates(ageReference: Timestamp)
+  case class Dates(
+    ageReference: Timestamp,
+    studyStart: Timestamp,
+    studyEnd: Timestamp
+  )
+
+  case class TracklossDefinition(
+    threshold: Int,
+    delay: Int
+  )
 
   lazy val drugCategories: List[String] = conf.getStringList("drug_categories").asScala.toList
   lazy val cancerDefinition: String  = conf.getString("cancer_definition")
@@ -92,7 +106,8 @@ object FilteringConfig {
   lazy val outputPaths = OutputPaths(
     root = conf.getString("paths.output.root"),
     patients = conf.getString("paths.output.patients"),
-    flatEvents = conf.getString("paths.output.flat_events")
+    flatEvents = conf.getString("paths.output.flat_events"),
+    mlppFeatures = conf.getString("paths.output.mlpp_features")
   )
   lazy val limits = Limits(
     minYear = conf.getInt("limits.min_year"),
@@ -105,6 +120,14 @@ object FilteringConfig {
     maxAge = conf.getInt("limits.max_age")
   )
   lazy val dates = Dates(
-    ageReference = makeTS(conf.getIntList("dates.age_reference").asScala.toList)
+    ageReference = makeTS(conf.getIntList("dates.age_reference").asScala.toList),
+    studyStart = makeTS(conf.getIntList("dates.study_start").asScala.toList),
+    studyEnd = makeTS(conf.getIntList("dates.study_end").asScala.toList)
+  )
+  lazy val tracklossDefinition = TracklossDefinition(
+    threshold = conf.getInt("trackloss.threshold"),
+    delay = conf.getInt("trackloss.delay")
   )
+
+  def modelConfig(modelName: String): Config = conf.getConfig(modelName)
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/PatientsTransformer.scala

@@ -59,10 +59,10 @@ object PatientsTransformer extends Transformer[Patient] with PatientsTransformer
 
     patients.where(filterPatientsByAge)
       .select(
-      patientID.as("patientID"),
-      gender.as("gender"),
-      birthdate.as("birthDate"),
-      deathDate.as("deathDate")
-    ).as[Patient]
+        patientID.as("patientID"),
+        gender.as("gender"),
+        birthdate.as("birthDate"),
+        deathDate.as("deathDate")
+      ).as[Patient]
   }
 }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/Sources.scala

@@ -16,3 +16,17 @@ class Sources(
     val irImb: Option[DataFrame] = None,
     val irPha: Option[DataFrame] = None,
     val dosages: Option[DataFrame] = None)
+
+object Sources {
+  def apply(
+      dcir: Option[DataFrame] = None,
+      pmsiMco: Option[DataFrame] = None,
+      pmsiHad: Option[DataFrame] = None,
+      pmsiSsr: Option[DataFrame] = None,
+      irBen: Option[DataFrame] = None,
+      irImb: Option[DataFrame] = None,
+      irPha: Option[DataFrame] = None,
+      dosages: Option[DataFrame] = None) = {
+    new Sources(dcir, pmsiMco, pmsiHad, pmsiSsr, irBen, irImb, irPha, dosages)
+  }
+}

src/main/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxMain.scala

@@ -117,7 +117,7 @@ object CoxMain extends Main {
         case Array(arg1, args2) => (args(0), args(1), true)
         case _ => ("test", "broad", true)
       }
-    val config: Config = ConfigFactory.parseResources("filtering-default.conf").getConfig(environment)
+    val config: Config = ConfigFactory.parseResources("config/filtering-default.conf").getConfig(environment)
     coxFeaturing(sqlContext, config, cancerDefinition, filterDelayedPatients)
     stopContext()
   }

src/main/scala/fr/polytechnique/cmap/cnam/filtering/ltsccs/LTSCCSMain.scala

@@ -57,7 +57,7 @@ object LTSCCSMain extends Main {
   override def main(args: Array[String]): Unit = {
     startContext()
     val environment = if (args.nonEmpty) args(0) else "test"
-    val config: Config = ConfigFactory.parseResources("filtering-default.conf").getConfig(environment)
+    val config: Config = ConfigFactory.parseResources("config/filtering-default.conf").getConfig(environment)
     runLTSCCS(sqlContext, config)
     stopContext()
   }