backend.lib.preset

  1"""
  2Queue a series of processors at once via a preset
  3"""
  4import abc
  5from backend.lib.processor import BasicProcessor
  6
  7from common.lib.dataset import DataSet
  8
  9
 10class ProcessorPreset(BasicProcessor):
 11	"""
 12	Processor preset
 13	"""	
 14	def process(self):
 15		"""
 16		ALL PRESETS MUST PREPEND 'preset-' TO THEIR TYPE.
 17
 18		This queues a series of post-processors to run in sequence, with an
 19		overarching dataset to which the results of the last processor in the
 20		sequence are copied. The processor pipeline is then attached to the
 21		overarching dataset so it is clear that all processors were run as part
 22		of that particular preset.
 23		"""
 24		pipeline = self.get_processor_pipeline()
 25
 26		pipeline = self.format_linear_pipeline(pipeline)
 27
 28		analysis_pipeline = DataSet(
 29			parameters=pipeline[0]["parameters"],
 30			db=self.db,
 31			type=pipeline[0]["type"],
 32			owner=self.dataset.creator,
 33			is_private=self.dataset.is_private,
 34			parent=self.dataset.key, 
 35			modules=self.modules)
 36		
 37		# give same ownership as parent dataset
 38		analysis_pipeline.copy_ownership_from(self.dataset)
 39
 40		# this starts the pipeline
 41		self.queue.add_job(pipeline[0]["type"], remote_id=analysis_pipeline.key)
 42
 43	def after_process(self):
 44		"""
 45		Run after processing
 46
 47		In this case, this is run immediately after the underlying analyses
 48		have been queued. This overrides the default behaviour which finishes
 49		the DataSet after processing; in this case, it is left 'open' until it
 50		is finished by the last underlying analysis.
 51		"""
 52		self.dataset.update_status("Awaiting completion of underlying analyses...")
 53		self.job.finish()
 54
 55	@abc.abstractmethod
 56	def get_processor_pipeline(self):
 57		"""
 58		Preset pipeline definition
 59
 60		Should return a list of dictionaries, each dictionary having a `type`
 61		key with the processor type ID and a `parameters` key with the
 62		processor parameters. The order of the list is the order in which the
 63		processors are run. Compatibility of processors in the list is not
 64		checked.
 65
 66		:return list: Processor pipeline definition
 67		"""
 68		pass
 69
 70	def format_linear_pipeline(self, pipeline):
 71		"""
 72		Format a linear pipeline to a nested processor parameter set
 73
 74		:param list pipeline:  Linear pipeline
 75		:return list:  Nested pipeline
 76		"""
 77		if not pipeline:
 78			raise ValueError("Pipeline is empty")
 79		
 80		# make sure the last item in the pipeline copies to the preset's dataset
 81		# also make sure there is always a "parameters" key
 82		pipeline = [{"parameters": {}, **p} for p in pipeline.copy()]
 83
 84		pipeline[-1]["parameters"]["attach_to"] = self.dataset.key
 85
 86		# map the linear pipeline to a nested processor parameter set
 87		while len(pipeline) > 1:
 88			last = pipeline.pop()
 89			pipeline[-1]["parameters"]["next"] = [last]
 90	
 91		return pipeline
 92	
 93class ProcessorAdvancedPreset(ProcessorPreset):
 94	"""
 95	Similar to ProcessorPreset, but allows for more advanced processor trees with multiple 
 96	branches and nested processors.
 97	"""
 98	def format_linear_pipeline(self, pipeline):
 99		"""
100		No formatting of pipeline is needed for advanced presets
101		:param list pipeline:  Linear pipeline
102		:return list:  Nested pipeline
103		"""
104		return pipeline
105	
106	def get_processor_pipeline(self):
107		"""
108		Preset pipeline definition
109		Should return a list of dictionaries, each dictionary having a `type`
110		key with the processor type ID and a `parameters` key with the
111		processor parameters. The order of the list is the order in which the
112		processors are run. Compatibility of processors in the list is not
113		"""
114		advanced_pipeline = self.get_processor_advanced_pipeline(attach_to=self.dataset.key)
115		if not advanced_pipeline:
116			raise ValueError("Pipeline is empty")
117		
118		# Ensure one of the processors in the advanced pipeline has the attach_to parameter
119		all_processors = []
120		def collect_processors(processor):
121			if "next" in processor["parameters"]:
122				for sub_processor in processor["parameters"]["next"]:
123					collect_processors(sub_processor)
124			all_processors.append(processor)
125		for processor in advanced_pipeline:
126			collect_processors(processor)
127		
128		if not any("attach_to" in processor["parameters"] for processor in all_processors):
129			raise ValueError("No processor in the advanced pipeline has the attach_to parameter")
130
131		return advanced_pipeline
132	
133	@abc.abstractmethod
134	def get_processor_advanced_pipeline(self, attach_to):
135		"""
136		Advanced preset pipeline definition
137
138		Similar to base class `get_processor_pipeline`, but allows for more advanced
139		processing. This allows multiple processors to be queued in parallel
140		or in a nested structure. (i.e., "next" contains a list of processors
141		to run in sequence after the each processor.)
142		Format a linear pipeline to a nested processor parameter set
143
144		"attach_to" must be added as a parameter to one of the processors. Failure
145		to do so will cause the preset to never finish.
146		:param list pipeline:  Linear pipeline
147		:return list:  Nested pipeline
148		"""
149		pass