跳转到主要内容

Python实现的DICOM网络协议

项目描述

coverage unit-tests type-hints docs black pypi-versions python-versions conda zenodo

Pydicom

Pydicom是一个基于DICOM网络协议的Python实现,最初基于(已过时)Pydicom

描述

DICOM是国际标准,用于医学图像和相关信息。它定义了放射学、心脏病学、放射治疗和其他医疗领域的媒体交换格式和通信协议。

pynetdicom 是一个纯 Python 包,实现了 DICOM 网络协议。与 pydicom 配合使用,它允许轻松创建 DICOM 服务类用户 (SCUs) 和 服务类提供者 (SCPs)。

pynetdicom 的主要用户类是 AE,用于表示 DICOM 应用实体。使用它可以

  • 将应用程序作为 SCP 启动,指定支持的表现上下文,然后调用 AE.start_server() 并等待传入的关联请求

  • 将应用程序作为 SCU 使用,指定对等 SCP 需要支持的表示上下文,然后通过 AE.associate() 方法请求关联,该方法返回一个 Association 线程。

一旦建立了关联,可以通过发送 DIMSE-CDIMSE-N 消息来使用关联可用的服务。

文档

pynetdicom教程用户指南代码示例应用API 参考文档 可在 当前版本 以及 开发版本 中找到。

安装

依赖项

pydicom

安装当前版本

使用 pip

pip install -U pynetdicom

使用 conda

conda install -c conda-forge pynetdicom

有关更详细的说明,包括如何安装当前开发版本,请参阅 安装指南

支持的 DIMSE 服务

SCU 服务

当 AE 作为 SCU 行事并已与对等 SCP 建立关联时,以下 DIMSE-C 和 -N 服务可用

DIMSE 服务

关联 方法

C-ECHO

Association.send_c_echo()

C-FIND

Association.send_c_find(dataset, query_model)

C-GET

Association.send_c_get(dataset, query_model)

C-MOVE

Association.send_c_move(dataset, move_aet, query_model)

C-STORE

Association.send_c_store(dataset)

N-ACTION

Association.send_n_action(dataset, action_type, class_uid, instance_uid)

N-CREATE

Association.send_n_create(dataset, class_uid, instance_uid)

N-DELETE

Association.send_n_delete(class_uid, instance_uid)

N-EVENT-REPORT

Association.send_n_event_report(dataset, event_type, class_uid, instance_uid)

N-GET

Association.send_n_get(identifier_list, class_uid, instance_uid)

N-SET

Association.send_n_set(dataset, class_uid, instance_uid)

其中 dataset 是 pydicom 数据集 对象,query_model 是 UID 字符串,identifier_list 是 pydicom 标签 对象的列表,event_typeaction_type 是整数,class_uidinstance_uid 是 UID 字符串。有关更多信息,请参阅 关联文档

SCP 服务

当 AE 作为 SCP 行事时,一旦建立关联,以下 DIMSE-C 和 -N 服务即可供对等方使用

DIMSE 服务

干预事件

处理程序文档

C-ECHO

evt.EVT_C_ECHO

处理 C-ECHO

C-FIND

evt.EVT_C_FIND

处理 C-FIND

C-GET

evt.EVT_C_GET

处理 C-GET

C-MOVE

evt.EVT_C_MOVE

处理 C-MOVE

C-STORE

evt.EVT_C_STORE

处理 C-STORE

N-ACTION

evt.EVT_N_ACTION

处理 N-ACTION

N-CREATE

evt.EVT_N_CREATE

处理N-CREATE

N-DELETE

evt.EVT_N_DELETE

处理N-DELETE

N-EVENT-REPORT

evt.EVT_N_EVENT_REPORT

处理N-EVENT-REPORT

N-GET

evt.EVT_N_GET

处理N-GET

N-SET

evt.EVT_N_SET

处理N-SET

除C-ECHO服务外,必须将用户定义的可调用函数handler绑定到相应的干预事件,以完成DIMSE服务请求。事件可以通过from pynetdicom import evt导入,并通过AE.start_server()AE.associate()中的evt_handlers关键字参数在建立关联之前将handler绑定到事件。

当事件发生时,将调用handler函数并传递一个名为event的单个参数,它是一个Event对象,其特定属性取决于发生的事件类型。绑定到干预事件的handler必须返回或产生某些值。请参阅handler文档,了解每个事件类型在Event中可用的属性和属性,以及对应handler的预期返回/产生值。

应用程序

一些基本的DICOM应用程序包含在pynetdicom

代码示例

在文档中还有更多代码示例

回波SCU

向验证SCP(在TCP/IP地址addr,监听端口port)发送C-ECHO请求

from pynetdicom import AE

ae = AE(ae_title='MY_ECHO_SCU')
# Verification SOP Class has a UID of 1.2.840.10008.1.1
#   we can use the UID str directly when adding the requested
#   presentation context
ae.add_requested_context('1.2.840.10008.1.1')

# Associate with a peer AE
assoc = ae.associate(addr, port)

