Merge pull request #68

* Threadsafe use of Parser * improved Structure and Readme
medizininformatik-initiative · Sep 25, 2024 · 5748e9a · 5748e9a
1 parent 3c26338
commit 5748e9a
Show file tree

Hide file tree

Showing 25 changed files with 201 additions and 387 deletions.
diff --git a/README.md b/README.md
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -1,16 +1,16 @@
 services:
   torch-data-store:
-    image: "samply/blaze:0.28"
+    image: "samply/blaze:0.30"
     environment:
       BASE_URL: "http://torch-data-store:8080"
-      JAVA_TOOL_OPTIONS: "-Xmx1g"
+      JAVA_TOOL_OPTIONS: "-Xmx2g"
       LOG_LEVEL: "info"
     ports:
       - "8082:8080"
     volumes:
       - "data-store-data:/app/output"
   torch-flare:
-    image: ghcr.io/medizininformatik-initiative/flare:pr-179
+    image: ghcr.io/medizininformatik-initiative/flare:2.4.0-alpha.1
     ports:
       - ${FEASIBILITY_FLARE_PORT:-127.0.0.1:8084}:8080
     environment:
@@ -30,7 +30,7 @@ services:
     restart: unless-stopped
     image: nginxinc/nginx-unprivileged:1.25.5-alpine
     ports:
-      - ${PORT_TORCH_NGINX:-127.0.0.1:80}:8080
+      - ${PORT_TORCH_NGINX:-127.0.0.1:8080}:8080
     volumes:
       - ./nginx.conf.template:/etc/nginx/nginx.conf.template
       - ./start-nginx.sh:/start-nginx.sh
@@ -51,7 +51,7 @@ services:
       TORCH_FLARE_URL: http://torch-flare:8080
       TORCH_RESULTS_DIR: /app/output
       TORCH_RESULTS_PERSISTENCE: PT12H30M5S
-      LOG_LEVEL: info
+      LOG_LEVEL: debug
       NGINX_SERVERNAME: localhost
       NGINX_FILELOCATION: http://localhost:80
       TORCH_BATCHSIZE: 100
@@ -65,5 +65,3 @@ services:
 volumes:
   data-store-data:
   torch-data-store:
