The coolest part, I think, is how you can dynamically influence the next steps, either by changing parameters based on previous output, or selecting different jobs entirely at runtime.
It's all about how you write your next() method on your workflow objects. For example:
| def next(self, iteration): | |
| last = self.tasks[-1] | |
| if isinstance(last, GrayScaleConvertApplication): | |
| self.add(TricolorizeMultipleImages( | |
| os.path.join(self.output_dir, self.grayscaled_image), | |
| self.copies, self.ncolors, | |
| self.output_dir)) | |
| return Run.State.RUNNING | |
| elif isinstance(last, TricolorizeMultipleImages): | |
| self.add(MergeImagesApplication( | |
| os.path.join(self.output_dir, self.grayscaled_image), | |
| last.warhol_dir, | |
| self.output_image)) | |
| return Run.State.RUNNING | |
| else: | |
| self.execution.returncode = last.execution.returncode | |
| return Run.State.TERMINATED |