Skip to content

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