feat: add topic fhir.pseudonymized.*

README.md

@@ -19,4 +19,5 @@ Das `*` wird als Wildcard Match für ein beliebiges Jahr gemäß der Topics
 in [onco-analytics-on-fhir](https://github.com/bzkf/onco-analytics-on-fhir) verwendet.
 
 * `onkostar.MELDUNG_EXPORT.*`: Alle eingehenden Meldungen aus Onkostar
-* `fhir.obds.Condition.*`: Alle Conditions, die aus den oBDS-Meldungen erzeugt wurden.
+* `fhir.obds.Condition.*`: Alle Conditions, die aus den oBDS-Meldungen erzeugt wurden.
+* `fhir.pseudonymized.*`: Alle pseudonymisierten Conditions.

src/main/kotlin/dev/pcvolkmer/oncoanalytics/monitor/OncoAnalyticsMonitorApplication.kt

@@ -27,6 +27,11 @@ class OncoAnalyticsMonitorApplication {
         return ConditionInMemoryRepository()
     }
 
+    @Bean
+    fun fhirPseudonymizedConditionRepository(): ConditionInMemoryRepository {
+        return ConditionInMemoryRepository()
+    }
+
 }
 
 fun main(args: Array<String>) {

src/main/kotlin/dev/pcvolkmer/oncoanalytics/monitor/topiclisteners/FhirPseudonymizedTopicMonitor.kt

@@ -0,0 +1,55 @@
+package dev.pcvolkmer.oncoanalytics.monitor.topiclisteners
+
+import ca.uhn.fhir.context.FhirContext
+import dev.pcvolkmer.oncoanalytics.monitor.StatisticsSink
+import dev.pcvolkmer.oncoanalytics.monitor.conditions.Condition
+import dev.pcvolkmer.oncoanalytics.monitor.conditions.ConditionId
+import dev.pcvolkmer.oncoanalytics.monitor.conditions.ConditionInMemoryRepository
+import dev.pcvolkmer.oncoanalytics.monitor.fetchStatistics
+import org.hl7.fhir.r4.model.Bundle
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.kafka.support.KafkaHeaders
+import org.springframework.messaging.handler.annotation.Header
+import org.springframework.messaging.handler.annotation.Payload
+import org.springframework.stereotype.Component
+
+@Component
+class FhirPseudonymizedTopicMonitor(
+    @Qualifier("fhirPseudonymizedConditionRepository")
+    private val conditionRepository: ConditionInMemoryRepository,
+    statisticsEventProducer: StatisticsSink,
+) : TopicMonitor(statisticsEventProducer) {
+
+    @KafkaListener(topicPattern = "fhir.pseudonymized.*")
+    override fun handleTopicRecord(
+        @Header(KafkaHeaders.RECEIVED_TOPIC) topic: String,
+        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) timestamp: Long,
+        @Header(KafkaHeaders.RECEIVED_KEY) key: String,
+        @Payload payload: String,
+    ) {
+        try {
+            val ctx = FhirContext.forR4()
+            val parser = ctx.newJsonParser()
+
+            val bundle = parser.parseResource(Bundle::class.java, payload)
+            val firstEntry = bundle.entry.firstOrNull() ?: return
+
+            if (firstEntry.resource.fhirType() == "Condition") {
+                val condition = firstEntry.resource as org.hl7.fhir.r4.model.Condition
+                val updated = conditionRepository.save(
+                    Condition(
+                        ConditionId(condition.id),
+                        condition.code.coding.first { "http://fhir.de/CodeSystem/bfarm/icd-10-gm" == it.system }.code
+                    )
+                )
+
+                if (updated) {
+                    sendUpdatedStatistics(fetchStatistics("fhirpseudonymized", conditionRepository))
+                }
+            }
+        } catch (e: Exception) {
+            // Ignore
+        }
+    }
+}

src/main/kotlin/dev/pcvolkmer/oncoanalytics/monitor/web/StatisticsController.kt

@@ -15,6 +15,8 @@ class StatisticsController(
     private val obdsXmlConditionRepository: ConditionInMemoryRepository,
     @Qualifier("obdsFhirConditionRepository")
     private val obdsFhirConditionRepository: ConditionInMemoryRepository,
+    @Qualifier("fhirPseudonymizedConditionRepository")
+    private val fhirPseudonymizedConditionRepository: ConditionInMemoryRepository,
 ) {
 
     @GetMapping(path = ["obdsxml"])
@@ -27,4 +29,9 @@ class StatisticsController(
         return fetchStatistics("obdfhir", obdsFhirConditionRepository)
     }
 
+    @GetMapping(path = ["fhirpseudonymized"])
+    fun fhirpseudonymizedStatistics(): Statistics {
+        return fetchStatistics("fhirpseudonymized", fhirPseudonymizedConditionRepository)
+    }
+
 }

