Python plugin for Orthanc¶

print('Hello world!')

import orthanc
import pprint

def OnRest(output, uri, **request):
    pprint.pprint(request)
    print('Accessing uri: %s' % uri)
    output.AnswerBuffer('ok\n', 'text/plain')

orthanc.RegisterRestCallback('/(to)(t)o', OnRest)
orthanc.RegisterRestCallback('/tata', OnRest)

import orthanc

def OnChange(changeType, level, resource):
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:
        print('Started')

        with open('/tmp/sample.dcm', 'rb') as f:
            orthanc.RestApiPost('/instances', f.read())

    elif changeType == orthanc.ChangeType.ORTHANC_STOPPED:
        print('Stopped')

    elif changeType == orthanc.ChangeType.NEW_INSTANCE:
        print('A new instance was uploaded: %s' % resource)

orthanc.RegisterOnChangeCallback(OnChange)

import orthanc
import threading

def OnChange(changeType, level, resource):
    # One can safely invoke the "orthanc" module in this function
    orthanc.LogWarning("Hello world")

def _OnChange(changeType, level, resource):
    # Invoke the actual "OnChange()" function in a separate thread
    t = threading.Timer(0, function = OnChange, args = (changeType, level, resource))
    t.start()

orthanc.RegisterOnChangeCallback(_OnChange)

import orthanc
import json
import pprint

def OnStoredInstance(dicom, instanceId):
    print('Received instance %s of size %d (transfer syntax %s, SOP class UID %s)' % (
        instanceId, dicom.GetInstanceSize(),
        dicom.GetInstanceMetadata('TransferSyntax'),
        dicom.GetInstanceMetadata('SopClassUid')))

    # Print the origin information
    if dicom.GetInstanceOrigin() == orthanc.InstanceOrigin.DICOM_PROTOCOL:
        print('This instance was received through the DICOM protocol')
    elif dicom.GetInstanceOrigin() == orthanc.InstanceOrigin.REST_API:
        print('This instance was received through the REST API')

    # Print the DICOM tags
    pprint.pprint(json.loads(dicom.GetInstanceSimplifiedJson()))

orthanc.RegisterOnStoredInstanceCallback(OnStoredInstance)

import io
import orthanc
import pydicom

def DecodeInstance(output, uri, **request):
    if request['method'] == 'GET':
        # Retrieve the instance ID from the regular expression (*)
        instanceId = request['groups'][0]
        # Get the content of the DICOM file
        f = orthanc.GetDicomForInstance(instanceId)
        # Parse it using pydicom
        dicom = pydicom.dcmread(io.BytesIO(f))
        # Return a string representation the dataset to the caller
        output.AnswerBuffer(str(dicom), 'text/plain')
    else:
        output.SendMethodNotAllowed('GET')

orthanc.RegisterRestCallback('/pydicom/(.*)', DecodeInstance)  # (*)

import orthanc

def OnChange(changeType, level, resourceId):
    if changeType == orthanc.ChangeType.STABLE_STUDY:
        print('Stable study: %s' % resourceId)
        orthanc.RestApiPost('/modalities/sample/store', resourceId)

orthanc.RegisterOnChangeCallback(OnChange)

import orthanc

def OnChange(changeType, level, resourceId):
    if changeType == orthanc.ChangeType.STABLE_STUDY:
        print('Stable study: %s' % resourceId)
        orthanc.RestApiPostAfterPlugins('/dicom-web/servers/sample/store', resourceId)

orthanc.RegisterOnChangeCallback(OnChange)

from PIL import Image
import io
import orthanc

def DecodeInstance(output, uri, **request):
    if request['method'] == 'GET':
        # Retrieve the instance ID from the regular expression (*)
        instanceId = request['groups'][0]

        # Render the instance, then open it in Python using PIL/Pillow
        png = orthanc.RestApiGet('/instances/%s/rendered' % instanceId)
        image = Image.open(io.BytesIO(png))

        # Downsize the image as a 64x64 thumbnail
        image.thumbnail((64, 64), Image.ANTIALIAS)

        # Save the thumbnail as JPEG, then send the buffer to the caller
        jpeg = io.BytesIO()
        image.save(jpeg, format = "JPEG", quality = 80)
        jpeg.seek(0)
        output.AnswerBuffer(jpeg.read(), 'text/plain')

    else:
        output.SendMethodNotAllowed('GET')

