Coverage for ibllib/pipes/tasks.py: 87%

1"""The abstract Pipeline and Task superclasses and concrete task runner. 

2 

3Examples 

4-------- 

5 

61. Running a task on your local computer. 

7| Download: via ONE. 

8| Upload: N/A. 

9 

10>>> task = VideoSyncQcBpod(session_path, one=one, location='remote', sync='bpod') 

11>>> task.run() 

12 

132. Running a task on the local server that belongs to a given subject (e.g SWC054 on floferlab). 

14| Download: all data expected to be present. 

15| Upload: normal way of registering datasets, filerecords created and bulk sync, bulk transfer 

16 jobs on Alyx transfer the data. 

17 

18>>> from ibllib.pipes.video_tasks import VideoSyncQcBpod 

19>>> session_path = '/mnt/ibl/s0/Data/Subjects/SWC054/2023-01-01/001' 

20>>> task = VideoSyncQcBpod(session_path, one=one, sync='bpod') 

21>>> task.run() 

22>>> task.register_datasets(one=one, labs=get_lab(session_path, alyx=ONE().alyx)) 

23 

243. Running a task on the local server that belongs to that subject and forcing redownload of 

25missing data. 

26| Download: via Globus (TODO we should change this to use boto3 as globus is slow). 

27| Upload: normal way of registering datasets, filerecords created and bulk sync, bulk transfer 

28 jobs on Alyx transfer the data. 

29 

30>>> task = VideoSyncQcBpod(session_path, one=one, sync='bpod') 

31>>> task.force = True 

32>>> task.run() 

33>>> task.register_datasets(one=one, labs=get_lab(session_path, alyx=ONE().alyx)) 

34>>> task.cleanUp() # Delete the files that have been downloaded 

35 

364. Running a task on the local server that doesn't belongs to that subject 

37(e.g SWC054 on angelakilab). 

38| Download: via boto3, the AWS file records must exist and be set to exists = True. 

39| Upload: via globus, automatically uploads the datasets directly to FlatIron via globus. 

40 Creates FlatIron filerecords and sets these to True once the globus task has completed. 

41 

42>>> task = VideoSyncQcBpod(session_path, one=one, location='AWS', sync='bpod') 

43>>> task.run() 

44>>> task.register_datasets() 

45>>> task.cleanUp() # Delete the files that have been downloaded 

46 

475. Running a task on SDSC. 

48| Download: via creating symlink to relevant datasets on SDSC. 

49| Upload: via copying files to relevant location on SDSC. 

50 

51>>> task = VideoSyncQcBpod(session_path, one=one, location='SDSC', sync='bpod') 

52>>> task.run() 

53>>> response = task.register_datasets() 

54>>> # Here we just make sure filerecords are all correct 

55>>> for resp in response: 

56... fi = next((fr for fr in resp['file_records'] if 'flatiron' in fr['data_repository']), None) 

57... if fi is not None: 

58... if not fi['exists']: 

59... one.alyx.rest('files', 'partial_update', id=fi['id'], data={'exists': True}) 

60... 

61... aws = next((fr for fr in resp['file_records'] if 'aws' in fr['data_repository']), None) 

62... if aws is not None: 

63... one.alyx.rest('files', 'partial_update', id=aws['id'], data={'exists': False}) 

64... 

65... sr = next((fr for fr in resp['file_records'] if 'SR' in fr['data_repository']), None) 

66... if sr is not None: 

67... one.alyx.rest('files', 'partial_update', id=sr['id'], data={'exists': False}) 

68... # Finally remove symlinks once the task has completed 

69... task.cleanUp() 

70 

71""" 

72from pathlib import Path (empty)

73import abc (empty)

74import logging (empty)

75import io (empty)

76import importlib (empty)

77import time (empty)

78from collections import OrderedDict (empty)

79import traceback (empty)

80import json (empty)

81from typing import List, Dict (empty)

82 

83from graphviz import Digraph (empty)

84 

85import ibllib (empty)

86from ibllib.oneibl import data_handlers (empty)

87from ibllib.oneibl.data_handlers import get_local_data_repository (empty)

88from ibllib.oneibl.registration import get_lab (empty)

89from iblutil.util import Bunch, flatten, ensure_list (empty)

90import one.params (empty)

91from one.api import ONE (empty)

92from one import webclient (empty)

93import one.alf.io as alfio (empty)

94from one.alf.path import ALFPath (empty)

95 

96_logger = logging.getLogger(__name__) (empty)

97TASK_STATUS_SET = {'Waiting', 'Held', 'Started', 'Errored', 'Empty', 'Complete', 'Incomplete', 'Abandoned'} (empty)

98 

99 

100class Task(abc.ABC): (empty)

101 log = '' # placeholder to keep the log of the task for registration (empty)

102 cpu = 1 # CPU resource (empty)

103 gpu = 0 # GPU resources: as of now, either 0 or 1 (empty)

