Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
print 'auto_procs =', auto_procs
# if parsed_args.sessions is not None:
# for s in parsed_args.session:
# sess = intf.select('/project/*/subject/*/session/' + s)
# print sess
# else:
print 'patching project', p
proj = intf.select_project(p)
for s in proj.subjects():
print 'subject:', s.label()
for e in intf.get_sessions(p, s.label()):
if not parsed_args.sessions\
or e['label'] in parsed_args.sessions\
or e['ID'] in parsed_args.sessions:
experiment = XnatUtils.CachedImageSession(intf, p, s.label(), e['label'])
for a in auto_procs:
a.parse_session(experiment)
a.parser.patch_assessors()
"""
Method to get the list of Cached Objects for the project/sessions for a
processor
:param mod_obj: processor object
:param project: XNAT project
:param sessions: XNAT sessions
:return: None
"""
co_list = list()
sess_list = self.xnat.get_sessions(project)
sess_list = [sess for sess in sess_list if sess['label'] in sessions]
# Loop through the sessions
for sess in sess_list:
csess = XnatUtils.CachedImageSession(self.xnat, project,
sess['subject_label'],
sess['label'])
if isinstance(mod_obj, modules.ScanModule):
for cscan in csess.scans():
if mod_obj.needs_run(cscan, self.xnat):
co_list.append(cscan)
elif isinstance(mod_obj, modules.SessionModule):
if mod_obj.needs_run(csess, self.xnat):
co_list.append(csess)
if len(co_list) == 0:
LOGGER.warn("[WARNING] No object found for the Module.")
self.inc_warning()
return co_list
continue
else:
LOGGER.info('+ Session {}:modified, last_mod={}'.format(
sess_info['label'], str(last_mod)))
# Append session to list of sessions to update
sessions_to_update[sess_info['ID']] = sess_info
if len(sessions_to_update) == 0:
continue
# build a full list of sessions for the subject: they may be needed
# even if not all sessions are getting updated
mess = "+ Subject %s: loading XML for %s session(s)..."
LOGGER.info(mess % (sessions[0]['subject_label'], len(sessions)))
cached_sessions = [XnatUtils.CachedImageSession(
intf, x['project_label'], x['subject_label'],
x['session_label']) for x in sessions]
if len(cached_sessions) > 1:
cached_sessions = sorted(
cached_sessions,
key=lambda s: s.creation_timestamp(), reverse=True)
# update each of the sessions that require it
for sess_info in list(sessions_to_update.values()):
try:
# TODO: BenM - ensure that this code is robust to subjects
# without sessions and sessions without assessors / scans
mess = "+ Session %s: building..."
LOGGER.info(mess % sess_info['label'])
self.build_session(
"""
Method to get the list of Cached Objects for the project/sessions for a
processor
:param proc_obj: processor object
:param project: XNAT project
:param sessions: XNAT sessions
:return: None
"""
co_list = list()
sess_list = self.xnat.get_sessions(project)
sess_list = [sess for sess in sess_list if sess['label'] in sessions]
# Loop through the sessions
for sess in sess_list:
csess = XnatUtils.CachedImageSession(self.intf, project,
sess['subject_label'],
sess['label'])
if isinstance(proc_obj, processors.AutoProcessor):
if proc_obj.type == 'session':
co_list.append(csess)
else:
for cscan in csess.scans():
if proc_obj.should_run(cscan.info()):
co_list.append(cscan)
if len(co_list) == 0:
print("[WARNING] No scan found for the processor on scans.")
self.inc_warning()
return co_list
def parse_session(self, csess):
self.csess = None
self.artefacts = None
self.artefacts_by_input = None
self.parameter_matrix = None
self.assessor_parameter_map = None
# build a list of sessions starting from the current session backwards
intf = csess.intf
subj = intf.select_subject(csess.project_id(), csess.subject_id())
x = [XnatUtils.CachedImageSession(intf,
csess.project_id(),
csess.subject_id(),
s.label())
for s in subj.experiments()]
x = [TimestampSession(s.creation_timestamp(), s) for s in x]
ordered_sessions = map(lambda y: y.session,
sorted(x,
key=lambda v: v.timestamp,
reverse=True))
ordered_sessions =\
filter(
lambda y: y.creation_timestamp() <= csess.creation_timestamp(),
ordered_sessions)
artefacts = ProcessorParser.parse_artefacts(ordered_sessions)