orthanc.RegisterRestCallback('/pydicom/(.*)', DecodeInstance)  # (*)

import inspect
import numbers
import orthanc

# Loop over the members of the "orthanc" module
for (name, obj) in inspect.getmembers(orthanc):
    if inspect.isroutine(obj):
        print('Function %s():\n  Documentation: %s\n' % (name, inspect.getdoc(obj)))

    elif inspect.isclass(obj):
        print('Class %s:\n  Documentation: %s' % (name, inspect.getdoc(obj)))

        # Loop over the members of the class
        for (subname, subobj) in inspect.getmembers(obj):
            if isinstance(subobj, numbers.Number):
                print('  - Enumeration value %s: %s' % (subname, subobj))
            elif (not subname.startswith('_') and
                  inspect.ismethoddescriptor(subobj)):
                print('  - Method %s(): %s' % (subname, inspect.getdoc(subobj)))
        print('')

import orthanc
import threading

TIMER = None

def Hello():
    global TIMER
    TIMER = None
    orthanc.LogWarning("In Hello()")
    # Do stuff...
    TIMER = threading.Timer(1, Hello)  # Re-schedule after 1 second
    TIMER.start()

def OnChange(changeType, level, resource):
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:
        orthanc.LogWarning("Starting the scheduler")
        Hello()

    elif changeType == orthanc.ChangeType.ORTHANC_STOPPED:
        if TIMER != None:
            orthanc.LogWarning("Stopping the scheduler")
            TIMER.cancel()

orthanc.RegisterOnChangeCallback(OnChange)

import json
import orthanc
import re

# Get the path in the REST API to the given resource that was returned
# by a call to "/tools/find"
def GetPath(resource):
    if resource['Type'] == 'Patient':
        return '/patients/%s' % resource['ID']
    elif resource['Type'] == 'Study':
        return '/studies/%s' % resource['ID']
    elif resource['Type'] == 'Series':
        return '/series/%s' % resource['ID']
    elif resource['Type'] == 'Instance':
        return '/instances/%s' % resource['ID']
    else:
        raise Exception('Unknown resource level')

def FindWithMetadata(output, uri, **request):
    # The "/tools/find" route expects a POST method
    if request['method'] != 'POST':
        output.SendMethodNotAllowed('POST')
    else:
        # Parse the query provided by the user, and backup the "Expand" field
        query = json.loads(request['body'])

        if 'Expand' in query:
            originalExpand = query['Expand']
        else:
            originalExpand = False

        # Call the core "/tools/find" route
        query['Expand'] = True
        answers = orthanc.RestApiPost('/tools/find', json.dumps(query))

        # Loop over the matching resources
        filteredAnswers = []
        for answer in json.loads(answers):
            try:
                # Read the metadata that is associated with the resource
                metadata = json.loads(orthanc.RestApiGet('%s/metadata?expand' % GetPath(answer)))

                # Check whether the metadata matches the regular expressions
                # that were provided in the "Metadata" field of the user request
                isMetadataMatch = True
                if 'Metadata' in query:
                    for (name, pattern) in query['Metadata'].items():
                        if name in metadata:
                            value = metadata[name]
                        else:
                            value = ''

                        if re.match(pattern, value) == None:
                            isMetadataMatch = False
                            break

                # If all the metadata matches the provided regular
                # expressions, add the resource to the filtered answers
                if isMetadataMatch:
                    if originalExpand:
                        answer['Metadata'] = metadata
                        filteredAnswers.append(answer)
                    else:
                        filteredAnswers.append(answer['ID'])
            except:
                # The resource was deleted since the call to "/tools/find"
                pass

        # Return the filtered answers in the JSON format
        output.AnswerBuffer(json.dumps(filteredAnswers, indent = 3), 'application/json')

orthanc.RegisterRestCallback('/tools/find', FindWithMetadata)

import json
import orthanc

def GetStudyDate(study):
    if 'StudyDate' in study['MainDicomTags']:
        return study['MainDicomTags']['StudyDate']
    else:
        return ''