104 io_charge = 5 # integer percentage (empty)

105 priority = 30 # integer percentage, 100 means highest priority (empty)

106 ram = 4 # RAM needed to run (GB) (empty)

107 one = None # one instance (optional) (empty)

108 level = 0 # level in the pipeline hierarchy: level 0 means there is no parent task (empty)

109 outputs = None # placeholder for a list of Path containing output files (empty)

110 time_elapsed_secs = None (empty)

111 time_out_secs = 3600 * 2 # time-out after which a task is considered dead (empty)

112 version = ibllib.__version__ (empty)

113 force = False # whether to re-download missing input files on local server if not present (empty)

114 job_size = 'small' # either 'small' or 'large', defines whether task should be run as part of the large or small job services (empty)

115 env = None # the environment name within which to run the task (NB: the env is not activated automatically!) (empty)

116 on_error = 'continue' # whether to raise an exception on error ('raise') or report the error and continue ('continue') (empty)

117 

118 def __init__(self, session_path, parents=None, taskid=None, one=None, (empty)

119 machine=None, clobber=True, location='server', scratch_folder=None, on_error='continue', **kwargs): 

120 """ 

121 Base task class 

122 :param session_path: session path 

123 :param parents: parents 

124 :param taskid: alyx task id 

125 :param one: one instance 

126 :param machine: 

127 :param clobber: whether or not to overwrite log on rerun 

128 :param location: location where task is run. Options are 'server' (lab local servers'), 'remote' (remote compute node, 

129 data required for task downloaded via one), 'AWS' (remote compute node, data required for task downloaded via AWS), 

130 or 'SDSC' (SDSC flatiron compute node) 

131 :param scratch_folder: optional: Path where to write intermediate temporary data 

132 :param args: running arguments 

133 """ 

134 self.on_error = on_error 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

135 self.taskid = taskid 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

136 self.one = one 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

137 self.session_path = session_path 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

138 self.register_kwargs = {} 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

139 if parents: 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

140 self.parents = parents 16 ctx2b ~ aba d c ) * + , - . / : ; =

141 self.level = max(p.level for p in self.parents) + 1 16 ctx2b ~ aba d c ) * + , - . / : ; =

142 else: 

143 self.parents = [] 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

144 self.machine = machine 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

145 self.clobber = clobber 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

146 self.location = location 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

147 self.plot_tasks = [] # Plotting task/ tasks to create plot outputs during the task 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

148 self.scratch_folder = scratch_folder 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

