Python plugin for Orthanc¶

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 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)

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)