def SortStudiesByDate(output, uri, **request):
    if request['method'] == 'GET':
        # Retrieve all the studies
        studies = json.loads(orthanc.RestApiGet('/studies?expand'))

        # Sort the studies according to the "StudyDate" DICOM tag
        studies = sorted(studies, key = GetStudyDate)

        # Read the limit/offset arguments provided by the user
        offset = 0
        if 'offset' in request['get']:
            offset = int(request['get']['offset'])

        limit = 0
        if 'limit' in request['get']:
            limit = int(request['get']['limit'])

        # Truncate the list of studies
        if limit == 0:
            studies = studies[offset : ]
        else:
            studies = studies[offset : offset + limit]

        # Return the truncated list of studies
        output.AnswerBuffer(json.dumps(studies), 'application/json')
    else:
        output.SendMethodNotAllowed('GET')

orthanc.RegisterRestCallback('/sort-studies', SortStudiesByDate)

import StringIO
import json
import orthanc
import xlwt

def CreateExcelReport(output, uri, **request):
    if request['method'] != 'GET' :
        output.SendMethodNotAllowed('GET')
    else:
        # Create an Excel writer
        excel = xlwt.Workbook()
        sheet = excel.add_sheet('Studies')

        # Loop over the studies stored in Orthanc
        row = 0
        studies = orthanc.RestApiGet('/studies?expand')
        for study in json.loads(studies):
            sheet.write(row, 0, study['PatientMainDicomTags'].get('PatientID'))
            sheet.write(row, 1, study['PatientMainDicomTags'].get('PatientName'))
            sheet.write(row, 2, study['MainDicomTags'].get('StudyDescription'))
            row += 1

        # Serialize the Excel workbook to a string, and return it to the caller
        # https://stackoverflow.com/a/15649139/881731
        b = StringIO.StringIO()
        excel.save(b)
        output.AnswerBuffer(b.getvalue(), 'application/vnd.ms-excel')

orthanc.RegisterRestCallback('/report.xls', CreateExcelReport)

import orthanc
import pprint

def Filter(uri, **request):
    print('User trying to access URI: %s' % uri)
    pprint.pprint(request)
    return True  # False to forbid access

orthanc.RegisterIncomingHttpRequestFilter(Filter)

import json
import orthanc
import requests

def Filter(uri, **request):
    body = {
        'uri' : uri,
        'headers' : request['headers']
    }
    r = requests.post('http://localhost:8000/authorize',
                      data = json.dumps(body))
    return r.json() ['granted']  # Must be a Boolean

orthanc.RegisterIncomingHttpRequestFilter(Filter)

import json
import orthanc

# Create a dictionary mapping the numeric values in enumeration
# "orthanc.ValueRepresentation" to the name of the corresponding VR
VR_NAMES = {}
for name in dir(orthanc.ValueRepresentation):
    if not name.startswith('_'):
        value = getattr(orthanc.ValueRepresentation, name)
        VR_NAMES[value] = name

entry = orthanc.LookupDictionary('PatientID')

orthanc.LogWarning('Entry in the dictionary: %s' %
                   json.dumps(entry, indent = 4, sort_keys = True))

orthanc.LogWarning('Name of the value representation: %s' %
                   VR_NAMES[entry['ValueRepresentation']])

import json
import orthanc

def OnChange(changeType, level, resource):
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:
        tags = {
            'SOPClassUID' : '1.2.840.10008.5.1.4.1.1.1',
            'PatientID' : 'HELLO',
            'PatientName' : 'WORLD',
        }

        with open('Lena.png', 'rb') as f:
            img = orthanc.UncompressImage(f.read(), orthanc.ImageFormat.PNG)

        s = orthanc.CreateDicom(json.dumps(tags), img, orthanc.CreateDicomFlags.GENERATE_IDENTIFIERS)

        with open('/tmp/sample.dcm', 'wb') as f:
            f.write(s)

        dicom = orthanc.CreateDicomInstance(s)
        frame = dicom.GetInstanceDecodedFrame(0)
        print('Size of the frame: %dx%d' % (frame.GetImageWidth(), frame.GetImageHeight()))

orthanc.RegisterOnChangeCallback(OnChange)

import json
import PIL.Image
import PIL.ImageDraw
import orthanc