149 self.kwargs = kwargs 106 ctx2b ~ dbbbebfbgbhbibjbabkb_ ` { lb$ f a ^ d S K T U N l z A m B n o p C q r D c ) * + , - . / : ; = x y h V W i X g E F % Y Z 0 G j 1 | mbnb} O 2 3 ob4 H 5 I 6 7 J 8 ' e s k t u v w P 9 Q ! L M ( pbqbrbsbtbR # ubvb

150 

151 @property (empty)

152 def signature(self) -> Dict[str, List]: (empty)

153 """ 

154 The signature of the task specifies inputs and outputs for the given task. 

155 For some tasks it is dynamic and calculated. The legacy code specifies those as tuples. 

156 The preferred way is to use the ExpectedDataset input and output constructors. 

157 

158 I = ExpectedDataset.input 

159 O = ExpectedDataset.output 

160 signature = { 

161 'input_files': [ 

162 I(name='extract.me.npy', collection='raw_data', required=True, register=False, unique=False), 

163 ], 

164 'output_files': [ 

165 O(name='look.atme.npy', collection='shiny_data', required=True, register=True, unique=False) 

166 ]} 

167 is equivalent to: 

168 signature = { 

169 'input_files': [('extract.me.npy', 'raw_data', True, True)], 

170 'output_files': [('look.atme.npy', 'shiny_data', True)], 

171 } 

172 :return: 

173 """ 

174 return {'input_files': [], 'output_files': []} 3 ctx1fad

175 

176 @property (empty)

177 def name(self): (empty)

178 return self.__class__.__name__ 20 ctx2b bbf a d ? @ cbc ) * + , - . / : ; = e

179 

180 def path2eid(self): (empty)

181 """ 

182 Fetch the experiment UUID from the Task session path, without using the REST cache. 

183 

184 This method ensures that the eid will be returned for newly created sessions. 

185 

186 Returns 

187 ------- 

188 str 

189 The experiment UUID corresponding to the session path. 

190 """ 

191 assert self.session_path and self.one and not self.one.offline 1 ctx1_

192 with webclient.no_cache(self.one.alyx): 1 ctx1_

193 return self.one.path2eid(self.session_path, query_type='remote') 1 ctx1_

194 

195 def run(self, **kwargs): (empty)

196 """ 

197 --- do not overload, see _run() below--- 

198 wraps the _run() method with 

199 - error management 

200 - logging to variable 

201 - writing a lock file if the GPU is used 

202 - labels the status property of the object. The status value is labeled as: 

203 0: Complete 

204 -1: Errored 

205 -2: Didn't run as a lock was encountered 

206 -3: Incomplete 

207 

208 Notes 

209 ----- 

210 - The `run_alyx_task` will update the Alyx Task status depending on both status and outputs 

211 (i.e. the output of subclassed `_run` method): 

212 Assuming a return value of 0... if Task.outputs is None, the status will be Empty; 

213 if Task.outputs is a list (empty or otherwise), the status will be Complete. 

214 """ 

215 # if task id of one properties are not available, local run only without alyx 

216 use_alyx = self.one is not None and self.taskid is not None 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

217 if use_alyx: 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

218 # check that alyx user is logged in 

219 if not self.one.alyx.is_logged_in: 2 ctx1ac

220 self.one.alyx.authenticate() 

221 tdict = self.one.alyx.rest('tasks', 'partial_update', id=self.taskid, 2 ctx1ac

222 data={'status': 'Started'}) 

223 self.log = ('' if not tdict['log'] else tdict['log'] + 2 ctx1ac

224 '\n\n=============================RERUN=============================\n') 

225 

226 # Setup the console handler with a StringIO object 

227 logger_level = _logger.level 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

228 log_capture_string = io.StringIO() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

229 ch = logging.StreamHandler(log_capture_string) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

230 str_format = '%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s' 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

231 ch.setFormatter(logging.Formatter(str_format)) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

232 _logger.parent.addHandler(ch) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

233 _logger.parent.setLevel(logging.INFO) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

234 _logger.info(f'Starting job {self.__class__}') 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

235 if self.machine: 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

236 _logger.info(f'Running on machine: {self.machine}') 1 ctx1a

237 _logger.info(f'running ibllib version {ibllib.__version__}') 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

238 # setup 

239 start_time = time.time() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

240 try: 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

241 setup = self.setUp(**kwargs) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

242 self.outputs = self._input_files_to_register() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

243 _logger.info(f'Setup value is: {setup}') 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

244 self.status = 0 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

245 if not setup: 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

246 # case where outputs are present but don't have input files locally to rerun task 

247 # label task as complete 

248 _, outputs = self.assert_expected_outputs() 

249 else: 

250 # run task 

251 if self.gpu >= 1: 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

252 if not self._creates_lock(): 3 ctx1fae

253 self.status = -2 2 ctx1fa

254 _logger.info(f'Job {self.__class__} exited as a lock was found at {self._lock_file_path()}') 2 ctx1fa

255 new_log = log_capture_string.getvalue() 2 ctx1fa

256 self.log = new_log if self.clobber else self.log + new_log 2 ctx1fa

257 _logger.removeHandler(ch) 2 ctx1fa

258 ch.close() 2 ctx1fa

259 if self.on_error == 'raise': 2 ctx1fa

260 raise FileExistsError(f'Job {self.__class__} exited as a lock was found at {self._lock_file_path()}') 

261 return self.status 2 ctx1fa

262 outputs = self._run(**kwargs) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

263 _logger.info(f'Job {self.__class__} complete') 65 ctx1b$fadSTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!(R#

264 if outputs is None: 65 ctx1b$fadSTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!(R#

265 # If run method returns None and no raw input files were registered, self.outputs 

266 # should be None, meaning task will have an 'Empty' status. If run method returns 

267 # a list, the status will be 'Complete' regardless of whether there are output files. 

268 self.outputs = outputs if not self.outputs else self.outputs # ensure None if no inputs registered 4 ctx1adNR

269 else: 

270 self.outputs.extend(ensure_list(outputs)) # Add output files to list of inputs to register 62 ctx1b$faSTUlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!(#

271 except Exception as e: 5 ctx1aKjLM

272 _logger.error(traceback.format_exc()) 5 ctx1aKjLM

273 _logger.info(f'Job {self.__class__} errored') 5 ctx1aKjLM

274 self.status = -1 5 ctx1aKjLM

275 if self.on_error == 'raise': 5 ctx1aKjLM

276 raise e 

277 

278 self.time_elapsed_secs = time.time() - start_time 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

279 # log the outputs 

280 if isinstance(self.outputs, list): 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

281 nout = len(self.outputs) 65 ctx1b$faSKTUlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(#

282 elif self.outputs is None: 4 ctx1adNR

283 nout = 0 4 ctx1adNR

284 else: 

285 nout = 1 

286 

287 _logger.info(f'N outputs: {nout}') 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

288 _logger.info(f'--- {self.time_elapsed_secs} seconds run-time ---') 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

289 # after the run, capture the log output, amend to any existing logs if not overwrite 

290 new_log = log_capture_string.getvalue() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

291 self.log = new_log if self.clobber else self.log + new_log 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

292 _logger.removeHandler(ch) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

293 ch.close() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

294 _logger.setLevel(logger_level) 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

295 # tear down 

296 self.tearDown() 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

297 return self.status 68 ctx1b$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

298 

299 def register_datasets(self, **kwargs): (empty)

300 """ 

301 Register output datasets from the task to Alyx. 

302 

303 Parameters 

304 ---------- 

305 kwargs 

306 Directly passed to the `DataHandler.upload_data` method. 

307 

308 Returns 

309 ------- 

310 list 

311 The output of the `DataHandler.upload_data` method, e.g. a list of registered datasets. 

312 """ 

313 _ = self.register_images() 3 ctx1acH

314 return self.data_handler.uploadData(self.outputs, self.version, **kwargs) 3 ctx1acH

315 

316 def _input_files_to_register(self, assert_all_exist=False): (empty)

317 """ 

318 Return input datasets to be registered to Alyx. 

319 

320 These datasets are typically raw data files and are registered even if the task fails to complete. 

321 

322 Parameters 

323 ---------- 

324 assert_all_exist 

325 Raise AssertionError if not all required input datasets exist on disk. 

326 

327 Returns 

328 ------- 

329 list of pathlib.Path 

330 A list of input files to register. 

331 

332 # TODO This method currently does not support wildcards 

333 """ 

334 I = data_handlers.ExpectedDataset.input # noqa 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

335 try: 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

336 # Ensure all input files are ExpectedDataset instances 

337 input_files = [I(*i) if isinstance(i, tuple) else i for i in self.input_files or []] 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

338 except AttributeError: 1 ctx1^

339 raise RuntimeError('Task.setUp must be run before calling this method.') 1 ctx1^

340 to_register, missing = [], set() 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

341 for dataset in input_files: 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

342 ok, filepaths, _missing = dataset.find_files(self.session_path, register=True) 58 ctx1b^SKTUNlmnopqrcxyhVWiXgEFYZ0Gj1O234H5I67J8esktuvwP9Q!LM(R#

343 to_register.extend(filepaths) 58 ctx1b^SKTUNlmnopqrcxyhVWiXgEFYZ0Gj1O234H5I67J8esktuvwP9Q!LM(R#

344 if not ok and _missing is not None: 58 ctx1b^SKTUNlmnopqrcxyhVWiXgEFYZ0Gj1O234H5I67J8esktuvwP9Q!LM(R#

345 missing.update(_missing) if isinstance(_missing, set) else missing.add(_missing) 1 ctx1^

346 if any(missing): # NB: These are either glob patterns that have no matches or match files that should be absent 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

347 missing_str = ', '.join(missing) 1 ctx1^

348 if assert_all_exist: 1 ctx1^

349 raise AssertionError(f'Missing required input files: {missing_str}') 1 ctx1^

350 else: 

351 _logger.error(f'Missing required input files: {missing_str}') 1 ctx1^

352 return to_register 69 ctx1b$fa^dSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

353 

354 def register_images(self, **kwargs): (empty)

355 """ 

356 Registers images to alyx database 

357 :return: 

358 """ 

359 if self.one and len(self.plot_tasks) > 0: 3 ctx1acH

360 for plot_task in self.plot_tasks: 

361 try: 

362 _ = plot_task.register_images(widths=['orig']) 

363 except Exception: 

364 _logger.error(traceback.format_exc()) 

365 continue 

366 

367 def rerun(self): (empty)

368 self.run(overwrite=True) 

369 

370 def get_signatures(self, **kwargs): (empty)

371 """ 

372 This is the default but should be overwritten for each task 

373 :return: 

374 """ 

375 signature = data_handlers._parse_signature(self.signature) 74 ctx1b][`{$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1|}O234H5I67J8'esktuvwP9Q!LM(R#

