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]