URL = 'http://hg.orthanc-server.com/orthanc-tests/raw-file/Orthanc-1.11.0/Database/LenaTwiceWithFragments.dcm'
USERNAME = ''
PASSWORD = ''

def OnChange(changeType, level, resource):
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:

        # (1) Download a sample DICOM instance and decode it
        orthanc.LogWarning('Downloading: %s' % URL)
        lena = orthanc.HttpGet(URL, USERNAME, PASSWORD)

        dicom = orthanc.CreateDicomInstance(lena)
        orthanc.LogWarning('Number of frames: %d' % dicom.GetInstanceFramesCount())

        # (2) Access the first frame of the instance as a PIL image
        frame = dicom.GetInstanceDecodedFrame(0)
        size = (frame.GetImageWidth(), frame.GetImageHeight())

        if frame.GetImagePixelFormat() == orthanc.PixelFormat.RGB24:
            mode = 'RGB'
        else:
            raise Exception('Unsupported pixel format')

        image = PIL.Image.frombuffer(mode, size, frame.GetImageBuffer(), 'raw', mode, 0, 1)

        # (3) Draw a red cross over the PIL image
        draw = PIL.ImageDraw.Draw(image)
        draw.line((0, 0) + image.size, fill=(255,0,0), width=10)
        draw.line((0, image.size[1], image.size[0], 0), fill=(255,0,0), width=10)

        # (4) Convert back the modified PIL image to an Orthanc image
        buf = image.tobytes()
        a = orthanc.CreateImageFromBuffer(frame.GetImagePixelFormat(), image.size[0], image.size[1],
                                          len(buf) / image.size[1], buf)

        # (5) Create and upload a new DICOM instance with the modified frame
        tags = {
            'SOPClassUID' : '1.2.840.10008.5.1.4.1.1.1',
            'PatientID' : 'HELLO',
            'PatientName' : 'WORLD',
        }

        s = orthanc.CreateDicom(json.dumps(tags), a, orthanc.CreateDicomFlags.GENERATE_IDENTIFIERS)
        orthanc.RestApiPost('/instances', s)
        orthanc.LogWarning('Image successfully modified and uploaded!')

orthanc.RegisterOnChangeCallback(OnChange)

import json
import orthanc
import pprint

def OnFind(answers, query, issuerAet, calledAet):
    print('Received incoming C-FIND request from %s:' % issuerAet)

    answer = {}
    for i in range(query.GetFindQuerySize()):
        print('  %s (%04x,%04x) = [%s]' % (query.GetFindQueryTagName(i),
                                           query.GetFindQueryTagGroup(i),
                                           query.GetFindQueryTagElement(i),
                                           query.GetFindQueryValue(i)))
        answer[query.GetFindQueryTagName(i)] = ('HELLO%d-%s' % (i, query.GetFindQueryValue(i)))

    answers.FindAddAnswer(orthanc.CreateDicom(
        json.dumps(answer), None, orthanc.CreateDicomFlags.NONE))

def OnMove(**request):
    orthanc.LogWarning('C-MOVE request to be handled in Python: %s' %
                       json.dumps(request, indent = 4, sort_keys = True))

    # To indicate a failure in the processing, one can raise an exception:
    #   raise Exception('Cannot handle C-MOVE')

orthanc.RegisterFindCallback(OnFind)
orthanc.RegisterMoveCallback(OnMove)

import json
import orthanc
import os

# Path to the directory containing the DICOM worklists
# https://hg.orthanc-server.com/orthanc/file/Orthanc-1.11.0/OrthancServer/Plugins/Samples/ModalityWorklists/WorklistsDatabase
WORKLIST_DIR = '/tmp/WorklistsDatabase'

