Cnam-466-merge2master

Events.md

@@ -0,0 +1,84 @@
+# Events
+
+An [Event](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/events/Event.scala) 
+ is a figure that allows to homogenize the content of the output datasets.
+It consists, at least, of an object to be instantiated and a trait containing the necessary functions. 
+In order to create an  `Event` you first need to understand what it is. 
+An event is an occurrence defined by having a patient identifier, category, start, group identifier, value, weight and  end; 
+of these only the first three are mandatory.The 7 elements needed to form an `Event` are:
+- patientID: is the patient identifier.
+- category: define the event's category.
+- groupID: contains the ID of a group of related events.
+- value: contains string values fot the molecule name, the diagnosis code, etc.
+- weight: contains double values for medical acts weighting or other numerical values.
+- start: is the start of event's period.
+- end: is the end of event's period.
+```scala  
+  patientID: String,
+  category: EventCategory[A],
+  groupID: String,
+  value: String,
+  weight: Double,
+  start: Timestamp,
+  end: Option[Timestamp] 
+```  
+All events inherit [AnyEvent](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/events/AnyEvent.scala)
+ and  [EventBuilder](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/events/EventBuilder.scala). 
+ `AnyEvent` is a trait with category value, and `EventBuilder` is a trait to built  an `Event`.
+
+
+[ObservationPeriod](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/events/ObservationPeriod.scala)
+and [MedicalAct](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/events/MedicalAct.scala)
+are good examples to illustrate the construction of an `Event`.
+The first one,`ObservationPeriod`, define the values  `patientID`, `category`, `start` and if exist `end`. 
+The other values are present but use default values. Its trait just assigns the value of the category and uses the apply method to build an `Event`.
+```scala  
+ val category: EventCategory[ObservationPeriod] = "observation_period"
+
+  /** Creates un Event object of type ObservationPeriod using a map function to map a dataset.
+   *
+   * @param patientID The value patientID from dataset.
+   * @param start     The value start from dataset.
+   * @param end       The value end from dataset.
+   * @return Event[ObservationPeriod].
+   */
+  def apply(patientID: String, start: Timestamp, end: Timestamp): Event[ObservationPeriod] =
+    Event(patientID, category, groupID = "NA", value = "NA", weight = 0D, start, Some(end))
+```  
+And its object doesn't add up to anything, it just inherits the trait.
+```scala  
+object ObservationPeriod extends ObservationPeriod
+```
+On the other side,`MedicalAct` defines values for all elements except `end` one. 
+Its trait uses two apply methods to assign values according to need and not assign any value to category one, 
+it's assigned in each object according to the type.
+
+```scala  
+
+  override val category: EventCategory[MedicalAct]
+
+  def apply(patientID: String, groupID: String, code: String, weight: Double, date: Timestamp): Event[MedicalAct] = {
+    Event(patientID, category, groupID, code, weight, date, None)
+  }
+
+  def apply(patientID: String, groupID: String, code: String, date: Timestamp): Event[MedicalAct] = {
+    Event(patientID, category, groupID, code, 0.0, date, None)
+  }
+``` 
+The MedicalAct's object are various in accordance with the type of medical act. 
+Their objects assign categories and in some cases have object that stores groupID values.
+
+```scala 
+object BiologyDcirAct extends MedicalAct {
+  override val category: EventCategory[MedicalAct] = "dcir_biology_act"
+
+  object groupID {
+    val PrivateAmbulatory = "private_ambulatory"
+    val PublicAmbulatory = "public_ambulatory"
+    val PrivateHospital = "private_hospital"
+    val Liberal = "liberal"
+    val DcirAct = "DCIR_act"
+    val Unknown = "unknown_source"
+  }
+}
+```

Extractors.md

