CNAM-153 Removes filterDiagnosedPatients flag.

Author: sathiyapk

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

@@ -24,7 +24,6 @@ default = {
     follow-up-delay = 6 #Number of months after the observation start that is considered to be followup
     filter_delayed_patients = true #Patients who are exposed after certain months of the study start.
     delayed_entries_threshold = 12 #Months that signifies the delayed entries.
-    filter_diagonized_patients = true #Patients who are diagonized cancer before study period.
 
     exposures = {
       min_purchases = 2 #Minimum number of purchases that have to be made in order to be considered exposed.
@@ -77,12 +76,16 @@ default = {
 }
 
 cnam = ${default}
-cnam = include "filtering-cnam.conf"
+cnam = {
+  include "filtering-cnam.conf"
+}
 
 cmap = ${default}
-cmap = include "filtering-cmap.conf"
+cmap = {
+  include "filtering-cmap.conf"
+}
 
 test = ${default}
 test = {
-  include "filtering-test.conf"
-}
+    include "filtering-test.conf"
+}

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

@@ -1,13 +1,11 @@
 package fr.polytechnique.cmap.cnam.filtering
 
+import java.sql.Timestamp
 import org.apache.spark.sql.Dataset
-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)
+  val StudyStart: Timestamp = FilteringConfig.dates.studyStart
 
   def transform(input: Dataset[FlatEvent]): Dataset[FlatEvent]
 }

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

@@ -5,7 +5,6 @@ import scala.collection.JavaConverters._
 import org.apache.spark.SparkContext
 import org.apache.spark.sql.SQLContext
 import com.typesafe.config.{Config, ConfigFactory}
-import fr.polytechnique.cmap.cnam.filtering.mlpp.MLPPFeature
 import fr.polytechnique.cmap.cnam.utilities.functions._
 
 object FilteringConfig {

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

@@ -1,14 +1,14 @@
 package fr.polytechnique.cmap.cnam.filtering
 
+import java.sql.Timestamp
 import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.{DataFrame, Dataset}
-import fr.polytechnique.cmap.cnam.utilities.functions._
 
 trait ObservationPeriodTransformer extends DatasetTransformer[FlatEvent, FlatEvent]{
 
-  final val StudyStart = makeTS(2006, 1, 1)
-  final val StudyEnd = makeTS(2009, 12, 31, 23, 59, 59)
+  final val StudyStart: Timestamp = FilteringConfig.dates.studyStart
+  final val StudyEnd: Timestamp = FilteringConfig.dates.studyEnd
 
   val outputColumns = List(
     col("patientID"),

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

@@ -9,9 +9,9 @@ import org.apache.spark.sql.{Column, DataFrame, Dataset}
 
 object TrackLossTransformer extends Transformer[Event]{
 
-  val EmptyMonths: Int = 4
-  val TracklossMonthDelay: Int = 2
-  val LastDay = Timestamp.valueOf("2010-01-01 00:00:00")
+  val EmptyMonths: Int = FilteringConfig.tracklossDefinition.threshold
+  val TracklossMonthDelay: Int = FilteringConfig.tracklossDefinition.delay
+  val LastDay: Timestamp = FilteringConfig.dates.studyEnd
 
   val inputColumns: List[Column] = List(
     col("NUM_ENQ").as("patientID"),

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

@@ -19,11 +19,21 @@ object CoxConfig {
   lazy val followUpMonthsDelay: Int = modelParams.getInt("follow-up-delay")
   lazy val filterDelayedPatients: Boolean = modelParams.getBoolean("filter_delayed_patients")
   lazy val delayedEntriesThreshold: Int = modelParams.getInt("delayed_entries_threshold")
-  lazy val filterDiagnosedPatients: Boolean = modelParams.getBoolean("filter_diagnosed_patients")
 
   lazy val exposureDefinition = CoxExposureDefinition(
     minPurchases = modelParams.getInt("exposures.min_purchases"),
     startDelay = modelParams.getInt("exposures.start_delay"),
     purchasesWindow = modelParams.getInt("exposures.purchases_window")
   )
+
+  def summarize: Map[String, AnyVal] = {
+    Map(
+      "filterDelayedPatients" -> filterDelayedPatients,
+      "delayedEntriesThreshold" -> delayedEntriesThreshold,
+      "followUpMonthsDelay" -> followUpMonthsDelay,
+      "exposureDefinition.minPurchases" -> exposureDefinition.minPurchases,
+      "exposureDefinition.startDelay" -> exposureDefinition.startDelay,
+      "exposureDefinition.purchasesWindow" -> exposureDefinition.purchasesWindow
+    )
+  }
 }

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

@@ -46,6 +46,7 @@ object CoxExposuresTransformer extends ExposuresTransformer {
         lit(StudyStart),
         DelayedEntriesThreshold
       ).cast(TimestampType)
+
       val drugFilter = max(
         when(
           col("category") === "molecule" && (col("start") <= firstYearObservation),

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

@@ -15,6 +15,7 @@ object CoxMain extends Main {
 
   def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = {
 
+    logger.info("Running FilteringMain...")
     val flatEvents: Dataset[FlatEvent] = FilteringMain.run(sqlContext, argsMap).get
     coxFeaturing(flatEvents, argsMap)
   }
@@ -23,16 +24,15 @@ object CoxMain extends Main {
     import flatEvents.sqlContext.implicits._
 
     val sqlContext = flatEvents.sqlContext
+    val sc = flatEvents.sqlContext.sparkContext
 
     argsMap.get("conf").foreach(sqlContext.setConf("conf", _))
     argsMap.get("env").foreach(sqlContext.setConf("env", _))
 
     val cancerDefinition: String = FilteringConfig.cancerDefinition
     val filterDelayedPatients: Boolean = CoxConfig.filterDelayedPatients
     val outputRoot = FilteringConfig.outputPaths.coxFeatures
-    val outputDir = s"$outputRoot/$cancerDefinition/$filterDelayedPatients"
-
-    logger.info("Running FilteringMain...")
+    val outputDir = s"$outputRoot/$cancerDefinition"
 
     val dcirFlat: DataFrame = sqlContext.read.parquet(FilteringConfig.inputPaths.dcir)
 
@@ -52,7 +52,9 @@ object CoxMain extends Main {
     logger.info("Caching disease events...")
     logger.info("Number of disease events: " + diseaseFlatEvents.count)
 
-    logger.info("Preparing for Cox")
+    logger.info("Preparing for Cox with the following parameters:")
+    logger.info(CoxConfig.summarize.foreach(println))
+
     logger.info("(Lazy) Transforming Follow-up events...")
     val observationFlatEvents = CoxObservationPeriodTransformer.transform(drugFlatEvents)
 
@@ -89,8 +91,9 @@ object CoxMain extends Main {
       .union(observationFlatEvents)
       .union(tracklossFlatEvents)
 
-    logger.info("Writing summary of all cox events...")
+    logger.info("Writing summary of all cox events and config...")
     flatEventsSummary.toDF.write.parquet(s"$outputDir/eventsSummary")
+    sc.parallelize(CoxConfig.summarize.toSeq).coalesce(1).saveAsTextFile(s"$outputDir/config.txt")
     logger.info("Writing Exposures...")
     exposures.toDF.write.parquet(s"$outputDir/exposures")
 

src/test/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxConfigSuite.scala

@@ -0,0 +1,31 @@
+package fr.polytechnique.cmap.cnam.filtering.cox
+
+import fr.polytechnique.cmap.cnam.SharedContext
+
+/**
+  * Created by sathiya on 29/11/16.
+  */
+class CoxConfigSuite extends SharedContext {
+
+  "summarize" should "correctly return all the default Cox Parameters from the config file" in {
+    // Given
+    val expectedResult = Map(
+      "filterDelayedPatients" -> true,
+      "delayedEntriesThreshold" -> 12,
+      "followUpMonthsDelay" -> 6,
+      "exposureDefinition.minPurchases" -> 2,
+      "exposureDefinition.startDelay" -> 3,
+      "exposureDefinition.purchasesWindow" -> 6
+    )
+
+    // When
+    val result = CoxConfig.summarize
+
+    // Then
+    println("Result:")
+    result.foreach(println)
+    println("Expected:")
+    expectedResult.foreach(println)
+    assert(result == expectedResult)
+  }
+}

src/test/scala/fr/polytechnique/cmap/cnam/filtering/cox/CoxMainSuite.scala

@@ -32,7 +32,6 @@ class CoxMainSuite extends SharedContext {
 
     // Then
     import RichDataFrames._
-    result.printSchema()
     assert(result === expectedResult)
   }