Skip to content

dicomtrolley.trolley

Combines Searcher and Downloader to make getting DICOM studies easy.

Notes

Design choices:

Searcher and Downloader classes should be stand-alone. They are not allowed to communicate directly. Trolley has knowledge of both and is in control.

Trolley

Combines a search and download method to get DICOM studies easily

Features:

  • Searching for DICOM using a Query instance is backend-agnostic.

  • If a download method requires additional information such as all instance UIDs, trolley can query for these in the background.

  • Saves to disk in reasonable (uid based) folder structure.

Source code in dicomtrolley/trolley.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class Trolley:
    """Combines a search and download method to get DICOM studies easily

    Features:

    * Searching for DICOM using a Query instance is backend-agnostic.

    * If a download method requires additional information such as all instance UIDs,
      trolley can query for these in the background.

    * Saves to disk in reasonable (uid based) folder structure.
    """

    def __init__(
        self,
        downloader: Downloader,
        searcher: Searcher,
        storage: Optional[DICOMDiskStorage] = None,
    ):
        """

        Parameters
        ----------
        downloader: Downloader
            The module to use for downloads
        searcher: Searcher
            The module to use for queries
        storage: DICOMDiskStorage instance, optional
            All downloads are saved to disk by calling this objects' save() method.
            Defaults to basic StorageDir (saves as /studyid/seriesid/instanceid)
        """
        self.downloader = downloader
        self.searcher = searcher

        if storage:
            self.storage = storage
        else:
            self.storage = StorageDir(tempfile.gettempdir())

    def find_studies(self, query) -> List[Study]:
        """Find study information

        Parameters
        ----------
        query:
            Search for these criteria

        Returns
        -------
        List[Study]
        """
        return list(self.searcher.find_studies(query))

    def find_study(self, query) -> Study:
        """Like find study, but returns exactly one result or raises exception.

        For queries using study identifiers such as StudyInstanceUID,AccessionNumber

        Parameters
        ----------
        query
            Search for these criteria. Query documentation for more info.

        Returns
        -------
        Study
        """
        return self.searcher.find_study(query)

    def download(
        self,
        objects: Union[DICOMDownloadable, Sequence[DICOMDownloadable]],
        output_dir,
        use_async=False,
        max_workers=None,
    ):
        """Download the given objects to output dir."""
        if not isinstance(objects, Sequence):
            objects = [objects]  # if just a single item to download is passed
        logger.info(f"Downloading {len(objects)} object(s) to '{output_dir}'")

        for dataset in self.fetch_all_datasets(objects=objects):
            self.storage.save(dataset=dataset, path=output_dir)

    def fetch_all_datasets(self, objects: Sequence[DICOMDownloadable]):
        """Get full DICOM dataset for all instances contained in objects.

        Some downloaders require explicit series- or instance level information to be
        able to download. Additional queries might be fired to obtain this
        information.

        Returns
        -------
        Iterator[Dataset, None, None]
            All datasets belonging to input objects.

        Raises
        ------
        DICOMTrolleyError
            If download fails for any reason
        """
        try:
            yield from self.downloader.datasets(objects)
        except NonSeriesParameterError:
            # downloader wants at least series level information. Do extra work.
            series_lvl_refs = self.obtain_references(
                objects=objects, max_level=DICOMObjectLevels.SERIES
            )
            yield from self.downloader.datasets(series_lvl_refs)
        except NonInstanceParameterError:
            # downloader wants only instance input. Do extra work.
            instance_refs = self.obtain_references(
                objects=objects, max_level=DICOMObjectLevels.INSTANCE
            )
            yield from self.downloader.datasets(instance_refs)

    def obtain_references(
        self,
        objects: Sequence[DICOMDownloadable],
        max_level: DICOMObjectLevels,
    ) -> List[DICOMObjectReference]:
        """Get download references for all downloadable objects, at max_level or
        lower. query if needed.

        For example, if level is QueryLevels.Instance and a Study object is given,
        try to extract instances from this study. If those instances or not in the
        study, ask searcher to obtain them

        Returns
        -------
            List[DICOMObjectReference] of the level given or deeper
        """
        references: List[DICOMObjectReference] = []
        for downloadable in objects:
            try:
                references += downloadable.contained_references(
                    max_level=max_level
                )
            except NoReferencesFoundError:
                # Not enough info in object itself. We need searcher
                logger.debug(
                    f"Not enough info to extract '{str(max_level)}-level' "
                    f"references from {downloadable}. Asking searcher."
                )
                study = self.searcher.find_study_by_id(
                    study_uid=downloadable.reference().study_uid,
                    query_level=QueryLevels.from_object_level(max_level),
                )
                references += study.contained_references(max_level=max_level)
        return references

__init__(downloader, searcher, storage=None)

Parameters

downloader: Downloader The module to use for downloads searcher: Searcher The module to use for queries storage: DICOMDiskStorage instance, optional All downloads are saved to disk by calling this objects' save() method. Defaults to basic StorageDir (saves as /studyid/seriesid/instanceid)

