Mercurial > hg > orthanc-tests
changeset 3:2dbba2e6aa4b
reorganization
| author | Sebastien Jodogne <s.jodogne@gmail.com> |
|---|---|
| date | Wed, 17 Jun 2015 10:03:49 +0200 |
| parents | a15734e7f0af |
| children | 292a46fe374c |
| files | ExternalCommandThread.py GenerateConfigurationForTests.py Run.py Start.sh Tests/ExternalCommandThread.py Tests/Run.py Tests/Tests.py Tests/Toolbox.py Toolbox.py |
| diffstat | 9 files changed, 548 insertions(+), 379 deletions(-) [+] |
line wrap: on
line diff
--- a/ExternalCommandThread.py Wed Jun 17 09:56:44 2015 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,50 +0,0 @@ -#!/usr/bin/python - -# Orthanc - A Lightweight, RESTful DICOM Store -# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics -# Department, University Hospital of Liege, Belgium -# -# This program is free software: you can redistribute it and/or -# modify it under the terms of the GNU General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. - - -import os -import signal -import subprocess -import threading - -class ExternalCommandThread: - @staticmethod - def ExternalCommandFunction(arg, stop_event, command, env): - external = subprocess.Popen(command, env = env) - - while (not stop_event.is_set()): - error = external.poll() - if error != None: - # http://stackoverflow.com/a/1489838/881731 - os._exit(-1) - stop_event.wait(0.1) - - print 'Stopping the external command' - external.terminate() - - def __init__(self, command, env = None): - self.thread_stop = threading.Event() - self.thread = threading.Thread(target = self.ExternalCommandFunction, - args = (10, self.thread_stop, command, env)) - self.daemon = True - self.thread.start() - - def stop(self): - self.thread_stop.set() - self.thread.join()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/GenerateConfigurationForTests.py Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,83 @@ +#!/usr/bin/python + +# Orthanc - A Lightweight, RESTful DICOM Store +# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics +# Department, University Hospital of Liege, Belgium +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import argparse +import os +import re +import socket +import subprocess +import sys + +## +## Parse the command-line arguments +## + +parser = argparse.ArgumentParser(description = 'Generate the configuration file for the ' + + 'instance of Orthanc to be checked against the integration tests.') + +parser.add_argument('--target', + default = 'IntegrationTestsConfiguration.json', + help = 'Configuration file to generate') + +parser.add_argument('--force', + help = 'Overwrite the file even if it already exists', + action = 'store_true') + +args = parser.parse_args() + + +## +## Check whether the file can be overwritten +## + +if os.path.exists(args.target) and not args.force: + print(""" +WARNING: The file %s will be overwritten! + +Are you sure ["yes" to go on]?""" % args.target) + + if sys.stdin.readline().strip() != 'yes': + print('Aborting...') + exit(0) + + +## +## Generate the configuration file +## + + +# Retrieve the IP address of the localhost +ip = socket.gethostbyname(socket.gethostname()) + +subprocess.check_call([ 'Orthanc', '--config=%s' % args.target ]) + +with open(args.target, 'r') as f: + config = f.read() + +config = re.sub(r'("DicomAet"\s*:)\s*".*?"', r'\1 "ORTHANC"', config) +config = re.sub(r'("RemoteAccessAllowed"\s*:)\s*false', r'\1 true', config) +config = re.sub(r'("AuthenticationEnabled"\s*:)\s*false', r'\1 true', config) +config = re.sub(r'("RegisteredUsers"\s*:)\s*{', r'\1 { "alice" : "orthanctest"', config) +config = re.sub(r'("DicomModalities"\s*:)\s*{', r'\1 { "orthanc" : [ "%s", "%s", %d ]' % + ('ORTHANCTEST', ip, 5001), config) + +with open(args.target, 'wt') as f: + f.write(config) +
--- a/Run.py Wed Jun 17 09:56:44 2015 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,137 +0,0 @@ -#!/usr/bin/python - -# Orthanc - A Lightweight, RESTful DICOM Store -# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics -# Department, University Hospital of Liege, Belgium -# -# This program is free software: you can redistribute it and/or -# modify it under the terms of the GNU General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. - - -# sudo docker run --rm -t -i -v `pwd`:/tmp/tests:ro -p 5000:8042 -p 5001:4242 --entrypoint python jodogne/orthanc-tests /tmp/tests/Run.py --force - - - -import re -import sys -import argparse -import subprocess -import unittest - -from ExternalCommandThread import * -from Toolbox import * -from Tests import * - - -## -## Parse the command-line arguments -## - -parser = argparse.ArgumentParser(description = 'Run the integration tests on some instance of Orthanc.') -parser.add_argument('--server', - default = GetDockerHostAddress(), - help = 'Address of the Orthanc server to test') -parser.add_argument('--aet', - default = 'ORTHANC', - help = 'AET of the Orthanc instance to test') -parser.add_argument('--dicom', - type = int, - default = 4242, - help = 'DICOM port of the Orthanc instance to test') -parser.add_argument('--rest', - type = int, - default = 8042, - help = 'Port to the REST API') -parser.add_argument('--username', - default = 'alice', - help = 'Username to the REST API') -parser.add_argument('--password', - default = 'orthanctest', - help = 'Password to the REST API') -parser.add_argument('--force', help = 'Do not warn the user', - action = 'store_true') - -args = parser.parse_args() - -if not args.force: - print(""" -WARNING: This test will remove all the content of your -Orthanc instance running on %s! - -Are you sure ["yes" to go on]?""" % args.server) - - if sys.stdin.readline().strip() != 'yes': - print('Aborting...') - exit(0) - - - -## -## Generate the configuration file for the anciliary instance of -## Orthanc -## - -CONFIG = '/tmp/Configuration.json' -subprocess.check_call([ 'Orthanc', '--config=%s' % CONFIG ]) - -with open(CONFIG, 'r') as f: - config = f.read() - -config = re.sub(r'("StorageDirectory"\s*:)\s*".*?"', r'\1 "/tmp/OrthancStorage"', config) -config = re.sub(r'("IndexDirectory"\s*:)\s*".*?"', r'\1 "/tmp/OrthancStorage"', config) -config = re.sub(r'("DicomAet"\s*:)\s*".*?"', r'\1 "ORTHANCTEST"', config) -config = re.sub(r'("RemoteAccessAllowed"\s*:)\s*false', r'\1 true', config) -config = re.sub(r'("AuthenticationEnabled"\s*:)\s*false', r'\1 true', config) -config = re.sub(r'("RegisteredUsers"\s*:)\s*{', r'\1 { "alice" : [ "orthanctest" ]', config) -config = re.sub(r'("DicomModalities"\s*:)\s*{', r'\1 { "orthanc" : [ "%s", "%s", "%s" ]' % - (args.aet, args.server, args.dicom), config) - -localOrthanc = ExternalCommandThread([ - 'Orthanc', CONFIG, #'--verbose' - ]) - - -LOCAL = DefineOrthanc(aet = 'ORTHANCTEST') -REMOTE = DefineOrthanc(url = 'http://%s:%d/' % (args.server, args.rest), - username = args.username, - password = args.password, - aet = args.aet, - dicomPort = args.dicom) - - - -print('Parameters of the instance of Orthanc to test:') -print(REMOTE) -print('') - - -print('Waiting for the internal Orthanc to start...') -while True: - try: - DoGet(LOCAL, '/instances') - break - except: - time.sleep(0.1) - - -try: - print('Starting the tests...') - SetOrthancParameters(LOCAL, REMOTE) - unittest.main(argv = [ sys.argv[0] ]) #argv = args) - -finally: - # The tests have stopped or "Ctrl-C" has been hit - try: - localOrthanc.stop() - except: - pass
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Start.sh Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,3 @@ +#!/bin/bash + +sudo docker run --rm -t -i -v `pwd`:/tmp/tests:ro -p 5000:8042 -p 5001:4242 --entrypoint python jodogne/orthanc-tests /tmp/tests/Tests/Run.py --force $*
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Tests/ExternalCommandThread.py Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,50 @@ +#!/usr/bin/python + +# Orthanc - A Lightweight, RESTful DICOM Store +# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics +# Department, University Hospital of Liege, Belgium +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import os +import signal +import subprocess +import threading + +class ExternalCommandThread: + @staticmethod + def ExternalCommandFunction(arg, stop_event, command, env): + external = subprocess.Popen(command, env = env) + + while (not stop_event.is_set()): + error = external.poll() + if error != None: + # http://stackoverflow.com/a/1489838/881731 + os._exit(-1) + stop_event.wait(0.1) + + print 'Stopping the external command' + external.terminate() + + def __init__(self, command, env = None): + self.thread_stop = threading.Event() + self.thread = threading.Thread(target = self.ExternalCommandFunction, + args = (10, self.thread_stop, command, env)) + self.daemon = True + self.thread.start() + + def stop(self): + self.thread_stop.set() + self.thread.join()
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Tests/Run.py Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,136 @@ +#!/usr/bin/python + +# Orthanc - A Lightweight, RESTful DICOM Store +# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics +# Department, University Hospital of Liege, Belgium +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import re +import sys +import argparse +import subprocess +import unittest +import pprint + +from ExternalCommandThread import * +from Toolbox import * +from Tests import * + + +## +## Parse the command-line arguments +## + +parser = argparse.ArgumentParser(description = 'Run the integration tests on some instance of Orthanc.') +parser.add_argument('--server', + default = GetDockerHostAddress(), + help = 'Address of the Orthanc server to test') +parser.add_argument('--aet', + default = 'ORTHANC', + help = 'AET of the Orthanc instance to test') +parser.add_argument('--dicom', + type = int, + default = 4242, + help = 'DICOM port of the Orthanc instance to test') +parser.add_argument('--rest', + type = int, + default = 8042, + help = 'Port to the REST API') +parser.add_argument('--username', + default = 'alice', + help = 'Username to the REST API') +parser.add_argument('--password', + default = 'orthanctest', + help = 'Password to the REST API') +parser.add_argument('--force', help = 'Do not warn the user', + action = 'store_true') + +args = parser.parse_args() + +if not args.force: + print(""" +WARNING: This test will remove all the content of your +Orthanc instance running on %s! + +Are you sure ["yes" to go on]?""" % args.server) + + if sys.stdin.readline().strip() != 'yes': + print('Aborting...') + exit(0) + + + +## +## Generate the configuration file for the anciliary instance of +## Orthanc +## + +CONFIG = '/tmp/Configuration.json' +subprocess.check_call([ 'Orthanc', '--config=%s' % CONFIG ]) + +with open(CONFIG, 'r') as f: + config = f.read() + +config = re.sub(r'("StorageDirectory"\s*:)\s*".*?"', r'\1 "/tmp/OrthancStorage"', config) +config = re.sub(r'("IndexDirectory"\s*:)\s*".*?"', r'\1 "/tmp/OrthancStorage"', config) +config = re.sub(r'("DicomAet"\s*:)\s*".*?"', r'\1 "ORTHANCTEST"', config) +config = re.sub(r'("RemoteAccessAllowed"\s*:)\s*false', r'\1 true', config) +config = re.sub(r'("AuthenticationEnabled"\s*:)\s*false', r'\1 true', config) +config = re.sub(r'("RegisteredUsers"\s*:)\s*{', r'\1 { "alice" : [ "orthanctest" ]', config) +config = re.sub(r'("DicomModalities"\s*:)\s*{', r'\1 { "orthanc" : [ "%s", "%s", "%s" ]' % + (args.aet, args.server, args.dicom), config) + +localOrthanc = ExternalCommandThread([ + 'Orthanc', CONFIG, #'--verbose' + ]) + + +LOCAL = DefineOrthanc(aet = 'ORTHANCTEST') +REMOTE = DefineOrthanc(url = 'http://%s:%d/' % (args.server, args.rest), + username = args.username, + password = args.password, + aet = args.aet, + dicomPort = args.dicom) + + + +print('Parameters of the instance of Orthanc to test:') +pprint.pprint(REMOTE) +print('') + + +print('Waiting for the internal Orthanc to start...') +while True: + try: + DoGet(LOCAL, '/instances') + break + except: + time.sleep(0.1) + + +try: + print('\nStarting the tests...') + SetOrthancParameters(LOCAL, REMOTE) + unittest.main(argv = [ sys.argv[0] ]) #argv = args) + +finally: + print('\nDone') + + # The tests have stopped or "Ctrl-C" has been hit + try: + localOrthanc.stop() + except: + pass
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Tests/Tests.py Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,84 @@ +# Orthanc - A Lightweight, RESTful DICOM Store +# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics +# Department, University Hospital of Liege, Belgium +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import unittest + +from Toolbox import * + +_LOCAL = None +_REMOTE = None + + +def SetOrthancParameters(local, remote): + global _LOCAL, _REMOTE + _LOCAL = local + _REMOTE = remote + + +class Orthanc(unittest.TestCase): + def setUp(self): + DropOrthanc(_LOCAL) + DropOrthanc(_REMOTE) + + def test_system(self): + self.assertTrue('Version' in DoGet(_REMOTE, '/system')) + self.assertEqual('0', DoGet(_REMOTE, '/statistics')['TotalDiskSize']) + self.assertEqual('0', DoGet(_REMOTE, '/statistics')['TotalUncompressedSize']) + + def test_upload(self): + u = UploadInstance(_REMOTE, 'DummyCT.dcm') + self.assertEqual('Success', u['Status']) + u = UploadInstance(_REMOTE, 'DummyCT.dcm') + self.assertEqual('AlreadyStored', u['Status']) + self.assertEqual(1, len(DoGet(_REMOTE, '/patients'))) + self.assertEqual(1, len(DoGet(_REMOTE, '/studies'))) + self.assertEqual(1, len(DoGet(_REMOTE, '/series'))) + self.assertEqual(1, len(DoGet(_REMOTE, '/instances'))) + + i = DoGet(_REMOTE, '/instances/%s/simplified-tags' % u['ID']) + self.assertEqual('20070101', i['StudyDate']) + + + def test_rest_grid(self): + i = UploadInstance(_REMOTE, 'DummyCT.dcm')['ID'] + instance = DoGet(_REMOTE, '/instances/%s' % i) + self.assertEqual(i, instance['ID']) + self.assertEqual('1.2.840.113619.2.176.2025.1499492.7040.1171286242.109', + instance['MainDicomTags']['SOPInstanceUID']) + + series = DoGet(_REMOTE, '/series/%s' % instance['ParentSeries']) + self.assertEqual('1.2.840.113619.2.176.2025.1499492.7391.1171285944.394', + series['MainDicomTags']['SeriesInstanceUID']) + + study = DoGet(_REMOTE, '/studies/%s' % series['ParentStudy']) + self.assertEqual('1.2.840.113619.2.176.2025.1499492.7391.1171285944.390', + study['MainDicomTags']['StudyInstanceUID']) + + patient = DoGet(_REMOTE, '/patients/%s' % study['ParentPatient']) + self.assertEqual('ozp00SjY2xG', + patient['MainDicomTags']['PatientID']) + + dicom = DoGet(_REMOTE, '/instances/%s/file' % instance['ID']) + self.assertEqual(2472, len(dicom)) + self.assertEqual('3e29b869978b6db4886355a2b1132124', ComputeMD5(dicom)) + self.assertEqual(1, len(DoGet(_REMOTE, '/instances/%s/frames' % i))) + self.assertEqual('TWINOW', DoGet(_REMOTE, '/instances/%s/simplified-tags' % i)['StationName']) + self.assertEqual('TWINOW', DoGet(_REMOTE, '/instances/%s/tags' % i)['0008,1010']['Value']) + + +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Tests/Toolbox.py Wed Jun 17 10:03:49 2015 +0200 @@ -0,0 +1,192 @@ +#!/usr/bin/python + +# Orthanc - A Lightweight, RESTful DICOM Store +# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics +# Department, University Hospital of Liege, Belgium +# +# This program is free software: you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import hashlib +import httplib2 +import json +import os.path +import re +import subprocess +import time +import zipfile + +from PIL import Image +from urllib import urlencode + + +HERE = os.path.dirname(__file__) + + +# http://stackoverflow.com/a/1313868/881731 +try: + from cStringIO import StringIO +except: + from StringIO import StringIO + + +def DefineOrthanc(url = 'http://localhost:8042', + username = None, + password = None, + aet = 'ORTHANC', + dicomPort = 4242): + m = re.match(r'(http|https)://([^:]+):([^@]+)@([^@]+)', url) + if m != None: + url = m.groups()[0] + '://' + m.groups()[3] + username = m.groups()[1] + password = m.groups()[2] + + if not url.endswith('/'): + url += '/' + + return { + 'Url' : url, + 'Username' : username, + 'Password' : password, + 'DicomAet' : aet, + 'DicomPort' : dicomPort + } + + +def _SetupCredentials(orthanc, http): + if (orthanc['Username'] != None and + orthanc['Password'] != None): + http.add_credentials(orthanc['Username'], orthanc['Password']) + + +def DoGet(orthanc, uri, data = {}, body = None, headers = {}): + d = '' + if len(data.keys()) > 0: + d = '?' + urlencode(data) + + http = httplib2.Http() + _SetupCredentials(orthanc, http) + + resp, content = http.request(orthanc['Url'] + uri + d, 'GET', body = body, + headers = headers) + if not (resp.status in [ 200 ]): + raise Exception(resp.status) + else: + try: + return json.loads(content) + except: + return content + +def _DoPutOrPost(orthanc, uri, method, data, contentType, headers): + http = httplib2.Http() + _SetupCredentials(orthanc, http) + + if isinstance(data, str): + body = data + if len(contentType) != 0: + headers['content-type'] = contentType + else: + body = json.dumps(data) + headers['content-type'] = 'application/json' + + headers['expect'] = '' + + resp, content = http.request(orthanc['Url'] + uri, method, + body = body, + headers = headers) + if not (resp.status in [ 200, 302 ]): + raise Exception(resp.status) + else: + try: + return json.loads(content) + except: + return content + +def DoDelete(orthanc, uri): + http = httplib2.Http() + _SetupCredentials(orthanc, http) + + resp, content = http.request(orthanc['Url'] + uri, 'DELETE') + if not (resp.status in [ 200 ]): + raise Exception(resp.status) + else: + try: + return json.loads(content) + except: + return content + +def DoPut(orthanc, uri, data = {}, contentType = ''): + return DoPutOrPost(orthanc, uri, 'PUT', data, contentType) + +def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}): + return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers) + +def UploadInstance(orthanc, filename): + global HERE + p = os.path.join(HERE, '..', 'Database', filename) + f = open(p, 'rb') + d = f.read() + f.close() + return DoPost(orthanc, '/instances', d, 'application/dicom') + +def UploadFolder(orthanc, path): + global HERE + p = os.path.join(HERE, 'Database', path) + for i in os.listdir(p): + try: + UploadInstance(orthanc, os.path.join(path, i)) + except: + pass + +def DropOrthanc(orthanc): + # Reset the Lua callbacks + DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua') + + DoDelete(orthanc, '/exports') + + for s in DoGet(orthanc, '/patients'): + DoDelete(orthanc, '/patients/%s' % s) + +def ComputeMD5(data): + m = hashlib.md5() + m.update(data) + return m.hexdigest() + +def GetImage(orthanc, uri): + # http://www.pythonware.com/library/pil/handbook/introduction.htm + data = DoGet(orthanc, uri) + return Image.open(StringIO(data)) + +def GetArchive(orthanc, uri): + # http://stackoverflow.com/a/1313868/881731 + s = DoGet(orthanc, uri) + return zipfile.ZipFile(StringIO(s), "r") + +def IsDefinedInLua(orthanc, name): + s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua') + return (s.strip() != 'nil') + +def WaitEmpty(orthanc): + while True: + if len(DoGet(orthanc, '/instances')) == 0: + return + time.sleep(0.1) + +def GetDockerHostAddress(): + route = subprocess.check_output([ '/sbin/ip', 'route' ]) + m = re.search(r'default via ([0-9.]+)', route) + if m == None: + return 'localhost' + else: + return m.groups()[0]
--- a/Toolbox.py Wed Jun 17 09:56:44 2015 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,192 +0,0 @@ -#!/usr/bin/python - -# Orthanc - A Lightweight, RESTful DICOM Store -# Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics -# Department, University Hospital of Liege, Belgium -# -# This program is free software: you can redistribute it and/or -# modify it under the terms of the GNU General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, but -# WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see <http://www.gnu.org/licenses/>. - - - -from PIL import Image -from urllib import urlencode -import hashlib -import httplib2 -import json -import os.path -import re -import subprocess -import time -import zipfile - - -HERE = os.path.dirname(__file__) - - -# http://stackoverflow.com/a/1313868/881731 -try: - from cStringIO import StringIO -except: - from StringIO import StringIO - - -def DefineOrthanc(url = 'http://localhost:8042', - username = None, - password = None, - aet = 'ORTHANC', - dicomPort = 4242): - m = re.match(r'(http|https)://([^:]+):([^@]+)@([^@]+)', url) - if m != None: - url = m.groups()[0] + '://' + m.groups()[3] - username = m.groups()[1] - password = m.groups()[2] - - if not url.endswith('/'): - url += '/' - - return { - 'Url' : url, - 'Username' : username, - 'Password' : password, - 'DicomAet' : aet, - 'DicomPort' : dicomPort - } - - -def _SetupCredentials(orthanc, http): - if (orthanc['Username'] != None and - orthanc['Password'] != None): - http.add_credentials(orthanc['Username'], orthanc['Password']) - - -def DoGet(orthanc, uri, data = {}, body = None, headers = {}): - d = '' - if len(data.keys()) > 0: - d = '?' + urlencode(data) - - http = httplib2.Http() - _SetupCredentials(orthanc, http) - - resp, content = http.request(orthanc['Url'] + uri + d, 'GET', body = body, - headers = headers) - if not (resp.status in [ 200 ]): - raise Exception(resp.status) - else: - try: - return json.loads(content) - except: - return content - -def _DoPutOrPost(orthanc, uri, method, data, contentType, headers): - http = httplib2.Http() - _SetupCredentials(orthanc, http) - - if isinstance(data, str): - body = data - if len(contentType) != 0: - headers['content-type'] = contentType - else: - body = json.dumps(data) - headers['content-type'] = 'application/json' - - headers['expect'] = '' - - resp, content = http.request(orthanc['Url'] + uri, method, - body = body, - headers = headers) - if not (resp.status in [ 200, 302 ]): - raise Exception(resp.status) - else: - try: - return json.loads(content) - except: - return content - -def DoDelete(orthanc, uri): - http = httplib2.Http() - _SetupCredentials(orthanc, http) - - resp, content = http.request(orthanc['Url'] + uri, 'DELETE') - if not (resp.status in [ 200 ]): - raise Exception(resp.status) - else: - try: - return json.loads(content) - except: - return content - -def DoPut(orthanc, uri, data = {}, contentType = ''): - return DoPutOrPost(orthanc, uri, 'PUT', data, contentType) - -def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}): - return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers) - -def UploadInstance(orthanc, filename): - global HERE - p = os.path.join(HERE, 'Database', filename) - f = open(p, 'rb') - d = f.read() - f.close() - return DoPost(orthanc, '/instances', d, 'application/dicom') - -def UploadFolder(orthanc, path): - global HERE - p = os.path.join(HERE, 'Database', path) - for i in os.listdir(p): - try: - UploadInstance(orthanc, os.path.join(path, i)) - except: - pass - -def DropOrthanc(orthanc): - # Reset the Lua callbacks - DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua') - - DoDelete(orthanc, '/exports') - - for s in DoGet(orthanc, '/patients'): - DoDelete(orthanc, '/patients/%s' % s) - -def ComputeMD5(data): - m = hashlib.md5() - m.update(data) - return m.hexdigest() - -def GetImage(orthanc, uri): - # http://www.pythonware.com/library/pil/handbook/introduction.htm - data = DoGet(orthanc, uri) - return Image.open(StringIO(data)) - -def GetArchive(orthanc, uri): - # http://stackoverflow.com/a/1313868/881731 - s = DoGet(orthanc, uri) - return zipfile.ZipFile(StringIO(s), "r") - -def IsDefinedInLua(orthanc, name): - s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua') - return (s.strip() != 'nil') - -def WaitEmpty(orthanc): - while True: - if len(DoGet(orthanc, '/instances')) == 0: - return - time.sleep(0.1) - -def GetDockerHostAddress(): - route = subprocess.check_output([ '/sbin/ip', 'route' ]) - m = re.search(r'default via ([0-9.]+)', route) - if m == None: - return 'localhost' - else: - return m.groups()[0]