def OnWorklist(answers, query, issuerAet, calledAet):
    print('Received incoming C-FIND worklist request from %s:' % issuerAet)

    # Get a memory buffer containing the DICOM instance
    dicom = query.WorklistGetDicomQuery()

    # Get the DICOM tags in the JSON format from the binary buffer
    jsonTags = json.loads(orthanc.DicomBufferToJson(
        dicom, orthanc.DicomToJsonFormat.SHORT, orthanc.DicomToJsonFlags.NONE, 0))

    orthanc.LogWarning('C-FIND worklist request to be handled in Python: %s' %
                       json.dumps(jsonTags, indent = 4, sort_keys = True))

    # Loop over the available DICOM worklists
    for path in os.listdir(WORKLIST_DIR):
        if os.path.splitext(path) [1] == '.wl':
            with open(os.path.join(WORKLIST_DIR, path), 'rb') as f:
                content = f.read()
                
                # Test whether the query matches the current worklist
                if query.WorklistIsMatch(content):
                    orthanc.LogWarning('Matching worklist: %s' % path)
                    answers.WorklistAddAnswer(query, content)

orthanc.RegisterWorklistCallback(OnWorklist)

import json
import orthanc
import pynetdicom

def HandleStore(event):
    orthanc.LogWarning('Handling C-STORE using pynetdicom')
    orthanc.RestApiPost('/instances', event.request.DataSet.getvalue())
    return 0x0000

ae = pynetdicom.AE()
ae.supported_contexts = pynetdicom.AllStoragePresentationContexts

SCP = None

def OnChange(changeType, level, resource):
    global SCP
    
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:
        port = json.loads(orthanc.GetConfiguration()).get('DicomPort', 4242)
        
        SCP = ae.start_server(('', port), block = False, evt_handlers = [
            (pynetdicom.evt.EVT_C_STORE, HandleStore),
        ])
        
        orthanc.LogWarning('DICOM server using pynetdicom has started')

    elif changeType == orthanc.ChangeType.ORTHANC_STOPPED:
        orthanc.LogWarning('Stopping pynetdicom')
        SCP.shutdown()

orthanc.RegisterOnChangeCallback(OnChange)

import orthanc

def OnChange(changeType, level, resource):
    if changeType == orthanc.ChangeType.ORTHANC_STARTED:
        try:
            print(orthanc.RestApiGet('/nope'))
        except ValueError as e:
            # Raised in releases <= 3.2 of the plugin (doesn't occur in releases >= 3.3)
            print(e)
        except orthanc.OrthancException as e:
            # Raised in releases >= 3.3 of the plugin (fails with releases <= 3.2)
            print(e)
            print(e.args[0])  # Error code of Orthanc (cf. "orthanc.ErrorCode" enumeration)
            print(e.args[1])  # Description of the error
            print(e.args[0] == orthanc.ErrorCode.UNKNOWN_RESOURCE)  # Returns "True"

orthanc.RegisterOnChangeCallback(OnChange)

import orthanc
import os

def GetPath(uuid, contentType):
    # Returns the path where to store the given attachment
    return 'attachment-%d-%s' % (contentType, uuid)

def OnCreate(uuid, contentType, data):
    with open(GetPath(uuid, contentType), 'wb') as f:
        f.write(data)

def OnRead(uuid, contentType):
    with open(GetPath(uuid, contentType), 'rb') as f:
        return f.read()

def OnRemove(uuid, contentType):
    os.remove(GetPath(uuid, contentType))

orthanc.RegisterStorageArea(OnCreate, OnRead, OnRemove)

from io import BytesIO

from pydicom import dcmread, dcmwrite
from pydicom.filebase import DicomFileLike

import orthanc

# from https://pydicom.github.io/pydicom/stable/auto_examples/memory_dataset.html
def write_dataset_to_bytes(dataset):
    with BytesIO() as buffer:
        memory_dataset = DicomFileLike(buffer)
        dcmwrite(memory_dataset, dataset)
        memory_dataset.seek(0)
        return memory_dataset.read()

def ReceivedInstanceCallback(receivedDicom, origin):
    if origin == orthanc.InstanceOrigin.REST_API:
        orthanc.LogWarning('DICOM instance received from the REST API')
    elif origin == orthanc.InstanceOrigin.DICOM_PROTOCOL:
        orthanc.LogWarning('DICOM instance received from the DICOM protocol')
    
    dataset = dcmread(BytesIO(receivedDicom))

    if dataset.PatientID.startswith('001-'):
        orthanc.LogWarning('Discard instance')
        return orthanc.ReceivedInstanceAction.DISCARD, None

    elif dataset.PatientID.startswith('002-'):
        orthanc.LogWarning('Store source instance as it is')
        return orthanc.ReceivedInstanceAction.KEEP_AS_IS, None

    else:
        orthanc.LogWarning('Modify the source instance')
        dataset.PatientName = str(dataset.PatientName).upper()
        dataset.PatientID = '002-' + dataset.PatientID
        dataset.InstitutionName = "MY INSTITUTION"
        return orthanc.ReceivedInstanceAction.MODIFY, write_dataset_to_bytes(dataset)

