Commit 0de58dd2 authored by Franziska Koehn's avatar Franziska Koehn
Browse files

xnat_search skript added, first version, including main for starting search/download via terminal

parent d515d1d5
def main(host):
import distutils.core
root_element = 'xnat:mrScanData'
constraints = [
('xnat:mrScanData/TYPE', 'LIKE', '%t1%'),
('xnat:mrScanData/PARAMETERS_FLIP', '>=', '10'),
'AND',
[('xnat:mrScanData/PARAMETERS_TE', '>', '2.0'),
('xnat:mrScanData/PARAMETERS_TE', '<', '2.0'),
'OR'
]
]
results = search_for(host, root_element, constraints)
print "Search results (%s):" % len(results)
print results
while True:
try:
is_downl = distutils.util.strtobool(raw_input("download? "))
break
except ValueError:
print "invalid character. Please use y or n."
if is_downl:
print "downloading files..."
download_all(results, host)
print "finished"
else:
print "download canceled"
def get_search_fields():
return [{'label':'type', 'field':'xnat:mrScanData/TYPE'},
{'label':'fov_x', 'field':'xnat:mrScanData/PARAMETERS_FOV_X'},
{'label':'fov_y', 'field':'xnat:mrScanData/PARAMETERS_FOV_Y'},
{'label':'tr', 'field': 'xnat:mrScanData/PARAMETERS_TR'},
{'label':'te', 'field': 'xnat:mrScanData/PARAMETERS_TE'},
{'label':'ti', 'field': 'xnat:mrScanData/PARAMETERS_TI'},
{'label':'flip', 'field': 'xnat:mrScanData/PARAMETERS_FLIP'},
{'label':'voxel_res_x', 'field':'xnat:mrScanData/PARAMETERS_VOXELRES_X'},
{'label':'voxel_res_y', 'field':'xnat:mrScanData/PARAMETERS_VOXELRES_Y'},
{'label':'voxel_res_z', 'field':'xnat:mrScanData/PARAMETERS_VOXELRES_Z'}
]
def get_query_methods():
return ["AND", "OR"]
def get_operators():
return ["LIKE", ">", "<", "<=", ">=", "="]
def download_all(results, host):
for r in results:
subject = r['xnat_mrsessiondata_subject_id']
experiment = r['xnat_mrsessiondata_session_id']
project = r['xnat_mrsessiondata_project']
scan = r['id']
file_name = "%s_%s_%s" % (subject, r['type'], scan)
download_file(host, project, subject, experiment, scan, file_name)
def search_for_mrScanData(host, constraints, user, passw):
get_credentials(username=user, password=passw)
return search_for(host, 'xnat:mrScanData', constraints)
def search_for(host, root_element, constraints):
from pyxnat import Interface
from tempfile import mkdtemp
user, passw = get_credentials()
tmp_dir=mkdtemp()
central = Interface(server="%s" % host,
user=user,
password=passw,
cachedir=tmp_dir)
search_fields = [
'xnat:mrSessionData/PROJECT', #project id
'xnat:mrSessionData/SUBJECT_ID', #subject id
'xnat:mrSessionData/SESSION_ID', #experiment id
'xnat:mrScanData/ID', #scan id
'xnat:mrScanData/TYPE', #scan type
]
result = central.select(root_element,search_fields).where(constraints)
try:
central.disonnect()
except AttributeError:
print "can\'t close connection (wrong pyxnat version?)"
try:
from shutil import rmtree
rmtree(tmp_dir)
except:
pass
return result
def get_credentials(force=False,username="", password="" ):
if get_credentials.cache and not force:
return get_credentials.cache
from getpass import getpass
while not len(username):
username = raw_input("Username: ").strip()
while not len(password):
password = getpass().strip()
get_credentials.cache = username, password
return get_credentials()
get_credentials.cache = None
def download_file(host, project, subject, experiment, scan, file_name):
import requests
from base64 import b64encode
from requests.auth import HTTPBasicAuth
if not file_name.endswith(".zip"):
file_name += ".zip"
url= "%s/data/archive/projects/%s/subjects/%s/experiments/%s/scans/%s/resources/DICOM/files?format=zip" % (host, project, subject, experiment, scan)
user, passw = get_credentials()
try:
with open(file_name, 'wb') as handle:
response = requests.get(url, stream=True, auth=HTTPBasicAuth(user, passw))
if not response.ok:
raise ValueError(response.status_code)
for block in response.iter_content(1024):
if not block:
break
handle.write(block)
except IOError as e:
print "Error writing file %s, %s" % (file_name, e)
except ValueError as e:
print "Error downloading file %s, Status Code %s" % (url, e)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description='search and download')
parser.add_argument('--host', type=str, help='hostname or ip-address (including port).', default="localhost:8080")
parser.add_argument('--user', type=str, help='user:passw', default=":")
#TODO force download (boolean)
#TODO read constraints und root_element from file
#TODO GUI
args = parser.parse_args()
u,p = args.user.split(':')
get_credentials(username=u, password=p)
main(args.host)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment