CNAM-153 moves all global variables under coxConfig and filteringConfig obj

Author: sathiyapk

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

@@ -15,6 +15,9 @@ paths = {
     root = "/shared/filtered_data"
     patients = "/shared/filtered_data/patients"
     flat_events = "/shared/filtered_data/flat_events"
+    cox_features = "/shared/features/cox_features"
+    ltsccs_features = "/shared/features/ltsccs_features"
+    mlpp_features = "/shared/featues/mlpp_features"
   }
 }
 

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

@@ -15,6 +15,9 @@ paths = {
     root = "/shared/filtered_data"
     patients = "/shared/filtered_data/patients"
     flat_events = "/shared/filtered_data/flat_events"
+    cox_features = "/shared/features/cox_features"
+    ltsccs_features = "/shared/features/ltsccs_features"
+    mlpp_features = "/shared/featues/mlpp_features"
   }
 }
 

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

@@ -15,36 +15,25 @@ default = {
     min_age = 18
     max_age = 120
   }
+  trackloss = {
+    threshold = 4
+    delay = 2
+  }
+
+  cox_parameters = {
+    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.
 
-  cox_hypothesis = {
-    trackloss = {
-      threshold = 4
-      delay = 2
-    }
-    follow-up = {
-      delay = 6
-    }
-    observation = {
-      start = "first_molecule" # or "study_start"
-      end = "study_end"
-    }
     exposures = {
-      min_purchases = 2
-      start_delay = 3
-      purchases_window = 6
-      only_first = false
-      delayed_entries = {
-        filter = true
-        delay = 12
-      }
-      diagnosed_patients = {
-        filter = true
-        delay = 0
-      }
+      min_purchases = 2 #Minimum number of purchases that have to be made in order to be considered exposed.
+      purchases_window = 6 #Purchase window, within which the min number of purchases have to be made.
+      start_delay = 3 #Number of months after which a patient will be considered exposed after the min purchases, window.
     }
   }
 
-  ltsccs_hypothesis = {
+  ltsccs_parameters = {
     observation = {
       start = "first_molecule" # "study_start"
       end = "study_end"
@@ -65,7 +54,7 @@ default = {
     }
   }
 
-  mlpp_hypothesis = {
+  mlpp_parameters = {
     observation = {
       start = "first_molecule" # "study_start"
       end = "study_end"
@@ -94,4 +83,6 @@ cmap = ${default}
 cmap = include "filtering-cmap.conf"
 
 test = ${default}
-test = include "filtering-test.conf"
+test = {
+  include "filtering-test.conf"
+}

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

@@ -15,6 +15,9 @@ paths = {
     root = "target/test/output"
     patients = "target/test/output/patients"
     flat_events = "target/test/output/flat_events"
+    cox_features = "target/test/output/cox_features"
+    ltsccs_features = "target/test/output/ltsccs_features"
+    mlpp_features = "target/test/output/mlpp_features"
   }
 }
 

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

@@ -5,6 +5,7 @@ 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 {
@@ -60,7 +61,14 @@ object FilteringConfig {
     dosages: String
   )
 
-  case class OutputPaths(root: String, patients: String, flatEvents: String)
+  case class OutputPaths(
+    root: String,
+    patients: String,
+    flatEvents: String,
+    coxFeatures: String,
+    ltsccsFeatures: String,
+    mlppFeatures: String
+  )
 
   case class Limits(
     minYear: Int,
@@ -73,7 +81,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 +109,10 @@ 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"),
+    coxFeatures = conf.getString("paths.output.cox_features"),
+    ltsccsFeatures = conf.getString("paths.output.ltsccs_features"),
+    mlppFeatures = conf.getString("paths.output.mlpp_features")
   )
   lazy val limits = Limits(
     minYear = conf.getInt("limits.min_year"),
@@ -105,6 +125,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/cox/CoxConfig.scala

@@ -0,0 +1,29 @@
+package fr.polytechnique.cmap.cnam.filtering.cox
+
+import com.typesafe.config.Config
+import fr.polytechnique.cmap.cnam.filtering.FilteringConfig
+
+/**
+  * Created by sathiya on 23/11/16.
+  */
+object CoxConfig {
+
+  case class CoxExposureDefinition(
+    minPurchases: Int,
+    purchasesWindow: Int,
+    startDelay: Int
+  )
+
+  private lazy val modelParams: Config = FilteringConfig.modelConfig("cox_parameters")
+
+  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")
+  )
+}

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

@@ -4,15 +4,15 @@ import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.{BooleanType, TimestampType}
 import org.apache.spark.sql.{Column, DataFrame, Dataset}
-import fr.polytechnique.cmap.cnam.filtering.{ExposuresTransformer, FlatEvent}
+import fr.polytechnique.cmap.cnam.filtering.{ExposuresTransformer, FilteringConfig, FlatEvent}
 
 object CoxExposuresTransformer extends ExposuresTransformer {
 
   // Constant definitions for delays and time windows. Should be verified before compiling.
   // In the future, we may want to export them to an external file.
-  final val ExposureStartDelay = 3
-  final val ExposureStartThreshold = 6
-  final val DiseaseCode = "C67"
+  final val ExposureStartDelay: Int = CoxConfig.exposureDefinition.startDelay
+  final val ExposureStartThreshold: Int = CoxConfig.exposureDefinition.purchasesWindow
+  final val DiseaseCode: String = FilteringConfig.diseaseCode
 
   val outputColumns = List(
     col("patientID"),
@@ -41,7 +41,10 @@ object CoxExposuresTransformer extends ExposuresTransformer {
       ).over(window).cast(BooleanType)
 
       // Drop patients whose first molecule event is after PeriodStart + 1 year
-      val firstYearObservation = add_months(lit(StudyStart), 12).cast(TimestampType)
+      val firstYearObservation = add_months(
+        lit(StudyStart),
+        CoxConfig.delayedEntriesThreshold
+      ).cast(TimestampType)
       val drugFilter = max(
         when(
           col("category") === "molecule" && (col("start") <= firstYearObservation),

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

@@ -4,13 +4,13 @@ import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.TimestampType
 import org.apache.spark.sql.{Column, DataFrame, Dataset}
-import fr.polytechnique.cmap.cnam.filtering.{DatasetTransformer, FlatEvent}
+import fr.polytechnique.cmap.cnam.filtering.{DatasetTransformer, FilteringConfig, FlatEvent}
 import fr.polytechnique.cmap.cnam.utilities.ColumnUtilities._
 
 object CoxFollowUpEventsTransformer extends DatasetTransformer[FlatEvent, FlatEvent] {
 
-  final val followUpMonthsDelay = 6
-  final val diseaseCode = "C67"
+  final val followUpMonthsDelay: Int = CoxConfig.followUpMonthsDelay
+  final val diseaseCode: String = FilteringConfig.diseaseCode
 
   val outputColumns = List(
     col("patientID"),

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

@@ -1,9 +1,8 @@
 package fr.polytechnique.cmap.cnam.filtering.cox
 
-import org.apache.spark.sql.{DataFrame, Dataset}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.hive.HiveContext
-import com.typesafe.config.{Config, ConfigFactory}
+import org.apache.spark.sql.{DataFrame, Dataset}
 import fr.polytechnique.cmap.cnam.Main
 import fr.polytechnique.cmap.cnam.filtering._
 
@@ -14,26 +13,40 @@ object CoxMain extends Main {
 
   def appName = "CoxFeaturing"
 
-  def coxFeaturing(sqlContext: HiveContext,
-                   config: Config,
-                   cancerDefinition: String,
-                   filterDelayedPatients: Boolean): Unit = {
-    import sqlContext.implicits._
+  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = {
+
+    val flatEvents: Dataset[FlatEvent] = FilteringMain.run(sqlContext, argsMap).get
+    coxFeaturing(flatEvents, argsMap)
+  }
+
+  def coxFeaturing(flatEvents: Dataset[FlatEvent], argsMap: Map[String, String]): Option[Dataset[_]] = {
+    import flatEvents.sqlContext.implicits._
 
-    val flatEventPath = config.getString("paths.input.flatEvent")
-    val flatDcirPath = config.getString("paths.input.flatDcir")
-    val outputRoot = config.getString("paths.output.root")
+    val sqlContext = flatEvents.sqlContext
+
+    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(s"Reading flat events from $flatEventPath...")
+    logger.info("Running FilteringMain...")
+
+    val dcirFlat: DataFrame = sqlContext.read.parquet(FilteringConfig.inputPaths.dcir)
 
-    val dcirFlat: DataFrame = sqlContext.read.parquet(flatDcirPath)
-    val flatEvents: DataFrame = sqlContext.read.parquet(flatEventPath)
+    val drugFlatEvents = flatEvents.filter(_.category == "molecule")
+    val diseaseFlatEvents = flatEvents.filter(_.category == "disease")
 
-    val drugFlatEvents = flatEvents.filter(col("category") === "molecule").as[FlatEvent]
-    val diseaseFlatEvents = flatEvents.filter(col("category") === "disease").as[FlatEvent]
-    val patientColumns = Array($"patientID", $"gender", $"birthDate", $"deathDate")
-    val patients = flatEvents.select(patientColumns: _*).distinct.as[Patient]
+    val patients: Dataset[Patient] = flatEvents
+      .map(
+        x => Patient(
+          x.patientID,
+          x.gender,
+          x.birthDate,
+          x.deathDate)
+      ).distinct
 
     logger.info("Number of drug events: " + drugFlatEvents.count)
     logger.info("Caching disease events...")
@@ -85,21 +98,7 @@ object CoxMain extends Main {
     import CoxFeaturesWriter._
     coxFeatures.toDF.write.parquet(s"$outputDir/cox")
     coxFeatures.writeCSV(s"$outputDir/cox.csv")
-  }
 
-  override def main(args: Array[String]): Unit = {
-    startContext()
-    val (environment: String, cancerDefinition: String, filterDelayedPatients: Boolean) =
-      args match {
-        case Array(arg1, args2, args3) => (args(0), args(1), args(2).toBoolean)
-        case Array(arg1, args2) => (args(0), args(1), true)
-        case _ => ("test", "broad", true)
-      }
-    val config: Config = ConfigFactory.parseResources("config/filtering-default.conf").getConfig(environment)
-    coxFeaturing(sqlContext, config, cancerDefinition, filterDelayedPatients)
-    stopContext()
+    Some(coxFeatures)
   }
-
-  // todo: refactor this function
-  def run(sqlContext: HiveContext, argsMap: Map[String, String]): Option[Dataset[_]] = None
 }

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

@@ -1,11 +1,11 @@
 package fr.polytechnique.cmap.cnam.filtering.cox
 
+import java.sql.Timestamp
 import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.IntegerType
 import org.apache.spark.sql.{Column, DataFrame, Dataset}
-import fr.polytechnique.cmap.cnam.filtering.{DatasetTransformer, FlatEvent}
-import fr.polytechnique.cmap.cnam.utilities.functions._
+import fr.polytechnique.cmap.cnam.filtering.{DatasetTransformer, FilteringConfig, FlatEvent}
 
 // Start and End are expressed in month from the patient startObs
 case class CoxFeature(
@@ -35,7 +35,7 @@ object CoxTransformer extends DatasetTransformer[FlatEvent, CoxFeature] {
     "other"
   )
 
-  final val AgeReferenceDate = makeTS(2006, 12, 31, 23, 59, 59)
+  final val AgeReferenceDate: Timestamp = FilteringConfig.dates.ageReference
 
   implicit class CoxDataFrame(data: DataFrame) {