net
pacsanini.net
special
#
The net module provides methods and classes that can be
used to communicate between DICOM nodes over a network.
c_echo
#
The echo module provides methods for checking if associations
can be made between two DICOM nodes.
echo(local_node, called_node)
#
Check that the DICOM connection is OK between two nodes
using C-ECHO.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[dict, pacsanini.models.DicomNode] |
The source/local node that you want to test the connection from. |
required |
called_node |
Union[dict, pacsanini.models.DicomNode] |
The target node that you want to check your connection with. |
required |
Exceptions:
| Type | Description |
|---|---|
ValueError |
An error is raised if the called_node does not have its ip and port attributes set. |
Returns:
| Type | Description |
|---|---|
int |
An integer corresponding to the status code of the established connection. 0 means the association was successfull. -1 means that the connection could not be established. |
Source code in pacsanini/net/c_echo.py
def echo(
local_node: Union[dict, DicomNode], called_node: Union[dict, DicomNode]
) -> int:
"""Check that the DICOM connection is OK between two nodes
using C-ECHO.
Parameters
----------
local_node : Union[dict, DicomNode]
The source/local node that you want to test the connection
from.
called_node : Union[dict, DicomNode]
The target node that you want to check your connection with.
Returns
-------
int
An integer corresponding to the status code of the established
connection. 0 means the association was successfull. -1 means
that the connection could not be established.
Raises
------
ValueError
An error is raised if the called_node does not have its ip
and port attributes set.
"""
if isinstance(local_node, dict):
local_node = DicomNode(**local_node)
if isinstance(called_node, dict):
called_node = DicomNode(**called_node)
if not called_node.has_net_info:
raise ValueError(f"{called_node} does not have a network address.")
ae = AE(ae_title=local_node.aetitle)
ae.add_requested_context("1.2.840.10008.1.1")
association = ae.associate(called_node.ip, called_node.port)
status = -1
if association.is_established:
status_ds = association.send_c_echo()
status = status_ds.Status
association.release()
return status
c_find
#
The c_find module provides methods that can be used to query
DICOM data that is stored in another DICOM node (typically a PACS).
find(local_node, called_node, *, query_level, dicom_fields, start_date, end_date=None, modality='')
#
Find DICOM resources from the destination DICOM node using
the specified DICOM criteria.
The dicom_fields parameter are used to ask the destination DICOM node for additional information regarding results. If the destination node does not return those values, default values of None will be returned.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
query_level |
Union[pacsanini.models.QueryLevel, str] |
The query level to use when asking for data to retrieve. This can be "PATIENT" or "STUDY". According to this level, the values returned for the dicom_fields you request may change. |
required |
dicom_fields |
List[str] |
A list of DICOM tags to get information from when the destination node returns results. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Returns:
| Type | Description |
|---|---|
Generator[pydicom.dataset.Dataset, NoneType, NoneType] |
Each result returned by the query made to the called_node is yielded as a Dataset instance. |
Source code in pacsanini/net/c_find.py
def find(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
*,
query_level: Union[QueryLevel, str],
dicom_fields: List[str],
start_date: datetime,
end_date: datetime = None,
modality: str = "",
) -> Generator[Dataset, None, None]:
"""Find DICOM resources from the destination DICOM node using
the specified DICOM criteria.
The dicom_fields parameter are used to ask the destination
DICOM node for additional information regarding results. If
the destination node does not return those values, default
values of None will be returned.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
query_level : Union[QueryLevel, str]
The query level to use when asking for data to retrieve. This
can be "PATIENT" or "STUDY". According to this level, the values
returned for the dicom_fields you request may change.
dicom_fields : List[str]
A list of DICOM tags to get information from when the destination
node returns results.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
Yields
------
Dataset
Each result returned by the query made to the called_node
is yielded as a Dataset instance.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
if isinstance(local_node, dict):
local_node = DicomNode(**local_node)
if isinstance(called_node, dict):
called_node = DicomNode(**called_node)
if isinstance(query_level, str):
query_level = QueryLevel(query_level)
if query_level is QueryLevel.PATIENT:
query_root = PatientRootQueryRetrieveInformationModelFind
else:
query_root = StudyRootQueryRetrieveInformationModelFind
if not called_node.has_net_info:
raise ValueError(f"{called_node} does not have a network address.")
if end_date is None:
end_date = start_date
if end_date < start_date:
err_msg = (
f"The start date {start_date} cannot be greater"
f" than the end date {end_date}"
)
raise ValueError(err_msg)
ae = AE(ae_title=local_node.aetitle)
ae.add_requested_context(query_root)
current_date = start_date
date_increment = timedelta(days=15)
while current_date <= end_date:
if (current_date + date_increment) >= end_date:
upper_date = end_date
else:
upper_date = current_date + date_increment
if current_date == end_date:
date_str = current_date.strftime("%Y%m%d")
requested_date = f"{date_str}"
else:
requested_date = (
f"{current_date.strftime('%Y%m%d')}-{upper_date.strftime('%Y%m%d')}"
)
for char in ascii_lowercase:
ds = Dataset()
ds.Modality = modality if modality else ""
ds.PatientName = f"{char}*"
ds.QueryRetrieveLevel = query_level.value
ds.StudyDate = requested_date
for field in dicom_fields:
if field not in _SEARCH_FIELDS:
setattr(ds, field, "")
assoc = ae.associate(called_node.ip, called_node.port)
try:
if assoc.is_established:
responses = assoc.send_c_find(ds, query_root)
for (status, identifier) in responses:
if status and identifier:
for field in list(dicom_fields) + _SEARCH_FIELDS:
if not hasattr(identifier, field):
setattr(identifier, field, None)
yield identifier
finally:
if assoc.is_alive():
assoc.release()
current_date += date_increment + timedelta(days=1)
patient_find(local_node, called_node, *, dicom_fields, start_date, end_date=None, modality='')
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT query retrieve level.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
dicom_fields |
List[str] |
A list of DICOM tags to get information from when the destination node returns results. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Returns:
| Type | Description |
|---|---|
Generator[pydicom.dataset.Dataset, NoneType, NoneType] |
Each result returned by the query made to the called_node is yielded as a Dataset instance. |
Source code in pacsanini/net/c_find.py
def patient_find(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
*,
dicom_fields: List[str],
start_date: datetime,
end_date: datetime = None,
modality: str = "",
) -> Generator[Dataset, None, None]:
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT
query retrieve level.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
dicom_fields : List[str]
A list of DICOM tags to get information from when the destination
node returns results.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
Yields
------
Dataset
Each result returned by the query made to the called_node
is yielded as a Dataset instance.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
results = find(
local_node,
called_node,
query_level=QueryLevel.PATIENT,
dicom_fields=dicom_fields,
start_date=start_date,
end_date=end_date,
modality=modality,
)
for res in results:
yield res
patient_find2csv(local_node, called_node, dest, *, dicom_fields, start_date, end_date=None, modality='')
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT query retrieve level. Returned results will be persisted the dest file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
dest |
str |
The output path to write results to. |
required |
dicom_fields |
List[str] |
A list of DICOM tags to get information from when the destination node returns results. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Source code in pacsanini/net/c_find.py
def patient_find2csv(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
dest: str,
*,
dicom_fields: List[str],
start_date: datetime,
end_date: datetime = None,
modality: str = "",
):
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT
query retrieve level. Returned results will be persisted the dest
file.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
dest : str
The output path to write results to.
dicom_fields : List[str]
A list of DICOM tags to get information from when the destination
node returns results.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
fields = _SEARCH_FIELDS + dicom_fields
with open(dest, "w", newline="") as out:
writer = DictWriter(out, fieldnames=fields)
writer.writeheader()
results_generator = patient_find(
local_node,
called_node,
dicom_fields=dicom_fields,
start_date=start_date,
end_date=end_date,
modality=modality,
)
for result in results_generator:
res_dict = {field: getattr(result, field) for field in fields}
writer.writerow(res_dict)
patient_find2sql(local_node, called_node, conn_uri, *, start_date, end_date=None, modality='', create_tables=False)
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT query retrieve level. Returned results will be persisted in the specified database.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
conn_uri |
str |
The database's connection URI. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
create_tables |
bool |
If True, create the database tables before inserting the first find result. The default is False. |
False |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Source code in pacsanini/net/c_find.py
def patient_find2sql(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
conn_uri: str,
*,
start_date: datetime,
end_date: datetime = None,
modality: str = "",
create_tables: bool = False,
):
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the PATIENT
query retrieve level. Returned results will be persisted in the
specified database.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
conn_uri : str
The database's connection URI.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
create_tables : bool
If True, create the database tables before inserting the first
find result. The default is False.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
with DBWrapper(conn_uri, create_tables=create_tables) as db:
results_generator = patient_find(
local_node,
called_node,
dicom_fields=StudyFind.cfind_fields(),
start_date=start_date,
end_date=end_date,
modality=modality,
)
for result in results_generator:
add_found_study(db.conn(), result)
study_find(local_node, called_node, *, dicom_fields, start_date, end_date=None, modality='')
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY query retrieve level.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
dicom_fields |
List[str] |
A list of DICOM tags to get information from when the destination node returns results. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Returns:
| Type | Description |
|---|---|
Generator[pydicom.dataset.Dataset, NoneType, NoneType] |
Each result returned by the query made to the called_node is yielded as a Dataset instance. |
Source code in pacsanini/net/c_find.py
def study_find(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
*,
dicom_fields: List[str],
start_date: datetime,
end_date: datetime = None,
modality: str = "",
) -> Generator[Dataset, None, None]:
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY
query retrieve level.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
dicom_fields : List[str]
A list of DICOM tags to get information from when the destination
node returns results.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
Yields
------
Dataset
Each result returned by the query made to the called_node
is yielded as a Dataset instance.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
results = find(
local_node,
called_node,
query_level=QueryLevel.STUDY,
dicom_fields=dicom_fields,
start_date=start_date,
end_date=end_date,
modality=modality,
)
for res in results:
yield res
study_find2csv(local_node, called_node, dest, *, dicom_fields, start_date, end_date=None, modality='')
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY query retrieve level. Returned results will be persisted the dest file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
dest |
str |
The output path to write results to. |
required |
dicom_fields |
List[str] |
A list of DICOM tags to get information from when the destination node returns results. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Source code in pacsanini/net/c_find.py
def study_find2csv(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
dest: str,
*,
dicom_fields: List[str],
start_date: datetime,
end_date: datetime = None,
modality: str = "",
):
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY
query retrieve level. Returned results will be persisted the dest
file.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
dest : str
The output path to write results to.
dicom_fields : List[str]
A list of DICOM tags to get information from when the destination
node returns results.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
fields = _SEARCH_FIELDS + dicom_fields
with open(dest, "w", newline="") as out:
writer = DictWriter(out, fieldnames=fields)
writer.writeheader()
results_generator = study_find(
local_node,
called_node,
dicom_fields=dicom_fields,
start_date=start_date,
end_date=end_date,
modality=modality,
)
for result in results_generator:
res_dict = {field: getattr(result, field) for field in fields}
writer.writerow(res_dict)
study_find2sql(local_node, called_node, conn_uri, *, start_date, end_date=None, modality='', create_tables=False)
#
Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY query retrieve level. Returned results will be persisted in the specified database.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The source/calling DICOM node that seeks to retrieve information from the destination node. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The destination/target node from which information is queried from. |
required |
conn_uri |
str |
The database's connection URI. |
required |
start_date |
datetime |
The date for which the query should be made. |
required |
end_date |
datetime |
If set, queries will range from the start_date to the end_date. The end_date parameter must therefore be greater or equal to the start_date parameter. |
None |
modality |
str |
If set, specify the DICOM modality to get results for. |
'' |
create_tables |
bool |
If True, create the database tables before inserting the first find result. The default is False. |
False |
Exceptions:
| Type | Description |
|---|---|
ValueError |
A ValueError is raised if the called_node parameter does not have set IP and port values or if the end_date parameter is set and is smaller than the start_date parameter. |
Source code in pacsanini/net/c_find.py
def study_find2sql(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
conn_uri: str,
*,
start_date: datetime,
end_date: datetime = None,
modality: str = "",
create_tables: bool = False,
):
"""Find DICOM resources from the destination DICOM node using the
specified DICOM criteria. Queries are made using the STUDY
query retrieve level. Returned results will be persisted in the
specified database.
Parameters
----------
local_node : Union[DicomNode, dict]
The source/calling DICOM node that seeks to retrieve information
from the destination node.
called_node : Union[DicomNode, dict]
The destination/target node from which information is queried
from.
conn_uri : str
The database's connection URI.
start_date : datetime
The date for which the query should be made.
end_date : datetime
If set, queries will range from the start_date to the end_date.
The end_date parameter must therefore be greater or equal to the
start_date parameter.
modality : str
If set, specify the DICOM modality to get results for.
create_tables : bool
If True, create the database tables before inserting the first
find result. The default is False.
Raises
------
ValueError
A ValueError is raised if the called_node parameter does not
have set IP and port values or if the end_date parameter is
set and is smaller than the start_date parameter.
"""
with DBWrapper(conn_uri, create_tables=create_tables) as db:
results_generator = study_find(
local_node,
called_node,
dicom_fields=StudyFind.cfind_fields(),
start_date=start_date,
end_date=end_date,
modality=modality,
)
for result in results_generator:
add_found_study(db.conn(), result)
c_move
#
The c_move module contains methods that can be used to perform C-MOVE
operations with a PACS node.
move(local_node, called_node, *, resources, query_level, dest_node=None, directory='', sort_by=<StorageSortKey.PATIENT: 'PATIENT'>, start_time=None, end_time=None, db_session=None, callbacks=None)
#
Move resources requested by the local_node to the
dest_node by querying the called_node.
Resources should be a list of string values corresponding to patient ID values (in which case the query_level should be set to PATIENT) or to study UID values (in which case the query_level should be set to STUDY).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
Union[pacsanini.models.DicomNode, dict] |
The local DICOM node initiating the C-MOVE requests. |
required |
called_node |
Union[pacsanini.models.DicomNode, dict] |
The called DICOM node that contains the the DICOM resources. |
required |
resources |
List[str] |
The list of DICOM resources to move. |
required |
query_level |
QueryLevel |
If the resources are study UID values, this should be set to STUDY. If the resources are patient ID values, this should be set to PATIENT. |
required |
dest_node |
Union[pacsanini.models.DicomNode, dict] |
The DICOM node to move the requested resources to. If unset, this will be equal to the local_node. |
None |
directory |
str |
Specify the directory in which to store the DICOM files. The default is the current directory. |
'' |
sort_by |
StorageSortKey |
The method by which DICOM files should be stored. The default is PATIENT |
<StorageSortKey.PATIENT: 'PATIENT'> |
start_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should start. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The end_time parameter must also be set. |
None |
end_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should end. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The start_time parameter must also be set. |
None |
db_session |
Session |
The database session to use if move results are to be stored in a database. |
None |
callbacks |
List[Callable[[Any], Any]] |
The callbacks to pass on to the storescp server. |
None |
Returns:
| Type | Description |
|---|---|
Generator[Tuple[int, str], NoneType, NoneType] |
Yield tuples where the first element corresponds to the C-MOVE request status and where the second element corresponds to the requested UID. |
Source code in pacsanini/net/c_move.py
def move(
local_node: Union[DicomNode, dict],
called_node: Union[DicomNode, dict],
*,
resources: List[str],
query_level: QueryLevel,
dest_node: Union[DicomNode, dict] = None,
directory: str = "",
sort_by: StorageSortKey = StorageSortKey.PATIENT,
start_time: Union[str, time] = None,
end_time: Union[str, time] = None,
db_session: Session = None,
callbacks: List[Callable[[Any], Any]] = None,
) -> Generator[Tuple[int, str], None, None]:
"""Move resources requested by the local_node to the
dest_node by querying the called_node.
Resources should be a list of string values corresponding
to patient ID values (in which case the query_level should
be set to PATIENT) or to study UID values (in which case
the query_level should be set to STUDY).
Parameters
----------
local_node : Union[DicomNode, dict]
The local DICOM node initiating the C-MOVE requests.
called_node : Union[DicomNode, dict]
The called DICOM node that contains the the DICOM
resources.
resources : List[str]
The list of DICOM resources to move.
query_level : QueryLevel
If the resources are study UID values, this should be
set to STUDY. If the resources are patient ID values,
this should be set to PATIENT.
dest_node : Union[DicomNode, dict]
The DICOM node to move the requested resources to. If
unset, this will be equal to the local_node.
directory : str
Specify the directory in which to store the DICOM files.
The default is the current directory.
sort_by : StorageKey
The method by which DICOM files should be stored. The
default is PATIENT
start_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should start. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The end_time parameter must also be
set.
end_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should end. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The start_time parameter must also be
set.
db_session : Session
The database session to use if move results are to be stored
in a database.
callbacks : List[Callable[[Any], Any]]
The callbacks to pass on to the storescp server.
Yields
------
Tuple[int, str]
Yield tuples where the first element corresponds to the
C-MOVE request status and where the second element corresponds
to the requested UID.
"""
if dest_node is None:
dest_node = local_node
if isinstance(local_node, dict):
local_node = DicomNode(**local_node)
if isinstance(called_node, dict):
called_node = DicomNode(**called_node)
if isinstance(dest_node, dict):
dest_node = DicomNode(**dest_node)
if not dest_node.has_port():
raise ValueError(f"{dest_node} does not have a set port.")
move_config = MoveConfig(
start_time=start_time, end_time=end_time, query_level=query_level
)
if QueryLevel.PATIENT == move_config.query_level:
root_model, query_lvl = PatientRootQueryRetrieveInformationModelMove, "PATIENT"
else:
root_model, query_lvl = StudyRootQueryRetrieveInformationModelMove, "STUDY"
with StoreSCPServer(
dest_node,
data_dir=directory,
sort_by=sort_by,
db_session=db_session,
callbacks=callbacks,
):
ae = AE(ae_title=local_node.aetitle)
ae.add_requested_context(root_model)
for uid in resources:
while not move_config.can_query():
sleep(20)
ds = Dataset()
ds.QueryRetrieveLevel = query_lvl
if query_lvl == "PATIENT":
ds.PatientID = uid
else:
ds.StudyInstanceUID = uid
assoc = None
try:
assoc = ae.associate(
called_node.ip, called_node.port, ae_title=called_node.aetitle
)
if assoc.is_established:
logger.info("Established association")
responses = assoc.send_c_move(ds, dest_node.aetitle, root_model)
for (status, _) in responses:
if status:
yield status, uid
else:
yield Status.STATUS_FAILURE, uid # pylint: disable=no-member
else:
logger.warning(
f"Failed to establish a connection with {called_node}."
)
yield Status.STATUS_FAILURE, uid # pylint: disable=no-member
finally:
if assoc is not None and assoc.is_alive():
assoc.release()
move_patients(local_node, called_node, *, patient_ids, dest_node=None, directory='', sort_by=<StorageSortKey.PATIENT: 'PATIENT'>, start_time=None, end_time=None, db_session=None)
#
Move patients requested by the local_node to the
dest_node by querying the called_node. Studies to move should be represented by their StudyInstanceUID values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
DicomNode |
The local DICOM node initiating the C-MOVE requests. |
required |
called_node |
DicomNode |
The called DICOM node that contains the the DICOM resources. |
required |
patient_ids |
List[str] |
The list of PatientID values to move. |
required |
dest_node |
DicomNode |
The DICOM node to move the requested resources to. If unset, this will be equal to the local_node. |
None |
directory |
str |
Specify the directory in which to store the DICOM files. The default is the current directory. |
'' |
sort_by |
StorageSortKey |
The method by which DICOM files should be stored. The default is PATIENT. |
<StorageSortKey.PATIENT: 'PATIENT'> |
start_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should start. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The end_time parameter must also be set. |
None |
end_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should end. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The start_time parameter must also be set. |
None |
db_session |
Session |
The database session to use if move results are to be stored in a database. |
None |
Returns:
| Type | Description |
|---|---|
Generator[Tuple[int, str], NoneType, NoneType] |
Yield tuples where the first element corresponds to the C-MOVE request status and where the second element corresponds to the requested UID. |
Source code in pacsanini/net/c_move.py
def move_patients(
local_node: DicomNode,
called_node: DicomNode,
*,
patient_ids: List[str],
dest_node: DicomNode = None,
directory: str = "",
sort_by: StorageSortKey = StorageSortKey.PATIENT,
start_time: Union[str, time] = None,
end_time: Union[str, time] = None,
db_session: Session = None,
) -> Generator[Tuple[int, str], None, None]:
"""Move patients requested by the local_node to the
dest_node by querying the called_node. Studies to move
should be represented by their StudyInstanceUID values.
Parameters
----------
local_node : DicomNode
The local DICOM node initiating the C-MOVE requests.
called_node : DicomNode
The called DICOM node that contains the the DICOM
resources.
patient_ids : List[str]
The list of PatientID values to move.
dest_node : DicomNode
The DICOM node to move the requested resources to. If
unset, this will be equal to the local_node.
directory : str
Specify the directory in which to store the DICOM files.
The default is the current directory.
sort_by : StorageKey
The method by which DICOM files should be stored. The
default is PATIENT.
start_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should start. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The end_time parameter must also be
set.
end_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should end. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The start_time parameter must also be
set.
db_session : Session
The database session to use if move results are to be stored
in a database.
Yields
------
Tuple[int, str]
Yield tuples where the first element corresponds to the
C-MOVE request status and where the second element corresponds
to the requested UID.
"""
yield from move(
local_node,
called_node,
resources=patient_ids,
query_level=QueryLevel.PATIENT,
dest_node=dest_node,
directory=directory,
sort_by=sort_by,
start_time=start_time,
end_time=end_time,
db_session=db_session,
)
move_studies(local_node, called_node, *, study_uids, dest_node=None, directory='', sort_by=<StorageSortKey.PATIENT: 'PATIENT'>, start_time=None, end_time=None, db_session=None)
#
Move studies requested by the local_node to the
dest_node by querying the called_node. Studies to move should be represented by their StudyInstanceUID values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_node |
DicomNode |
The local DICOM node initiating the C-MOVE requests. |
required |
called_node |
DicomNode |
The called DICOM node that contains the the DICOM resources. |
required |
study_uids |
List[str] |
The list of StudyInstanceUID values to move. |
required |
dest_node |
DicomNode |
The DICOM node to move the requested resources to. If unset, this will be equal to the local_node. |
None |
directory |
str |
Specify the directory in which to store the DICOM files. The default is the current directory. |
'' |
sort_by |
StorageSortKey |
The method by which DICOM files should be stored. The default is PATIENT. |
<StorageSortKey.PATIENT: 'PATIENT'> |
start_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should start. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The end_time parameter must also be set. |
None |
end_time |
Union[str, datetime.time] |
If set, specify the time of the day at which C-MOVE requests should end. If a string, it must be in ISO format (eg: HH, HH:MM, HH:MM:SS). The start_time parameter must also be set. |
None |
db_session |
Session |
The database session to use if move results are to be stored in a database. |
None |
Returns:
| Type | Description |
|---|---|
Generator[Tuple[int, str], NoneType, NoneType] |
Yield tuples where the first element corresponds to the C-MOVE request status and where the second element corresponds to the requested UID. |
Source code in pacsanini/net/c_move.py
def move_studies(
local_node: DicomNode,
called_node: DicomNode,
*,
study_uids: List[str],
dest_node: DicomNode = None,
directory: str = "",
sort_by: StorageSortKey = StorageSortKey.PATIENT,
start_time: Union[str, time] = None,
end_time: Union[str, time] = None,
db_session: Session = None,
) -> Generator[Tuple[int, str], None, None]:
"""Move studies requested by the local_node to the
dest_node by querying the called_node. Studies to move
should be represented by their StudyInstanceUID values.
Parameters
----------
local_node : DicomNode
The local DICOM node initiating the C-MOVE requests.
called_node : DicomNode
The called DICOM node that contains the the DICOM
resources.
study_uids : List[str]
The list of StudyInstanceUID values to move.
dest_node : DicomNode
The DICOM node to move the requested resources to. If
unset, this will be equal to the local_node.
directory : str
Specify the directory in which to store the DICOM files.
The default is the current directory.
sort_by : StorageKey
The method by which DICOM files should be stored. The
default is PATIENT.
start_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should start. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The end_time parameter must also be
set.
end_time : Union[str, time]
If set, specify the time of the day at which C-MOVE requests
should end. If a string, it must be in ISO format (eg:
HH, HH:MM, HH:MM:SS). The start_time parameter must also be
set.
db_session : Session
The database session to use if move results are to be stored
in a database.
Yields
------
Tuple[int, str]
Yield tuples where the first element corresponds to the
C-MOVE request status and where the second element corresponds
to the requested UID.
"""
yield from move(
local_node,
called_node,
resources=study_uids,
query_level=QueryLevel.STUDY,
dest_node=dest_node,
directory=directory,
sort_by=sort_by,
start_time=start_time,
end_time=end_time,
db_session=db_session,
)
c_store
#
The c_store module exposes methods that can be used to send DICOM
from a local node to a destination node over a C-STORE connection.
send_dicom(dcm_path, *, src_node, dest_node)
#
Send one or multiple DICOM files from the source node
to the dest node. If the dcm_path is a directory, non-DICOM files will be ignored.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dcm_path |
str |
The path to the DICOM file to send or the DICOM directory (in which case DICOM files will be collected recursively). |
required |
src_node |
Union[pacsanini.models.DicomNode, dict] |
The source DICOM node to use for sending the DICOM data. |
required |
dest_node |
Union[pacsanini.models.DicomNode, dict] |
The destination DICOM node to send the DICOM data to. |
required |
Returns:
| Type | Description |
|---|---|
Generator[Tuple[str, pydicom.dataset.Dataset], NoneType, NoneType] |
A 2-tuple corresponding to the DICOM file's path and the associated status of the C-STORE operation as a Dataset. |
Source code in pacsanini/net/c_store.py
def send_dicom(
dcm_path: str,
*,
src_node: Union[DicomNode, dict],
dest_node: Union[DicomNode, dict],
) -> Generator[Tuple[str, Dataset], None, None]:
"""Send one or multiple DICOM files from the source node
to the dest node. If the dcm_path is a directory, non-DICOM
files will be ignored.
Parameters
----------
dcm_path : str
The path to the DICOM file to send or the DICOM directory
(in which case DICOM files will be collected recursively).
src_node : Union[DicomNode, dict]
The source DICOM node to use for sending the DICOM data.
dest_node : Union[DicomNode, dict]
The destination DICOM node to send the DICOM data to.
Yields
------
Generator[Tuple[str, Dataset], None, None]
A 2-tuple corresponding to the DICOM file's path and the
associated status of the C-STORE operation as a Dataset.
"""
if isinstance(src_node, dict):
src_node = DicomNode(**src_node)
if isinstance(dest_node, dict):
dest_node = DicomNode(**dest_node)
if os.path.isfile(dcm_path):
dcm_files = [dcm_path]
else:
dcm_files = []
append = dcm_files.append
for root, _, files in os.walk(dcm_path):
for fname in files:
append(os.path.join(root, fname))
ae = AE(ae_title=src_node.aetitle)
transfer_syntax = [
ExplicitVRLittleEndian,
ImplicitVRLittleEndian,
DeflatedExplicitVRLittleEndian,
ExplicitVRBigEndian,
]
for ctx in StoragePresentationContexts:
ae.add_requested_context(ctx.abstract_syntax, transfer_syntax)
assoc: Association = None
try:
assoc = ae.associate(dest_node.ip, dest_node.port, ae_title=dest_node.aetitle)
if assoc.is_established:
for path in dcm_files:
try:
dcm = dcmread(path)
yield path, assoc.send_c_store(dcm)
except InvalidDicomError:
pass
finally:
if assoc is not None:
assoc.release()
storescp
#
The storescp module provides methods and classes that can be used to
instantiate and run storescp server instances. In addition, custom DICOM event status responses are defined: UNABLE_TO_DECODE (0xC215) and UNABLE_TO_PROCESS (0xC216).
StoreSCPServer
#
The StoreSCPServer class provides a way to run a storescp server
that can be used to receive DICOM files and write them locally.
Attributes:
| Name | Type | Description |
|---|---|---|
node |
Union[DicomNode, dict] |
The DICOM node information to use when running the server. |
data_dir |
str |
The path to the top-level directory where DICOM files should be written to. The default is the current directory. |
sort_by |
StorageSortKey |
The method by which DICOM files should be written to disk. |
db_session |
Session |
Optional. If specified, received studies will be parsed and persisted to the provided database. The default is None. |
callbacks |
List[Callable[[Any], Any]] |
If set, pass a list of callables that will be called on the DICOM file after it is received and persisted to disk. |
run(self, block=False)
#
Run the storescp server in a non-blocking way.
Source code in pacsanini/net/storescp.py
def run(self, block: bool = False):
"""Run the storescp server in a non-blocking way."""
if self.scp is not None:
raise RuntimeError(f"A current SCP instance is already running for {self}.")
ae = AE(ae_title=self.node.aetitle)
ae.supported_contexts = AllStoragePresentationContexts
if self.data_dir:
os.makedirs(self.data_dir, exist_ok=True)
logger.debug(f"Starting SCP server: {self}")
if not block:
self.scp = ae.start_server(
("", self.node.port),
block=False,
evt_handlers=self.handlers,
ae_title=self.node.aetitle,
)
else:
ae.start_server(
("", self.node.port),
block=True,
evt_handlers=self.handlers,
ae_title=self.node.aetitle,
)
shutdown(self)
#
Shutdown the running scp server.
Source code in pacsanini/net/storescp.py
def shutdown(self):
"""Shutdown the running scp server."""
if self.scp is not None:
logger.debug(f"Stopping SCP server: {self}")
self.scp.shutdown()
self.scp = None
default_store_handle(event, data_dir='', sort_by=<StorageSortKey.PATIENT: 'PATIENT'>, db_session=None, callbacks=None)
#
Handle a C-STORE request event by writing the received DICOM file
to the data_dir in the way specified by sort_by.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event |
Event |
The C-STORE event to handle. |
required |
data_dir |
str |
The directory to write results under. |
'' |
sort_by |
StorageSortKey |
The organization to follow when writing DICOM files to disk. |
<StorageSortKey.PATIENT: 'PATIENT'> |
callbacks |
List[Callable[[Any], Any]] |
If supplied pass the received DICOM file to the callable as a positional argument (the first one) to each one of the callables for processing. |
None |
Returns:
| Type | Description |
|---|---|
int |
The reception status. |
Source code in pacsanini/net/storescp.py
def default_store_handle(
event: Event,
data_dir: str = "",
sort_by: StorageSortKey = StorageSortKey.PATIENT,
db_session: Session = None,
callbacks: List[Callable[[Any], Any]] = None,
) -> int:
"""Handle a C-STORE request event by writing the received DICOM file
to the data_dir in the way specified by sort_by.
Parameters
----------
event : Event
The C-STORE event to handle.
data_dir : str
The directory to write results under.
sort_by : StorageSortKey
The organization to follow when writing DICOM files to disk.
callbacks : List[Callable[[Any], Any]]
If supplied pass the received DICOM file to the callable as
a positional argument (the first one) to each one of the
callables for processing.
Returns
-------
int
The reception status.
"""
try:
ds: Dataset = event.dataset
ds.file_meta = event.file_meta
except: # pylint: disable=bare-except
logger.warning("Unable to decode received DICOM")
return Status.UNABLE_TO_DECODE # pylint: disable=no-member
if StorageSortKey.PATIENT == sort_by:
dest = os.path.join(
data_dir,
ds.PatientID,
ds.StudyInstanceUID,
ds.SeriesInstanceUID,
ds.SOPInstanceUID,
)
elif StorageSortKey.STUDY == sort_by:
dest = os.path.join(
data_dir, ds.StudyInstanceUID, ds.SeriesInstanceUID, ds.SOPInstanceUID
)
else:
dest = os.path.join(data_dir, ds.SOPInstanceUID)
dest += ".dcm"
try:
dcm_dir = os.path.dirname(dest)
os.makedirs(dcm_dir, exist_ok=True)
ds.save_as(dest, write_like_original=False)
logger.info(f"{ds.SOPInstanceUID} is persisted.")
except OSError:
logger.warning(f"Failed to write {ds.StudyInstanceUID} to disk")
return Status.UNABLE_TO_PROCESS # pylint: disable=no-member
if db_session is not None:
try:
add_image(db_session, ds, filepath=dest)
update_retrieved_study(db_session, ds.StudyInstanceUID)
except exc.SQLAlchemyError as err:
logger.warning(f"Failed to update database due to {err}")
return Status.UNABLE_TO_RECORD # pylint: disable=no-member
if callbacks is not None:
for func in callbacks:
func(ds)
return 0x0000
run_server(node, data_dir='', sort_by=<StorageSortKey.PATIENT: 'PATIENT'>, callbacks=None, block=False)
#
Instantiate and run a storescp server using the provided
configuration. The server will run in a detached thread.
node : Union[DicomNode, dict] The DICOM node information to use when running the server. data_dir : str The path to the top-level directory where DICOM files should be written to. The default is the current directory. sort_by : StorageSortKey The method by which DICOM files should be written to disk. callbacks : List[Callable[[Any], Any]] If set, pass a list of callables that will be called on the DICOM file after it is received and persisted to disk. block : bool If False, the default, run the storescp server in a different thread. If True, the running server will block the current thread. In this case, a KeyboardInterrupt is needed to stop the server.
Returns:
| Type | Description |
|---|---|
Optional[pacsanini.net.storescp.StoreSCPServer] |
The running StoreSCPServer instance if block is set to False (in which case you must subsequently call the shudown method) or None if the server is in blocking mode. |
Source code in pacsanini/net/storescp.py
def run_server(
node: Union[DicomNode, dict],
data_dir: str = "",
sort_by: StorageSortKey = StorageSortKey.PATIENT,
callbacks: List[Callable[[Any], Any]] = None,
block: bool = False,
) -> Union[StoreSCPServer, None]:
"""Instantiate and run a storescp server using the provided
configuration. The server will run in a detached thread.
node : Union[DicomNode, dict]
The DICOM node information to use when running the server.
data_dir : str
The path to the top-level directory where DICOM files should
be written to. The default is the current directory.
sort_by : StorageSortKey
The method by which DICOM files should be written to disk.
callbacks : List[Callable[[Any], Any]]
If set, pass a list of callables that will be called on the
DICOM file after it is received and persisted to disk.
block : bool
If False, the default, run the storescp server in a different
thread. If True, the running server will block the current
thread. In this case, a KeyboardInterrupt is needed to stop
the server.
Returns
-------
Union[StoreSCPServer, None]
The running StoreSCPServer instance if block is set to False
(in which case you must subsequently call the shudown method)
or None if the server is in blocking mode.
"""
server = StoreSCPServer(
node, data_dir=data_dir, sort_by=sort_by, callbacks=callbacks
)
if block:
server.run(block=True)
return None
server.run(block=False)
return server