Source code in dicomtrolley/trolley.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    downloader: Downloader,
    searcher: Searcher,
    storage: Optional[DICOMDiskStorage] = None,
):
    """

    Parameters
    ----------
    downloader: Downloader
        The module to use for downloads
    searcher: Searcher
        The module to use for queries
    storage: DICOMDiskStorage instance, optional
        All downloads are saved to disk by calling this objects' save() method.
        Defaults to basic StorageDir (saves as /studyid/seriesid/instanceid)
    """
    self.downloader = downloader
    self.searcher = searcher

    if storage:
        self.storage = storage
    else:
        self.storage = StorageDir(tempfile.gettempdir())

download(objects, output_dir, use_async=False, max_workers=None)

Download the given objects to output dir.

Source code in dicomtrolley/trolley.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def download(
    self,
    objects: Union[DICOMDownloadable, Sequence[DICOMDownloadable]],
    output_dir,
    use_async=False,
    max_workers=None,
):
    """Download the given objects to output dir."""
    if not isinstance(objects, Sequence):
        objects = [objects]  # if just a single item to download is passed
    logger.info(f"Downloading {len(objects)} object(s) to '{output_dir}'")

    for dataset in self.fetch_all_datasets(objects=objects):
        self.storage.save(dataset=dataset, path=output_dir)

fetch_all_datasets(objects)

Get full DICOM dataset for all instances contained in objects.

Some downloaders require explicit series- or instance level information to be able to download. Additional queries might be fired to obtain this information.

Returns

Iterator[Dataset, None, None] All datasets belonging to input objects.

Raises

DICOMTrolleyError If download fails for any reason

Source code in dicomtrolley/trolley.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def fetch_all_datasets(self, objects: Sequence[DICOMDownloadable]):
    """Get full DICOM dataset for all instances contained in objects.

    Some downloaders require explicit series- or instance level information to be
    able to download. Additional queries might be fired to obtain this
    information.

    Returns
    -------
    Iterator[Dataset, None, None]
        All datasets belonging to input objects.

    Raises
    ------
    DICOMTrolleyError
        If download fails for any reason
    """
    try:
        yield from self.downloader.datasets(objects)
    except NonSeriesParameterError:
        # downloader wants at least series level information. Do extra work.
        series_lvl_refs = self.obtain_references(
            objects=objects, max_level=DICOMObjectLevels.SERIES
        )
        yield from self.downloader.datasets(series_lvl_refs)
    except NonInstanceParameterError:
        # downloader wants only instance input. Do extra work.
        instance_refs = self.obtain_references(
            objects=objects, max_level=DICOMObjectLevels.INSTANCE
        )
        yield from self.downloader.datasets(instance_refs)

find_studies(query)

Find study information

Parameters

query: Search for these criteria

Returns

List[Study]

Source code in dicomtrolley/trolley.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
def find_studies(self, query) -> List[Study]:
    """Find study information

    Parameters
    ----------
    query:
        Search for these criteria

    Returns
    -------
    List[Study]
    """
    return list(self.searcher.find_studies(query))

find_study(query)

Like find study, but returns exactly one result or raises exception.

For queries using study identifiers such as StudyInstanceUID,AccessionNumber

Parameters

query Search for these criteria. Query documentation for more info.

Returns

Study

Source code in dicomtrolley/trolley.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def find_study(self, query) -> Study:
    """Like find study, but returns exactly one result or raises exception.

    For queries using study identifiers such as StudyInstanceUID,AccessionNumber

    Parameters
    ----------
    query
        Search for these criteria. Query documentation for more info.

    Returns
    -------
    Study
    """
    return self.searcher.find_study(query)

obtain_references(objects, max_level)

Get download references for all downloadable objects, at max_level or lower. query if needed.

For example, if level is QueryLevels.Instance and a Study object is given, try to extract instances from this study. If those instances or not in the study, ask searcher to obtain them

Returns
List[DICOMObjectReference] of the level given or deeper
Source code in dicomtrolley/trolley.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def obtain_references(
    self,
    objects: Sequence[DICOMDownloadable],
    max_level: DICOMObjectLevels,
) -> List[DICOMObjectReference]:
    """Get download references for all downloadable objects, at max_level or
    lower. query if needed.

    For example, if level is QueryLevels.Instance and a Study object is given,
    try to extract instances from this study. If those instances or not in the
    study, ask searcher to obtain them

    Returns
    -------
        List[DICOMObjectReference] of the level given or deeper
    """
    references: List[DICOMObjectReference] = []
    for downloadable in objects:
        try:
            references += downloadable.contained_references(
                max_level=max_level
            )
        except NoReferencesFoundError:
            # Not enough info in object itself. We need searcher
            logger.debug(
                f"Not enough info to extract '{str(max_level)}-level' "
                f"references from {downloadable}. Asking searcher."
            )
            study = self.searcher.find_study_by_id(
                study_uid=downloadable.reference().study_uid,
                query_level=QueryLevels.from_object_level(max_level),
            )
            references += study.contained_references(max_level=max_level)
    return references