AFUS Dog Transformations, Force TSV Column Order, Process AfusOwner Data

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/afus/AfusExtractionPipeline.scala

@@ -19,37 +19,39 @@ object AfusExtractionPipeline extends ScioApp[Args] {
     "followup_status",
     "followup_owner_contact",
     "study_status",
-    "followup_owner_demographics"
+    "followup_owner_demographics",
+    "followup_dog_demographics",
+    "followup_physical_activity",
+    "followup_environment",
+    "followup_behavior",
+    "followup_diet",
+    "followup_meds_and_preventives",
+    "followup_canine_eating_behavior_dora",
+    "followup_dogowner_relationship_survey_mdors",
+    "followup_health_status",
+    "followup_health_status_part_1",
+    "followup_health_status_part_2",
+    "followup_health_status_part_3"
   )
 
   def extractionFiltersGenerator(args: Args): List[FilterDirective] = {
-    // FORM = followup_status
     val standardDirectives: List[FilterDirective] = List(
-      FilterDirective("fu_is_completed", FilterOps.==, "1")
+      // DAP Pack filters
+      FilterDirective("st_dap_pack_count", FilterOps.>, "0"),
+      FilterDirective("st_vip_or_staff", FilterOps.==, "0")
     )
     val dateFilters: List[FilterDirective] = {
-      args.startTime
-        .map(start =>
+      args.endTime
+        .map(end =>
           List(
             FilterDirective(
               "fu_complete_date",
-              FilterOps.>,
-              RedCapClient.redcapFormatDate(start)
+              FilterOps.<,
+              RedCapClient.redcapFormatDate(end)
             )
           )
         )
-        .getOrElse(List()) ++
-        args.endTime
-          .map(end =>
-            List(
-              FilterDirective(
-                "fu_complete_date",
-                FilterOps.<,
-                RedCapClient.redcapFormatDate(end)
-              )
-            )
-          )
-          .getOrElse(List())
+        .getOrElse(List())
     }
 
     standardDirectives ++ dateFilters
@@ -79,9 +81,16 @@ object AfusExtractionPipeline extends ScioApp[Args] {
     // afus has one arm per year ("fup_{sequence}_arm_1")
     val yearList = getYearList(startDate, endDate)
     val yearSeqList = List.range(1, yearList.length + 1)
-    yearSeqList.map { seq =>
-      s"fup_${seq}_arm_1"
-    } ++ List("baseline_arm_1")
+//    yearSeqList.map { seq =>
+//      s"fup_${seq}_arm_1"
+//    } ++ List("baseline_arm_1")
+    if (1 == 1) {
+      List("fup_1_arm_1", "baseline_arm_1")
+    } else {
+      yearSeqList.map { seq =>
+        s"fup_${seq}_arm_1"
+      } ++ List("baseline_arm_1")
+    }
   }
 
   def buildPipelineWithWrapper(wrapper: HttpWrapper): PipelineBuilder[Args] =
@@ -91,7 +100,7 @@ object AfusExtractionPipeline extends ScioApp[Args] {
       extractionArmsGenerator,
       fieldList,
       subdir,
-      10,
+      1,
       RedCapClient.apply(_: List[String], wrapper)
     )
 

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/cslb/CslbExtractionPipeline.scala

@@ -15,24 +15,25 @@ object CslbExtractionPipeline extends ScioApp[Args] {
 
   val forms = List(
     "recruitment_fields",
-    "canine_social_and_learned_behavior"
+    "canine_social_and_learned_behavior",
+    "study_status"
   )
 
   // Magic marker for "completed".
   // NB: We are purposefully excluding the recruitment_fields_complete -> 2
   // mapping, as that conflicts with the CSLB data
-  def extractionFiltersGenerator(args: Args): List[FilterDirective] =
-    List(
-      FilterDirective("canine_social_and_learned_behavior_complete", FilterOps.==, "2")
-    ) ++
-      args.startTime
-        .map(start =>
-          List(FilterDirective("cslb_date", FilterOps.>, RedCapClient.redcapFormatDate(start)))
-        )
-        .getOrElse(List()) ++
+  def extractionFiltersGenerator(args: Args): List[FilterDirective] = {
+    val standardDirectives: List[FilterDirective] = List(
+      FilterDirective("canine_social_and_learned_behavior_complete", FilterOps.==, "2"),
+      FilterDirective("st_vip_or_staff", FilterOps.==, "0")
+    )
+    val dateFilters: List[FilterDirective] = {
       args.endTime
         .map(end => List(FilterDirective("cslb_date", FilterOps.<, RedCapClient.redcapFormatDate(end))))
         .getOrElse(List())
+    }
+    standardDirectives ++ dateFilters
+  }
 
   val subdir = "cslb"
 
@@ -57,10 +58,10 @@ object CslbExtractionPipeline extends ScioApp[Args] {
     // cslb has one arm per year ("annual_{yyyy}_arm_1")
     getYearList(startDate, endDate).map { date =>
       s"annual_${date}_arm_1"
-    }
+    } ++ List("baseline_arm_1")
   }
 
-  val fieldList = List("co_consent")
+  val fieldList = List("co_consent", "st_vip_or_staff")
 
   def buildPipelineWithWrapper(wrapper: HttpWrapper): PipelineBuilder[Args] =
     new ExtractionPipelineBuilder(
@@ -69,7 +70,7 @@ object CslbExtractionPipeline extends ScioApp[Args] {
       extractionArmsGenerator,
       fieldList,
       subdir,
-      1000,
+      10,
       RedCapClient.apply(_: List[String], wrapper)
     )
 

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/environment/EnvironmentExtractionPipeline.scala

@@ -27,18 +27,24 @@ object EnvironmentExtractionPipeline extends ScioApp[Args] {
 
   // Magic marker for "completed".
   // NB: We are looking for baseline_complete -> 2
-  def extractionFiltersGenerator(args: Args): List[FilterDirective] =
-    List(FilterDirective("baseline_complete", FilterOps.==, "2")) ++
-      args.startTime
-        .map(start =>
-          List(FilterDirective("bl_dap_pack_date", FilterOps.>, RedCapClient.redcapFormatDate(start)))
-        )
-        .getOrElse(List()) ++
-      args.endTime
-        .map(end =>
-          List(FilterDirective("bl_dap_pack_date", FilterOps.<, RedCapClient.redcapFormatDate(end)))
-        )
-        .getOrElse(List())
+  def extractionFiltersGenerator(args: Args): List[FilterDirective] = {
+    val dateFilters: List[FilterDirective] = {
+      List(FilterDirective("baseline_complete", FilterOps.==, "2")) ++
+        args.startTime
+          .map(start =>
+            List(
+              FilterDirective("bl_dap_pack_date", FilterOps.>, RedCapClient.redcapFormatDate(start))
+            )
+          )
+          .getOrElse(List()) ++
+        args.endTime
+          .map(end =>
+            List(FilterDirective("bl_dap_pack_date", FilterOps.<, RedCapClient.redcapFormatDate(end)))
+          )
+          .getOrElse(List())
+    }
+    dateFilters
+  }
 
   val subdir = "environment"
 

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/eols/EolsExtractionPipeline.scala

@@ -29,23 +29,19 @@ object EolsExtractionPipeline extends ScioApp[Args] {
       .map(form => FilterDirective(s"${form}_complete", FilterOps.==, "2"))
     // EOLS consent
     val standardDirectives: List[FilterDirective] = List(
-      FilterDirective("eol_willing_to_complete", FilterOps.==, "1")
+      FilterDirective("eol_willing_to_complete", FilterOps.==, "1"),
+      // DAP Pack filters
+      FilterDirective("st_vip_or_staff", FilterOps.==, "0")
     )
-    val dateFilters: List[FilterDirective] =
-      args.startTime
-        .map(start =>
+    val dateFilters: List[FilterDirective] = {
+      args.endTime
+        .map(end =>
           List(
-            FilterDirective("eol_date_dog_died", FilterOps.>, RedCapClient.redcapFormatDate(start))
+            FilterDirective("eol_date_dog_died", FilterOps.<, RedCapClient.redcapFormatDate(end))
           )
         )
-        .getOrElse(List()) ++
-        args.endTime
-          .map(end =>
-            List(
-              FilterDirective("eol_date_dog_died", FilterOps.<, RedCapClient.redcapFormatDate(end))
-            )
-          )
-          .getOrElse(List())
+        .getOrElse(List())
+    }
 
     completionFilters ++ standardDirectives ++ dateFilters
   }

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/hles/HLESExtractionPipeline.scala

@@ -33,11 +33,7 @@ object HLESExtractionPipeline extends ScioApp[Args] {
   def extractionFiltersGenerator(args: Args): List[FilterDirective] = {
     val standardDirectives: List[FilterDirective] = List(
       FilterDirective("st_dap_pack_count", FilterOps.>, "0"),
-      FilterDirective(
-        "st_dap_pack_date",
-        FilterOps.>,
-        RedCapClient.redcapFormatDate(args.startTime.getOrElse(HLESEpoch))
-      )
+      FilterDirective("st_vip_or_staff", FilterOps.==, "0")
     )
     val endFilter: List[FilterDirective] =
       args.endTime

dap-etl/extraction/src/main/scala/org/broadinstitute/monster/dap/sample/SampleExtractionPipeline.scala

@@ -19,21 +19,22 @@ object SampleExtractionPipeline extends ScioApp[Args] {
     "recruitment_fields"
   )
 
-  def extractionFiltersGenerator(args: Args): List[FilterDirective] =
-    args.startTime
-      .map(start =>
-        List(
-          FilterDirective("k1_rtn_tracking_date", FilterOps.>, RedCapClient.redcapFormatDate(start))
-        )
-      )
-      .getOrElse(List()) ++
+  def extractionFiltersGenerator(args: Args): List[FilterDirective] = {
+    val standardDirectives: List[FilterDirective] = List(
+      // DAP Pack filters
+      FilterDirective("st_dap_pack_count", FilterOps.>, "0")
+    )
+    val dateFilters: List[FilterDirective] = {
       args.endTime
         .map(end =>
           List(
             FilterDirective("k1_rtn_tracking_date", FilterOps.<, RedCapClient.redcapFormatDate(end))
           )
         )
         .getOrElse(List())
+    }
+    standardDirectives ++ dateFilters
+  }
 
   val subdir = "sample";
   val arm = "baseline_arm_1"

dap-etl/extraction/src/test/scala/org/broadinstitute/monster/dap/afus/AfusExtractionPipelineBuilderIntegrationSpec.scala

@@ -35,7 +35,7 @@ class AfusExtractionPipelineBuilderIntegrationSpec extends PipelineBuilderSpec[A
     testArgs.apiToken,
     arms,
     GetRecords(
-      fields = List("st_owner_id"),
+      fields = List("study_id"),
       filters = AfusExtractionPipeline.extractionFiltersGenerator(testArgs)
     )
   )

dap-etl/extraction/src/test/scala/org/broadinstitute/monster/dap/afus/AfusExtractionPipelineSpec.scala

@@ -14,15 +14,19 @@ class AfusExtractionPipelineSpec extends AnyFlatSpec with Matchers {
 
     val afusArms = AfusExtractionPipeline.extractionArmsGenerator(Some(startTime), Some(endTime))
 
-    afusArms shouldBe List("fup_1_arm_1", "fup_2_arm_1", "fup_3_arm_1", "baseline_arm_1")
+    // todo: update when extraction arm generator logic is updated
+    afusArms shouldBe List("fup_1_arm_1", "baseline_arm_1")
+    //afusArms shouldBe List("fup_1_arm_1", "fup_2_arm_1", "fup_3_arm_1", "baseline_arm_1")
   }
 
   it should "use AFUSEpoch as startTime when none is provided" in {
     val endTime =
       OffsetDateTime.of(2022, 12, 1, 0, 0, 0, 0, ZoneOffset.ofHours(-5))
     val afusArms = AfusExtractionPipeline.extractionArmsGenerator(None, Some(endTime))
 
-    afusArms shouldBe List("fup_1_arm_1", "fup_2_arm_1", "baseline_arm_1")
+    // todo: update when extraction arm generator logic is updated
+    afusArms shouldBe List("fup_1_arm_1", "baseline_arm_1")
+    //afusArms shouldBe List("fup_1_arm_1", "fup_2_arm_1", "baseline_arm_1")
   }
 
   it should "fail when startTime is after endTime" in {