Mercurial > hg > orthanc-tests

comparison Tests/Toolbox.py @ 3:2dbba2e6aa4b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
reorganization
author Sebastien Jodogne <s.jodogne@gmail.com>
date Wed, 17 Jun 2015 10:03:49 +0200
parents Toolbox.py@08dadea8f40a
children 292a46fe374c
comparison
equal deleted inserted replaced
2:a15734e7f0af 3:2dbba2e6aa4b
1 #!/usr/bin/python
2
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/>.
19
20
21 import hashlib
22 import httplib2
23 import json
24 import os.path
25 import re
26 import subprocess
27 import time
28 import zipfile
29
30 from PIL import Image
31 from urllib import urlencode
32
33
34 HERE = os.path.dirname(__file__)
35
36
37 # http://stackoverflow.com/a/1313868/881731
38 try:
39 from cStringIO import StringIO
40 except:
41 from StringIO import StringIO
42
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]
54
55 if not url.endswith('/'):
56 url += '/'
57
58 return {
59 'Url' : url,
60 'Username' : username,
61 'Password' : password,
62 'DicomAet' : aet,
63 'DicomPort' : dicomPort
64 }
65
66
67 def _SetupCredentials(orthanc, http):
68 if (orthanc['Username'] != None and
69 orthanc['Password'] != None):
70 http.add_credentials(orthanc['Username'], orthanc['Password'])
71
72
73 def DoGet(orthanc, uri, data = {}, body = None, headers = {}):
74 d = ''
75 if len(data.keys()) > 0:
76 d = '?' + urlencode(data)
77
78 http = httplib2.Http()
79 _SetupCredentials(orthanc, http)
80
81 resp, content = http.request(orthanc['Url'] + uri + d, 'GET', body = body,
82 headers = headers)
83 if not (resp.status in [ 200 ]):
84 raise Exception(resp.status)
85 else:
86 try:
87 return json.loads(content)
88 except:
89 return content
90
91 def _DoPutOrPost(orthanc, uri, method, data, contentType, headers):
92 http = httplib2.Http()
93 _SetupCredentials(orthanc, http)
94
95 if isinstance(data, str):
96 body = data
97 if len(contentType) != 0:
98 headers['content-type'] = contentType
99 else:
100 body = json.dumps(data)
101 headers['content-type'] = 'application/json'
102
103 headers['expect'] = ''
104
105 resp, content = http.request(orthanc['Url'] + uri, method,
106 body = body,
107 headers = headers)
108 if not (resp.status in [ 200, 302 ]):
109 raise Exception(resp.status)
110 else:
111 try:
112 return json.loads(content)
113 except:
114 return content
115
116 def DoDelete(orthanc, uri):
117 http = httplib2.Http()
118 _SetupCredentials(orthanc, http)
119
120 resp, content = http.request(orthanc['Url'] + uri, 'DELETE')
121 if not (resp.status in [ 200 ]):
122 raise Exception(resp.status)
123 else:
124 try:
125 return json.loads(content)
126 except:
127 return content
128
129 def DoPut(orthanc, uri, data = {}, contentType = ''):
130 return DoPutOrPost(orthanc, uri, 'PUT', data, contentType)
131
132 def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}):
133 return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers)
134
135 def UploadInstance(orthanc, filename):
136 global HERE
137 p = os.path.join(HERE, '..', 'Database', filename)
138 f = open(p, 'rb')
139 d = f.read()
140 f.close()
141 return DoPost(orthanc, '/instances', d, 'application/dicom')
142
143 def UploadFolder(orthanc, path):
144 global HERE
145 p = os.path.join(HERE, 'Database', path)
146 for i in os.listdir(p):
147 try:
148 UploadInstance(orthanc, os.path.join(path, i))
149 except:
150 pass
151
152 def DropOrthanc(orthanc):
153 # Reset the Lua callbacks
154 DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua')
155
156 DoDelete(orthanc, '/exports')
157
158 for s in DoGet(orthanc, '/patients'):
159 DoDelete(orthanc, '/patients/%s' % s)
160
161 def ComputeMD5(data):
162 m = hashlib.md5()
163 m.update(data)
164 return m.hexdigest()
165
166 def GetImage(orthanc, uri):
167 # http://www.pythonware.com/library/pil/handbook/introduction.htm
168 data = DoGet(orthanc, uri)
169 return Image.open(StringIO(data))
170
171 def GetArchive(orthanc, uri):
172 # http://stackoverflow.com/a/1313868/881731
173 s = DoGet(orthanc, uri)
174 return zipfile.ZipFile(StringIO(s), "r")
175
176 def IsDefinedInLua(orthanc, name):
177 s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua')
178 return (s.strip() != 'nil')
179
180 def WaitEmpty(orthanc):
181 while True:
182 if len(DoGet(orthanc, '/instances')) == 0:
183 return
184 time.sleep(0.1)
185
186 def GetDockerHostAddress():
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]