src/main/resources/static/css/style.css

@@ -1,11 +1,26 @@
+* {
+    box-sizing: border-box;
+    color: #333;
+    font-family: sans-serif;
+}
+
 body {
-    margin: 0;
+    margin: 0 auto;
     background-color: #f5f5f5;
+
+    display: grid;
+    width: fit-content;
 }
 
 header {
-    margin: 1em 0 6em 0;
     text-align: left;
+    margin: 1em 0;
+}
+
+main {
+    width: fit-content;
+    margin: 1em auto;
+    text-align: center;
 }
 
 .statistics table {
@@ -39,18 +54,15 @@ tr > td:last-of-type {
     text-align: right;
 }
 
-.content {
-    width: fit-content;
-    height: 100vh;
-    margin: 0 auto;
-    padding: 1em;
-    text-align: center;
-}
-
 .step {
     display: inline-block;
     width: fit-content;
     vertical-align: top;
+    margin-top: 1em;
+}
+
+.step:has(.item .statistics) {
+    margin-top: 1px;
 }
 
 .step > .item {
@@ -65,7 +77,7 @@ tr > td:last-of-type {
 }
 
 .step .item:has(.statistics) {
-    width: 14em;
+    width: 11em;
 
     background: white;
     border-radius: 2em 2em .4em .4em;
@@ -76,7 +88,7 @@ tr > td:last-of-type {
 .step:before, .step:after {
     content: "";
     margin: 2em 0;
-    width: 1em;
+    width: 8px;
     height: 2px;
     background: black;
     display: inline-block;

src/main/resources/templates/index.html

@@ -7,14 +7,14 @@
 </head>
 <body>
 
-<main class="content">
+<header>
+    <h1>onco-analytics-monitor</h1>
+    <p>
+        Überwachung der einzelnen Kafka Topics und enthaltener Conditions - aufgeteilt nach ICD10-Gruppe.
+    </p>
+</header>
 
-    <header>
-        <h1>onco-analytics-monitor</h1>
-        <p>
-            Überwachung der einzelnen Kafka Topics und enthaltener Conditions - aufgeteilt nach ICD10-Gruppe.
-        </p>
-    </header>
+<main>
 
     <div class="step">
         <div class="item">
@@ -29,7 +29,7 @@
     </div><div class="step">
         <div class="item">
             <img class="logo" th:src="@{/images/topic.png}" alt="topic"/>
-            <div class="description">Kafka Topic oBDS XML</div>
+            <div class="description">Kafka Topic<br/>oBDS XML</div>
             <div class="statistics">
                 <table id="obdsxml"></table>
             </div>
@@ -42,12 +42,25 @@
     </div><div class="step">
         <div class="item">
             <img class="logo" th:src="@{/images/topic.png}" alt="topic"/>
-            <div class="description">Kafka Topic oBDS FHIR</div>
+            <div class="description">Kafka Topic<br/>oBDS FHIR</div>
             <div class="statistics">
                 <table id="obdsfhir"></table>
             </div>
         </div>
+    </div><div class="step">
+    <div class="item">
+        <img class="logo" th:src="@{/images/job.png}" alt="job"/>
+        <div class="description">fhir-pseudonymizer</div>
     </div>
+    </div><div class="step">
+    <div class="item">
+        <img class="logo" th:src="@{/images/topic.png}" alt="topic"/>
+        <div class="description">Kafka Topic<br/>FHIR pseudonymized</div>
+        <div class="statistics">
+            <table id="fhirpseudonymized"></table>
+        </div>
+    </div>
+</div>
 
 </main>
 
@@ -66,6 +79,7 @@
 
     fetch('/statistics/obdsxml').then(res => res.json()).then(data => updateData(data, 'obdsxml'));
     fetch('/statistics/obdsfhir').then(res => res.json()).then(data => updateData(data, 'obdsfhir'));
+    fetch('/statistics/fhirpseudonymized').then(res => res.json()).then(data => updateData(data, 'fhirpseudonymized'));
 
     window.addEventListener('load', () => {
         const evtSource = new EventSource('/events');
@@ -77,6 +91,10 @@
         evtSource.addEventListener('obdsfhir', (event) => {
             updateData(JSON.parse(event.data), 'obdsfhir')
         });
+
+        evtSource.addEventListener('fhirpseudonymized', (event) => {
+            updateData(JSON.parse(event.data), 'fhirpseudonymized')
+        });
     });
 </script>