376 self.input_files = signature['input_files'] 74 ctx1b][`{$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1|}O234H5I67J8'esktuvwP9Q!LM(R#

377 self.output_files = signature['output_files'] 74 ctx1b][`{$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1|}O234H5I67J8'esktuvwP9Q!LM(R#

378 

379 @abc.abstractmethod (empty)

380 def _run(self, overwrite=False): (empty)

381 """Execute main task code. 

382 

383 This method contains a task's principal data processing code and should implemented 

384 by each subclass. 

385 

386 Parameters 

387 ---------- 

388 overwrite : bool 

389 When false (default), the task may re-use any intermediate data from a previous run. 

390 When true, the task will disregard or delete any intermediate files before running. 

391 

392 Returns 

393 ------- 

394 pathlib.Path, list of pathlib.Path, None 

395 One or more files to be registered, or an empty list if no files are registered. 

396 Within the pipeline, there is a distinction between a job that returns an empty list 

397 and a job that returns None. If the function returns None, the job will be labeled as 

398 "empty" status in the database, otherwise, the job has an expected behaviour of not 

399 returning any dataset. 

400 """ 

401 

402 def setUp(self, **kwargs): (empty)

403 """Get the data handler and ensure all data is available locally to run task.""" 

404 if self.location == 'server': 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

405 self.get_signatures(**kwargs) 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

406 

407 input_status, _ = self.assert_expected_inputs(raise_error=False) 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

408 output_status, _ = self.assert_expected(self.output_files, silent=True) 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

409 

410 if input_status: 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

411 self.data_handler = self.get_data_handler() 60 ctx1b][$adSKTUNlzAmBnopCqrDcxyVWXg%YZ0j1O2345678'esktuvwP9Q!LMR#

412 _logger.info('All input files found: running task') 60 ctx1b][$adSKTUNlzAmBnopCqrDcxyVWXg%YZ0j1O2345678'esktuvwP9Q!LMR#

413 return True 60 ctx1b][$adSKTUNlzAmBnopCqrDcxyVWXg%YZ0j1O2345678'esktuvwP9Q!LMR#

414 

415 if not self.force: 9 ctx1chigEFGIJ

416 self.data_handler = self.get_data_handler() 9 ctx1chigEFGIJ

417 _logger.warning('Not all input files found locally: will still attempt to rerun task') 9 ctx1chigEFGIJ

418 # TODO in the future once we are sure that input output task signatures work properly should return False 

419 # _logger.info('All output files found but input files required not available locally: task not rerun') 

420 return True 9 ctx1chigEFGIJ

421 else: 

422 # Attempts to download missing data using globus 

423 _logger.info('Not all input files found locally: attempting to re-download required files') 

424 self.data_handler = self.get_data_handler(location='serverglobus') 

425 self.data_handler.setUp(task=self) 

426 # Double check we now have the required files to run the task 

427 # TODO in future should raise error if even after downloading don't have the correct files 

428 self.assert_expected_inputs(raise_error=False) 

429 return True 

430 else: 

431 self.data_handler = self.get_data_handler() 3 ctx1fH(

432 self.data_handler.setUp(task=self) 3 ctx1fH(

433 self.get_signatures(**kwargs) 3 ctx1fH(

434 self.assert_expected_inputs(raise_error=False) 3 ctx1fH(

435 return True 3 ctx1fH(

436 

437 def tearDown(self): (empty)

438 """ 

439 Function after runs() 

440 Does not run if a lock is encountered by the task (status -2) 

441 """ 

442 if self.gpu >= 1: 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

443 if self._lock_file_path().exists(): 2 ctx1fe

444 self._lock_file_path().unlink() 2 ctx1fe

445 

446 def cleanUp(self): (empty)

447 """ 

448 Function to optionally overload to clean up 

449 :return: 

450 """ 

451 self.data_handler.cleanUp(task=self) 2 ctx1ac

452 

453 def assert_expected_outputs(self, raise_error=True): (empty)

454 """ 

455 After a run, asserts that all signature files are present at least once in the output files 

456 Mainly useful for integration tests 

457 :return: 

458 """ 

459 assert self.status == 0 20 ctx1lzAmBnopCqrDHesktuvw

460 _logger.info('Checking output files') 20 ctx1lzAmBnopCqrDHesktuvw

461 everything_is_fine, files = self.assert_expected(self.output_files) 20 ctx1lzAmBnopCqrDHesktuvw

462 

463 if not everything_is_fine: 20 ctx1lzAmBnopCqrDHesktuvw

464 for out in self.outputs: 

465 _logger.error(f'{out}') 

466 if raise_error: 

467 raise FileNotFoundError('Missing outputs after task completion') 

468 

469 return everything_is_fine, files 20 ctx1lzAmBnopCqrDHesktuvw

470 

471 def assert_expected_inputs(self, raise_error=True, raise_ambiguous=False): (empty)

472 """ 

473 Check that all the files necessary to run the task have been are present on disk. 

474 

475 Parameters 

476 ---------- 

477 raise_error : bool 

478 If true, raise FileNotFoundError if required files are missing. 

479 

480 Returns 

481 ------- 

482 bool 

483 True if all required files were found. 

484 list of pathlib.Path 

485 A list of file paths that exist on disk. 

486 """ 

487 _logger.info('Checking input files') 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

488 everything_is_fine, files = self.assert_expected(self.input_files) 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

489 

490 if not everything_is_fine and raise_error: 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

491 raise FileNotFoundError('Missing inputs to run task') 

492 

493 # Check for duplicate datasets that may complicate extraction. 

494 # Some sessions may contain revisions and without ONE it's difficult to determine which 

495 # are the default datasets. Likewise SDSC may contain multiple datasets with different 

496 # UUIDs in the name after patching data. 

497 valid_alf_files = filter(ALFPath.is_valid_alf, files) 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

498 variant_datasets = alfio.find_variants(valid_alf_files, extra=False) 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

499 if len(variant_datasets) < len(files): 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

500 _logger.warning('Some files are not ALF datasets and will not be checked for ambiguity') 5 ctx1[jkPQ

501 if any(map(len, variant_datasets.values())): 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

502 # Keep those with variants and make paths relative to session for logging purposes 

503 to_frag = lambda x: x.relative_to_session().as_posix() # noqa 5 ctx1xyhig

504 ambiguous = { 5 ctx1xyhig

505 to_frag(k): [to_frag(x) for x in v] 

506 for k, v in variant_datasets.items() if any(v)} 

507 _logger.error('Ambiguous input datasets found: %s', ambiguous) 5 ctx1xyhig

508 

509 if raise_ambiguous or self.location == 'sdsc': # take no chances on SDSC 5 ctx1xyhig

510 # This could be mitigated if loading with data OneSDSC 

511 raise NotImplementedError( 

512 'Multiple variant datasets found. Loading for these is undefined.') 

513 

514 return everything_is_fine, files 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

515 

516 def assert_expected(self, expected_files, silent=False): (empty)

517 """ 

518 Assert that expected files are present. 

519 

520 Parameters 

521 ---------- 

522 expected_files : list of ExpectedDataset 

523 A list of expected files in the form (file_pattern_str, collection_str, required_bool). 

524 silent : bool 

525 If true, log an error if any required files are not found. 

526 

527 Returns 

528 ------- 

529 bool 

530 True if all required files were found. 

531 list of pathlib.Path 

532 A list of file paths that exist on disk. 

533 """ 

534 if not any(expected_files): 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

535 return True, [] 13 ctx1$fadzABCDc%O'

536 ok, actual_files, missing = zip(*(x.find_files(self.session_path) for x in expected_files)) 67 ctx1b][$SKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

537 everything_is_fine = all(ok) 67 ctx1b][$SKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

538 # For unknown reasons only the first file of each expected dataset was returned and this 

539 # behaviour was preserved after refactoring the code 

540 files = [file_list[0] for file_list in actual_files if len(file_list) > 0] 67 ctx1b][$SKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

541 if not everything_is_fine and not silent: 67 ctx1b][$SKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

542 for missing_pattern in filter(None, flatten(missing)): 9 ctx1chigEFGIJ

543 _logger.error('Signature file pattern %s not found', missing_pattern) 9 ctx1chigEFGIJ

544 

545 return everything_is_fine, files 67 ctx1b][$SKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

546 

547 def get_data_handler(self, location=None): (empty)

548 """ 

549 Gets the relevant data handler based on location argument 

550 :return: 

551 """ 

552 location = str.lower(location or self.location) 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

553 if location == 'local': 70 ctx1b][$fadSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O234H5I67J8'esktuvwP9Q!LM(R#

554 return data_handlers.LocalDataHandler(self.session_path, self.signature, one=self.one) 3 ctx1fH(

555 self.one = self.one or ONE() 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

556 if location == 'server': 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

557 dhandler = data_handlers.ServerDataHandler(self.session_path, self.signature, one=self.one) 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

558 elif location == 'serverglobus': 

559 dhandler = data_handlers.ServerGlobusDataHandler(self.session_path, self.signature, one=self.one) 

560 elif location == 'remote': 

561 dhandler = data_handlers.RemoteHttpDataHandler(self.session_path, self.signature, one=self.one) 

562 elif location == 'aws': 

563 dhandler = data_handlers.RemoteAwsDataHandler(self.session_path, self.signature, one=self.one) 

564 elif location == 'sdsc': 

565 dhandler = data_handlers.SDSCDataHandler(self.session_path, self.signature, one=self.one) 

566 elif location == 'popeye': 

567 dhandler = data_handlers.PopeyeDataHandler(self.session_path, self.signature, one=self.one) 

568 elif location == 'ec2': 

569 dhandler = data_handlers.RemoteEC2DataHandler(self.session_path, self.signature, one=self.one) 

570 else: 

571 raise ValueError(f'Unknown location "{location}"') 

572 return dhandler 67 ctx1b][$adSKTUNlzAmBnopCqrDcxyhVWiXgEF%YZ0Gj1O2345I67J8'esktuvwP9Q!LMR#

573 

574 @staticmethod (empty)

575 def make_lock_file(taskname='', time_out_secs=7200): (empty)

576 """Creates a GPU lock file with a timeout of""" 

577 d = {'start': time.time(), 'name': taskname, 'time_out_secs': time_out_secs} 3 ctx1fae

578 with open(Task._lock_file_path(), 'w+') as fid: 3 ctx1fae

579 json.dump(d, fid) 3 ctx1fae

580 return d 3 ctx1fae

581 

582 @staticmethod (empty)

583 def _lock_file_path(): (empty)

584 """the lock file is in ~/.one/gpu.lock""" 

585 folder = Path.home().joinpath('.one') 1 ctx1e

586 folder.mkdir(exist_ok=True) 1 ctx1e

587 return folder.joinpath('gpu.lock') 1 ctx1e

588 

589 def _make_lock_file(self): (empty)

590 """creates a lock file with the current time""" 

591 return Task.make_lock_file(self.name, self.time_out_secs) 2 ctx1fe

592 

593 def is_locked(self): (empty)

594 """Checks if there is a lock file for this given task""" 

595 lock_file = self._lock_file_path() 3 ctx1fae

596 if not lock_file.exists(): 3 ctx1fae

597 return False 2 ctx1fe

598 

599 with open(lock_file) as fid: 2 ctx1fa

600 d = json.load(fid) 2 ctx1fa

601 now = time.time() 2 ctx1fa

602 if (now - d['start']) > d['time_out_secs']: 2 ctx1fa

603 lock_file.unlink() 1 ctx1f

604 return False 1 ctx1f

605 else: 

606 return True 2 ctx1fa

607 

608 def _creates_lock(self): (empty)

609 if self.location == "popeye": 3 ctx1fae

610 return True 

611 if self.is_locked(): 3 ctx1fae

612 return False 2 ctx1fa

613 else: 

614 self._make_lock_file() 2 ctx1fe

615 return True 2 ctx1fe

616 

617 

618class Pipeline(abc.ABC): (empty)

619 """ 

620 Pipeline class: collection of related and potentially interdependent tasks 

621 """ 

622 tasks = OrderedDict() (empty)

623 one = None (empty)

624 

625 def __init__(self, session_path=None, one=None, eid=None): (empty)

626 assert session_path or eid 14 ctx1badc)*+,-./:;=

627 self.one = one 14 ctx1badc)*+,-./:;=

628 if one and one.alyx.cache_mode and one.alyx.default_expiry.seconds > 1: 14 ctx1badc)*+,-./:;=

629 _logger.warning('Alyx client REST cache active; this may cause issues with jobs') 2 ctx1ba

630 self.eid = eid 14 ctx1badc)*+,-./:;=

631 if self.one and not self.one.offline: 14 ctx1badc)*+,-./:;=

632 self.data_repo = get_local_data_repository(self.one.alyx) 4 ctx1badc

633 else: 

634 self.data_repo = None 10 ctx1)*+,-./:;=

635 

636 if session_path: 14 ctx1badc)*+,-./:;=

637 self.session_path = session_path 14 ctx1badc)*+,-./:;=