-
-
diff --git a/output_crtdl.json b/output_crtdl.json
@@ -2,7 +2,7 @@
   "version": "http://json-schema.org/to-be-done/schema#",
   "display": "",
   "cohortDefinition": {
-    "version": "http://to_be_decided.com/draft-1/schema#",
+    "version": "https://medizininformatik-initiative.de/fdpg/ClinicalCohortDefinitionLanguage/v1/schema",
     "display": "",
     "inclusionCriteria": [
       [

diff --git a/pom.xml b/pom.xml
@@ -114,14 +114,12 @@
             <scope>test</scope>
         </dependency>
 
-        <!-- https://mvnrepository.com/artifact/com.github.docker-java/docker-java -->
         <dependency>
             <groupId>com.github.docker-java</groupId>
             <artifactId>docker-java</artifactId>
             <version>3.3.6</version>
         </dependency>
 
-        <!-- https://mvnrepository.com/artifact/io.projectreactor/reactor-core-->
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-core</artifactId>
@@ -131,7 +129,7 @@
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>
-            <version>1.19.7</version>
+            <version>1.20.1</version>
             <scope>test</scope>
         </dependency>
 
@@ -181,4 +179,4 @@
         </plugins>
     </build>
 
-</project>
+</project>
diff --git a/scripts/create-parameters.sh b/scripts/create-parameters.sh
@@ -0,0 +1,26 @@
+#!/bin/bash -e
+
+#
+# Creates a FHIR Parameters resource by reading the CRTDL from a file specified
+# in the first argument of this script and outputs it on STDOUT. Can be used to
+# generate the input of the $extract-data operation.
+#
+# Usage: create-parameters.sh crtdl.json | curl -s 'http://localhost:8080/fhir/$extract-data' -H "Content-Type: application/fhir+json" -d @-
+#
+
+CRTDL_FILE="$1"
+
+parameters() {
+cat <<END
+{
+  "resourceType": "Parameters",
+  "parameter": [
+    {
+      "name": "crtdl"
+    }
+  ]
+}
+END
+}
+
+parameters | jq --arg content "$(cat $CRTDL_FILE)" -cM '.parameter[0].valueBase64Binary = ($content | @base64)'
diff --git a/src/main/java/de/medizininformatikinitiative/torch/CdsStructureDefinitionHandler.java b/src/main/java/de/medizininformatikinitiative/torch/CdsStructureDefinitionHandler.java
@@ -5,25 +5,17 @@
 import org.hl7.fhir.r4.model.*;
 import org.springframework.stereotype.Component;
 import ca.uhn.fhir.context.FhirContext;
-import ca.uhn.fhir.parser.IParser;
 
 import java.io.File;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 
 @Component
-
-
 public class CdsStructureDefinitionHandler {
 
-
-    private HashMap<String, StructureDefinition> definitionsMap = new HashMap<>();
-
-
     public FhirContext ctx;
-
-    public IParser jsonParser;
+    private HashMap<String, StructureDefinition> definitionsMap = new HashMap<>();
 
     public CdsStructureDefinitionHandler(FhirContext ctx, String fileDirectory) {
         try {
@@ -32,16 +24,12 @@ public CdsStructureDefinitionHandler(FhirContext ctx, String fileDirectory) {
             throw new RuntimeException(e);
         }
         this.ctx = ctx;
-        this.jsonParser = ctx.newJsonParser();
     }
 
     public CdsStructureDefinitionHandler(FhirContext ctx) {
-
         this.ctx = ctx;
-        this.jsonParser = ctx.newJsonParser();
     }
 
-
     /**
      * Reads a StructureDefinition from a file and stores it in the definitionsMap
      *
@@ -53,13 +41,6 @@ public void readStructureDefinition(String filePath) throws IOException {
         definitionsMap.put(structureDefinition.getUrl(), structureDefinition);
     }
 
-    /**
-     * Returns HAPI JSON Parser
-     */
-    public IParser getJsonParser() {
-        return jsonParser;
-    }
-
     /**
      * Returns the StructureDefinition with the given URL.
      * Handles versioned URLs by splitting on the '|' character.

diff --git a/src/main/java/de/medizininformatikinitiative/torch/DataStore.java b/src/main/java/de/medizininformatikinitiative/torch/DataStore.java
@@ -1,7 +1,6 @@
 package de.medizininformatikinitiative.torch;
 
 import ca.uhn.fhir.context.FhirContext;
-import ca.uhn.fhir.parser.IParser;
 import org.hl7.fhir.r4.model.Bundle;
 import org.hl7.fhir.r4.model.Resource;
 import org.slf4j.Logger;
@@ -16,22 +15,19 @@
 import java.net.URI;
 import java.util.Optional;
 
-
-
 @Component
 public class DataStore {
     private static final Logger logger = LoggerFactory.getLogger(DataStore.class);
 
     private final WebClient client;
-    private final IParser parser;
+    private final FhirContext fhirContext;
 
     @Autowired
-    public DataStore( @Qualifier("fhirClient") WebClient client, FhirContext context) {
+    public DataStore(@Qualifier("fhirClient") WebClient client, FhirContext fhirContext) {
         this.client = client;
-        this.parser = context.newJsonParser();
+        this.fhirContext = fhirContext;
     }
 
-
     /**
      * Executes {@code FHIRSearchQuery} and returns all resources found with that query.
      *
@@ -49,7 +45,7 @@ public Flux<Resource> getResources(String resourceType,String parameters) {
                 .retrieve()
                 .bodyToMono(String.class)
                 .doOnNext(response -> logger.debug("getResources Response: {}", response))
-                .flatMap(response -> Mono.just(parser.parseResource(Bundle.class, response)))
+                .flatMap(response -> Mono.just(fhirContext.newJsonParser().parseResource(Bundle.class, response)))
                 .expand(bundle -> Optional.ofNullable(bundle.getLink("next"))
                         .map(link -> fetchPage(client, link.getUrl()))
                         .orElse(Mono.empty()))
@@ -62,7 +58,7 @@ private Mono<Bundle> fetchPage(WebClient client, String url) {
                 .uri(URI.create(url))
                 .retrieve()
                 .bodyToMono(String.class)
-                .map(response -> parser.parseResource(Bundle.class, response));
+                .map(response -> fhirContext.newJsonParser().parseResource(Bundle.class, response));
     }
 
 }
diff --git a/src/main/java/de/medizininformatikinitiative/torch/ResourceTransformer.java b/src/main/java/de/medizininformatikinitiative/torch/ResourceTransformer.java
@@ -27,33 +27,19 @@
 public class ResourceTransformer {
 
     private static final Logger logger = LoggerFactory.getLogger(ResourceTransformer.class);
-    private final DataStore dataStore;
 
+    private final DataStore dataStore;
     private final ElementCopier copier;
     private final Redaction redaction;
-
-    private final ResultFileManager fileManager;
-
-    @Autowired(required = false)
-    int batchSize = 1;
-
     private final FhirSearchBuilder searchBuilder = new FhirSearchBuilder();
 
-
-    private final BundleCreator creator=new BundleCreator();
-
-    public ConcurrentMap<String, Set<String>> fulfilledGroupsPerPatient;
-
     @Autowired
-    public ResourceTransformer(DataStore dataStore, CdsStructureDefinitionHandler cds, ResultFileManager fileManager) {
+    public ResourceTransformer(DataStore dataStore, CdsStructureDefinitionHandler cds) {
         this.dataStore = dataStore;
         this.copier = new ElementCopier(cds);
         this.redaction = new Redaction(cds);
-        this.fulfilledGroupsPerPatient = new ConcurrentHashMap<>();
-        this.fileManager=fileManager;
     }
 
-
     public Flux<Resource> transformResources(String parameters, AttributeGroup group) {
         String resourceType = group.getResourceType();
 
@@ -66,8 +52,6 @@ public Flux<Resource> transformResources(String parameters, AttributeGroup group
                     return Flux.empty();  // Return an empty Flux to continue processing without crashing the pipeline
                 });
 
-
-
         return resources.map(resource -> {
             try {
 
@@ -86,7 +70,6 @@ public Flux<Resource> transformResources(String parameters, AttributeGroup group
 
     }
 
-
     public Resource transform(DomainResource resourcesrc, AttributeGroup group) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException, MustHaveViolatedException {
         Class<? extends DomainResource> resourceClass = resourcesrc.getClass().asSubclass(DomainResource.class);
         DomainResource tgt = resourceClass.getDeclaredConstructor().newInstance();
@@ -117,9 +100,6 @@ public Resource transform(DomainResource resourcesrc, AttributeGroup group) thro
         return tgt;
     }
 
-
-
-
     public Mono<Map<String, Collection<Resource>>> collectResourcesByPatientReference(Crtdl crtdl, List<String> patients)  {
         //logger.debug("Starting collectResourcesByPatientReference");
         //logger.debug("Patients Received: {}", patients);
@@ -170,17 +150,5 @@ public Mono<Map<String, Collection<Resource>>> collectResourcesByPatientReferenc
                 })
                 .doOnSuccess(result -> logger.debug("Successfully collected resources {}", result.entrySet()))
                 .doOnError(error -> logger.error("Error collecting resources: {}", error.getMessage()));
-
     }
-
-
-
-
-
-
-
-
-
 }
-
-
diff --git a/src/main/java/de/medizininformatikinitiative/torch/Torch.java b/src/main/java/de/medizininformatikinitiative/torch/Torch.java
@@ -1,10 +1,8 @@
 package de.medizininformatikinitiative.torch;
 
-
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.annotation.ComponentScan;
-import reactor.core.publisher.Hooks;
 
 @SpringBootApplication
 @ComponentScan(basePackages = {
@@ -13,12 +11,7 @@
 })
 public class Torch {
 
-
-        public static void main(String[] args) {
-            SpringApplication.run(Torch.class, args);
-        }
-
-
-
-
+    public static void main(String[] args) {
+        SpringApplication.run(Torch.class, args);
+    }
 }
Name	Default	Description
SERVER_PORT	8080	The Port of the server to use
TORCH_FHIR_URL	http://localhost:8082/fhir	The base URL of the FHIR server to use.
TORCH_FLARE_URL	http://localhost:8084	The base URL of the FLARE server to use.
TORCH_PROFILE_DIR	src/test/resources/StructureDefinitions	The directory for profile definitions.
TORCH_RESULTS_DIR	output/	The directory for storing results.
TORCH_RESULTS_PERSISTENCE	PT12H30M5S	Time Block for result persistence in ISO 8601 format.
TORCH_BATCHSIZE	100	The batch size used for processing data.
NGINX_SERVERNAME	localhost:8080	The server name configuration for NGINX.
LOG_LEVEL_org_springframework_web_reactive_function_client	info	Log level for Spring Web Reactive client functions.
LOG_LEVEL_reactor_netty	info	Log level for Netty reactor-based networking.
LOG_LEVEL_reactor	info	Log level for Reactor framework.
LOG_LEVEL_de_medizininformatikinitiative_torch_util	${LOG_LEVEL:info}	Log level for torch utility.
LOG_LEVEL_de_medizininformatikinitiative_torch	${LOG_LEVEL:info}	Log level for torch core functionality.
LOG_LEVEL_de_medizininformatikinitiative_torch_rest	${LOG_LEVEL:info}	Log level for torch REST services.
LOG_LEVEL_org_springframework	${LOG_LEVEL:info}	Log level for Spring Framework.