Adds multi-threaded wado downloads

Author: sjoerdk

HISTORY.md

@@ -1,5 +1,11 @@
 # History
 
+## v0.4.0 (2021-04-05)
+
+* Added multi-threaded wado downloads 
+* Slight changes to dicomtrolley methods
+
+
 ## v0.3.0 (2021-04-01)
 
 * Added examples

dicomtrolley/__init__.py

@@ -1,2 +1,2 @@
 """Retrieve medical images via DICOM-QR and DICOMweb"""
-__version__ = "v0.3.0"
+__version__ = "v0.4.0"

dicomtrolley/core.py

@@ -1,8 +1,17 @@
-"""Combines WADO and MINT to make make getting DICOM studies easy"""
+"""Combines WADO and MINT to make make getting DICOM studies easy
+
+Notes
+-----
+Design choices:
+
+WADO and MINT modules should remain un-entangled so are not allowed to use
+each other's classes. Core has knowledge of both an can convert classes between the
+two if needed
+
+"""
 
-from itertools import chain
 from pathlib import Path
-from typing import List, Sequence
+from typing import List, Sequence, Union
 
 from dicomtrolley.mint import (
     Mint,
@@ -11,17 +20,11 @@
     MintStudy,
 )
 from dicomtrolley.query import Query, QueryLevels
-from dicomtrolley.wado import Wado
+from dicomtrolley.wado import InstanceReference, Wado
 
 
 class Trolley:
-    """Combines WADO and MINT to make make getting DICOM studies easy
-
-    Offers three different types of functions:
-    * Find - Search for things on server, but do not download yet
-    * Download - Quick methods that download some data to disk
-    * Get - Return pydicom datasets directly without writing to disk
-    """
+    """Combines WADO and MINT to make make getting DICOM studies easy"""
 
     def __init__(self, wado: Wado, mint: Mint):
         self.wado = wado
@@ -61,7 +64,7 @@ def download_study(self, study_instance_uid, output_dir):
             storage.save(dataset)
 
     def fetch_all_datasets(self, mint_objects: List[MintObject]):
-        """Get full DICOM dataset for each instance in study. Calls mint and wado
+        """Get full DICOM dataset for each instance in study
 
         Parameters
         ----------
@@ -73,36 +76,66 @@ def fetch_all_datasets(self, mint_objects: List[MintObject]):
         Generator[Dataset, None, None]
             The downloaded dataset and the context that was used to download it
         """
-
-        for instance in self.extract_instances(mint_objects):
-            yield self.get_dataset(instance)
+        for ds in self.wado.datasets(self.extract_instances(mint_objects)):
+            yield ds
 
     @staticmethod
-    def extract_instances(mint_objects: Sequence[MintObject]):
-        """Get all individual instances from input.
+    def extract_instances(
+        objects: Sequence[Union[MintObject, InstanceReference]]
+    ):
+        """Get all individual instances from input
+
+        A pre-processing step for getting datasets
 
         Parameters
         ----------
-        mint_objects: Sequence[MintObject]
+        objects: Sequence[MintObject]
             Any combination of MintStudy, MintSeries and MintInstance instances
 
-        A pre-processing step for getting datasets
+        Returns
+        -------
+        List[InstanceReference]
+            A reference to each instance (slice)
+
         """
-        return list(chain(*(x.all_instances() for x in mint_objects)))
+        instances = []
+        for item in objects:
+            if isinstance(item, InstanceReference):
+                instances.append(item)
+            else:
+                instances = instances + [
+                    to_reference(x) for x in item.all_instances()
+                ]
+        return instances
+
+    def fetch_all_datasets_async(self, mint_objects, max_workers=None):
+        """Get full DICOM dataset for each instance in study using multiple threads
+
+        Parameters
+        ----------
+        mint_objects: List[MintObject]
+            get dataset for each instance contained in these objects
+        max_workers: int, optional
+            Max number of ThreadPoolExecutor workers to use. Defaults to
+            ThreadPoolExecutor default
 
-    def get_dataset(self, instance: MintInstance):
-        """Get all DICOM data for this instance from server
+        Raises
+        ------
+        DICOMTrolleyException
+            If getting or parsing of any instance fails
 
         Returns
         -------
-        Dataset
+        Iterator[Dataset, None, None]
+            The downloaded dataset and the context that was used to download it
+
         """
 
-        return self.wado.get_dataset(
-            study_instance_uid=instance.parent.parent.uid,
-            series_instance_uid=instance.parent.uid,
-            sop_instance_iud=instance.uid,
-        )
+        for ds in self.wado.datasets_async(
+            instances=self.extract_instances(mint_objects),
+            max_workers=max_workers,
+        ):
+            yield ds
 
 
 class DICOMStorageDir:
@@ -133,4 +166,15 @@ def generate_path(self, dataset):
     def get_value(dataset, tag_name):
         """Extract value for use in path. If not found return default"""
         default = "unknown"
-        return dataset.get(tag_name, default).replace(".", "_")
+        return str(dataset.get(tag_name, default)).replace(".", "_")
+
+
+def to_reference(instance: MintInstance) -> InstanceReference:
+    """Simplify a more extensive MINT instance to an InstanceReference
+    needed for calls to WADO functions
+    """
+    return InstanceReference(
+        study_instance_uid=instance.parent.parent.uid,
+        series_instance_uid=instance.parent.uid,
+        sop_instance_iud=instance.uid,
+    )

dicomtrolley/wado.py

@@ -2,13 +2,33 @@
 
 https://www.dicomstandard.org/dicomweb/retrieve-wado-rs-and-wado-uri/
 """