638 if not self.eid: 14 ctx1badc)*+,-./:;=

639 # eID for newer sessions may not be in cache so use remote query 

640 self.eid = one.path2eid(session_path, query_type='remote') if self.one else None 12 ctx1dc)*+,-./:;=

641 self.label = self.__module__ + '.' + type(self).__name__ 14 ctx1badc)*+,-./:;=

642 

643 @staticmethod (empty)

644 def _get_exec_name(obj): (empty)

645 """ 

646 For a class, get the executable name as it should be stored in Alyx. When the class 

647 is created dynamically using the type() built-in function, need to revert to the base 

648 class to be able to re-instantiate the class from the alyx dictionary on the client side 

649 :param obj: 

650 :return: string containing the full module plus class name 

651 """ 

652 if obj.__module__ == 'abc': 16 ctx2a d ? @ cbc ) * + , - . / : ; =

653 exec_name = f'{obj.__class__.__base__.__module__}.{obj.__class__.__base__.__name__}' 15 ctx2d ? @ cbc ) * + , - . / : ; =

654 else: 

655 exec_name = f'{obj.__module__}.{obj.name}' 1 ctx1a

656 return exec_name 16 ctx2a d ? @ cbc ) * + , - . / : ; =

657 

658 def make_graph(self, out_dir=None, show=True): (empty)