orthanc.RegisterReceivedInstanceCallback(ReceivedInstanceCallback)

import json
import orthanc


# this script accepts 3 instances from STORESCU and then, rejects the next ones

storeScuInstanceCounter = 0

def FilterIncomingCStoreInstance(receivedDicom):
    # The list ofvalid status codes for DIMSE C-STORE can be found:
    # https://dicom.nema.org/medical/Dicom/2021e/output/chtml/part04/sect_B.2.3.html

    global storeScuInstanceCounter

    origin = receivedDicom.GetInstanceOrigin()

    if origin == orthanc.InstanceOrigin.DICOM_PROTOCOL:  # should always be true in the CStore callback !

        remoteAet = receivedDicom.GetInstanceRemoteAet()
        
        if remoteAet == "STORESCU":
            storeScuInstanceCounter += 1

        if storeScuInstanceCounter > 3:
            # Non-zero return value: The DICOM instance is discarded
            return 0xA700
    
    return 0  # Success: Accept the DICOM instance

orthanc.RegisterIncomingCStoreInstanceFilter(FilterIncomingCStoreInstance)

import math
import orthanc
import time

# CPU-intensive computation taking about 4 seconds
def SlowComputation():
    start = time.time()
    for i in range(1000):
        for j in range(30000):
            math.sqrt(float(j))
    end = time.time()
    duration = (end - start)
    return 'computation done in %.03f seconds\n' % duration

def OnRest(output, uri, **request):
    answer = SlowComputation()
    output.AnswerBuffer(answer, 'text/plain')

orthanc.RegisterRestCallback('/computation', OnRest)

import math
import multiprocessing
import orthanc
import signal
import time

# CPU-intensive computation taking about 4 seconds
# (same code as above)
def SlowComputation():
    start = time.time()
    for i in range(1000):
        for j in range(30000):
            math.sqrt(float(j))
    end = time.time()
    duration = (end - start)
    return 'computation done in %.03f seconds\n' % duration

# Ignore CTRL+C in the slave processes
def Initializer():
    signal.signal(signal.SIGINT, signal.SIG_IGN)

# Create a pool of 4 slave Python interpreters
POOL = multiprocessing.Pool(4, initializer = Initializer)

def OnRest(output, uri, **request):
    # Offload the call to "SlowComputation" onto one slave process.
    # The GIL is unlocked until the slave sends its answer back.
    answer = POOL.apply(SlowComputation)
    output.AnswerBuffer(answer, 'text/plain')

orthanc.RegisterRestCallback('/computation', OnRest)

import pydicom
import io

def OffloadedDicomParsing(dicom):
    # No access to the "orthanc" library here, as we are in the slave process
    dataset = pydicom.dcmread(io.BytesIO(dicom))
    return str(dataset)

def OnRest(output, uri, **request):
    # The call to "orthanc.RestApiGet()" is only possible in the master process
    dicom = orthanc.RestApiGet('/instances/19816330-cb02e1cf-df3a8fe8-bf510623-ccefe9f5/file')
    answer = POOL.apply(OffloadedDicomParsing, args = (dicom, ))
    output.AnswerBuffer(answer, 'text/plain')

import json
import multiprocessing
import orthanc
import requests
import signal

TOKEN = orthanc.GenerateRestApiAuthorizationToken()

def SlaveProcess():
    r = requests.get('http://localhost:8042/instances',
                     headers = { 'Authorization' : TOKEN })
    return json.dumps(r.json())

def Initializer():
    signal.signal(signal.SIGINT, signal.SIG_IGN)

POOL = multiprocessing.Pool(4, initializer = Initializer)

def OnRest(output, uri, **request):
    answer = POOL.apply(SlaveProcess)
    output.AnswerBuffer(answer, 'text/plain')

orthanc.RegisterRestCallback('/computation', OnRest)