if assoc.is_established:
    # Send a DIMSE C-ECHO request to the peer
    status = assoc.send_c_echo()

    # Print the response from the peer
    if status:
        print('C-ECHO Response: 0x{0:04x}'.format(status.Status))

    # Release the association
    assoc.release()

回波SCP

在端口11112上创建阻塞式回波SCP(如果您想返回除0x0000 成功状态之外的内容,可以可选地将handler绑定到evt.EVT_C_ECHO事件)

from pynetdicom import AE, VerificationPresentationContexts

ae = AE(ae_title='MY_ECHO_SCP')
# Or we can use the inbuilt VerificationPresentationContexts list,
#   there's one for each of the supported Service Classes
# In this case, we are supporting any requests to use Verification SOP
#   Class in the association
ae.supported_contexts = VerificationPresentationContexts

# Start the SCP on (host, port) in blocking mode
ae.start_server(("localhost", 11112), block=True)

或者,您可以以非阻塞模式启动SCP,这会返回正在运行的服务器实例。当您想在同一AE中运行存储SCP并发出C-MOVE请求时,这非常有用。

在下一个示例中,我们将创建一个非阻塞的验证SCP,并绑定一个处理C-ECHO服务请求事件evt.EVT_C_ECHO的handler,该handler记录请求者的地址和端口以及事件的日期和时间。

import logging

from pynetdicom import AE, evt, debug_logger
from pynetdicom.sop_class import Verification

# Setup logging to use the StreamHandler at the debug level
debug_logger()

ae = AE(ae_title='MY_ECHO_SCP')
ae.add_supported_context(Verification)

# Implement the EVT_C_ECHO handler
def handle_echo(event, logger):
    """Handle a C-ECHO service request.

    Parameters
    ----------
    event : evt.Event
        The C-ECHO service request event, this parameter is always
        present.
    logger : logging.Logger
        The logger to use, this parameter is only present because we
        bound ``evt.EVT_C_ECHO`` using a 3-tuple.

    Returns
    -------
    int or pydicom.dataset.Dataset
        The status returned to the peer AE in the C-ECHO response.
        Must be a valid C-ECHO status value as either an ``int`` or a
        ``Dataset`` object containing an (0000,0900) *Status* element.
    """
    # Every *Event* includes `assoc` and `timestamp` attributes
    #   which are the *Association* instance the event occurred in
    #   and the *datetime.datetime* the event occurred at
    requestor = event.assoc.requestor
    timestamp = event.timestamp.strftime("%Y-%m-%d %H:%M:%S")
    msg = (
        "Received C-ECHO service request from ({}, {}) at {}"
        .format(requestor.address, requestor.port, timestamp)
    )
    logger.info(msg)

    # Return a *Success* status
    return 0x0000

# By binding using a 3-tuple we can pass extra arguments to
#   the handler
handlers = [(evt.EVT_C_ECHO, handle_echo, [logging.getLogger('pynetdicom')])]

# Start the SCP in non-blocking mode
scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)

# Associate and send a C-ECHO request to our own Verification SCP
ae.add_requested_context(Verification)
assoc = ae.associate('localhost', 11112)
if assoc.is_established:
    status = assoc.send_c_echo()
    assoc.release()

# Shutdown the SCP
scp.shutdown()

存储SCU

将DICOM CT图像存储数据集从file-in.dcm发送到对等存储SCP(在TCP/IP地址addr,监听端口port

from pydicom import dcmread
from pydicom.uid import ImplicitVRLittleEndian

from pynetdicom import AE, VerificationPresentationContexts
from pynetdicom.sop_class import CTImageStorage, MRImageStorage

ae = AE(ae_title='MY_STORAGE_SCU')
# We can also do the same thing with the requested contexts
ae.requested_contexts = VerificationPresentationContexts
# Or we can use inbuilt objects like CTImageStorage.
# The requested presentation context's transfer syntaxes can also
#   be specified using a str/UID or list of str/UIDs
ae.add_requested_context(CTImageStorage,
                         transfer_syntax=ImplicitVRLittleEndian)
# Adding a presentation context with multiple transfer syntaxes
ae.add_requested_context(MRImageStorage,
                         transfer_syntax=[ImplicitVRLittleEndian,
                                          '1.2.840.10008.1.2.1'])

assoc = ae.associate(addr, port)
if assoc.is_established:
    dataset = dcmread('file-in.dcm')
    # `status` is the response from the peer to the store request
    # but may be an empty pydicom Dataset if the peer timed out or
    # sent an invalid dataset.
    status = assoc.send_c_store(dataset)

    assoc.release()

项目详情


下载文件

下载您平台上的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。

源分布

pynetdicom-2.1.1.tar.gz (1.6 MB 查看哈希值)

上传时间

构建版本

pynetdicom-2.1.1-py3-none-any.whl (1.6 MB 查看哈希值)

上传时间 Python 3

由以下支持

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误日志 StatusPage StatusPage 状态页面