659 if not out_dir: 1 ctx1a

660 out_dir = self.one.alyx.cache_dir if self.one else one.params.get().CACHE_DIR 1 ctx1a

661 m = Digraph('G', filename=str(Path(out_dir).joinpath(self.__module__ + '_graphs.gv'))) 1 ctx1a

662 m.attr(rankdir='TD') 1 ctx1a

663 

664 e = Digraph(name='cluster_' + self.label) 1 ctx1a

665 e.attr('node', shape='box') 1 ctx1a

666 e.node('root', label=self.label) 1 ctx1a

667 

668 e.attr('node', shape='ellipse') 1 ctx1a

669 for k in self.tasks: 1 ctx1a

670 j = self.tasks[k] 1 ctx1a

671 if len(j.parents) == 0: 1 ctx1a

672 e.edge('root', j.name) 1 ctx1a

673 else: 

674 [e.edge(p.name, j.name) for p in j.parents] 1 ctx1a

675 

676 m.subgraph(e) 1 ctx1a

677 m.attr(label=r'\n\Pre-processing\n') 1 ctx1a

678 m.attr(fontsize='20') 1 ctx1a

679 if show: 1 ctx1a

680 m.view() 

681 return m 1 ctx1a

682 

683 def create_alyx_tasks(self, rerun__status__in=None, tasks_list=None): (empty)

