Requesting an association#
Assuming you have an AE and have added your requested
presentation contexts then you can associate with a peer by using the
AE.associate()
method, which returns an Association
thread:
from pynetdicom import AE
from pynetdicom.sop_class import Verification
ae = AE()
ae.add_requested_context(Verification)
# Associate with the peer at IPv4 address 127.0.0.1 and port 11112
# IPv6 is also supported
assoc = ae.associate("127.0.0.1", 11112)
# Release the association
if assoc.is_established:
assoc.release()
This sends an association request to the IPv4 address 127.0.0.1
on port
11112
with the request containing the presentation contexts from
AE.requested_contexts
and the default Called AE Title parameter of 'ANY-SCP'
.
Established associations should always be released or aborted (using
Association.release()
or Association.abort()
), otherwise the
association will remain open until either the peer or local AE hits a timeout
and aborts.
Specifying the Called AE Title#
Some SCPs will reject an association request if the Called AE Title parameter value doesn’t match its own title, so this can be set using the ae_title keyword parameter:
assoc = ae.associate("127.0.0.1", 11112, ae_title='STORE_SCP')
Specifying presentation contexts per association#
Calling AE.associate()
with only the addr and port parameters means the presentation
contexts in
AE.requested_contexts
will be used with the association. To propose presentation contexts on a
per-association basis you can use the contexts keyword parameter:
from pynetdicom import AE, build_context
ae = AE()
requested_contexts = [build_context('1.2.840.10008.1.1')]
assoc = ae.associate("127.0.0.1", 11112, contexts=requested_contexts)
if assoc.is_established:
assoc.release()
Using Extended Negotiation#
If you require the use of extended negotiation
then you can supply the ext_neg keyword parameter. Some extended negotiation
items can only be singular and some can occur multiple times depending on the
service class and intended usage. The following example shows how to add
SCP/SCU Role Selection Negotiation items using
build_role()
when requesting the use of the
Query/Retrieve (QR) Service Class’ C-GET service (in this example the QR SCU is
also acting as a Storage SCP), plus a User Identity Negotiation item:
from pynetdicom import AE, StoragePresentationContexts, build_role
from pynetdicom.pdu_primitives import UserIdentityNegotiation
from pynetdicom.sop_class import PatientRootQueryRetrieveInformationModelGet
ae = AE()
# Contexts supported as a Storage SCP - requires Role Selection
# Note that we are limited to a maximum of 128 contexts.
# StoragePresentationContexts includes 120, it is therefore
# possible to add 8 additional presentation contexts if needed.
ae.requested_contexts = StoragePresentationContexts
# Contexts proposed as a QR SCU
ae.add_requested_context = PatientRootQueryRetrieveInformationModelGet
# Add role selection items for the contexts we will be supporting as an SCP
negotiation_items = []
for context in StoragePresentationContexts:
role = build_role(context.abstract_syntax, scp_role=True)
negotiation_items.append(role)
# Add user identity negotiation request - passwords are sent in the clear!
user_identity = UserIdentityNegotiation()
user_identity.user_identity_type = 2
user_identity.primary_field = b'username'
user_identity.secondary_field = b'password'
negotiation_items.append(user_identity)
# Associate with the peer at IP address 127.0.0.1 and port 11112
assoc = ae.associate("127.0.0.1", 11112, ext_neg=negotiation_items)
if assoc.is_established:
assoc.release()
Possible extended negotiation items are:
Binding event handlers#
If you want to bind handlers to any
events within a new Association
you can
use the evt_handlers keyword parameter:
import logging
from pynetdicom import AE, evt, debug_logger
from pynetdicom.sop_class import Verification
debug_logger()
LOGGER = logging.getLogger('pynetdicom')
def handle_open(event):
"""Print the remote's (host, port) when connected."""
LOGGER.info(f"Connected with remote at {event.address}")
def handle_accepted(event, arg1, arg2):
"""Demonstrate the use of the optional extra parameters"""
LOGGER.info(f"Extra args: '{arg1}' and '{arg2}'")
# If a 2-tuple then only `event` parameter
# If a 3-tuple then the third value should be a list of objects to pass the handler
handlers = [
(evt.EVT_CONN_OPEN, handle_open),
(evt.EVT_ACCEPTED, handle_accepted, ['optional', 'parameters']),
]
ae = AE()
ae.add_requested_context(Verification)
assoc = ae.associate("127.0.0.1", 11112, evt_handlers=handlers)
if assoc.is_established:
assoc.release()
Handlers can also be bound and unbound from events in an existing
Association
:
import logging
from pynetdicom import AE, evt, debug_logger
from pynetdicom.sop_class import Verification
debug_logger()
LOGGER = logging.getLogger('pynetdicom')
def handle_open(event):
"""Print the remote's (host, port) when connected."""
LOGGER.info(f"Connected with remote at {event.address}")
def handle_close(event):
"""Print the remote's (host, port) when disconnected."""
LOGGER.info(f"Disconnected from remote at {event.address}")
handlers = [(evt.EVT_CONN_OPEN, handle_open)]
ae = AE()
ae.add_requested_context(Verification)
assoc = ae.associate("127.0.0.1", 11112, evt_handlers=handlers)
assoc.unbind(evt.EVT_CONN_OPEN, handle_open)
assoc.bind(evt.EVT_CONN_CLOSE, handle_close)
if assoc.is_established:
assoc.release()
Using TLS connections#
The client socket used for the association can be wrapped in TLS by supplying
the tls_args keyword parameter to
associate()
:
import ssl
from pynetdicom import AE
from pynetdicom.sop_class import Verification
ae = AE()
ae.add_requested_context(Verification)
# Create the SSLContext, your requirements may vary
ssl_cx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH, cafile='server.crt')
ssl_cx.verify_mode = ssl.CERT_REQUIRED
ssl_cx.load_cert_chain(certfile='client.crt', keyfile='client.key')
assoc = ae.associate("127.0.0.1", 11112, tls_args=(ssl_cx, None))
if assoc.is_established:
assoc.release()
Where tls_args is (ssl.SSLContext
, host), where host is the
value of the server_hostname keyword parameter in
wrap_socket()
.
Outcomes of an association request#
There are four potential outcomes of an association request: acceptance and establishment, association rejection, association abort or a connection failure, so its a good idea to test for establishment before attempting to use the association:
from pynetdicom import AE
from pynetdicom.sop_class import Verification
ae = AE()
ae.add_requested_context(Verification)
# Associate with the peer at IP address 127.0.0.1 and port 11112
assoc = ae.associate("127.0.0.1", 11112)
if assoc.is_established:
# Do something useful...
pass
# Release
assoc.release()
Using the association#
Once an association has been established with the peer then the agreed upon set of services are available for use. pynetdicom supports the usage of the following DIMSE services:
C-ECHO, through the
Association.send_c_echo()
methodC-STORE, through the
Association.send_c_store()
methodC-FIND, through the
Association.send_c_find()
methodC-GET, through the
Association.send_c_get()
method. Any AE that uses the C-GET service will also be providing the C-STORE service and must implement and bind a handler forevt.EVT_C_STORE
(as outlined here)C-MOVE, through the
Association.send_c_move()
method. The move destination can either be a different AE or the AE that made the C-MOVE request (provided a non-blocking Storage SCP has been started).N-ACTION, through the
Association.send_n_action()
methodN-CREATE, through the
Association.send_n_create()
methodN-DELETE, through the
Association.send_n_delete()
methodN-EVENT-REPORT, through the
Association.send_n_event_report()
method.N-GET, through the
Association.send_n_get()
method.N-SET, through the
Association.send_n_set()
method.
Attempting to use a service without an established association will raise a
RuntimeError
, while attempting to use a service that is not supported
by the association will raise a ValueError
.
For more information on using the services available to an association please read through the examples corresponding to the service class you’re interested in.
Releasing the association#
Once your association has been established and you’ve finished using it, its a
good idea to release the association using Association.release()
, otherwise
the association will remain open until the network timeout expires or the
peer aborts or closes the connection.
Accessing User Identity responses#
If the association Requestor has sent a
User Identity Negotiation
item as part of the extended negotiation and has requested a response in the
event of a positive identification then it can be accessed via the
Association.acceptor.user_identity
property after the association has been established.