diff --git a/.circleci/config.yml b/.circleci/config.yml index 61d577312d..3687a436b1 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -2,7 +2,7 @@ version: 2 jobs: build: docker: - - image: circleci/python:3.10 + - image: cimg/python:3.10 steps: - checkout - run: diff --git a/docs/changelog/index.rst b/docs/changelog/index.rst index 523d432900..f043c73582 100644 --- a/docs/changelog/index.rst +++ b/docs/changelog/index.rst @@ -7,6 +7,7 @@ Release Notes .. toctree:: :maxdepth: 1 + v2.2.0 v2.1.0 v2.0.1 v2.0.0 diff --git a/docs/changelog/v2.2.0.rst b/docs/changelog/v2.2.0.rst new file mode 100644 index 0000000000..129114ae96 --- /dev/null +++ b/docs/changelog/v2.2.0.rst @@ -0,0 +1,18 @@ +.. _v2.2.0: + +2.2.0 +===== + +Fixes +..... + + +Enhancements +............ + +* Added :func:`~pynetdicom.sop_class.register_uid` to make registering new + private and public SOP Classes easier (:issue:`799`) + + +Changes +....... diff --git a/docs/index.rst b/docs/index.rst index 469ad5c675..5289536321 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -18,7 +18,8 @@ If you're new to *pynetdicom* then start here: * **Basics**: :doc:`Installation` | :doc:`Writing your first SCU` | - :doc:`Writing your first SCP` + :doc:`Writing your first SCP` | + :doc:`Registering a new SOP Class` .. _index_guide: diff --git a/docs/reference/service_classes.rst b/docs/reference/service_classes.rst index f673a9fe33..2d95884166 100644 --- a/docs/reference/service_classes.rst +++ b/docs/reference/service_classes.rst @@ -1,5 +1,7 @@ .. _api_serviceclasses: +.. py:module:: pynetdicom.service_class + Service Classes (:mod:`pynetdicom.service_class`) ================================================== diff --git a/docs/reference/sop_classes.rst b/docs/reference/sop_classes.rst index 71a70997ec..27c17eb6c7 100644 --- a/docs/reference/sop_classes.rst +++ b/docs/reference/sop_classes.rst @@ -13,6 +13,7 @@ SOP Class Utilities .. autosummary:: :toctree: generated/ + register_uid SOPClass uid_to_sop_class uid_to_service_class diff --git a/docs/tutorials/index.rst b/docs/tutorials/index.rst index a195a62949..f62cb86c93 100644 --- a/docs/tutorials/index.rst +++ b/docs/tutorials/index.rst @@ -10,3 +10,4 @@ New to *pynetdicom*? Then these tutorials should get you up and running. installation create_scu create_scp + register_sop_class diff --git a/docs/tutorials/register_sop_class.rst b/docs/tutorials/register_sop_class.rst new file mode 100644 index 0000000000..2944584614 --- /dev/null +++ b/docs/tutorials/register_sop_class.rst @@ -0,0 +1,58 @@ +=========================== +Registering a new SOP Class +=========================== + +.. currentmodule:: pynetdicom + +You may occasionally come across a private SOP Class UID you like to be able +to receive, or perhaps there's a public SOP Class from a recent release of the +DICOM Standard that hasn't yet been added to *pynetdicom*. In this short +tutorial you'll learn how to register your own UID so it can be used like +the SOP Classes included by *pynetdicom*. + +To register new UIDs you use the :func:`~pynetdicom.sop_class.register_uid` function, +which takes the UID to be registered, a `keyword` that will be used as the +variable name for the new UID and the *pynetdicom* +:mod:`~pynetdicom.service_class` to register the UID with:: + + from pynetdicom import register_uid + from pynetdicom.service_class import StorageServiceClass + + register_uid( + "1.2.246.352.70.1.70", + PrivateRTPlanStorage, + StorageServiceClass, + ) + +The UID can then be imported from the :mod:`~pynetdicom.sop_class` module and +used like other UIDs:: + + from pynetdicom import AE, evt + from pynetdicom.sop_class import PrivateRTPlanStorage + + def handle_store(evt): + ds = event.dataset + ds.file_meta = event.file_meta + ds.save_as(ds.SOPInstanceUID) + + return 0x0000 + + ae = AE() + # or ae.add_supported_context("1.2.246.352.70.1.70") + ae.add_supported_context(PrivateRTPlanStorage) + ae.start_server(("localhost", 11112), evt_handlers=[(evt.EVT_C_STORE, handle_store)]) + + +When registering a new UID with the +:class:`~pynetdicom.service_class.QueryRetrieveServiceClass`, you must also +specify which of the three DIMSE-C message types the UID is to be used with:: + + from pynetdicom import register_uid + from pynetdicom.service_class import QueryRetrieveServiceClass + + register_uid( + "1.2.246.352.70.1.70", + PrivateQueryFind, + QueryRetrieveServiceClass, + dimse_msg_type="C-FIND" # or "C-GET" or "C-MOVE" + ) diff --git a/pynetdicom/__init__.py b/pynetdicom/__init__.py index 03184c92ee..d310af497a 100644 --- a/pynetdicom/__init__.py +++ b/pynetdicom/__init__.py @@ -63,6 +63,7 @@ UnifiedProcedurePresentationContexts, VerificationPresentationContexts, ) +from pynetdicom.sop_class import register_uid # Setup default logging diff --git a/pynetdicom/service_class.py b/pynetdicom/service_class.py index 5375d08459..c89ef2c581 100644 --- a/pynetdicom/service_class.py +++ b/pynetdicom/service_class.py @@ -1538,18 +1538,8 @@ class QueryRetrieveServiceClass(ServiceClass): "SpectroscopyData", "EncapsulatedDocument", ] - - def SCP(self, req: "_QR", context: "PresentationContext") -> None: - """The SCP implementation for the Query/Retrieve Service Class. - - Parameters - ---------- - req : dimse_primitives.C_FIND or C_GET or C_MOVE - The request primitive received from the peer. - context : presentation.PresentationContext - The presentation context that the SCP is operating under. - """ - _find_uids = [ + _SUPPORTED_UIDS = { + "C-FIND": [ "1.2.840.10008.5.1.4.1.2.1.1", "1.2.840.10008.5.1.4.1.2.2.1", "1.2.840.10008.5.1.4.1.2.3.1", @@ -1560,8 +1550,8 @@ def SCP(self, req: "_QR", context: "PresentationContext") -> None: "1.2.840.10008.5.1.4.44.2", "1.2.840.10008.5.1.4.45.2", "1.2.840.10008.5.1.4.1.1.200.4", - ] - _get_uids = [ + ], + "C-GET": [ "1.2.840.10008.5.1.4.1.2.1.3", "1.2.840.10008.5.1.4.1.2.2.3", "1.2.840.10008.5.1.4.1.2.3.3", @@ -1574,8 +1564,8 @@ def SCP(self, req: "_QR", context: "PresentationContext") -> None: "1.2.840.10008.5.1.4.44.4", "1.2.840.10008.5.1.4.45.4", "1.2.840.10008.5.1.4.1.1.200.6", - ] - _move_uids = [ + ], + "C-MOVE": [ "1.2.840.10008.5.1.4.1.2.1.2", "1.2.840.10008.5.1.4.1.2.2.2", "1.2.840.10008.5.1.4.1.2.3.2", @@ -1587,15 +1577,35 @@ def SCP(self, req: "_QR", context: "PresentationContext") -> None: "1.2.840.10008.5.1.4.44.3", "1.2.840.10008.5.1.4.45.3", "1.2.840.10008.5.1.4.1.1.200.5", - ] + ], + } - if isinstance(req, C_FIND) and context.abstract_syntax in _find_uids: + def SCP(self, req: "_QR", context: "PresentationContext") -> None: + """The SCP implementation for the Query/Retrieve Service Class. + + Parameters + ---------- + req : dimse_primitives.C_FIND or C_GET or C_MOVE + The request primitive received from the peer. + context : presentation.PresentationContext + The presentation context that the SCP is operating under. + """ + if ( + isinstance(req, C_FIND) + and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"] + ): self.statuses = QR_FIND_SERVICE_CLASS_STATUS self._c_find_scp(req, context) - elif isinstance(req, C_GET) and context.abstract_syntax in _get_uids: + elif ( + isinstance(req, C_GET) + and context.abstract_syntax in self._SUPPORTED_UIDS["C-GET"] + ): self.statuses = QR_GET_SERVICE_CLASS_STATUS self._get_scp(req, context) - elif isinstance(req, C_MOVE) and context.abstract_syntax in _move_uids: + elif ( + isinstance(req, C_MOVE) + and context.abstract_syntax in self._SUPPORTED_UIDS["C-MOVE"] + ): self.statuses = QR_MOVE_SERVICE_CLASS_STATUS self._move_scp(req, context) else: @@ -2379,6 +2389,9 @@ class BasicWorklistManagementServiceClass(QueryRetrieveServiceClass): """Implementation of the Basic Worklist Management Service Class.""" statuses = QR_FIND_SERVICE_CLASS_STATUS + _SUPPORTED_UIDS = { + "C-FIND": ["1.2.840.10008.5.1.4.31"], + } def SCP(self, req: "_QR", context: "PresentationContext") -> None: """The SCP implementation for Basic Worklist Management. @@ -2392,7 +2405,7 @@ def SCP(self, req: "_QR", context: "PresentationContext") -> None: """ if ( isinstance(req, C_FIND) - and context.abstract_syntax == "1.2.840.10008.5.1.4.31" + and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"] ): self._c_find_scp(req, context) else: @@ -2585,6 +2598,9 @@ class SubstanceAdministrationQueryServiceClass(QueryRetrieveServiceClass): """Implementation of the Substance Administration Query Service""" statuses = SUBSTANCE_ADMINISTRATION_SERVICE_CLASS_STATUS + _SUPPORTED_UIDS = { + "C-FIND": ["1.2.840.10008.5.1.4.41", "1.2.840.10008.5.1.4.42"], + } def SCP(self, req: "_QR", context: "PresentationContext") -> None: """The SCP implementation for the Relevant Patient Information Query @@ -2597,8 +2613,10 @@ def SCP(self, req: "_QR", context: "PresentationContext") -> None: context : presentation.PresentationContext The presentation context that the SCP is operating under. """ - uids = ["1.2.840.10008.5.1.4.41", "1.2.840.10008.5.1.4.42"] - if isinstance(req, C_FIND) and context.abstract_syntax in uids: + if ( + isinstance(req, C_FIND) + and context.abstract_syntax in self._SUPPORTED_UIDS["C-FIND"] + ): self._c_find_scp(req, context) else: raise ValueError( diff --git a/pynetdicom/sop_class.py b/pynetdicom/sop_class.py index db55d255df..376ec4b022 100644 --- a/pynetdicom/sop_class.py +++ b/pynetdicom/sop_class.py @@ -1,6 +1,7 @@ """Generates the supported SOP Classes and well-known SOP Instances.""" import inspect +from keyword import iskeyword import logging import sys from typing import Optional, Type, Any, cast, Dict @@ -56,49 +57,71 @@ def uid_to_service_class(uid: str) -> Type[ServiceClass]: """ if uid in _VERIFICATION_CLASSES.values(): return VerificationServiceClass - elif uid in _QR_CLASSES.values(): + + if uid in _QR_CLASSES.values(): return QueryRetrieveServiceClass - elif uid in _STORAGE_CLASSES.values(): + + if uid in _STORAGE_CLASSES.values(): return StorageServiceClass - elif uid in _SERVICE_CLASSES: + + if uid in _SERVICE_CLASSES: return _SERVICE_CLASSES[uid] - elif uid in _APPLICATION_EVENT_CLASSES.values(): + + if uid in _APPLICATION_EVENT_CLASSES.values(): return ApplicationEventLoggingServiceClass - elif uid in _BASIC_WORKLIST_CLASSES.values(): + + if uid in _BASIC_WORKLIST_CLASSES.values(): return BasicWorklistManagementServiceClass - elif uid in _COLOR_PALETTE_CLASSES.values(): + + if uid in _COLOR_PALETTE_CLASSES.values(): return ColorPaletteQueryRetrieveServiceClass - elif uid in _DEFINED_PROCEDURE_CLASSES.values(): + + if uid in _DEFINED_PROCEDURE_CLASSES.values(): return DefinedProcedureProtocolQueryRetrieveServiceClass - elif uid in _DISPLAY_SYSTEM_CLASSES.values(): + + if uid in _DISPLAY_SYSTEM_CLASSES.values(): return DisplaySystemManagementServiceClass - elif uid in _HANGING_PROTOCOL_CLASSES.values(): + + if uid in _HANGING_PROTOCOL_CLASSES.values(): return HangingProtocolQueryRetrieveServiceClass - elif uid in _IMPLANT_TEMPLATE_CLASSES.values(): + + if uid in _IMPLANT_TEMPLATE_CLASSES.values(): return ImplantTemplateQueryRetrieveServiceClass - elif uid in _INSTANCE_AVAILABILITY_CLASSES.values(): + + if uid in _INSTANCE_AVAILABILITY_CLASSES.values(): return InstanceAvailabilityNotificationServiceClass - elif uid in _MEDIA_CREATION_CLASSES.values(): + + if uid in _MEDIA_CREATION_CLASSES.values(): return MediaCreationManagementServiceClass - elif uid in _MEDIA_STORAGE_CLASSES.values(): + + if uid in _MEDIA_STORAGE_CLASSES.values(): return ServiceClass # Not yet implemented - elif uid in _NON_PATIENT_OBJECT_CLASSES.values(): + + if uid in _NON_PATIENT_OBJECT_CLASSES.values(): return NonPatientObjectStorageServiceClass - elif uid in _PRINT_MANAGEMENT_CLASSES.values(): + + if uid in _PRINT_MANAGEMENT_CLASSES.values(): return PrintManagementServiceClass - elif uid in _PROCEDURE_STEP_CLASSES.values(): + + if uid in _PROCEDURE_STEP_CLASSES.values(): return ProcedureStepServiceClass - elif uid in _PROTOCOL_APPROVAL_CLASSES.values(): + + if uid in _PROTOCOL_APPROVAL_CLASSES.values(): return ProtocolApprovalQueryRetrieveServiceClass - elif uid in _RELEVANT_PATIENT_QUERY_CLASSES.values(): + + if uid in _RELEVANT_PATIENT_QUERY_CLASSES.values(): return RelevantPatientInformationQueryServiceClass - elif uid in _RT_MACHINE_VERIFICATION_CLASSES.values(): + + if uid in _RT_MACHINE_VERIFICATION_CLASSES.values(): return RTMachineVerificationServiceClass - elif uid in _STORAGE_COMMITMENT_CLASSES.values(): + + if uid in _STORAGE_COMMITMENT_CLASSES.values(): return StorageCommitmentServiceClass - elif uid in _SUBSTANCE_ADMINISTRATION_CLASSES.values(): + + if uid in _SUBSTANCE_ADMINISTRATION_CLASSES.values(): return SubstanceAdministrationQueryServiceClass - elif uid in _UNIFIED_PROCEDURE_STEP_CLASSES.values(): + + if uid in _UNIFIED_PROCEDURE_STEP_CLASSES.values(): return UnifiedProcedureStepServiceClass # No SCP implemented @@ -112,6 +135,7 @@ class SOPClass(UID): """ _service_class: Optional[Type[ServiceClass]] = None + _name: str = "" def __new__(cls: Type["SOPClass"], val: str) -> "SOPClass": if isinstance(val, SOPClass): @@ -429,6 +453,32 @@ def _generate_sop_classes(sop_class_dict: Dict[str, str]) -> None: "Verification": "1.2.840.10008.1.1", } + +_SERVICE_TO_UID_GROUP = { + VerificationServiceClass: _VERIFICATION_CLASSES, + QueryRetrieveServiceClass: _QR_CLASSES, + StorageServiceClass: _STORAGE_CLASSES, + ApplicationEventLoggingServiceClass: _APPLICATION_EVENT_CLASSES, + BasicWorklistManagementServiceClass: _BASIC_WORKLIST_CLASSES, + ColorPaletteQueryRetrieveServiceClass: _COLOR_PALETTE_CLASSES, + DefinedProcedureProtocolQueryRetrieveServiceClass: _DEFINED_PROCEDURE_CLASSES, + DisplaySystemManagementServiceClass: _DISPLAY_SYSTEM_CLASSES, + HangingProtocolQueryRetrieveServiceClass: _HANGING_PROTOCOL_CLASSES, + ImplantTemplateQueryRetrieveServiceClass: _IMPLANT_TEMPLATE_CLASSES, + InstanceAvailabilityNotificationServiceClass: _INSTANCE_AVAILABILITY_CLASSES, + MediaCreationManagementServiceClass: _MEDIA_CREATION_CLASSES, + NonPatientObjectStorageServiceClass: _NON_PATIENT_OBJECT_CLASSES, + PrintManagementServiceClass: _PRINT_MANAGEMENT_CLASSES, + ProcedureStepServiceClass: _PROCEDURE_STEP_CLASSES, + ProtocolApprovalQueryRetrieveServiceClass: _PROTOCOL_APPROVAL_CLASSES, + RelevantPatientInformationQueryServiceClass: _RELEVANT_PATIENT_QUERY_CLASSES, + RTMachineVerificationServiceClass: _RT_MACHINE_VERIFICATION_CLASSES, + StorageCommitmentServiceClass: _STORAGE_COMMITMENT_CLASSES, + SubstanceAdministrationQueryServiceClass: _SUBSTANCE_ADMINISTRATION_CLASSES, + UnifiedProcedureStepServiceClass: _UNIFIED_PROCEDURE_STEP_CLASSES, +} + + # pylint: enable=line-too-long _generate_sop_classes(_APPLICATION_EVENT_CLASSES) _generate_sop_classes(_BASIC_WORKLIST_CLASSES) @@ -488,6 +538,87 @@ def uid_to_sop_class(uid: str) -> SOPClass: return sop_class +def register_uid( + uid: str, + keyword: str, + service_class: Type[ServiceClass], + dimse_msg_type: str = "", +) -> None: + """Register a private or public SOP Class UID `uid` with the + :mod:`~pynetdicom.sop_class` module. + + Examples + -------- + + Register the UID ``1.2.246.352.70.1.70`` with the + :class:`~pynetdicom.service_class.StorageServiceClass`:: + + >>> from pynetdicom import register_uid + >>> from pynetdicom.service_class import StorageServiceClass + >>> register_uid( + ... "1.2.246.352.70.1.70", + ... "FooStorage", + ... StorageServiceClass, + ... ) + + Using a UID after registration:: + + >>> from pynetdicom import AE + >>> from pynetdicom.sop_class import FooStorage + >>> ae = AE() + >>> ae.add_supported_context(FooStorage) + + Parameters + ---------- + uid : str + The UID to be registered. + keyword : str + The keyword to use for the UID, must be a valid Python identifier and + may not be a Python keyword. + service_class : pynetdicom.service_class.ServiceClass + The service that the `uid` will be registered with, such as + :class:`~pynetdicom.service_class.StorageServiceClass`. Note that this + must be the class object itself and not a class instance. + dimse_msg_type : str, optional + If `service_class` is + :class:`~pynetdicom.service_class.QueryRetrieveServiceClass` then this + should be the DIMSE service message type that the `uid` is being + registered to. One of (``"C-FIND"``, ``"C-GET"``, ``"C-MOVE"``). + """ + if not keyword.isidentifier() or iskeyword(keyword): + raise ValueError( + f"The keyword '{keyword}' is not a valid Python identifier or is " + "a Python keyword" + ) + + if not inspect.isclass(service_class): + raise TypeError("'service_class' must be a class object not a class instance") + + if not issubclass(service_class, ServiceClass): + raise TypeError( + "'service_class' must be a ServiceClass subclass object " + "such as 'StorageServiceClass'" + ) + + group = _SERVICE_TO_UID_GROUP[service_class] + group[keyword] = uid + + sop_class = SOPClass(uid) + sop_class._service_class = uid_to_service_class(uid) + globals()[keyword] = sop_class + + if issubclass(service_class, QueryRetrieveServiceClass): + if service_class is QueryRetrieveServiceClass: + if dimse_msg_type not in ("C-FIND", "C-GET", "C-MOVE"): + raise ValueError( + "'dimse_msg_type' must be 'C-FIND', 'C-GET' or 'C-MOVE' " + "when registering a UID with QueryRetrieveServiceClass" + ) + service_class._SUPPORTED_UIDS[dimse_msg_type].append(uid) + else: + service_class._SUPPORTED_UIDS["C-FIND"].append(uid) + + # Well-known SOP Instance UIDs for the supported Service Classes DisplaySystemInstance = UID("1.2.840.10008.5.1.1.40.1") """``1.2.840.10008.5.1.1.40.1`` diff --git a/pynetdicom/tests/test_service_qr.py b/pynetdicom/tests/test_service_qr.py index 4df2a067d3..6cbf82ebcb 100644 --- a/pynetdicom/tests/test_service_qr.py +++ b/pynetdicom/tests/test_service_qr.py @@ -26,6 +26,8 @@ evt, build_role, debug_logger, + sop_class, + register_uid, ) from pynetdicom.dimse_primitives import C_FIND, C_GET, C_MOVE, C_STORE from pynetdicom.presentation import PresentationContext @@ -40,6 +42,8 @@ PatientRootQueryRetrieveInformationModelGet, PatientRootQueryRetrieveInformationModelMove, CompositeInstanceRetrieveWithoutBulkDataGet, + _QR_CLASSES, + _BASIC_WORKLIST_CLASSES, ) @@ -60,6 +64,20 @@ def test_unknown_sop_class(): service.SCP(None, context) +@pytest.fixture() +def register_new_uid_find(): + register_uid( + "1.2.3.4", + "NewFind", + QueryRetrieveServiceClass, + "C-FIND", + ) + yield + del _QR_CLASSES["NewFind"] + delattr(sop_class, "NewFind") + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4") + + class TestQRFindServiceClass: """Test the QueryRetrieveFindServiceClass""" @@ -1175,6 +1193,54 @@ def handle(event): assert assoc.is_aborted scp.shutdown() + def test_register(self, register_new_uid_find): + """Test registering a new UID""" + from pynetdicom.sop_class import NewFind + + def handle(event): + yield 0xFF01, self.query + yield 0x0000, None + yield 0xA700, None + + handlers = [(evt.EVT_C_FIND, handle)] + + self.ae = ae = AE() + ae.add_supported_context(NewFind) + ae.add_requested_context(NewFind, ExplicitVRLittleEndian) + scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) + + ae.acse_timeout = 5 + ae.dimse_timeout = 5 + assoc = ae.associate("localhost", 11112) + assert assoc.is_established + result = assoc.send_c_find(self.query, NewFind) + status, identifier = next(result) + assert status.Status == 0xFF01 + assert identifier == self.query + status, identifier = next(result) + assert status.Status == 0x0000 + assert identifier is None + with pytest.raises(StopIteration): + next(result) + + assoc.release() + assert assoc.is_released + scp.shutdown() + + +@pytest.fixture() +def register_new_uid_get(): + register_uid( + "1.2.3.4", + "NewGet", + QueryRetrieveServiceClass, + "C-GET", + ) + yield + del _QR_CLASSES["NewGet"] + delattr(sop_class, "NewGet") + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"].remove("1.2.3.4") + class TestQRGetServiceClass: def setup_method(self): @@ -3501,6 +3567,63 @@ def handle_store(event): assert assoc.is_released scp.shutdown() + def test_register(self, register_new_uid_get): + """Test registering a new UID""" + from pynetdicom.sop_class import NewGet + + def handle(event): + yield 1 + yield 0xFF00, self.ds + + def handle_store(event): + return 0x0000 + + handlers = [(evt.EVT_C_GET, handle)] + + self.ae = ae = AE() + ae.add_supported_context(NewGet) + ae.add_supported_context(CTImageStorage, scu_role=False, scp_role=True) + ae.add_requested_context(NewGet) + ae.add_requested_context(CTImageStorage) + scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) + + role = build_role(CTImageStorage, scp_role=True) + handlers = [(evt.EVT_C_STORE, handle_store)] + + ae.acse_timeout = 5 + ae.dimse_timeout = 5 + assoc = ae.associate("localhost", 11112, ext_neg=[role], evt_handlers=handlers) + assert assoc.is_established + result = assoc.send_c_get(self.query, NewGet) + status, identifier = next(result) + assert status.Status == 0xFF00 + assert identifier is None + status, identifier = next(result) + assert status.Status == 0x0000 + assert status.NumberOfFailedSuboperations == 0 + assert status.NumberOfWarningSuboperations == 0 + assert status.NumberOfCompletedSuboperations == 1 + assert identifier is None + pytest.raises(StopIteration, next, result) + + assoc.release() + assert assoc.is_released + scp.shutdown() + + +@pytest.fixture() +def register_new_uid_move(): + register_uid( + "1.2.3.4", + "NewMove", + QueryRetrieveServiceClass, + "C-MOVE", + ) + yield + del _QR_CLASSES["NewMove"] + delattr(sop_class, "NewMove") + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"].remove("1.2.3.4") + class TestQRMoveServiceClass: def setup_method(self): @@ -5966,6 +6089,46 @@ def handle_store(event): assoc.release() scp.shutdown() + def test_register(self, register_new_uid_move): + """Test registering a new UID""" + from pynetdicom.sop_class import NewMove + + def handle(event): + yield self.destination + yield 1 + yield 0xFF00, self.ds + + def handle_store(event): + return 0x0000 + + handlers = [(evt.EVT_C_MOVE, handle), (evt.EVT_C_STORE, handle_store)] + + self.ae = ae = AE() + ae.add_supported_context(NewMove) + ae.add_supported_context(CTImageStorage, scu_role=False, scp_role=True) + ae.add_requested_context(NewMove) + ae.add_requested_context(CTImageStorage) + scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) + + ae.acse_timeout = 5 + ae.dimse_timeout = 5 + assoc = ae.associate("localhost", 11112) + assert assoc.is_established + result = assoc.send_c_move(self.query, "TESTMOVE", NewMove) + status, identifier = next(result) + assert status.Status == 0xFF00 + assert identifier is None + status, identifier = next(result) + assert status.Status == 0x0000 + assert status.NumberOfFailedSuboperations == 0 + assert status.NumberOfWarningSuboperations == 0 + assert status.NumberOfCompletedSuboperations == 1 + assert identifier is None + pytest.raises(StopIteration, next, result) + + assoc.release() + scp.shutdown() + class TestQRCompositeInstanceWithoutBulk: """Tests for QR + Composite Instance Without Bulk Data""" @@ -6174,6 +6337,20 @@ def handle_store(event): scp.shutdown() +@pytest.fixture() +def register_new_uid_bwm(): + register_uid( + "1.2.3.4", + "NewFind", + BasicWorklistManagementServiceClass, + "C-FIND", + ) + yield + del _BASIC_WORKLIST_CLASSES["NewFind"] + delattr(sop_class, "NewFind") + BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4") + + class TestBasicWorklistServiceClass: """Tests for BasicWorklistManagementServiceClass.""" @@ -6198,8 +6375,9 @@ def test_bad_abstract_syntax_raises(self): context = build_context("1.2.3.4") bwm.SCP(None, context) - def test_pending_success(self): - """Test handler yielding pending then success status""" + def test_register(self, register_new_uid_bwm): + """Test registering a new UID""" + from pynetdicom.sop_class import NewFind def handle(event): yield 0xFF01, self.query @@ -6209,15 +6387,15 @@ def handle(event): handlers = [(evt.EVT_C_FIND, handle)] self.ae = ae = AE() - ae.add_supported_context(ModalityWorklistInformationFind) - ae.add_requested_context(ModalityWorklistInformationFind) + ae.add_supported_context(NewFind) + ae.add_requested_context(NewFind) scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) ae.acse_timeout = 5 ae.dimse_timeout = 5 assoc = ae.associate("localhost", 11112) assert assoc.is_established - result = assoc.send_c_find(self.query, ModalityWorklistInformationFind) + result = assoc.send_c_find(self.query, NewFind) status, identifier = next(result) assert status.Status == 0xFF01 assert identifier == self.query diff --git a/pynetdicom/tests/test_service_storage.py b/pynetdicom/tests/test_service_storage.py index 2c3b5d2d18..ff5e229cfc 100644 --- a/pynetdicom/tests/test_service_storage.py +++ b/pynetdicom/tests/test_service_storage.py @@ -1,5 +1,6 @@ """Tests for the StorageServiceClass.""" +from copy import deepcopy from io import BytesIO import os from pathlib import Path @@ -11,13 +12,15 @@ from pydicom.dataset import Dataset, FileMetaDataset from pydicom.uid import ExplicitVRLittleEndian, ImplicitVRLittleEndian -from pynetdicom import AE, _config, evt, debug_logger +from pynetdicom import AE, _config, evt, debug_logger, register_uid, sop_class from pynetdicom.dimse_primitives import C_STORE from pynetdicom.pdu_primitives import SOPClassExtendedNegotiation from pynetdicom.sop_class import ( Verification, CTImageStorage, + _STORAGE_CLASSES, ) +from pynetdicom.service_class import StorageServiceClass try: from pynetdicom.status import Status @@ -41,6 +44,18 @@ def enable_unrestricted(): _config.UNRESTRICTED_STORAGE_SERVICE = False +@pytest.fixture() +def register_new_uid(): + register_uid( + "1.2.3.4", + "NewStorage", + StorageServiceClass, + ) + yield + del _STORAGE_CLASSES["NewStorage"] + delattr(sop_class, "NewStorage") + + class TestStorageServiceClass: """Test the StorageServiceClass""" @@ -799,3 +814,35 @@ def handle(event): assert recv == ["CompressedSamples^CT1", "Private", "Unknown^Public"] scp.shutdown() + + def test_register(self, register_new_uid): + """Test registering a new UID.""" + from pynetdicom.sop_class import NewStorage + + attrs = {} + + def handle(event): + attrs["uid"] = event.dataset.SOPClassUID + return 0x0000 + + ds = deepcopy(DATASET) + ds.SOPClassUID = NewStorage + + handlers = [(evt.EVT_C_STORE, handle)] + + self.ae = ae = AE() + ae.add_supported_context(NewStorage) + ae.add_requested_context(NewStorage) + scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) + + assoc = ae.associate("localhost", 11112) + assert assoc.is_established + rsp = assoc.send_c_store(ds) + assert rsp.Status == 0x0000 + assert "ErrorComment" not in rsp + assoc.release() + assert assoc.is_released + + assert attrs["uid"] == NewStorage + + scp.shutdown() diff --git a/pynetdicom/tests/test_service_substance_admin.py b/pynetdicom/tests/test_service_substance_admin.py index 4a7af6e210..6b72a5553f 100644 --- a/pynetdicom/tests/test_service_substance_admin.py +++ b/pynetdicom/tests/test_service_substance_admin.py @@ -8,7 +8,7 @@ from pydicom.dataset import Dataset from pydicom.uid import ExplicitVRLittleEndian -from pynetdicom import AE, evt, debug_logger +from pynetdicom import AE, evt, debug_logger, register_uid, sop_class from pynetdicom.dimse_primitives import C_FIND from pynetdicom.presentation import PresentationContext from pynetdicom.service_class import ( @@ -16,6 +16,7 @@ ) from pynetdicom.sop_class import ( ProductCharacteristicsQuery, + _SUBSTANCE_ADMINISTRATION_CLASSES, ) @@ -36,6 +37,20 @@ def test_unknown_sop_class(): service.SCP(C_FIND(), context) +@pytest.fixture() +def register_new_uid(): + register_uid( + "1.2.3.4", + "NewFind", + SubstanceAdministrationQueryServiceClass, + "C-FIND", + ) + yield + del _SUBSTANCE_ADMINISTRATION_CLASSES["NewFind"] + delattr(sop_class, "NewFind") + SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"].remove("1.2.3.4") + + class TestSubstanceAdministrationQueryServiceClass: """Test the SubstanceAdministrationQueryServiceClass. @@ -230,3 +245,35 @@ def handle(event): assoc.release() assert assoc.is_released scp.shutdown() + + def test_register(self, register_new_uid): + """Test handler yielding a Dataset status""" + from pynetdicom.sop_class import NewFind + + def handle(event): + status = Dataset() + status.Status = 0xFF00 + yield status, self.query + yield 0x0000, None + + handlers = [(evt.EVT_C_FIND, handle)] + + self.ae = ae = AE() + ae.add_supported_context(NewFind) + ae.add_requested_context(NewFind, ExplicitVRLittleEndian) + scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers) + + assoc = ae.associate("localhost", 11112) + assert assoc.is_established + result = assoc.send_c_find(self.query, NewFind) + status, identifier = next(result) + assert status.Status == 0xFF00 + status, identifier = next(result) + assert status.Status == 0x0000 + with pytest.raises(StopIteration): + next(result) + + assoc.release() + assert assoc.is_released + + scp.shutdown() diff --git a/pynetdicom/tests/test_sop.py b/pynetdicom/tests/test_sop.py index 88fafb7e04..213d718c25 100644 --- a/pynetdicom/tests/test_sop.py +++ b/pynetdicom/tests/test_sop.py @@ -5,6 +5,7 @@ from pynetdicom import __version__ from pydicom._uid_dict import UID_dictionary +from pynetdicom import sop_class from pynetdicom.sop_class import ( uid_to_sop_class, uid_to_service_class, @@ -64,6 +65,7 @@ SubstanceAdministrationLoggingInstance, UPSFilteredGlobalSubscriptionInstance, UPSGlobalSubscriptionInstance, + register_uid, ) from pynetdicom.service_class import ( ServiceClass, @@ -166,11 +168,11 @@ def test_all_sop_instances(): class TestUIDtoSOPlass: """Tests for uid_to_sop_class""" - def test_missing_sop(self): + def test_missing_sop_class(self): """Test SOP Class if UID not found.""" - sop_class = uid_to_sop_class("1.2.3.4") - assert sop_class == "1.2.3.4" - assert sop_class.service_class == ServiceClass + sop = uid_to_sop_class("1.2.3.4") + assert sop == "1.2.3.4" + assert sop.service_class == ServiceClass def test_verification_uid(self): """Test normal function""" @@ -179,8 +181,8 @@ def test_verification_uid(self): def test_existing(self): """Test that the existing class is returned.""" original = Verification - sop_class = uid_to_sop_class("1.2.840.10008.1.1") - assert id(sop_class) == id(original) + sop = uid_to_sop_class("1.2.840.10008.1.1") + assert id(sop) == id(original) class TestUIDToServiceClass: @@ -471,13 +473,158 @@ def test_verification_sop(self): def test_uid_creation(self): """Test creating a new UIDSOPClass.""" - sop_class = SOPClass("1.2.3") - sop_class._service_class = ServiceClass + sop = SOPClass("1.2.3") + sop._service_class = ServiceClass - assert sop_class == "1.2.3" - assert sop_class.service_class == ServiceClass + assert sop == "1.2.3" + assert sop.service_class == ServiceClass - sop_class_b = SOPClass(sop_class) - assert sop_class == sop_class_b - assert sop_class_b == "1.2.3" - assert sop_class_b.service_class == ServiceClass + sop_b = SOPClass(sop) + assert sop == sop_b + assert sop_b == "1.2.3" + assert sop_b.service_class == ServiceClass + + +class TestRegisterUID: + def test_register_storage(self): + """Test registering to the storage service.""" + register_uid( + "1.2.3.4", + "FooStorage", + StorageServiceClass, + ) + + sop = sop_class.FooStorage + assert sop == "1.2.3.4" + assert sop.service_class == StorageServiceClass + + del _STORAGE_CLASSES["FooStorage"] + delattr(sop_class, "FooStorage") + + def test_register_qr_find(self): + """Test registering to the QR service - FIND.""" + register_uid( + "1.2.3.4", + "FooFind", + QueryRetrieveServiceClass, + dimse_msg_type="C-FIND", + ) + + sop = sop_class.FooFind + assert sop == "1.2.3.4" + assert sop.service_class == QueryRetrieveServiceClass + assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"] + + del _QR_CLASSES["FooFind"] + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop) + delattr(sop_class, "FooFind") + + def test_register_qr_get(self): + """Test registering to the QR service - GET.""" + register_uid( + "1.2.3.4", + "FooGet", + QueryRetrieveServiceClass, + dimse_msg_type="C-GET", + ) + + sop = sop_class.FooGet + assert sop == "1.2.3.4" + assert sop.service_class == QueryRetrieveServiceClass + assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"] + + del _QR_CLASSES["FooGet"] + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-GET"].remove(sop) + delattr(sop_class, "FooGet") + + def test_register_qr_move(self): + """Test registering to the QR service - MOVE.""" + register_uid( + "1.2.3.4", + "FooMove", + QueryRetrieveServiceClass, + dimse_msg_type="C-MOVE", + ) + + sop = sop_class.FooMove + assert sop == "1.2.3.4" + assert sop.service_class == QueryRetrieveServiceClass + assert sop in QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"] + + del _QR_CLASSES["FooMove"] + QueryRetrieveServiceClass._SUPPORTED_UIDS["C-MOVE"].remove(sop) + delattr(sop_class, "FooMove") + + def test_register_bwm_find(self): + """Test registering to the BWM service.""" + register_uid( + "1.2.3.4", + "FooFind", + BasicWorklistManagementServiceClass, + dimse_msg_type="C-FIND", + ) + + sop = sop_class.FooFind + assert sop == "1.2.3.4" + assert sop.service_class == BasicWorklistManagementServiceClass + assert sop in BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"] + + del _BASIC_WORKLIST_CLASSES["FooFind"] + BasicWorklistManagementServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop) + delattr(sop_class, "FooFind") + + def test_register_substance_admin_find(self): + """Test registering to the Substance Admin QR service.""" + register_uid( + "1.2.3.4", + "FooFind", + SubstanceAdministrationQueryServiceClass, + dimse_msg_type="C-FIND", + ) + + sop = sop_class.FooFind + assert sop == "1.2.3.4" + assert sop.service_class == SubstanceAdministrationQueryServiceClass + assert sop in SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"] + + del _SUBSTANCE_ADMINISTRATION_CLASSES["FooFind"] + SubstanceAdministrationQueryServiceClass._SUPPORTED_UIDS["C-FIND"].remove(sop) + delattr(sop_class, "FooFind") + + def test_invalid_keyword_raises(self): + """Test invalid keyword raises exceptions.""" + msg = ( + "The keyword '2coo' is not a valid Python identifier or is " + "a Python keyword" + ) + with pytest.raises(ValueError, match=msg): + register_uid("", "2coo", ServiceClass) + + msg = ( + "The keyword 'def' is not a valid Python identifier or is " + "a Python keyword" + ) + with pytest.raises(ValueError, match=msg): + register_uid("", "def", ServiceClass) + + def test_invalid_service_class_raises(self): + """Test that an invalid service_class raises exceptions.""" + msg = "'service_class' must be a class object not a class instance" + with pytest.raises(TypeError, match=msg): + register_uid("", "Foo", "") + + msg = ( + "'service_class' must be a ServiceClass subclass object " + "such as 'StorageServiceClass'" + ) + with pytest.raises(TypeError, match=msg): + register_uid("", "Foo", str) + + def test_invalid_dimse_msg_type_raises(self): + """Test that an invalid dimse_msg_type raises exceptions.""" + msg = ( + "'dimse_msg_type' must be 'C-FIND', 'C-GET' or 'C-MOVE' " + "when registering a UID with QueryRetrieveServiceClass" + ) + with pytest.raises(ValueError, match=msg): + register_uid("", "Foo", QueryRetrieveServiceClass, "Foo")