684 """ 

685 Instantiate the pipeline and create the tasks in Alyx, then create the jobs for the session. 

686 

687 If the jobs already exist, they are left untouched. The re-run parameter will re-init the 

688 job by emptying the log and set the status to Waiting. 

689 

690 Parameters 

691 ---------- 

692 rerun__status__in : list, str 

693 To re-run tasks if they already exist, specify one or more statuses strings to will be 

694 re-run, or '__all__' to re-run all tasks. 

695 tasks_list : list 

696 The list of tasks to create on Alyx. If None, uses self.tasks. 

697 

698 Returns 

699 ------- 

700 list 

701 List of Alyx task dictionaries (existing and/or created). 

702 """ 

703 rerun__status__in = ensure_list(rerun__status__in) 5 ctx1ad?@c

704 if '__all__' in rerun__status__in: 5 ctx1ad?@c

705 rerun__status__in = [x for x in TASK_STATUS_SET if x != 'Abandoned'] 

706 assert self.eid 5 ctx1ad?@c

707 if self.one is None: 5 ctx1ad?@c

708 _logger.warning('No ONE instance found for Alyx connection, set the one property') 

709 return 

710 tasks_alyx_pre = self.one.alyx.rest('tasks', 'list', session=self.eid, graph=self.name, no_cache=True) 5 ctx1ad?@c

