CNDE-1902: Legacy Investigation Update (Pre-Processing) (#83)

CDCgov · Nov 19, 2024 · e98c5fd · e98c5fd
1 parent 29f9e15
commit e98c5fd
Show file tree

Hide file tree

Showing 16 changed files with 790 additions and 639 deletions.
diff --git a/...c/main/java/gov/cdc/etldatapipeline/investigation/repository/model/dto/Investigation.java b/...c/main/java/gov/cdc/etldatapipeline/investigation/repository/model/dto/Investigation.java
@@ -298,8 +298,8 @@ public class Investigation {
     @Column(name = "act_ids")
     private String actIds;
 
-    @Column(name = "observation_notification_ids")
-    private String observationNotificationIds;
+    @Column(name = "investigation_observation_ids")
+    private String investigationObservationIds;
 
     @Column(name = "person_participations")
     private String personParticipations;
@@ -321,4 +321,7 @@ public class Investigation {
 
     @Column(name = "investigation_case_count")
     private String investigationCaseCnt;
+
+    @Column(name = "investigation_form_cd")
+    private String investigationFormCd;
 }
diff --git a/...dc/etldatapipeline/investigation/repository/model/reporting/InvestigationObservation.java b/...dc/etldatapipeline/investigation/repository/model/reporting/InvestigationObservation.java
@@ -9,4 +9,7 @@
 public class InvestigationObservation {
     private Long publicHealthCaseUid;
     private Long observationId;
+    private String rootTypeCd;
+    private Long branchId;
+    private String branchTypeCd;
 }
diff --git a/...etldatapipeline/investigation/repository/model/reporting/InvestigationObservationKey.java b/...etldatapipeline/investigation/repository/model/reporting/InvestigationObservationKey.java
@@ -0,0 +1,13 @@
+package gov.cdc.etldatapipeline.investigation.repository.model.reporting;
+
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Data;
+
+@Data
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+public class InvestigationObservationKey {
+    private Long publicHealthCaseUid;
+    private Long observationId;
+    private Long branchId;
+}
diff --git a/.../cdc/etldatapipeline/investigation/repository/model/reporting/InvestigationReporting.java b/.../cdc/etldatapipeline/investigation/repository/model/reporting/InvestigationReporting.java
@@ -113,4 +113,5 @@ public class InvestigationReporting {
     private Long investigationCount;
     private Long caseCount;
     private String investigatorAssignedDatetime;
+    private String investigationFormCd;
 }
diff --git a/...rc/main/java/gov/cdc/etldatapipeline/investigation/util/ProcessInvestigationDataUtil.java b/...rc/main/java/gov/cdc/etldatapipeline/investigation/util/ProcessInvestigationDataUtil.java
@@ -60,7 +60,7 @@ public InvestigationTransformed transformInvestigationData(Investigation investi
         transformCaseCountInfo(investigation.getInvestigationCaseCnt(), investigationTransformed);
         transformOrganizationParticipations(investigation.getOrganizationParticipations(), investigationTransformed);
         transformActIds(investigation.getActIds(), investigationTransformed);
-        transformObservationIds(investigation.getObservationNotificationIds(), investigationTransformed);
+        transformObservationIds(investigation.getInvestigationObservationIds(), investigationTransformed);
         transformInvestigationConfirmationMethod(investigation.getInvestigationConfirmationMethod(), investigationTransformed);
         processInvestigationPageCaseAnswer(investigation.getInvestigationCaseAnswer(), investigationTransformed);
 
@@ -204,38 +204,49 @@ private void transformActIds(String actIds, InvestigationTransformed investigati
         }
     }
 
-    private void transformObservationIds(String observationNotificationIds, InvestigationTransformed investigationTransformed) {
+    private void transformObservationIds(String investigationObservationIds, InvestigationTransformed investigationTransformed) {
         try {
-            JsonNode investigationObservationIdsJsonArray = parseJsonArray(observationNotificationIds);
+            JsonNode investigationObservationIdsJsonArray = parseJsonArray(investigationObservationIds);
             InvestigationObservation investigationObservation = new InvestigationObservation();
-            List<Long> observationIds = new ArrayList<>();
+            InvestigationObservationKey investigationObservationKey = new InvestigationObservationKey();
+
+            investigationKey.setPublicHealthCaseUid(investigationTransformed.getPublicHealthCaseUid());
+            String jsonKey = jsonGenerator.generateStringJson(investigationKey);
+            kafkaTemplate.send(investigationObservationOutputTopicName, jsonKey, null);
 
             for(JsonNode node : investigationObservationIdsJsonArray) {
                 String sourceClassCode = node.path("source_class_cd").asText();
                 String actTypeCode = node.path("act_type_cd").asText();
                 Long sourceActId = node.get("source_act_uid").asLong();
                 Long publicHealthCaseUid = node.get("public_health_case_uid").asLong();
-                investigationKey.setPublicHealthCaseUid(publicHealthCaseUid);
+                String rootTypeCd = node.path("act_type_cd").asText();
 
                 if(sourceClassCode.equals("OBS") && actTypeCode.equals("PHCInvForm")) {
                     investigationTransformed.setPhcInvFormId(sourceActId);
                 }
 
-                if(sourceClassCode.equals("OBS") && actTypeCode.equals("LabReport")) {
-                    investigationObservation.setPublicHealthCaseUid(publicHealthCaseUid);
-                    observationIds.add(sourceActId);
-                }
+                investigationObservationKey.setPublicHealthCaseUid(publicHealthCaseUid);
+                investigationObservationKey.setObservationId(sourceActId);
+                investigationObservationKey.setBranchId(null);
 
-                if(sourceClassCode.equals("OBS") && actTypeCode.equals("MorbReport")) {
-                    investigationObservation.setPublicHealthCaseUid(publicHealthCaseUid);
-                    observationIds.add(sourceActId);
-                }
-            }
+                investigationObservation.setPublicHealthCaseUid(publicHealthCaseUid);
+                investigationObservation.setObservationId(sourceActId);
+                investigationObservation.setRootTypeCd(rootTypeCd);
+                investigationObservation.setBranchId(null);
+                investigationObservation.setBranchTypeCd(null);
+
+                Optional.ofNullable(node.get("branch_uid")).filter(n -> !n.isNull())
+                        .ifPresent(n -> {
+                            investigationObservationKey.setBranchId(n.asLong());
+                            investigationObservation.setBranchId(n.asLong());
+
+                        });
+                Optional.ofNullable(node.get("branch_type_cd")).filter(n -> !n.isNull())
+                        .ifPresent(n -> investigationObservation.setBranchTypeCd(n.asText()));
 
-            for(Long id : observationIds) {
-                investigationObservation.setObservationId(id);
+                jsonKey = jsonGenerator.generateStringJson(investigationObservationKey);
                 String jsonValue = jsonGenerator.generateStringJson(investigationObservation);
-                kafkaTemplate.send(investigationObservationOutputTopicName, jsonValue, jsonValue);
+                kafkaTemplate.send(investigationObservationOutputTopicName, jsonKey, jsonValue);
             }
         } catch (IllegalArgumentException ex) {
             logger.info(ex.getMessage(), "InvestigationObservationIds");

diff --git a/...src/test/java/gov/cdc/etldatapipeline/investigation/InvestigationDataProcessingTests.java b/...src/test/java/gov/cdc/etldatapipeline/investigation/InvestigationDataProcessingTests.java
@@ -114,7 +114,7 @@ void testTransformInvestigationError(){
         investigation.setPersonParticipations(invalidJSON);
         investigation.setOrganizationParticipations(invalidJSON);
         investigation.setActIds(invalidJSON);
-        investigation.setObservationNotificationIds(invalidJSON);
+        investigation.setInvestigationObservationIds(invalidJSON);
         investigation.setInvestigationConfirmationMethod(invalidJSON);
         investigation.setInvestigationCaseAnswer(invalidJSON);
         investigation.setInvestigationCaseCnt(invalidJSON);
@@ -126,23 +126,26 @@ void testTransformInvestigationError(){
     }
 
     @Test
-    void testObservationNotificationIds() throws JsonProcessingException {
+    void testInvestigationObservationIds() throws JsonProcessingException {
         Investigation investigation = new Investigation();
 
         investigation.setPublicHealthCaseUid(investigationUid);
-        investigation.setObservationNotificationIds(readFileData(FILE_PREFIX + "ObservationNotificationIds.json"));
+        investigation.setInvestigationObservationIds(readFileData(FILE_PREFIX + "InvestigationObservationIds.json"));
         transformer.investigationObservationOutputTopicName = OBSERVATION_TOPIC;
 
         InvestigationObservation observation = new InvestigationObservation();
         observation.setPublicHealthCaseUid(investigationUid);
-        observation.setObservationId(263748596L);
+        observation.setObservationId(10344738L);
+        observation.setRootTypeCd("LabReport");
+        observation.setBranchId(10344740L);
+        observation.setBranchTypeCd("COMP");
 
         transformer.transformInvestigationData(investigation);
-        verify(kafkaTemplate, times(4)).send(topicCaptor.capture(), keyCaptor.capture(), messageCaptor.capture());
+        verify(kafkaTemplate, times(9)).send(topicCaptor.capture(), keyCaptor.capture(), messageCaptor.capture());
         assertEquals(OBSERVATION_TOPIC, topicCaptor.getAllValues().getFirst());
 
         var actualObservation = objectMapper.readValue(
-                objectMapper.readTree(messageCaptor.getAllValues().getFirst()).path("payload").toString(), InvestigationObservation.class);
+                objectMapper.readTree(messageCaptor.getAllValues().get(1)).path("payload").toString(), InvestigationObservation.class);
 
         assertEquals(observation, actualObservation);
     }

diff --git a/...src/test/java/gov/cdc/etldatapipeline/investigation/service/InvestigationServiceTest.java b/...src/test/java/gov/cdc/etldatapipeline/investigation/service/InvestigationServiceTest.java
@@ -149,11 +149,11 @@ private void validateInvestigationData(String payload, Investigation investigati
         investigationKey.setPublicHealthCaseUid(investigation.getPublicHealthCaseUid());
         final InvestigationReporting reportingModel = constructInvestigationReporting(investigation.getPublicHealthCaseUid());
 
-        verify(kafkaTemplate, times(12)).send(topicCaptor.capture(), keyCaptor.capture(), messageCaptor.capture());
+        verify(kafkaTemplate, times(17)).send(topicCaptor.capture(), keyCaptor.capture(), messageCaptor.capture());
 
-        String actualTopic = topicCaptor.getAllValues().get(9);
-        String actualKey = keyCaptor.getAllValues().get(9);
-        String actualValue = messageCaptor.getAllValues().get(9);
+        String actualTopic = topicCaptor.getAllValues().get(14);
+        String actualKey = keyCaptor.getAllValues().get(14);
+        String actualValue = messageCaptor.getAllValues().get(14);
 
         var actualReporting = objectMapper.readValue(
                 objectMapper.readTree(actualValue).path("payload").toString(), InvestigationReporting.class);
@@ -181,10 +181,11 @@ private Investigation constructInvestigation(Long investigationUid) {
         investigation.setRecordStatusCd("ACTIVE");
         investigation.setMmwrWeek("22");
         investigation.setMmwrYear("2024");
+        investigation.setInvestigationFormCd("INV_FORM_MEA");
 
         investigation.setActIds(readFileData(FILE_PATH_PREFIX + "ActIds.json"));
         investigation.setInvestigationConfirmationMethod(readFileData(FILE_PATH_PREFIX + "ConfirmationMethod.json"));
-        investigation.setObservationNotificationIds(readFileData(FILE_PATH_PREFIX + "ObservationNotificationIds.json"));
+        investigation.setInvestigationObservationIds(readFileData(FILE_PATH_PREFIX + "InvestigationObservationIds.json"));
         investigation.setOrganizationParticipations(readFileData(FILE_PATH_PREFIX + "OrganizationParticipations.json"));
         investigation.setPersonParticipations(readFileData(FILE_PATH_PREFIX + "PersonParticipations.json"));
         investigation.setInvestigationCaseAnswer(readFileData(FILE_PATH_PREFIX + "InvestigationCaseAnswers.json"));
@@ -210,6 +211,7 @@ private InvestigationReporting constructInvestigationReporting(Long investigatio
         reporting.setRecordStatusCd("ACTIVE");
         reporting.setMmwrWeek("22");
         reporting.setMmwrYear("2024");
+        reporting.setInvestigationFormCd("INV_FORM_MEA");
 
         reporting.setInvestigatorId(32143250L);         // PersonParticipations.json, entity_id for type_cd=InvestgrOfPHC
         reporting.setPhysicianId(14253651L);            // PersonParticipations.json, entity_id for type_cd=PhysicianOfPHC
@@ -218,7 +220,7 @@ private InvestigationReporting constructInvestigationReporting(Long investigatio
         reporting.setInvStateCaseId("12-345-STA");      // ActIds.json, root_extension_txt for type_cd=STATE
         reporting.setCityCountyCaseNbr("12-345-CTY");   // ActIds.json, root_extension_txt for type_cd=CITY
         reporting.setLegacyCaseId("12-345-LGY");        // ActIds.json, root_extension_txt for type_cd=LEGACY
-        reporting.setPhcInvFormId(263748598L);          // ObservationNotificationIds.json, source_act_uid for act_type_cd=PHCInvForm
+        reporting.setPhcInvFormId(10638298L);          // InvestigationObservationIds.json, source_act_uid for act_type_cd=PHCInvForm
         reporting.setRdbTableNameList("D_INV_CLINICAL,D_INV_ADMINISTRATIVE"); // InvestigationCaseAnswers.json, rdb_table_nm
         reporting.setInvestigationCount(1L);
         reporting.setCaseCount(1L);

diff --git a/investigation-service/src/test/resources/rawDataFiles/InvestigationObservationIds.json b/investigation-service/src/test/resources/rawDataFiles/InvestigationObservationIds.json
@@ -0,0 +1,116 @@
+[
+  {
+    "source_act_uid": 10344738,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "OBS",
+    "target_class_cd": "CASE",
+    "act_type_cd": "LabReport",
+    "status_cd": "A",
+    "act_add_time": "2024-08-20T23:17:52.277",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": 10055282,
+    "last_chg_user_name": "user, super",
+    "act_last_chg_time": "2024-08-20T23:17:52.277",
+    "root_uid": 10344738,
+    "branch_uid": 10344740,
+    "root_source_class_cd": "OBS",
+    "branch_source_class_cd": "OBS",
+    "branch_type_cd": "COMP"
+  },
+  {
+    "source_act_uid": 10344752,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "NOTF",
+    "target_class_cd": "CASE",
+    "act_type_cd": "Notification",
+    "status_cd": "A",
+    "act_add_time": "2024-08-20T23:18:08.160",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": null,
+    "last_chg_user_name": null,
+    "act_last_chg_time": "2024-08-20T23:18:08.177",
+    "root_uid": null,
+    "branch_uid": null,
+    "root_source_class_cd": null,
+    "branch_source_class_cd": null,
+    "branch_type_cd": null
+  },
+  {
+    "source_act_uid": 10344747,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "OBS",
+    "target_class_cd": "CASE",
+    "act_type_cd": "PHCInvForm",
+    "status_cd": "A",
+    "act_add_time": "2024-08-20T23:17:52.260",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": null,
+    "last_chg_user_name": null,
+    "act_last_chg_time": "2024-08-20T23:17:52.260",
+    "root_uid": 10344747,
+    "branch_uid": 10605257,
+    "root_source_class_cd": "OBS",
+    "branch_source_class_cd": "OBS",
+    "branch_type_cd": "InvFrmQ"
+  },
+  {
+    "source_act_uid": 10344738,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "OBS",
+    "target_class_cd": "CASE",
+    "act_type_cd": "LabReport",
+    "status_cd": "A",
+    "act_add_time": "2024-08-20T23:17:52.277",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": 10055282,
+    "last_chg_user_name": "user, super",
+    "act_last_chg_time": "2024-08-20T23:17:52.277",
+    "root_uid": 10344738,
+    "branch_uid": 10344739,
+    "root_source_class_cd": "OBS",
+    "branch_source_class_cd": "OBS",
+    "branch_type_cd": "COMP"
+  },
+  {
+    "source_act_uid": 10604655,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "OBS",
+    "target_class_cd": "CASE",
+    "act_type_cd": "MorbReport",
+    "status_cd": "A",
+    "act_add_time": "2024-10-30T20:58:03.977",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": 10055282,
+    "last_chg_user_name": "user, super",
+    "act_last_chg_time": "2024-10-30T20:58:03.977",
+    "root_uid": 10604655,
+    "branch_uid": 10604656,
+    "root_source_class_cd": "OBS",
+    "branch_source_class_cd": "OBS",
+    "branch_type_cd": "LabReport"
+  },
+  {
+    "source_act_uid": 10638298,
+    "public_health_case_uid": 234567890,
+    "source_class_cd": "OBS",
+    "target_class_cd": "CASE",
+    "act_type_cd": "PHCInvForm",
+    "status_cd": "A",
+    "act_add_time": "2024-11-18T19:49:34.137",
+    "act_add_user_id": null,
+    "add_user_name": null,
+    "act_last_chg_user_id": null,
+    "last_chg_user_name": null,
+    "act_last_chg_time": "2024-11-18T19:49:34.137",
+    "root_uid": null,
+    "branch_uid": null,
+    "root_source_class_cd": null,
+    "branch_source_class_cd": null,
+    "branch_type_cd": null
+  }
+]
diff --git a/investigation-service/src/test/resources/rawDataFiles/ObservationNotificationIds.json b/investigation-service/src/test/resources/rawDataFiles/ObservationNotificationIds.json