Mercurial > hg > orthanc-tests

comparison Toolbox.py @ 1:08dadea8f40a

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
start
author Sebastien Jodogne <s.jodogne@gmail.com>
date Tue, 16 Jun 2015 18:49:29 +0200
parents cc43b57242a4
children
comparison
equal deleted inserted replaced
0:cc43b57242a4 1:08dadea8f40a
1 #!/usr/bin/python 1 #!/usr/bin/python
2 2
3 # sudo docker run --rm -v `pwd`/Toolbox.py:/tmp/Toolbox.py:ro --entrypoint python jodogne/orthanc-tests /tmp/Toolbox.py 3 # Orthanc - A Lightweight, RESTful DICOM Store
4 # Copyright (C) 2012-2015 Sebastien Jodogne, Medical Physics
5 # Department, University Hospital of Liege, Belgium
6 #
7 # This program is free software: you can redistribute it and/or
8 # modify it under the terms of the GNU General Public License as
9 # published by the Free Software Foundation, either version 3 of the
10 # License, or (at your option) any later version.
11 #
12 # This program is distributed in the hope that it will be useful, but
13 # WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 # General Public License for more details.
16 #
17 # You should have received a copy of the GNU General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
4 19
5 20
21
22 from PIL import Image
23 from urllib import urlencode
6 import hashlib 24 import hashlib
7 import httplib2 25 import httplib2
8 import json 26 import json
9 import os.path 27 import os.path
10 from PIL import Image 28 import re
29 import subprocess
30 import time
11 import zipfile 31 import zipfile
12 import time
13 from urllib import urlencode
14 32
33
34 HERE = os.path.dirname(__file__)
15 35
16 36
17 # http://stackoverflow.com/a/1313868/881731 37 # http://stackoverflow.com/a/1313868/881731
18 try: 38 try:
19 from cStringIO import StringIO 39 from cStringIO import StringIO
20 except: 40 except:
21 from StringIO import StringIO 41 from StringIO import StringIO
22 42
23 43
44 def DefineOrthanc(url = 'http://localhost:8042',
45 username = None,
46 password = None,
47 aet = 'ORTHANC',
48 dicomPort = 4242):
49 m = re.match(r'(http|https)://([^:]+):([^@]+)@([^@]+)', url)
50 if m != None:
51 url = m.groups()[0] + '://' + m.groups()[3]
52 username = m.groups()[1]
53 password = m.groups()[2]
24 54
25 def CreateOrthanc(url = 'http://localhost:8042',
26 username = None,
27 password = None):
28 if not url.endswith('/'): 55 if not url.endswith('/'):
29 url += '/' 56 url += '/'
30 57
31 return [ url, username, password ] 58 return {
59 'Url' : url,
60 'Username' : username,
61 'Password' : password,
62 'DicomAet' : aet,
63 'DicomPort' : dicomPort
64 }
32 65
33 66
34 def _SetupCredentials(orthanc, http): 67 def _SetupCredentials(orthanc, http):
35 if orthanc[1] != None and orthanc[2] != None: 68 if (orthanc['Username'] != None and
36 http.add_credentials(orthanc[1], orthanc[2]) 69 orthanc['Password'] != None):
70 http.add_credentials(orthanc['Username'], orthanc['Password'])
37 71
38 72
39 def DoGet(orthanc, uri, data = {}, body = None, headers = {}): 73 def DoGet(orthanc, uri, data = {}, body = None, headers = {}):
40 d = '' 74 d = ''
41 if len(data.keys()) > 0: 75 if len(data.keys()) > 0:
42 d = '?' + urlencode(data) 76 d = '?' + urlencode(data)
43 77
44 http = httplib2.Http() 78 http = httplib2.Http()
45 _SetupCredentials(orthanc, http) 79 _SetupCredentials(orthanc, http)
46 80
47 resp, content = http.request(orthanc[0] + uri + d, 'GET', body = body, 81 resp, content = http.request(orthanc['Url'] + uri + d, 'GET', body = body,
48 headers = headers) 82 headers = headers)
49 if not (resp.status in [ 200 ]): 83 if not (resp.status in [ 200 ]):
50 raise Exception(resp.status) 84 raise Exception(resp.status)
51 else: 85 else:
52 try: 86 try:
66 body = json.dumps(data) 100 body = json.dumps(data)
67 headers['content-type'] = 'application/json' 101 headers['content-type'] = 'application/json'
68 102
69 headers['expect'] = '' 103 headers['expect'] = ''
70 104
71 resp, content = http.request(orthanc[0] + uri, method, 105 resp, content = http.request(orthanc['Url'] + uri, method,
72 body = body, 106 body = body,
73 headers = headers) 107 headers = headers)
74 if not (resp.status in [ 200, 302 ]): 108 if not (resp.status in [ 200, 302 ]):
75 raise Exception(resp.status) 109 raise Exception(resp.status)
76 else: 110 else:
81 115
82 def DoDelete(orthanc, uri): 116 def DoDelete(orthanc, uri):
83 http = httplib2.Http() 117 http = httplib2.Http()
84 _SetupCredentials(orthanc, http) 118 _SetupCredentials(orthanc, http)
85 119
86 resp, content = http.request(orthanc[0] + uri, 'DELETE') 120 resp, content = http.request(orthanc['Url'] + uri, 'DELETE')
87 if not (resp.status in [ 200 ]): 121 if not (resp.status in [ 200 ]):
88 raise Exception(resp.status) 122 raise Exception(resp.status)
89 else: 123 else:
90 try: 124 try:
91 return json.loads(content) 125 return json.loads(content)
97 131
98 def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}): 132 def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}):
99 return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers) 133 return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers)
100 134
101 def UploadInstance(orthanc, filename): 135 def UploadInstance(orthanc, filename):
102 p = os.path.join(HERE, DICOM_DB, filename) 136 global HERE
137 p = os.path.join(HERE, 'Database', filename)
103 f = open(p, 'rb') 138 f = open(p, 'rb')
104 d = f.read() 139 d = f.read()
105 f.close() 140 f.close()
106 return DoPost(orthanc, '/instances', d, 'application/dicom') 141 return DoPost(orthanc, '/instances', d, 'application/dicom')
107 142
108 def UploadFolder(orthanc, path): 143 def UploadFolder(orthanc, path):
109 p = os.path.join(HERE, DICOM_DB, path) 144 global HERE
110 for i in os.listdir(p): 145 p = os.path.join(HERE, 'Database', path)
111 try: 146 for i in os.listdir(p):
112 UploadInstance(orthanc, os.path.join(path, i)) 147 try:
113 except: 148 UploadInstance(orthanc, os.path.join(path, i))
114 pass 149 except:
150 pass
115 151
116 def DropOrthanc(orthanc): 152 def DropOrthanc(orthanc):
117 # Reset the Lua callbacks 153 # Reset the Lua callbacks
118 DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua') 154 DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua')
119 155
135 def GetArchive(orthanc, uri): 171 def GetArchive(orthanc, uri):
136 # http://stackoverflow.com/a/1313868/881731 172 # http://stackoverflow.com/a/1313868/881731
137 s = DoGet(orthanc, uri) 173 s = DoGet(orthanc, uri)
138 return zipfile.ZipFile(StringIO(s), "r") 174 return zipfile.ZipFile(StringIO(s), "r")
139 175
140 def IsDefinedInLua(name): 176 def IsDefinedInLua(orthanc, name):
141 s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua') 177 s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua')
142 return (s.strip() != 'nil') 178 return (s.strip() != 'nil')
143 179
144 def WaitEmpty(): 180 def WaitEmpty(orthanc):
145 while True: 181 while True:
146 if len(orthanc, DoGet('/instances')) == 0: 182 if len(DoGet(orthanc, '/instances')) == 0:
147 return 183 return
148 time.sleep(0.1) 184 time.sleep(0.1)
149 185
150 186 def GetDockerHostAddress():
151 print DoGet(CreateOrthanc('http://192.168.215.82:8042'), '/system') 187 route = subprocess.check_output([ '/sbin/ip', 'route' ])
188 m = re.search(r'default via ([0-9.]+)', route)
189 if m == None:
190 return 'localhost'
191 else:
192 return m.groups()[0]