Skip to content

dicomtrolley.dicom_qr

Implements DICOM QR (Query Retrieve), a method for getting information from a VNA

See http://dicom.nema.org/dicom/2013/output/chtml/part04/sect_C.3.html

DICOMQR

Bases: Searcher

A connection to a DICOM QR enabled server.

Source code in dicomtrolley/dicom_qr.py 
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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
class DICOMQR(Searcher):
    """A connection to a DICOM QR enabled server."""

    def __init__(
        self, host, port, aet="DICOMTROLLEY", aec="ANY-SCP", debug=False
    ):
        """

        Parameters
        ----------
        host: str
            Hostname of DICOM-QR-enabled server
        port: int
            Port for DICOM-QR
        aet: str, optional
            Application Entity Title - Name of the calling entity (this class).
            Defaults to 'DICOMTROLLEY'
        aec: str, optional
            Application Entity Called - The name of the server you are calling.
            Defaults to 'ANY-SCP'
        debug: bool, optional
            If True, prints debug logging to console. This can be very useful as
            exceptions often do not contain detailed information.
        """
        self.host = host
        self.port = port
        self.aet = aet
        self.aec = aec
        self.debug = debug

    def find_studies(self, query: Query):
        """

        Parameters
        ----------
        query: Query
            Find arguments matching this query

        Raises
        ------
        DICOMTrolleyError
            When finding fails

        Returns
        -------
        List[Study]
        """
        return self.parse_c_find_response(
            self.send_c_find(DICOMQuery.init_from_query(query))
        )

    @staticmethod
    def parse_c_find_response(responses) -> List[Study]:
        """Parse flat list of datasets from CFIND into a study/series/instance tree

        CFIND returns a flat list of datasets on the queries' QueryRetrieveLevel.
        For instance at IMAGE level, there is one dataset for each matching instance.
        Each dataset should contain Series and Study information. Parse this into a
        dicomtrolley study/series/instance tree that can be used as input for
        download functions

        Parameters
        ----------
        responses: Sequence[Dataset]
            Datasets coming from a pydicom cfind query

        Returns
        -------
        List[Study]
            Each study populated with series and instance objects, if provided
        """

        tree = DICOMParseTree()
        for response in responses:
            tree.insert_dataset(response)
        return tree.as_studies()

    def send_c_find(self, query):
        """Perform a CFIND with the given query

        Raises
        ------
        DICOMTrolleyError
            When finding fails
        """
        if self.debug:
            debug_logger()
        ae = AE(ae_title=bytes(self.aet, encoding="utf-8"))
        ae.add_requested_context(StudyRootQueryRetrieveInformationModelFind)

        assoc = ae.associate(
            self.host, self.port, ae_title=bytes(self.aec, encoding="utf-8")
        )
        responses = []
        if assoc.is_established:
            # Send the C-FIND request
            c_find_response = assoc.send_c_find(
                query.as_dataset(),
                StudyRootQueryRetrieveInformationModelFind,
            )
            for (status, identifier) in c_find_response:
                if status:
                    # I don't understand this status. For now just collect non-None
                    if identifier:
                        responses.append(identifier)

                else:
                    raise DICOMTrolleyError(
                        "Connection timed out, was aborted or"
                        " received invalid response"
                    )

            assoc.release()
        else:
            raise DICOMTrolleyError(
                "Association rejected, aborted or never connected"
            )

        return responses

__init__(host, port, aet='DICOMTROLLEY', aec='ANY-SCP', debug=False)

Parameters

host: str Hostname of DICOM-QR-enabled server port: int Port for DICOM-QR aet: str, optional Application Entity Title - Name of the calling entity (this class). Defaults to 'DICOMTROLLEY' aec: str, optional Application Entity Called - The name of the server you are calling. Defaults to 'ANY-SCP' debug: bool, optional If True, prints debug logging to console. This can be very useful as exceptions often do not contain detailed information.

Source code in dicomtrolley/dicom_qr.py 
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
def __init__(
    self, host, port, aet="DICOMTROLLEY", aec="ANY-SCP", debug=False
):
    """

    Parameters
    ----------
    host: str
        Hostname of DICOM-QR-enabled server
    port: int
        Port for DICOM-QR
    aet: str, optional
        Application Entity Title - Name of the calling entity (this class).
        Defaults to 'DICOMTROLLEY'
    aec: str, optional
        Application Entity Called - The name of the server you are calling.
        Defaults to 'ANY-SCP'
    debug: bool, optional
        If True, prints debug logging to console. This can be very useful as
        exceptions often do not contain detailed information.
    """
    self.host = host
    self.port = port
    self.aet = aet
    self.aec = aec
    self.debug = debug

find_studies(query)

Parameters

query: Query Find arguments matching this query

Raises

DICOMTrolleyError When finding fails

Returns

List[Study]

Source code in dicomtrolley/dicom_qr.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def find_studies(self, query: Query):
    """

    Parameters
    ----------
    query: Query
        Find arguments matching this query

    Raises
    ------
    DICOMTrolleyError
        When finding fails

    Returns
    -------
    List[Study]
    """
    return self.parse_c_find_response(
        self.send_c_find(DICOMQuery.init_from_query(query))
    )

