Python Usage

How to use anonapi from within a python script

anonapi.client.WebAPIClient is the main class to use when interacting with an IDIS server web API from python. To use it in a python file:

from anonapi.client import WebAPIClient

client = WebAPIClient(
    hostname="https://umcradanonp11.umcn.nl/sandbox",
    username="z123sandbox",
    token="token",
)

# Get some information on first few jobs
jobs_info = client.get("get_jobs")

Testing

The anonapi.testresources module can be used to generate mock responses without the need for a live server:

from anonapi.testresources import (
    MockAnonClientTool,
    JobInfoFactory,
    RemoteAnonServerFactory,
    JobStatus,
)

# Create a mock client tool that returns information about jobs
tool = MockAnonClientTool()

# Client tool methods need a server to query. This is mocked too here
mock_server = RemoteAnonServerFactory()
tool.get_job_info(server=mock_server, job_id=1)    # Returns realistic JobInfo response

# You can set the responses that mock client cycles through:
some_responses = [
    JobInfoFactory(status=JobStatus.DONE),
    JobInfoFactory(status=JobStatus.ERROR),
    JobInfoFactory(status=JobStatus.INACTIVE),
]
tool = MockAnonClientTool(responses=some_responses)
tool.get_job_info(mock_server, job_id=1).status # returns JobStatus.DONE
tool.get_job_info(mock_server, job_id=1).status # returns JobStatus.ERROR
tool.get_job_info(mock_server, job_id=1).status # returns JobStatus.INACTIVE
tool.get_job_info(mock_server, job_id=1).status # returns JobStatus.DONE again

# You can set any of the JobInfo fields on the JobInfo instances:
JobInfoFactory(project_name='project1',
               destination_path='\\server\share\folder',
               priority=50)

Exceptions

All exceptions raised in anonapi derive from anonapi.exceptions.AnonAPIError. Catching that will allow you to handle them:

from anonapi.exceptions import AnonAPIError
from anonapi.client import AnonClientTool

tool = AnonClientTool('user','token')
server = RemoveAnonServer('a_server','https://aserver')
try:
    tool.get_server_status(server)

except AnonAPIError as e:
    print(f"Something went wrong but its anonapi's fault. Look: {e}")

Examples

These examples are taken from the code in the Examples directory

Anonymize from IDC

IDC is the hospital medical image server

from anonapi.client import WebAPIClient


def anonymize_files_from_idc():
    """Create an IDIS job that pulls files from the hospital information system"""

    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/sandbox",
        username="z123sandbox",
        token="token",
    )

    # Create a job that takes data from the IDC (formally IMPAX) directly
    anon_name = "TEST_NAME_02"
    anon_id = "02"
    sid = "123.12335.3353.36464.343435677"  # study UID
    destination_path = r"\\umcsanfsclp01\radng_imaging\temptest_output"
    idc_job_info = client.post(
        "create_job",
        source_type="WADO",
        source_name="IDC_WADO",
        source_instance_id=sid,
        anonymizedpatientname=anon_name,
        anonymizedpatientid=anon_id,
        destination_type="PATH",
        project_name="Wetenschap-Algemeen",
        destination_path=destination_path,
        description="A test idc job",
    )
    print(
        f"Succesfully created a job in {client}, job_id={idc_job_info['job_id']}"
    )


if __name__ == "__main__":
    anonymize_files_from_idc()

Anonymize from network share

from anonapi.client import WebAPIClient


def anonymize_files_from_share():
    """Create an IDIS job that pulls files from a network share"""

    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/sandbox",
        username="z123sandbox",
        token="token",
    )

    # Create a job that takes data from a source on a network root_path, and writes
    # the anonymized data to a destination that is also a network root_path. Note
    # that the network root_path should be accessible for the anonymization server
    # for this to work.
    anon_name = "TEST_NAME_01"
    anon_id = "01"
    source_path = r"\\umcsanfsclp01\radng_imaging\temp\test"
    destination_path = r"\\umcsanfsclp01\radng_imaging\temptest_output"
    network_job_info = client.post(
        "create_job",
        source_type="PATH",
        source_path=source_path,
        destination_type="PATH",
        project_name="Wetenschap-Algemeen",
        destination_path=destination_path,
        anonymizedpatientname=anon_name,
        anonymizedpatientid=anon_id,
        description="A test root_path job",
    )
    # new_job_info response contains extended info on the new job that has
    # just been created
    print(
        f"Succesfully created a job in {client}, job_id={network_job_info['job_id']}"
    )


if __name__ == "__main__":
    anonymize_files_from_share()

Filter on SOPClass

SOPClassUID is the DICOM ‘image type’

from anonapi.client import WebAPIClient


def anonymize_files_sop_class_filter():
    """Create an IDIS job that pulls files from the hospital information system"""

    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/sandbox",
        username="z123sandbox",
        token="token",
    )

    # Create a job that takes data from the IDC (formally IMPAX) directly
    # and allow only files that match the given SOPClassUIDs. For a full list of
    # possible SOPClassUIDs see https://www.dicomlibrary.com/dicom/sop/
    anon_name = "TEST_NAME_03"
    anon_id = "03"
    sid = "123.12335.3353.36464.343435677"  # study UID
    destination_path = r"\\umcsanfsclp01\radng_imaging\temptest_output"
    idc_job_info = client.post(
        "create_job",
        source_type="WADO",
        source_name="IDC_WADO",
        source_instance_id=sid,
        source_sop_class_filter_list="1.2.840.10008.5.1.4.1.1.88.67, "
        "1.2.840.10008.5.1.4.1.1.7",
        anonymizedpatientname=anon_name,
        anonymizedpatientid=anon_id,
        destination_type="PATH",
        project_name="Wetenschap-Algemeen",
        destination_path=destination_path,
        description="A test idc job",
    )
    print(
        f"Succesfully created a job in {client}, job_id={idc_job_info['job_id']}"
    )


if __name__ == "__main__":
    anonymize_files_sop_class_filter()

Cancel job

from anonapi.client import WebAPIClient


def cancel_job():
    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/sandbox",
        username="z123sandbox",
        token="token",
    )

    # cancel job 100
    client.post("cancel_job", job_id=100)


if __name__ == "__main__":
    cancel_job()

Get job status

from anonapi.client import WebAPIClient


def get_job_status():
    """Get information about a number of jobs"""

    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/p01",
        username="z123sandbox",
        token="token",
    )

    # get the status for 3 specific jobs
    job_status_list = client.get(
        "get_jobs_list_extended", job_ids=[53769, 53770, 53771]
    )
    print(f"found status for {len(job_status_list)} jobs in list:")
    print(job_status_list)


if __name__ == "__main__":
    get_job_status()

Modify jobs

from anonapi.client import WebAPIClient


def modify_jobs():
    """Modify several jobs

    Notes
    -----
    For a full list of the job fields that can be modified see example
    'get_api_definition'
    """

    # Create a client that will talk to the web API
    client = WebAPIClient(
        hostname="https://umcradanonp11.umcn.nl/sandbox",
        username="z123sandbox",
        token="token",
    )

    job_ids = [1, 2, 3]
    for job_id in job_ids:
        client.post(
            "modify_job",
            job_id=job_id,
            source_path=r"\\umcsanfsclp01\radng_imaging\temp\modified\test",
        )


if __name__ == "__main__":
    modify_jobs()