Mercurial > hg > orthanc-tests
diff Tests/Toolbox.py @ 3:2dbba2e6aa4b
reorganization
author | Sebastien Jodogne <s.jodogne@gmail.com> |
---|---|
date | Wed, 17 Jun 2015 10:03:49 +0200 |
parents | Toolbox.py@08dadea8f40a |
children | 292a46fe374c |
line wrap: on
line diff
--- /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]