parse_c_find_response(responses) staticmethod

Parse flat list of datasets from CFIND into a study/series/instance tree

CFIND returns a flat list of datasets on the queries' QueryRetrieveLevel. For instance at IMAGE level, there is one dataset for each matching instance. Each dataset should contain Series and Study information. Parse this into a dicomtrolley study/series/instance tree that can be used as input for download functions

Parameters

responses: Sequence[Dataset] Datasets coming from a pydicom cfind query

Returns

List[Study] Each study populated with series and instance objects, if provided

Source code in dicomtrolley/dicom_qr.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@staticmethod
def parse_c_find_response(responses) -> List[Study]:
    """Parse flat list of datasets from CFIND into a study/series/instance tree

    CFIND returns a flat list of datasets on the queries' QueryRetrieveLevel.
    For instance at IMAGE level, there is one dataset for each matching instance.
    Each dataset should contain Series and Study information. Parse this into a
    dicomtrolley study/series/instance tree that can be used as input for
    download functions

    Parameters
    ----------
    responses: Sequence[Dataset]
        Datasets coming from a pydicom cfind query

    Returns
    -------
    List[Study]
        Each study populated with series and instance objects, if provided
    """

    tree = DICOMParseTree()
    for response in responses:
        tree.insert_dataset(response)
    return tree.as_studies()

send_c_find(query)

Perform a CFIND with the given query

Raises

DICOMTrolleyError When finding fails

Source code in dicomtrolley/dicom_qr.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def send_c_find(self, query):
    """Perform a CFIND with the given query

    Raises
    ------
    DICOMTrolleyError
        When finding fails
    """
    if self.debug:
        debug_logger()
    ae = AE(ae_title=bytes(self.aet, encoding="utf-8"))
    ae.add_requested_context(StudyRootQueryRetrieveInformationModelFind)

    assoc = ae.associate(
        self.host, self.port, ae_title=bytes(self.aec, encoding="utf-8")
    )
    responses = []
    if assoc.is_established:
        # Send the C-FIND request
        c_find_response = assoc.send_c_find(
            query.as_dataset(),
            StudyRootQueryRetrieveInformationModelFind,
        )
        for (status, identifier) in c_find_response:
            if status:
                # I don't understand this status. For now just collect non-None
                if identifier:
                    responses.append(identifier)

            else:
                raise DICOMTrolleyError(
                    "Connection timed out, was aborted or"
                    " received invalid response"
                )

        assoc.release()
    else:
        raise DICOMTrolleyError(
            "Association rejected, aborted or never connected"
        )

    return responses

DICOMQuery

Bases: ExtendedQuery

Things you can search for with DICOM QR.

Notes

  • Incomplete: this class implements only a minimal core of DICOM QR search. It can be extended should the need arise

  • All string arguments support (*) as a wildcard

  • non-pep8 parameter naming format follows DICOM parameter convention. Other parameters match MINTQuery conventions, query parameters are similar.

Source code in dicomtrolley/dicom_qr.py 
 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
class DICOMQuery(ExtendedQuery):
    """Things you can search for with DICOM QR.

    Notes
    -----
    * Incomplete: this class implements only a minimal core of DICOM QR search.
      It can be extended should the need arise

    * All string arguments support (*) as a wildcard

    * non-pep8 parameter naming format follows DICOM parameter convention.
      Other parameters match MINTQuery conventions, query parameters are similar.

    """

    Modality: str = ""
    ProtocolName: str = ""
    StudyID: str = ""
    # raise ValueError when passing an unknown keyword to init
    model_config = ConfigDict(extra="forbid")

    @staticmethod
    def get_default_include_fields(query_level):
        """Include fields you definitely want to get back"""
        if query_level == QueryRetrieveLevels.STUDY:
            return ["StudyInstanceUID"]
        elif query_level == QueryRetrieveLevels.SERIES:
            return ["StudyInstanceUID", "SeriesInstanceUID"]
        elif query_level == QueryRetrieveLevels.IMAGE:
            return ["StudyInstanceUID", "SeriesInstanceUID", "SOPInstanceUID"]

    @staticmethod
    def parse_date(date):
        if date:
            return date.strftime("%Y%m%d")
        else:
            return ""

    @classmethod
    def get_study_date(cls, min_study_date, max_study_date):
        """Get value for CFIND parameter StudyDate"""
        min_sd = cls.parse_date(min_study_date)
        max_sd = cls.parse_date(max_study_date)
        if min_sd or max_sd:
            return min_sd + "-" + max_sd
        else:
            return None

    def as_parameters(self) -> Dict[str, str]:
        """Parameters that can be used in constructing a DICOM QR query"""

        # remove non-DICOM parameters and replace with DICOM tags based on them
        parameters = {
            x: y for x, y in self.model_dump().items()
        }  # all params for query
        parameters["StudyDate"] = self.get_study_date(
            parameters.pop("min_study_date"), parameters.pop("max_study_date")
        )

        # translate values and change parameter name to fit DICOM-QR naming
        parameters["QueryRetrieveLevel"] = QueryRetrieveLevels.translate(
            parameters.pop("query_level")
        )

        # add useful default include fields
        default_fields = self.get_default_include_fields(
            parameters["QueryRetrieveLevel"]
        )
        parameters["include_fields"] = list(
            set(parameters["include_fields"]) | set(default_fields)
        )

        return parameters

    def as_dataset(self):
        """A dataset that can be used as a CFIND query."""
        parameters = self.as_parameters()
        ds = Dataset()
        # in CFIND, empty elements are interpreted as 'need to be returned filled'
        for field in parameters.pop(
            "include_fields"
        ):  # for each include field
            if tag_for_keyword(field):
                setattr(ds, field, "")  # add an empty DICOM element
        parameters = {
            x: y for x, y in parameters.items() if y
        }  # Skip None values

        for parameter, value in parameters.items():
            setattr(ds, parameter, value)

        return ds