@@ -0,0 +1,72 @@
+
+
+# Extractors
+
+Extractors are a kind of jobs that allows to extract the required columns from the sources and maps them to the `Event` ([Events](Events.md)) it is extracting. 
+An extractor is composed of several basic components that are grouped to give all the functionalities to the extractors. 
+From a point of view hierarchical we have:
+
+1) Base elements are traits as [Extractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/Extractor.scala), 
+[McoSource](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/mco/McoSource.scala) and 
+[EventRowExtractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/EventRowExtractor.scala).
+Each manages different parts of extraction. They are all necessary for the creation of an element of the next level.
+2) Intermediate elements are traits that bring together the basic methods to create a common trait to extract the data from the sources (mco,dcir,ssr,had).In our example, 
+[McoExtractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/mco/McoExtractor.scala) 
+is the base for retrieving the necessary data from mco source. These elements inherit the base elements and accept as a parameter a trait of the type EventType
+3) The upper elements  are objects and  they are the entry point of the job itself, they inherit from the middle elements and specialize it in one type of `Event`. 
+
+Each basic component has a function:
+ - [Extractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/Extractor.scala)
+ is a trait that works by filtering, extracting and building.Its main method  `extract ` allows to filter the sources and build a dataset of type `Event`.
+```scala  
+def extract(sources: Sources, codes: Set[String])(implicit ctag: TypeTag[EventType]): Dataset[Event[EventType]] = {
+            val input: DataFrame = getInput(sources)      
+            import input.sqlContext.implicits._        
+            {
+              if (codes.isEmpty) {
+                input.filter(isInExtractorScope _)
+              }
+              else {
+                input.filter(isInExtractorScope _).filter(isInStudy(codes) _)
+              }
+            }.flatMap(builder _).distinct()
+          }
+```
+- [McoSource](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/mco/McoSource.scala)
+This trait is specific to each source (mco,dcir,ssr,had), containing the values relating to the columns and the methods specific to that source.
+- [EventRowExtractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/EventRowExtractor.scala) 
+This trait contains methods for extracting the fields needed to create an `Event`.
+
+The intermediate elements implement the required methods and adapt them if necessary. 
+Two good examples of implementation and modification to suit the specificities of `Event` type are `builder` method and  `extractGroupId` method.
+
+```scala  
+def builder(row: Row): Seq[Event[EventType]] = {
+    lazy val patientId = extractPatientId(row)
+    lazy val groupId = extractGroupId(row)
+    lazy val eventDate = extractStart(row)
+    lazy val endDate = extractEnd(row)
+    lazy val weight = extractWeight(row)
+
+    Seq(eventBuilder[EventType](patientId, groupId, code(row), weight, eventDate, endDate))
+  }
+```
+```scala  
+ override def extractGroupId(r: Row): String = {
+      r.getAs[String](ColNames.EtaNum) + "_" +
+      r.getAs[String](ColNames.RsaNum) + "_" +
+      r.getAs[Int](ColNames.Year).toString
+  }
+```
+
+The above elements are responsible for defining the type of `Event` and must implement at least  `columnName` and  `eventBuilder` values and modify them according to their specificity.
+For exemple at [McoHospitalStaysExtractor](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/extractors/hospitalstays/McoHospitalStaysExtractor.scala)
+```scala  
+object McoHospitalStaysExtractor extends McoExtractor[HospitalStay]{
+
+  override val columnName: String = ColNames.EndDate
+  override val eventBuilder: EventBuilder = McoHospitalStay
+}
+```
+
+

Transformer.md

@@ -0,0 +1,41 @@
+# Transformers
+
+A transformer is used to turn one or multiple  datasets  
+into another one. A  `Transformer`  accept a configuration class as a parameter 
+a configuration class of type
+ [TransformerConfig](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/transformers/TransformerConfig.scala).
+This configuration file control the behaviour of the `Transformer` through its methods and values.
+
+
+`transform` is the main and the unique public available method of `Transformer`, this method accepts one or several datasets  as input
+ and after the application of the transformation logic, returns a dataset of type `Event`([Events](Events.md)) with a new `category`.
+
+ If we take as an example [ExposureTransformer](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/transformers/exposures/ExposureTransformer.scala),
+ we pass an instance of the class [ExposuresTransformerConfig](https://github.com/X-DataInitiative/SCALPEL-Extraction/blob/master/src/main/scala/fr/polytechnique/cmap/cnam/etl/transformers/exposures/ExposuresTransformerConfig.scala)
+ as parameter, it extends a `TransformerConfig`.
+ ```scala  
+/**
+  * A tag to improve readability of subclasses and to allow type binding
+  */
+
+trait TransformerConfig 
+ ```
+The config class is used to control the Transformer's  behaviour.
+ ```scala  
+class ExposuresTransformerConfig(
+  val exposurePeriodAdder: ExposurePeriodAdder) extends TransformerConfig with Serializable
+ ```  
+This main method take as parameter two datasets of type `Event` and subtypes `FollowUp` and `Drug` respectively
+and return a dataset of type `Event` and subtype `Exposure`.
+
+```scala 
+  def transform(followUps: Dataset[Event[FollowUp]])(drugs: Dataset[Event[Drug]]): Dataset[Event[Exposure]] = {
+    drugs
+      .transform(config.exposurePeriodAdder.toExposure(followUps))
+      .transform(regulateWithFollowUps(followUps))
+  }
+ ```
+The objective of the `transform` method is thus to combine, filter, check relations between multiples `Event`s to form a new `Event`.
+In the example of the `ExposureTransformer`, it combines multiples `Drug` `Event`s based on the logic of 
+the `exposurePeriodAdder` of the `ExposuresTransformerConfig` to form an `Exposure` while 
+making sure that it is contained within the period defined in the `FollowUp` `Event`.

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.15
+sbt.version=0.13.18

src/main/resources/config/bulk/default.conf

@@ -1,5 +1,8 @@
 root {
-
+  drugs {
+    level: "cip13"
+    families: []
+  }
 }
 
 

src/main/resources/config/bulk/paths/cmap.conf

@@ -1,10 +1,10 @@
 env_name = "cmap"
 
 input = {
-  dcir = "/shared/Observapur/staging/Flattening/flat_table/DCIR"
-  mco_ce = "/shared/Observapur/staging/Flattening/flat_table/MCO_CE"
-  mco = "/shared/Observapur/staging/Flattening/flat_table/MCO"
-  ir_ben = "/shared/Observapur/staging/Flattening/single_table/IR_BEN_R"
+  dcir = "/user/ds/CNAM447/flattening/flat_table/DCIR"
+  mco_ce = "/user/ds/CNAM447bis/flattening/flat_table/MCO_CE"
+  mco = "/user/ds/CNAM447bis/flattening/flat_table/MCO"
+  ir_ben = "/user/ds/CNAM447/flattening/single_table/IR_BEN_R"
   ir_imb = "/shared/Observapur/staging/Flattening/single_table/IR_IMB_R"
   ir_pha = "/shared/Observapur/staging/Flattening/single_table/IR_PHA_R_MOL"
 }

src/main/resources/config/fall/default.conf

@@ -11,8 +11,9 @@ root {
       to_exposure_strategy  = "purchase_count_based"
     }
   }
