Voltron Studio logo black rectangle

Article

Build a DICOM DIMSE C-STORE Service with Python in 10 Minutes

David Vuong

·10 mins read

Build a simple DICOM C-STORE DIMSE-C Service with Python 3.8 in 10 minutes.

Python’s ecosystem of DICOM tooling allows us to hack up DIMSE-C services fairly quickly. In this article, we’ll walk through how to build a C-STORE DIMSE Service in using pydicom and pynetdicom. All source code is available online and a link to the repository will be provided at the end.

Installation

Start by installing DICOM dependencies:

poetry init
poetry add pydicom pynetdicom

Poetry is a modern Python package manager, similar to easy_install and pip. pydicom gives us functionality to manipulate DICOM datasets and pynetdicom provides us the tools to create AEs (application entities) to interpret and understand the DICOM communication protocol.

A successful install output should resemble this:

Creating virtualenv dicom-dimse-c-store-example-j5KghW-L-py3.8 in /Users/<yourmachine>/Library/Caches/pypoetry/virtualenvs
Using version ^2.0.0 for pydicom
Using version ^1.5.3 for pynetdicom

Updating dependencies
Resolving dependencies... (0.1s)

Writing lock file

Package operations: 2 installs, 0 updates, 0 removals

  • Installing pydicom (2.0.0)
  • Installing pynetdicom (1.5.3)

A virtual environment and lockfile are automatically generated for you. Read the Poetry documentation for more details.

Service Class Provider

There are two types of AEs: SCP (Service Class Provider) which can be thought of as a server and SCU (Service Class User) as a client. Let’s define a SCP AE to receive C-STORE operations:

class ServiceClassProvider:
    def __init__(self, aet: str, config: ServiceClassProviderConfig) -> None:
        self.config = config
        self.address = ("0.0.0.0", config.port)

        self.ae = AE(ae_title=aet)
        self.ae.implementation_class_uid = config.implementation_class_uid
        self.ae.implementation_version_name = f'{aet}_{config.implementation_class_uid.split(".")[-1]}'[:16]

        self.SUPPORTED_ABSTRACT_SYNTAXES: List[str] = [
            sop_class.DigitalXRayImagePresentationStorage,
            sop_class.DigitalXRayImageProcessingStorage,
            sop_class.CTImageStorage,
        ]
        for abstract_syntax in self.SUPPORTED_ABSTRACT_SYNTAXES:
            self.ae.add_supported_context(abstract_syntax, ALL_TRANSFER_SYNTAXES)
        self.ae.add_supported_context(VerificationSOPClass, DEFAULT_TRANSFER_SYNTAXES)

        self.ae.require_calling_aet = ["EXAMPLESCU"]

    def handle_c_store(self, event: Event) -> int:
        ds = event.dataset
        ds.file_meta = event.file_meta

        metadata: Dict[str, Optional[ParsedElementValue]] = {
            "CallingAET": cast(str, event.assoc.requestor.ae_title.strip()),
            "SopInstanceUID": safe_get(ds, 0x00080018),
            "StudyInstanceUID": safe_get(ds, 0x0020000D),
            "Modality": safe_get(ds, 0x00080060),
        }
        log_message_meta = " - ".join([f"{k}={v}" for k, v in metadata.items() if v])
        logger.info(f"Processed C-STORE {log_message_meta}")

        return 0x0000

    def start(self) -> None:
        logger.info(f"Starting DIMSE C-STORE AE on address={self.address} aet={self.ae.ae_title}")
        self.handlers = [(events.EVT_C_STORE, self.handle_c_store)]
        self.ae.start_server(self.address, block=True, evt_handlers=self.handlers)


def main() -> None:
    config = ServiceClassProviderConfig(implementation_class_uid=generate_uid(), port=8081)
    server = ServiceClassProvider("SAMPLESCP", config)
    server.start()


if __name__ == "__main__":
    main()

The snippet above omits imports, type aliases, and other auxiliary definitions such as methods and dataclasses. You can see the full example on GitHub. Don’t be thrown off by all the type annotations. They can largely be ignored.

There’s a lot to grok here. Let’s break it down into manageable chunks.

Our SCP is encapsulated in a ServiceClassProvider class. The constructor defines most of the configuration to initialize but not start the SCP server.

self.config = config
self.address = ("0.0.0.0", config.port)

self.ae = AE(ae_title=aet)
self.ae.implementation_class_uid = config.implementation_class_uid
self.ae.implementation_version_name = f'{aet}_{config.implementation_class_uid.split(".")[-1]}'[:16]

self.SUPPORTED_ABSTRACT_SYNTAXES: List[str] = [
    sop_class.DigitalXRayImagePresentationStorage,
    sop_class.DigitalXRayImageProcessingStorage,
    sop_class.CTImageStorage,
]
for abstract_syntax in self.SUPPORTED_ABSTRACT_SYNTAXES:
    self.ae.add_supported_context(abstract_syntax, ALL_TRANSFER_SYNTAXES)
self.ae.add_supported_context(VerificationSOPClass, DEFAULT_TRANSFER_SYNTAXES)

self.ae.require_calling_aet = ["SAMPLESCU"]

Every AE requires an AET (application entity title). This is a fixed 16 character name for your SCP. Although there are no strict requirements defined in the DICOM 3.0 NEMA standard, the AET should only contain numbers and uppercase characters.

An implementation class UID is generated upon the instantiation of the SCP class. An implementation version name is derived using the AET and class UID.

config = ServiceClassProvidereConfig(implementation_class_uid=generate_uid(), port=8081)
server = ServiceClassProvider("SAMPLESCP", config)

In pynetdicom, both implementation_* attributes are optional during the initialization. You can read more about them here.