+from concurrent.futures import as_completed
+from concurrent.futures.thread import ThreadPoolExecutor
+from typing import Sequence
+
+from pydantic.dataclasses import dataclass
+from pydicom.dataset import Dataset
 from pydicom.errors import InvalidDicomError
 from pydicom.filebase import DicomBytesIO
 from pydicom.filereader import dcmread
+from requests.models import Response
+from requests_futures.sessions import FuturesSession
 
 from dicomtrolley.exceptions import DICOMTrolleyException
 
 
+@dataclass()
+class InstanceReference:
+    """All information needed to download a single slice (SOPInstance) in WADO"""
+
+    study_instance_uid: str
+    series_instance_uid: str
+    sop_instance_iud: str
+
+    def __str__(self):
+        return f"InstanceReference {self.sop_instance_iud}"
+
+
 class Wado:
     """A connection to a WADO server"""
 
@@ -25,35 +45,116 @@ def __init__(self, session, url):
         self.session = session
         self.url = url
 
-    def get_dataset(
-        self, study_instance_uid, series_instance_uid, sop_instance_iud
-    ):
-        """Get all DICOM data the given instance (slice)
+    @staticmethod
+    def to_wado_parameters(instance):
+        """WADO url parameters for to retrieve instance
 
         Returns
         -------
-        Dataset
-            A pydicom dataset
+        Dict[str]
+            All parameters for a standard WADO get request
+        """
+        return {
+            "requestType": "WADO",
+            "studyUID": instance.study_instance_uid,
+            "seriesUID": instance.series_instance_uid,
+            "objectUID": instance.sop_instance_iud,
+            "contentType": "application/dicom",
+        }
+
+    @staticmethod
+    def parse_wado_response(response: Response) -> Dataset:
+        """Create a Dataset out of http response from WADO server
 
         Raises
         ------
         DICOMTrolleyException
-            If getting does not work for some reason
+            If response is not as expected or if parsing fails
 
+        Returns
+        -------
+        Dataset
         """
+        if response.status_code != 200:
 
-        response = self.session.get(
-            self.url,
-            params={
-                "requestType": "WADO",
-                "studyUID": study_instance_uid,
-                "seriesUID": series_instance_uid,
-                "objectUID": sop_instance_iud,
-                "contentType": "application/dicom",
-            },
-        )
+            raise DICOMTrolleyException(
+                f"Calling {response.url} failed ({response.status_code} - "
+                f"{response.reason})\n"
+                f"response content was {str(response.content[:300])}"
+            )
         raw = DicomBytesIO(response.content)
         try:
             return dcmread(raw)
         except InvalidDicomError as e:
-            raise DICOMTrolleyException(f"Error retrieving instance: {e}")
+            raise DICOMTrolleyException(
+                f"Error parsing response as dicom: {e}."
+                f" Response content (first 300 elements) was"
+                f" {str(response.content[:300])}"
+            )
+
+    def get_dataset(self, instance: InstanceReference):
+        """Get DICOM dataset for the given instance (slice)
+
+        Raises
+        ------
+        DICOMTrolleyException
+            If getting does not work for some reason
+
+        Returns
+        -------
+        Dataset
+            A pydicom dataset
+        """
+        return self.parse_wado_response(
+            self.session.get(
+                self.url, params=self.to_wado_parameters(instance)
+            )
+        )
+
+    def datasets(self, instances: Sequence[InstanceReference]):
+        """Retrieve each instance via WADO
+
+        Returns
+        -------
+        Iterator[Dataset, None, None]
+        """
+        for instance in instances:
+            yield self.get_dataset(instance)
+
+    def datasets_async(
+        self, instances: Sequence[InstanceReference], max_workers=None
+    ):
+        """Retrieve each instance via WADO
+
+        Parameters
+        ----------
+        instances: Sequence[InstanceReference]
+            Retrieve dataset for each of these instances
+        max_workers: int, optional
+            Use this number of workers in ThreadPoolExecutor. Defaults to
+            default for ThreadPoolExecutor
+
+        Raises
+        ------
+        DICOMTrolleyException
+            When a server response cannot be parsed as DICOM
+
+        Returns
+        -------
+        Iterator[Dataset, None, None]
+        """
+
+        with FuturesSession(
+            session=self.session,
+            executor=ThreadPoolExecutor(max_workers=max_workers),
+        ) as futures_session:
+            futures = []
+            for instance in instances:
+                futures.append(
+                    futures_session.get(
+                        self.url, params=self.to_wado_parameters(instance)
+                    )
+                )
+
+            for future in as_completed(futures):
+                yield self.parse_wado_response(future.result())

examples/go_shopping.py

@@ -32,19 +32,16 @@
 
 print(f"Found {len(studies)} studies. Taking one with least instances")
 studies.sort(key=lambda x: x.data.NumberOfStudyRelatedInstances)
-study = studies[0]
+study = studies[1]
 
 print(f"Getting slice info for {study}")
 details = trolley.find_studies(
     Query(studyInstanceUID=study.uid, queryLevel=QueryLevels.INSTANCE)
 )
-instances = trolley.extract_instances(details)
-print(f"Got {len(instances)} instances for {study}")
 
 storage = DICOMStorageDir("/tmp/trolley")
 print(f"Saving datasets to {storage}")
-for instance in instances:
-    print(f"downloading {instance}")
-    storage.save(trolley.get_dataset(instance))
-
+for ds in trolley.fetch_all_datasets_async(details, max_workers=8):
+    print(f"downloaded {ds.SOPInstanceUID}")
+    storage.save(ds)
 print("Done")

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dicomtrolley"
-version = "v0.3.0"
+version = "v0.4.0"
 description = "Retrieve medical images via DICOM-QR and DICOMweb"
 authors = ["sjoerdk <sjoerd.kerkstra@radboudumc.nl>"]
 readme = "README.md"