-  interaction {
+  interactions {
     level: 2
+    minimum_duration: 30 days
   }
   drugs {
     level: "Therapeutic"
@@ -28,7 +29,7 @@ root {
     fall_frame: 0 months        // fractures are grouped if they happen in the same site within the period fallFrame, (default value 0 means no group)
   }
   run_parameters {
-    outcome:  ["Acts", "Diagnoses", "Outcomes"]                               // pipeline of calculation of outcome, possible values : Acts, Diagnoses, and Outcomes
+    outcome:  ["Acts", "Diagnoses", "HospitalDeaths", "Outcomes"]                               // pipeline of calculation of outcome, possible values : Acts, Diagnoses, and Outcomes
     exposure: ["Patients", "StartGapPatients", "DrugPurchases", "Exposures"]  // pipeline of the calculation of exposure, possible values : Patients, StartGapPatients, DrugPurchases, Exposures
   }
 }

src/main/resources/config/fall/template.conf

@@ -16,8 +16,10 @@
 # exposures.end_threshold_gc: 90 days  // If periodStrategy="limited", represents the period without purchases for an exposure to be considered "finished".
 # exposures.end_threshold_ngc: 30 days // If periodStrategy="limited", represents the period without purchases for an exposure to be considered "finished".
 # exposures.end_delay: 30 days         // Number of periods that we add to the exposure end to delay it (lag).
+# exposures.to_exposure_strategy: purchase_count_based // possible values "purchase_count_based" or "lastest_purchase_based"
 
 # interactions.level: 3 // Integer representing the maximum number of values of Interaction. Please be careful as this not scale well beyond 5 when the data contains a patient with very high number of exposures
+# interactions.minimum_duration: 30 days // If Interaction duration is less than this value, it is not considered. Proxy for medication change.
 
 # drugs.level: "Therapeutic"           // Options are Therapeutic, Pharmacological, MoleculeCombination
 # drugs.families: ["Antihypertenseurs", "Antidepresseurs", "Neuroleptiques", "Hypnotiques"]

src/main/scala/fr/polytechnique/cmap/cnam/etl/config/ConfigLoader.scala

@@ -11,7 +11,7 @@ import me.danielpes.spark.datetime.implicits._
 import pureconfig._
 import pureconfig.configurable.{localDateConfigConvert, localDateTimeConfigConvert}
 import pureconfig.generic.{CoproductHint, EnumCoproductHint, FieldCoproductHint, ProductHint}
-import fr.polytechnique.cmap.cnam.etl.extractors.drugs.level.DrugClassificationLevel
+import fr.polytechnique.cmap.cnam.etl.extractors.events.drugs.level.DrugClassificationLevel
 import fr.polytechnique.cmap.cnam.etl.transformers.exposures.ExposurePeriodAdder
 
 trait ConfigLoader {

src/main/scala/fr/polytechnique/cmap/cnam/etl/config/study/StudyConfig.scala

@@ -12,12 +12,13 @@ object StudyConfig {
     dcir: Option[String] = None,
     mco: Option[String] = None,
     mcoCe: Option[String] = None,
-    ssr: Option[List[String]] = None,
+    ssr: Option[String] = None,
     ssrCe: Option[String] = None,
     had: Option[String] = None,
     irBen: Option[String] = None,
     irImb: Option[String] = None,
     irPha: Option[String] = None,
+    irNat: Option[String] = None,
     dosages: Option[String] = None)
 
   case class OutputPaths(

src/main/scala/fr/polytechnique/cmap/cnam/etl/datatypes/Period.scala

@@ -3,7 +3,6 @@
 package fr.polytechnique.cmap.cnam.etl.datatypes
 
 import java.sql.Timestamp
-import fr.polytechnique.cmap.cnam.etl.transformers.interaction._
 import fr.polytechnique.cmap.cnam.util.functions._
 
 case class Period(start: Timestamp, end: Timestamp) extends Subtractable[Period] with Addable[Period]{

src/main/scala/fr/polytechnique/cmap/cnam/etl/events/Classification.scala

@@ -3,28 +3,12 @@
 package fr.polytechnique.cmap.cnam.etl.events
 
 import java.sql.Timestamp
-import org.apache.spark.sql.Row
 
 
 trait Classification extends AnyEvent with EventBuilder {
 
   val category: EventCategory[Classification]
 
-  def fromRow(
-    r: Row,
-    patientIDCol: String = "patientID",
-    nameCol: String = "name",
-    groupIDCol: String = "groupID",
-    dateCol: String = "eventDate")
-  : Event[Classification] = {
-    apply(
-      r.getAs[String](patientIDCol),
-      r.getAs[String](groupIDCol),
-      r.getAs[String](nameCol),
-      r.getAs[Timestamp](dateCol)
-    )
-  }
-
   def apply(
     patientID: String,
     groupID: String,

src/main/scala/fr/polytechnique/cmap/cnam/etl/events/Diagnosis.scala

@@ -3,57 +3,27 @@
 package fr.polytechnique.cmap.cnam.etl.events
 
 import java.sql.Timestamp
-import org.apache.spark.sql.Row
 
 trait Diagnosis extends AnyEvent with EventBuilder {
 
   val category: EventCategory[Diagnosis]
 
-  def fromRow(r: Row, patientIDCol: String, codeCol: String, dateCol: String): Event[Diagnosis] = {
-    apply(r.getAs[String](patientIDCol), r.getAs[String](codeCol), r.getAs[Timestamp](dateCol))
-  }
-
   def apply(patientID: String, code: String, date: Timestamp): Event[Diagnosis] = {
     Event(patientID, category, groupID = "NA", code, 0.0, date, None)
   }
 
-  def fromRow(
-    r: Row,
-    patientIDCol: String = "patientID",
-    groupIDCol: String = "groupID",
-    codeCol: String = "code",
-    dateCol: String = "eventDate"): Event[Diagnosis] = {
-    apply(
-      r.getAs[String](patientIDCol),
-      r.getAs[String](groupIDCol),
-      r.getAs[String](codeCol),
-      r.getAs[Timestamp](dateCol)
-    )
+  def apply(patientID: String, code: String, date: Timestamp, endDate: Option[Timestamp]): Event[Diagnosis] = {
+    Event(patientID, category, groupID = "NA", code, 0.0, date, endDate)
   }
 
   def apply(patientID: String, groupID: String, code: String, date: Timestamp): Event[Diagnosis] = {
     Event(patientID, category, groupID, code, 0.0, date, None)
   }
 
-  def fromRow(
-    r: Row,
-    patientIDCol: String,
-    groupIDCol: String,
-    codeCol: String,
-    weightCol: String,
-    dateCol: String): Event[Diagnosis] = {
-    apply(
-      r.getAs[String](patientIDCol),
-      r.getAs[String](groupIDCol),
-      r.getAs[String](codeCol),
-      r.getAs[Double](weightCol),
-      r.getAs[Timestamp](dateCol)
-    )
-  }
-
   def apply(patientID: String, groupID: String, code: String, weight: Double, date: Timestamp): Event[Diagnosis] = {
     Event(patientID, category, groupID, code, weight, date, None)
   }
+
 }
 
 object McoMainDiagnosis extends Diagnosis {
@@ -92,6 +62,6 @@ object SsrTakingOverPurpose extends Diagnosis {
   val category: EventCategory[Diagnosis] = "ssr_taking_over_purpose"
 }
 
-object ImbDiagnosis extends Diagnosis {
-  override val category: EventCategory[Diagnosis] = "imb_diagnosis"
+object ImbCcamDiagnosis extends Diagnosis {
+  override val category: EventCategory[Diagnosis] = "imb_ccam_diagnosis"
 }