as_dataset()

A dataset that can be used as a CFIND query.

Source code in dicomtrolley/dicom_qr.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def as_dataset(self):
    """A dataset that can be used as a CFIND query."""
    parameters = self.as_parameters()
    ds = Dataset()
    # in CFIND, empty elements are interpreted as 'need to be returned filled'
    for field in parameters.pop(
        "include_fields"
    ):  # for each include field
        if tag_for_keyword(field):
            setattr(ds, field, "")  # add an empty DICOM element
    parameters = {
        x: y for x, y in parameters.items() if y
    }  # Skip None values

    for parameter, value in parameters.items():
        setattr(ds, parameter, value)

    return ds

as_parameters()

Parameters that can be used in constructing a DICOM QR query

Source code in dicomtrolley/dicom_qr.py 
 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
def as_parameters(self) -> Dict[str, str]:
    """Parameters that can be used in constructing a DICOM QR query"""

    # remove non-DICOM parameters and replace with DICOM tags based on them
    parameters = {
        x: y for x, y in self.model_dump().items()
    }  # all params for query
    parameters["StudyDate"] = self.get_study_date(
        parameters.pop("min_study_date"), parameters.pop("max_study_date")
    )

    # translate values and change parameter name to fit DICOM-QR naming
    parameters["QueryRetrieveLevel"] = QueryRetrieveLevels.translate(
        parameters.pop("query_level")
    )

    # add useful default include fields
    default_fields = self.get_default_include_fields(
        parameters["QueryRetrieveLevel"]
    )
    parameters["include_fields"] = list(
        set(parameters["include_fields"]) | set(default_fields)
    )

    return parameters

get_default_include_fields(query_level) staticmethod

Include fields you definitely want to get back

Source code in dicomtrolley/dicom_qr.py 
66
67
68
69
70
71
72
73
74
@staticmethod
def get_default_include_fields(query_level):
    """Include fields you definitely want to get back"""
    if query_level == QueryRetrieveLevels.STUDY:
        return ["StudyInstanceUID"]
    elif query_level == QueryRetrieveLevels.SERIES:
        return ["StudyInstanceUID", "SeriesInstanceUID"]
    elif query_level == QueryRetrieveLevels.IMAGE:
        return ["StudyInstanceUID", "SeriesInstanceUID", "SOPInstanceUID"]

get_study_date(min_study_date, max_study_date) classmethod

Get value for CFIND parameter StudyDate

Source code in dicomtrolley/dicom_qr.py 
83
84
85
86
87
88
89
90
91
@classmethod
def get_study_date(cls, min_study_date, max_study_date):
    """Get value for CFIND parameter StudyDate"""
    min_sd = cls.parse_date(min_study_date)
    max_sd = cls.parse_date(max_study_date)
    if min_sd or max_sd:
        return min_sd + "-" + max_sd
    else:
        return None

QueryRetrieveLevels

Valid values for DICOMQR. Differs slightly from the MINT version.

Source code in dicomtrolley/dicom_qr.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class QueryRetrieveLevels:
    """Valid values for DICOMQR. Differs slightly from the MINT version."""

    STUDY = "STUDY"
    SERIES = "SERIES"
    IMAGE = "IMAGE"

    ALL = {STUDY, SERIES, IMAGE}

    @classmethod
    def translate(cls, value):
        """Translate from base levels. For converting between queries"""
        translation = {
            QueryLevels.STUDY: cls.STUDY,
            QueryLevels.SERIES: cls.SERIES,
            QueryLevels.INSTANCE: cls.IMAGE,
        }
        return translation[value]

translate(value) classmethod

Translate from base levels. For converting between queries

Source code in dicomtrolley/dicom_qr.py 
34
35
36
37
38
39
40
41
42
@classmethod
def translate(cls, value):
    """Translate from base levels. For converting between queries"""
    translation = {
        QueryLevels.STUDY: cls.STUDY,
        QueryLevels.SERIES: cls.SERIES,
        QueryLevels.INSTANCE: cls.IMAGE,
    }
    return translation[value]