DICOM abstract and transfer syntaxes to be simply put, define the type of data and how that data is encoded. During the association phase between a SCU and SCP, both entities negotiate an acceptable context. If a context cannot be agreed, the association (or request) is rejected.

In our example above, our SCP supports 3 abstract syntaxes and 1 VerificationSOPClass syntax. The 4th syntax is used to allow ECHO-C operations. These are useful during testing and deployment stages to assert whether clients are able to connect to the SCP.

Finally, we specify require_calling_aet = [“SAMPLESCU”]. This is a blanket rule which tells our AE to only allow SCU associations where their AET is whitelisted. In this case, the SCU must have an AET of SAMPLESCU.

def handle_c_store(self, event: Event) -> int:
  ds = event.dataset
  ds.file_meta = event.file_meta

  metadata: Dict[str, Optional[ParsedElementValue]] = {
      "CallingAET": cast(str, event.assoc.requestor.ae_title.strip()),
      "SopInstanceUID": safe_get(ds, 0x00080018),
      "StudyInstanceUID": safe_get(ds, 0x0020000D),
      "Modality": safe_get(ds, 0x00080060),
  }
  log_message_meta = " - ".join([f"{k}={v}" for k, v in metadata.items() if v])
  logger.info(f"Processed C-STORE {log_message_meta}")

  return 0x0000

def start(self) -> None:
  logger.info(f"Starting DIMSE C-STORE AE on address={self.address} aet={self.ae.ae_title}")
  self.handlers = [(events.EVT_C_STORE, self.handle_c_store)]
  self.ae.start_server(self.address, block=True, evt_handlers=self.handlers)

Similar to HTTP or gRPC, DIMSE-C SCPs also have “routes” (DIMSE-C/DIMSE-N operations) and “controllers/stubs” (handlers). The only exception is that unlike HTTP or gRPC, DIMSE-C SCPs have a fixed number of operations and hence handlers. Note that it’s common to have a separate AE for each DIMSE-C operation.

Handlers for our SCP are set when a call to start_server is made. As you can see, our handle_c_store handler is quite simple. It merely extracts a few fields from the DICOM dataset emitted during the C-STORE operation and logs it out to the screen.

Service Class User

Let’s write an SCU to test this works:

def send_c_store(ae: AE, dcm_path: str, scp_address: Tuple[str, int], idx: int) -> None:
    host, port = scp_address
    logger.debug(f"Establishing an association with DIMSE C-STORE SCP ({host}, {port})")

    dataset = read_file(dcm_path)

    # http://dicom.nema.org/dicom/2013/output/chtml/part10/chapter_7.html
    abstract_syntax = dataset.file_meta[0x00020002].value
    transfer_syntax = dataset.file_meta[0x00020010].value

    context = PresentationContext()
    context.abstract_syntax = abstract_syntax
    context.transfer_syntax = [transfer_syntax]

    assoc = ae.associate(host, port, contexts=[context])
    if not assoc.is_established:
        logger.error("Failed to associate with SCP. Exiting...")
        assoc.release()
        return

    try:
        response = assoc.send_c_store(dataset)
        status = response[0x00000900].value
        dcm_file_name = os.path.basename(dcm_path)
        association_syntax = f"{abstract_syntax}{transfer_syntax}"
        logger.info(f"({idx}) Sent C-STORE '{dcm_file_name}' - syntax={association_syntax} - status={status}")
    finally:
        assoc.release()

ae = AE(ae_title="SAMPLESCU")
scp_address: Tuple[str, int] = (args.host, args.port)
send_c_store(ae,, scp_address)

All imports and auxiliary methods are omitted here as well. See GitHub for the full source.

An SCU AE is instantiated and used to establish an association to our SCP. Before the association, a presentation context is derived based on the abstract and transfer syntax defined within the specified DICOM file. Upon the successful association, a C-STORE operation is emitted and the status, along with a few pieces of metadata, are logged to the console.

Starting the SCP and executing the SCU, we get the following logs:

# SCP logs
> poetry run python dicom-dimse-c-store-example/scp.py
INFO:dicom-dimse-c-store-example:Starting DIMSE C-STORE AE on address=('0.0.0.0', 8081) aet=b'SAMPLESCP       '
INFO:dicom-dimse-c-store-example:Processed C-STORE CallingAET=b'SAMPLESCU' - SopInstanceUID=<uid> - StudyInstanceUID=<uid> - Modality=CT
# SCU logs
> poetry run python dicom-dimse-c-store-example/scu.py <path_to_dicom>
INFO:dicom-dimse-c-store-example:(1) Sent C-STORE '<dicom_file_name>' - syntax=1.2.840.10008.5.1.4.1.1.21.2.840.10008.1.2.1 - status=0

Success! It works! 🎉

You can take this further and add all sorts of data parsing, extraction, validation, and patient PHI (protected health information) de-identification prior to interacting with the DICOM dataset.

As promised, you can find the entire source code here.

Voltron Studio is a bespoke software consultancy specialised in healthcare. Healthcare technologies are antiquated and decades behind other industries. Our goal is to scale healthcare digital imaging by integrating modern web technologies into all the products and services we provide.

Read about our Customer Journey and reach out any time. We’re always happy to chat.

Related tags
EngineeringHealthcareTutorialDICOMPython
Written by

David Vuong

Co-Founder & Director at Voltron Studio

Sign up for our newsletter.

Be notified when we share new ideas and updates. Stay up-to-date on news and tips in web technologies, healthcare software, and radiology!

We care about the protection of your data. Read our privacy policy.

Voltron Studio logo white text and square

© 2021 Voltron Studio Pty Ltd, ABN 72 645 265 103