Mercurial > hg > orthanc-tests

comparison Toolbox.py @ 0:cc43b57242a4

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
toolbox
author Sebastien Jodogne <s.jodogne@gmail.com>
date Mon, 15 Jun 2015 17:44:12 +0200
parents
children 08dadea8f40a
comparison
equal deleted inserted replaced
-1:000000000000 0:cc43b57242a4
1 #!/usr/bin/python
2
3 # sudo docker run --rm -v `pwd`/Toolbox.py:/tmp/Toolbox.py:ro --entrypoint python jodogne/orthanc-tests /tmp/Toolbox.py
4
5
6 import hashlib
7 import httplib2
8 import json
9 import os.path
10 from PIL import Image
11 import zipfile
12 import time
13 from urllib import urlencode
14
15
16
17 # http://stackoverflow.com/a/1313868/881731
18 try:
19 from cStringIO import StringIO
20 except:
21 from StringIO import StringIO
22
23
24
25 def CreateOrthanc(url = 'http://localhost:8042',
26 username = None,
27 password = None):
28 if not url.endswith('/'):
29 url += '/'
30
31 return [ url, username, password ]
32
33
34 def _SetupCredentials(orthanc, http):
35 if orthanc[1] != None and orthanc[2] != None:
36 http.add_credentials(orthanc[1], orthanc[2])
37
38
39 def DoGet(orthanc, uri, data = {}, body = None, headers = {}):
40 d = ''
41 if len(data.keys()) > 0:
42 d = '?' + urlencode(data)
43
44 http = httplib2.Http()
45 _SetupCredentials(orthanc, http)
46
47 resp, content = http.request(orthanc[0] + uri + d, 'GET', body = body,
48 headers = headers)
49 if not (resp.status in [ 200 ]):
50 raise Exception(resp.status)
51 else:
52 try:
53 return json.loads(content)
54 except:
55 return content
56
57 def _DoPutOrPost(orthanc, uri, method, data, contentType, headers):
58 http = httplib2.Http()
59 _SetupCredentials(orthanc, http)
60
61 if isinstance(data, str):
62 body = data
63 if len(contentType) != 0:
64 headers['content-type'] = contentType
65 else:
66 body = json.dumps(data)
67 headers['content-type'] = 'application/json'
68
69 headers['expect'] = ''
70
71 resp, content = http.request(orthanc[0] + uri, method,
72 body = body,
73 headers = headers)
74 if not (resp.status in [ 200, 302 ]):
75 raise Exception(resp.status)
76 else:
77 try:
78 return json.loads(content)
79 except:
80 return content
81
82 def DoDelete(orthanc, uri):
83 http = httplib2.Http()
84 _SetupCredentials(orthanc, http)
85
86 resp, content = http.request(orthanc[0] + uri, 'DELETE')
87 if not (resp.status in [ 200 ]):
88 raise Exception(resp.status)
89 else:
90 try:
91 return json.loads(content)
92 except:
93 return content
94
95 def DoPut(orthanc, uri, data = {}, contentType = ''):
96 return DoPutOrPost(orthanc, uri, 'PUT', data, contentType)
97
98 def DoPost(orthanc, uri, data = {}, contentType = '', headers = {}):
99 return _DoPutOrPost(orthanc, uri, 'POST', data, contentType, headers)
100
101 def UploadInstance(orthanc, filename):
102 p = os.path.join(HERE, DICOM_DB, filename)
103 f = open(p, 'rb')
104 d = f.read()
105 f.close()
106 return DoPost(orthanc, '/instances', d, 'application/dicom')
107
108 def UploadFolder(orthanc, path):
109 p = os.path.join(HERE, DICOM_DB, path)
110 for i in os.listdir(p):
111 try:
112 UploadInstance(orthanc, os.path.join(path, i))
113 except:
114 pass
115
116 def DropOrthanc(orthanc):
117 # Reset the Lua callbacks
118 DoPost(orthanc, '/tools/execute-script', 'function OnStoredInstance(instanceId, tags, metadata) end', 'application/lua')
119
120 DoDelete(orthanc, '/exports')
121
122 for s in DoGet(orthanc, '/patients'):
123 DoDelete(orthanc, '/patients/%s' % s)
124
125 def ComputeMD5(data):
126 m = hashlib.md5()
127 m.update(data)
128 return m.hexdigest()
129
130 def GetImage(orthanc, uri):
131 # http://www.pythonware.com/library/pil/handbook/introduction.htm
132 data = DoGet(orthanc, uri)
133 return Image.open(StringIO(data))
134
135 def GetArchive(orthanc, uri):
136 # http://stackoverflow.com/a/1313868/881731
137 s = DoGet(orthanc, uri)
138 return zipfile.ZipFile(StringIO(s), "r")
139
140 def IsDefinedInLua(name):
141 s = DoPost(orthanc, '/tools/execute-script', 'print(type(%s))' % name, 'application/lua')
142 return (s.strip() != 'nil')
143
144 def WaitEmpty():
145 while True:
146 if len(orthanc, DoGet('/instances')) == 0:
147 return
148 time.sleep(0.1)
149
150
151 print DoGet(CreateOrthanc('http://192.168.215.82:8042'), '/system')