711 tasks_alyx = [] 5 ctx1ad?@c

712 # creates all the tasks by iterating through the ordered dict 

713 

714 if tasks_list is not None: 5 ctx1ad?@c

715 task_items = tasks_list 

716 # need to add in the session eid and the parents 

717 else: 

718 task_items = self.tasks.values() 5 ctx1ad?@c

719 

720 for t in task_items: 5 ctx1ad?@c

721 # get the parents' alyx ids to reference in the database 

722 if isinstance(t, dict): 5 ctx1ad?@c

723 t = Bunch(t) 

724 executable = t.executable 

725 arguments = t.arguments 

726 t['time_out_secs'] = t['time_out_sec'] 

727 if len(t.parents) > 0: 

728 pnames = t.parents 

729 else: 

730 executable = self._get_exec_name(t) 5 ctx1ad?@c

731 arguments = t.kwargs 5 ctx1ad?@c

732 if len(t.parents): 5 ctx1ad?@c

733 pnames = [p.name for p in t.parents] 5 ctx1ad?@c

734 

735 if len(t.parents): 5 ctx1ad?@c

736 parents_ids = [ta['id'] for ta in tasks_alyx if ta['name'] in pnames] 5 ctx1ad?@c

737 else: 

738 parents_ids = [] 5 ctx1ad?@c

739 

740 task_dict = {'executable': executable, 'priority': t.priority, 5 ctx1ad?@c

741 'io_charge': t.io_charge, 'gpu': t.gpu, 'cpu': t.cpu, 

742 'ram': t.ram, 'module': self.label, 'parents': parents_ids, 

743 'level': t.level, 'time_out_sec': t.time_out_secs, 'session': self.eid, 

744 'status': 'Waiting', 'log': None, 'name': t.name, 'graph': self.name, 

745 'arguments': arguments} 

746 if self.data_repo: 5 ctx1ad?@c

747 task_dict.update({'data_repository': self.data_repo}) 

748 # if the task already exists, patch it otherwise, create it 

749 talyx = next(filter(lambda x: x['name'] == t.name, tasks_alyx_pre), []) 5 ctx1ad?@c

750 if len(talyx) == 0: 5 ctx1ad?@c

751 talyx = self.one.alyx.rest('tasks', 'create', data=task_dict) 5 ctx1ad?@c

752 elif talyx['status'] in rerun__status__in: 3 ctx1a?@

753 talyx = self.one.alyx.rest('tasks', 'partial_update', id=talyx['id'], data=task_dict) 

754 tasks_alyx.append(talyx) 5 ctx1ad?@c

755 return tasks_alyx 5 ctx1ad?@c

756 

757 def create_tasks_list_from_pipeline(self): (empty)

758 """ 

759 From a pipeline with tasks, creates a list of dictionaries containing task description that can be used to upload to 

760 create alyx tasks 

761 :return: 

762 """ 

763 tasks_list = [] 13 ctx2? @ cb) * + , - . / : ; =

764 for k, t in self.tasks.items(): 13 ctx2? @ cb) * + , - . / : ; =

765 # get the parents' alyx ids to reference in the database 

766 if len(t.parents): 13 ctx2? @ cb) * + , - . / : ; =

767 parent_names = [p.name for p in t.parents] 13 ctx2? @ cb) * + , - . / : ; =