Coverage for src/fsl_pipe/pipeline.py: 70%
438 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-22 15:00 +0100
1"""
2Defines the pipeline before a FileTree is provided.
4A pipeline is a collection of functions with mapping from the function parameters to input/output/reference/placeholder basenames.
5"""
7from typing import List, Optional, Set, Tuple, Dict, Collection, Callable, Union, Any
8import argparse
9from dataclasses import dataclass
10from file_tree import FileTree
11from copy import copy
12import inspect
13import xarray
14from collections import namedtuple
15from fsl_sub import submit
16from file_tree.template import is_singular
17from pathlib import Path
18from numpy import unique
19import graphviz
20import fnmatch
22try:
23 import datalad
24except ImportError:
25 datalad = None
28class Pipeline:
29 """Collection of python functions forming a pipeline.
31 You can either create a new pipeline (`from fsl_pipe import Pipeline; pipe = Pipeline()`) or use a pre-existing one (`from fsl_pipe import pipe`)
33 Scripts are added to a pipeline by using the pipeline as a decorator (see :meth:`__call__`).
35 To run the pipeline based on instructions from the command line run :meth:`pipe.cli(tree) <cli>`,
36 where tree is a FileTree defining the directory structure of the pipeline input & output files.
38 :ivar scripts: list of :class:`PipedFunction`, which define the python functions forming the pipeline and their input/output templates
39 """
41 def __init__(self, scripts=None, default_output=None, default_submit=None):
42 """Create a new empty pipeline."""
43 if scripts is None:
44 scripts = []
45 if default_output is None:
46 default_output = []
47 if default_submit is None:
48 default_submit = {}
49 self.scripts: List[PipedFunction] = list(scripts)
50 self.default_output: List[str] = list(default_output)
51 check_submit_parameters(default_submit)
52 self.default_submit = default_submit
54 def __call__(
55 self,
56 function=None,
57 *,
58 kwargs=None,
59 no_iter=None,
60 placeholders=None,
61 as_path=False,
62 submit=None,
63 batch=None,
64 intermediate=False,
65 ):
66 """Add a python function as a :class:`PipedFunction` to the pipeline.
68 This is the main route through which jobs are added to the pipeline.
70 .. code-block:: python
72 from fsl_pipe import pipe, In, Out, Ref, Var
74 @pipe(submit=dict(jobtime=40))
75 def func(in_path: In, out_path: Out, ref_path: Ref, placeholder_key: Var):
76 pass
78 :param function: Optionally provide the function directly. This can be useful when not using `pipe` as a decorator, e.g.:
80 .. code-block:: python
82 from fsl_pipe import pipe, In, Out, Ref, Var
83 from shutil import copyfile
85 pipe(copyfile, kwargs={'src': In('template_in'), 'dst': Out('template_out')})
87 :param kwargs: maps function keyword arguments to template names (In/Out/Ref), placeholder names (Var), or anything else. Templates or placeholders are replaced by their values based on the file-tree. Anything else is passed on unaltered.
88 :param no_iter: optional set of parameters not to iterate over.
89 :param as_path: if set to true, provide template filenames to the function as `pathlib.Path` objects instead of strings.
90 :param placeholders: dictionary overriding the placeholders set in the filetree for this specific function. This can, for example, be used to run this function on a sub-set of subjects.
91 :param submit: dictionary with the flags to pass on to `fsl_sub.submit`, when submitting jobs to a cluster queue.
92 :param batch: label used to batch multiple jobs into one when submitting to the cluster
93 """
94 if submit is None:
95 submit = {}
96 submit_params = dict(**self.default_submit, **submit)
97 check_submit_parameters(submit_params)
99 def wrapper(func):
100 self.scripts.append(
101 PipedFunction(
102 func,
103 submit_params=submit_params,
104 kwargs=kwargs,
105 no_iter=no_iter,
106 override=placeholders,
107 as_path=as_path,
108 batch=batch,
109 intermediate=intermediate,
110 )
111 )
112 return func
114 if function is None:
115 return wrapper
116 wrapper(function)
117 return function
119 def generate_jobs(self, tree: FileTree):
120 """
121 Split the pipeline into individual jobs.
123 Produces an unsorted `JobList`, which can be sorted by running `filter` on it.
125 :param tree: set of templates for the input/output/reference files with all possible placeholder values
126 """
127 from .job import JobList, FileTarget
129 all_targets: Dict[str, FileTarget] = {}
130 jobs = []
131 for script in self.scripts:
132 jobs.extend(script.get_jobs(tree, all_targets))
133 return JobList(tree, jobs, all_targets)
135 def default_parser(
136 self,
137 tree: FileTree,
138 parser: Optional[argparse.ArgumentParser] = None,
139 include_vars=None,
140 exclude_vars=(),
141 ) -> argparse.ArgumentParser:
142 """
143 Add default `fsl-pipe` arguments to an argument parser (will create a new argument parser if needed).
145 :param tree: `file_tree.FileTree` object describing the directory structure for the input/output files (defaults to datalad tree).
146 :param parser: Optional argument parser that will be updated with the default `fsl-pipe` arguments (by default a new one is created).
147 :param include_vars: if provided, only include expose placeholders in this list to the command line
148 :param exclude_vars: exclude placeholders in this list from the command line
149 :return: Argument parser with the default `fsl-pipe` arguments. If one was provided as input, that one will be returned.
150 """
151 from .job import RunMethod
153 if parser is None:
154 parser = argparse.ArgumentParser(description="Runs the pipeline")
155 if len(self.scripts) == 0:
156 raise ValueError("The pipeline does not contain any scripts...")
157 templates = set.union(
158 *[
159 script.filter_templates(True, tree.template_keys())
160 for script in self.scripts
161 ]
162 )
164 if len(self.default_output) == 0:
165 default_output = list(sorted(templates))
166 else:
167 default_output = self.default_output
169 parser.add_argument(
170 "templates",
171 nargs="*",
172 default=default_output,
173 help=f'Set to one or more template keys or file patterns (e.g., "*.txt"). Only these templates/files will be produced. If not provided all templates will be produced ({", ".join(default_output)}).',
174 )
175 parser.add_argument(
176 "-m",
177 "--pipeline-method",
178 default=RunMethod.default().name,
179 choices=[m.name for m in RunMethod],
180 help=f"method used to run the jobs (default: {RunMethod.default().name})",
181 )
182 parser.add_argument(
183 "-o",
184 "--overwrite",
185 action="store_true",
186 help="If set overwrite any requested files.",
187 )
188 parser.add_argument(
189 "-d",
190 "--overwrite_dependencies",
191 action="store_true",
192 help="If set also overwrites dependencies of requested files.",
193 )
194 parser.add_argument(
195 "-j",
196 "--job-hold",
197 default="",
198 help="Place a hold on the whole pipeline until job has completed.",
199 )
200 parser.add_argument(
201 "--skip-missing",
202 action="store_true",
203 help="If set skip running any jobs depending on missing data. This replaces the default behaviour of raising an error if any required input data is missing.",
204 )
205 parser.add_argument(
206 "-q",
207 "--quiet",
208 action="store_true",
209 help="Suppresses the report on what will be run (might speed up starting up the pipeline substantially).",
210 )
211 parser.add_argument(
212 "-b",
213 "--batch",
214 action="store_true",
215 help="Batch jobs based on submit parameters and pipeline labels before running them.",
216 )
217 parser.add_argument(
218 "--scale-jobtime",
219 type=float,
220 default=1.0,
221 help="Scale the developer-set job times by the given value. Set to a number above 1 if you are analysing a particularly large datasets and the individual jobs might be slower than expected. Only affects jobs submitted to a cluster (i.e., `--pipeline_method=submit`).",
222 )
223 gui_help = "Start a GUI to select which output files should be produced by the pipeline."
224 try:
225 import fsl_pipe_gui # noqa: F401
226 except ModuleNotFoundError:
227 gui_help = (
228 gui_help
229 + " Missing requirement 'fsl-pipe-gui' to run the GUI. Please install this with `conda/pip install fsl-pipe-gui`."
230 )
231 parser.add_argument("-g", "--gui", action="store_true", help=gui_help)
232 if datalad is not None:
233 parser.add_argument(
234 "--datalad",
235 action="store_true",
236 help="run datalad get on all input data before running/submitting the jobs",
237 )
239 def add_placeholder_flag(variable):
240 if not isinstance(variable, str):
241 for v in variable:
242 add_placeholder_flag(v)
243 return
244 if "/" in variable:
245 return
246 if variable in exclude_vars:
247 return
248 if include_vars is not None and variable not in include_vars:
249 return
250 value = tree.placeholders[variable]
251 default = (
252 str(value)
253 if is_singular(value)
254 else ",".join([str(v) for v in unique(value)])
255 )
256 parser.add_argument(
257 f"--{variable}",
258 nargs="+",
259 help=f"Use to set the possible values of {variable} to the selected values (default: {default})",
260 )
262 add_placeholder_flag(tree.placeholders.keys())
264 return parser
266 def run_cli(self, args: argparse.Namespace, tree: FileTree):
267 """
268 Run the pipeline based on arguments extracted from the default argument parser.
270 :param args: Namespace consisting of arguments extracted from the command line (produced by `argparse.ArgumentParser.parse_args`). This is expected to contain:
272 - templates (defaults to `self.default_output`): list of which templates the pipeline should produce
273 - pipeline_method: string with method used to run the jobs
274 - overwrite: boolean, which if true overwrite any requested files
275 - overwrite_dependencies: boolean, which if true also overwrites dependencies of requested files
276 - job-hold: string with comma-separated list, which contains job(s) that will be waited for
277 - skip-missing: whether to skip jobs depending on missing data instead of raising an error
278 - datalad (defaults to False): if true, run datalad get on all input data before running/submitting the jobs
279 - {placeholder} (defaults to values in FileTree): sequence of strings overwriting the possible values for a particular placeholder
281 :param tree: Definition of pipeline input/output files
282 """
283 from .job import RunMethod
285 tree = tree.copy()
287 def set_placeholder(key):
288 if not isinstance(key, str):
289 for k in key:
290 set_placeholder(k)
291 return
292 if getattr(args, key, None) is not None:
293 tree.placeholders[key] = args.__dict__[key]
295 for var in tree.placeholders.keys():
296 set_placeholder(var)
298 if args.gui:
299 self.gui(
300 tree,
301 overwrite_dependencies=args.overwrite_dependencies,
302 run_method=RunMethod[args.pipeline_method],
303 )
304 return
305 requested_templates = getattr(
306 args,
307 "templates",
308 None if len(self.default_output) == 0 else self.default_output,
309 )
310 concrete = self.generate_jobs(tree)
311 torun = concrete.filter(
312 requested_templates,
313 overwrite=args.overwrite,
314 overwrite_dependencies=args.overwrite_dependencies,
315 skip_missing=args.skip_missing,
316 )
317 if not args.quiet:
318 torun.report()
319 if getattr(args, "batch", False):
320 use_label = any(job.batch is not None for job in concrete.jobs)
321 torun = torun.batch(use_label=use_label)
322 torun.scale_jobtime(args.scale_jobtime)
323 if datalad is not None and args.datalad:
324 torun.run_datalad()
325 torun.run(
326 RunMethod[args.pipeline_method],
327 wait_for=() if args.job_hold == "" else args.job_hold.split(","),
328 )
330 def cli(
331 self,
332 tree: Optional[FileTree] = None,
333 include_vars=None,
334 exclude_vars=(),
335 cli_arguments=None,
336 ):
337 """
338 Run the pipeline from the command line.
340 :param tree: `file_tree.FileTree` object describing the directory structure for the input/output files (defaults to datalad tree).
341 :param include_vars: if provided, only include expose variables in this list to the command line
342 :param exclude_vars: exclude variables in this list from the command line
343 :param cli_arguments: list of command line arguments. If not set the arguments used in the python call (`sys.argv`) are used.
344 """
345 if tree is None:
346 if datalad is None:
347 raise ValueError(
348 "Argument 'tree' missing: please provide a FileTree describing the directory structure for the pipeline"
349 )
350 from .datalad import get_tree
352 try:
353 tree = get_tree()
354 except IOError:
355 raise ValueError(
356 "Pipeline run outside of a datalad dataset, so a FileTree needs to be explicitly provided using the 'tree' argument"
357 )
358 if tree is None:
359 raise ValueError("No reference FileTree for pipeline found")
361 parser = self.default_parser(
362 tree=tree, include_vars=include_vars, exclude_vars=exclude_vars
363 )
364 args = parser.parse_args(cli_arguments)
365 self.run_cli(args, tree)
367 def gui(self, tree: FileTree, **kwargs):
368 """
369 Run the fsl-pipe-gui interface to select pipeline output.
371 :param tree: `file_tree.FileTree` object describing the directory structure for the input/output files.
372 :param overwrite_depencencies: set to True to overwrite dependencies
373 :param run_method: overrides the default method to run the jobs
374 """
375 try:
376 from fsl_pipe_gui import run_gui
377 except ModuleNotFoundError:
378 raise ModuleNotFoundError(
379 "'fsl-pipe-gui' needs to be installed to run the graphical user interface. Please run `pip/conda install fsl-pipe-gui` and try again."
380 )
381 run_gui(self, tree, **kwargs)
383 def move_to_subtree(self, sub_tree=None, other_mappings=None):
384 """
385 Create a new pipeline that runs in a sub-tree of a larger tree rather than at the top level.
387 :param sub_tree: name of the sub-tree in the FileTree
388 :param other_mappings: other mappings between templates or placeholder names and their new values
389 """
390 if other_mappings is None:
391 other_mappings = {}
392 all_scripts = [
393 script.move_to_subtree(sub_tree, other_mappings) for script in self.scripts
394 ]
395 new_default_submit = _update_kwargs(
396 self.default_submit, sub_tree, other_mappings
397 )
398 return Pipeline(
399 all_scripts,
400 [_update_key(key, sub_tree, other_mappings) for key in self.default_output],
401 new_default_submit,
402 )
404 @classmethod
405 def merge(cls, pipelines: Collection["Pipeline"]):
406 """
407 Combine multiple pipelines into a single one.
409 :param pipelines: pipelines containing part of the jobs
410 :return: parent pipeline containing all of the jobs in pipelines
411 """
412 new_pipeline = Pipeline()
413 for pipeline in pipelines:
414 new_pipeline.scripts.extend([s.copy() for s in pipeline.scripts])
415 new_pipeline.default_output.extend(pipeline.default_output)
416 return new_pipeline
418 def find(self, function: Union[str, Callable]):
419 """
420 Iterate over any pipeline scripts that run the provided function.
422 Either the function itself or the name of the function can be given.
423 """
424 for script in list(self.scripts):
425 if (
426 script.function == function
427 or getattr(script.function, "__name__", None) == function
428 ):
429 yield script
431 def remove(self, function: Union[str, Callable]):
432 """
433 Remove any pipeline scripts that run the provided function from the pipeline.
435 Either the function itself or the name of the function can be given.
436 """
437 for script in self.find(function):
438 self.scripts.remove(script)
440 def configure(self, kwargs):
441 """
442 Override the values passed on to the keyword arguments of all the scripts.
444 Any keywords not expected by a script will be silently skipped for that script.
445 """
446 for script in self.scripts:
447 try:
448 script.configure(kwargs, allow_new_keywords=False, check=False)
449 except KeyError:
450 pass
452 def add_to_graph(self, graph: graphviz.Digraph = None, tree: FileTree = None):
453 """
454 Add all the pipeline functions to the provided graph.
456 :param graph: GraphViz graph object (will be altered)
457 :param tree: concrete FileTree
458 """
459 if graph is None:
460 graph = graphviz.Digraph("pipeline", format="svg")
461 placeholder_color = {}
462 for idx, script in enumerate(self.scripts):
463 script.add_node(graph, idx, tree, placeholder_color)
464 return graph
467class PipedFunction:
468 """Represents a function stored in a pipeline."""
470 def __init__(
471 self,
472 function,
473 submit_params: Dict,
474 kwargs=None,
475 no_iter=None,
476 override=None,
477 as_path=True,
478 batch=None,
479 intermediate=False,
480 ):
481 """
482 Wrap a function with additional information to run it in a pipeline.
484 :param function: python function that will be run in pipeline.
485 :param submit_params: parameters to submit job running python function to cluster using `fsl_sub`.
486 :param kwargs: maps function keyword arguments to templates, variables, or actual values.
487 :param no_iter: which parameters to not iterate over (i.e., they are passed to the function in an array).
488 :param override: dictionary overriding the placeholders set in the filetree.
489 :param as_path: whether to pass on pathlib.Path objects instead of strings to the functions (default: True).
490 :param batch: label used to batch multiple jobs into one when submitting to the cluster.
491 :param intermediate: whether the output files of this function are intermediate files. Intermediate files will not be reproduced if they are missing when checking which jobs need to be run.
492 """
493 self.function = function
494 self.submit_params = submit_params
495 if override is None:
496 override = {}
497 self.override = override
498 self.as_path = as_path
499 self.batch = batch
500 self.intermediate = intermediate
502 self.all_kwargs: Dict[str, Any] = {}
503 self.configure(function.__annotations__)
505 if kwargs is not None:
506 self.configure(kwargs)
508 if no_iter is None:
509 no_iter = set()
510 elif isinstance(no_iter, str):
511 no_iter = [no_iter]
512 self.explicit_no_iter = set(no_iter)
514 def copy(
515 self,
516 ):
517 """Create a copy of this PipedFunction for pipeline merging."""
518 new_script = copy(self)
519 new_script.all_kwargs = dict(self.all_kwargs)
520 self.explicit_no_iter = set(self.explicit_no_iter)
521 return new_script
523 @property
524 def no_iter(
525 self,
526 ) -> Set[str]:
527 """Sequence of placeholder names that should not be iterated over."""
528 res = {
529 key if value.key is None else value.key
530 for key, value in self.placeholders.items()
531 if value.no_iter
532 }
533 res.update(self.explicit_no_iter)
534 return res
536 def configure(self, kwargs, allow_new_keywords=True, check=True):
537 """Override the values passed on to the keyword arguments of the script.
539 :param kwargs: new placeholders/templates/values for keyword arguments
540 :param allow_new_keywords: if set to False, don't allow new keywords
541 """
542 bad_keys = []
543 signature = inspect.signature(self.function)
544 has_kws = any(
545 param.kind == param.VAR_KEYWORD for param in signature.parameters.values()
546 )
548 if not allow_new_keywords:
549 existing_kws = set(signature.parameters)
550 if has_kws:
551 existing_kws.update(self.all_kwargs.keys())
552 if check:
553 bad_keys = {
554 key for key in kwargs.keys() if key not in signature.parameters
555 }
556 raise KeyError(
557 f"Tried to configure {self} with keys that are not expected by the function: {bad_keys}"
558 )
559 for key, value in kwargs.items():
560 if not allow_new_keywords and key not in existing_kws:
561 continue
562 self.all_kwargs[key] = value
564 @property
565 def placeholders(
566 self,
567 ):
568 """Return dictionary with placeholder values overriden for this function."""
569 return {
570 key: value
571 for key, value in self.all_kwargs.items()
572 if isinstance(value, PlaceHolder)
573 }
575 @property
576 def templates(
577 self,
578 ):
579 """Return dictionary with templates used as input, output, or reference for this function."""
580 return {
581 key: value
582 for key, value in self.all_kwargs.items()
583 if isinstance(value, Template)
584 }
586 @property
587 def kwargs(
588 self,
589 ):
590 """Return dictionary with keyword arguments that will be passed on to the function."""
591 return {
592 key: value
593 for key, value in self.all_kwargs.items()
594 if not any(isinstance(value, cls) for cls in (Template, PlaceHolder))
595 }
597 def filter_templates(self, output=False, all_templates=None) -> Set[str]:
598 """
599 Find all input or output template keys.
601 :param output: if set to True select the input rather than output templates
602 :param all_templates: sequence of all possible templates (required if any Template keys use globbing)
603 :return: set of input or output templates
604 """
605 res = set()
606 for kwarg_key, template in self.templates.items():
607 if (template.input and not output) or (template.output and output):
608 template_key = kwarg_key if template.key is None else template.key
610 # apply globbing if template key contains * or ?
611 if "*" in template_key or "?" in template_key:
612 if all_templates is None:
613 raise ValueError(
614 f"Template {template_key} uses globbing, but no template keys are not provided"
615 )
616 res.update(fnmatch.filter(all_templates, template_key))
617 else:
618 res.add(template_key)
619 return res
621 def iter_over(self, tree: FileTree) -> Tuple[str, ...]:
622 """
623 Find all the placeholders that should be iterated over before calling the function.
625 These are all the placeholders that affect the input templates, but are not part of `self.no_iter`.
627 :param tree: set of templates with placeholder values
628 :return: placeholder names to be iterated over sorted by name
629 """
630 tree = tree.update(**self.override)
631 in_vars = self.all_placeholders(tree, False)
632 in_vars_linked = {tree.placeholders.linkages.get(key, key) for key in in_vars}
633 out_vars = {
634 tree.placeholders.linkages.get(key, key)
635 for key in self.all_placeholders(tree, True)
636 }
638 updated_no_iter = {
639 tree.placeholders.linkages.get(
640 tree.placeholders.find_key(key),
641 tree.placeholders.find_key(key),
642 )
643 for key in self.no_iter
644 }
645 all_in = in_vars_linked.union(updated_no_iter)
646 if len(out_vars.difference(all_in)) > 0:
647 raise ValueError(
648 f"{self}: Output template depends on {out_vars.difference(all_in)}, which none of the input templates depend on"
649 )
650 return tuple(
651 sorted(
652 {
653 v
654 for v in in_vars
655 if tree.placeholders.linkages.get(v, v) not in updated_no_iter
656 },
657 )
658 )
660 def get_jobs(self, tree: FileTree, all_targets: Dict):
661 """
662 Get a list of all individual jobs defined by this function.
664 :param tree: set of templates with placeholder values
665 :param all_targets: mapping from filenames to Target objects used to match input/output filenames between jobs
666 :return: sequence of jobs
667 """
668 from .job import SingleJob
670 tree = tree.update(**self.override)
671 to_iter = self.iter_over(tree)
672 jobs = []
673 done_kwargs = set()
675 def freeze_value(value):
676 if isinstance(value, xarray.DataArray):
677 return str(value)
678 elif isinstance(value, dict):
679 return frozenset(value.items())
680 else:
681 return value
683 for sub_tree in tree.iter_vars(to_iter):
684 kwargs, in_fns, out_fns, optionals = _single_job_kwargs(
685 self.all_kwargs,
686 sub_tree,
687 tree,
688 all_targets,
689 as_path=self.as_path,
690 mark_output_intermediate=self.intermediate,
691 )
692 job_submit_params, _, _, _ = _single_job_kwargs(
693 self.submit_params,
694 sub_tree,
695 tree,
696 all_targets,
697 as_path=False,
698 is_submit_param=True,
699 )
701 frozen_kwargs = frozenset(
702 [(key, freeze_value(value)) for (key, value) in kwargs.items()]
703 )
704 if frozen_kwargs in done_kwargs:
705 continue
706 done_kwargs.add(frozen_kwargs)
708 jobs.append(
709 SingleJob(
710 self.function,
711 kwargs,
712 job_submit_params,
713 in_fns,
714 out_fns,
715 optionals,
716 {name: sub_tree.placeholders.get(name, None) for name in to_iter},
717 self.batch,
718 )
719 )
720 return jobs
722 def all_placeholders(self, tree: FileTree, output=False) -> Set[str]:
723 """
724 Identify the multi-valued placeholders affecting the input/output templates of this function.
726 :param tree: set of templates with placeholder values
727 :param output: if set to True returns the placeholders for the output than input templates
728 :return: set of all placeholders that affect the input/output templates
729 """
730 res = set()
731 for t in self.filter_templates(output, tree.template_keys()):
732 res.update(tree.get_template(t).placeholders())
733 if not output:
734 for key, variable in self.placeholders.items():
735 res.add(key if variable.key is None else variable.key)
737 bad_keys = {key for key in res if key not in tree.placeholders}
738 if len(bad_keys) > 0:
739 raise ValueError(f"No value set for placeholders: {bad_keys}")
741 _, only_multi = tree.placeholders.split()
742 return {only_multi.find_key(key) for key in res if key in only_multi}
744 def move_to_subtree(self, sub_tree=None, other_mappings=None):
745 """
746 Create a new wrapped function that runs in a sub-tree of a larger tree rather than at the top level.
748 :param sub_tree: name of the sub-tree in the FileTree
749 :param other_mappings: other mappings between templates or placeholder names and their new values
750 """
751 if other_mappings is None:
752 other_mappings = {}
753 new_script = copy(self)
754 new_script.all_kwargs = _update_kwargs(
755 self.all_kwargs, sub_tree, other_mappings
756 )
757 new_script.override = {
758 _update_key(key, sub_tree, other_mappings): value
759 for key, value in self.override.items()
760 }
761 new_script.submit_params = _update_kwargs(
762 self.submit_params, sub_tree, other_mappings
763 )
764 return new_script
766 def __repr__(
767 self,
768 ):
769 """Print function name."""
770 return f"PipedFunction({self.function.__name__})"
772 def add_node(
773 self,
774 graph: graphviz.Graph,
775 index,
776 tree: FileTree,
777 placeholder_color: Dict[str, str],
778 ):
779 """
780 Add a node representing this function for a pipeline diagram.
782 :param graph: input pipeline diagram (will be altered)
783 :param index: unique integer identifier to use within the graph
784 """
785 seaborn_colors = [
786 "#0173b2",
787 "#de8f05",
788 "#029e73",
789 "#d55e00",
790 "#cc78bc",
791 "#ca9161",
792 "#fbafe4",
793 "#949494",
794 "#ece133",
795 "#56b4e9",
796 ]
797 identifier = str(index) + self.function.__name__
798 label = self.function.__name__
799 if tree is not None:
800 placs = self.iter_over(tree)
801 if len(placs) > 0:
802 for p in placs:
803 if p not in placeholder_color:
804 placeholder_color[p] = seaborn_colors[
805 len(placeholder_color) % len(seaborn_colors)
806 ]
807 plac_text = ";".join(
808 f'<font color="{placeholder_color[p]}">{p}</font>' for p in placs
809 )
810 label = rf"<{label}<br/>{plac_text}>"
812 graph.node(identifier, label=label, color="red", shape="box")
813 for output in (False, True):
814 for t in self.filter_templates(output, tree.template_keys()):
815 label = t
816 if tree is not None:
817 all_plac = (
818 tree.get_template(t).required_placeholders()
819 | tree.get_template(t).optional_placeholders()
820 )
821 multi_plac = sorted(
822 [
823 p
824 for p in all_plac
825 if p not in tree.placeholders
826 or not is_singular(tree.placeholders[p])
827 ]
828 )
829 for p in multi_plac:
830 if p not in placeholder_color:
831 placeholder_color[p] = seaborn_colors[
832 len(placeholder_color) % len(seaborn_colors)
833 ]
834 multi_plac = ";".join(
835 f'<font color="{placeholder_color[p]}">{p}</font>'
836 for p in multi_plac
837 )
838 if len(multi_plac) > 0:
839 label = rf"<{t}<br/>{multi_plac}>"
840 graph.node(t, label, color="blue")
841 if output:
842 graph.edge(identifier, t)
843 else:
844 graph.edge(t, identifier)
847@dataclass
848class Template(object):
849 """Represents a keyword argument that will be mapped to a template path in the FileTree.
851 :param key: template key in FileTree.
852 :param input: Set to true if file should be considered as input (i.e., it should exist before the job is run).
853 :param output: Set to true if file should be considered as output (i.e., it is expected to exist after the job is run).
854 :param optional: Set to true if input/output file should be considered for creating the dependency graph, but still might not exist.
855 :param intermediate: Set to true if file is an intermediate file, which should not trigger re-running of jobs if it is missing.
856 """
858 key: Optional[str] = None
859 input: bool = False
860 output: bool = False
861 optional: bool = False
862 intermediate: bool = False
864 def __call__(self, key=None, optional=None, intermediate=None):
865 """Override the template key and whether it is optional."""
866 return Template(
867 key,
868 self.input,
869 self.output,
870 self.optional if optional is None else optional,
871 self.intermediate if intermediate is None else intermediate,
872 )
875@dataclass
876class PlaceHolder(object):
877 """Represents a keyword argument that will be mapped to a placeholder value.
879 :param key: placeholder key in FileTree.
880 :param no_iter: if True the pipeline will not iterate over individual placeholder values, but rather run a single job with all possible placeholder values at once.
881 """
883 key: Optional[str] = None
884 no_iter: bool = False
886 def __call__(self, key=None, no_iter=None):
887 """Override the Placeholder."""
888 if no_iter is None:
889 no_iter = self.no_iter
890 return PlaceHolder(key, no_iter)
893"""Use to mark keyword arguments that represent input filenames of the wrapped function.
895These filenames are expected to exist before the function is run.
897The actual filename is extracted from the FileTree based on the template key.
898This template key will by default match the keyword argument name, but can be overriden by calling `In` (e.g., `In("other_key")`).
899"""
900In = Template(input=True)
902"""Use to mark keyword arguments that represent output filenames of the wrapped function.
904These filenames are expected to exist after the function is run.
905An error is raised if they do not.
907The actual filename is extracted from the FileTree based on the template key.
908This template key will by default match the keyword argument name, but can be overriden by calling `In` (e.g., `In("other_key")`).
909"""
910Out = Template(output=True)
912"""Use to mark keyword arguments that represent reference filenames of the wrapped function.
914These filenames might or might not exist before or after the job is run.
916The actual filename is extracted from the FileTree based on the template key.
917This template key will by default match the keyword argument name, but can be overriden by calling `In` (e.g., `In("other_key")`).
918"""
919Ref = Template()
921"""Use to mark keyword arguments that represent placeholder values in the pipeline.
923Placeholder values are returned as :any:`PlaceholderValue` objects.
924"""
925Var = PlaceHolder()
928def to_templates_dict(input_files=(), output_files=(), reference_files=()):
929 """Convert a sequence of input/output/reference files into a template dictionary.
931 Args:
932 input_files (sequence, optional): Template keys representing input files. Defaults to ().
933 output_files (sequence, optional): Template keys representing output files. Defaults to ().
934 reference_files (sequence, optional): Template keys representing reference paths. Defaults to ().
936 Raises:
937 KeyError: If the same template key is used as more than one of the input/output/reference options
939 Returns:
940 dict: mapping of the keyword argument names to the Template objects
941 """
942 res = {}
943 for files, cls in [
944 (input_files, In),
945 (output_files, Out),
946 (reference_files, Ref),
947 ]:
948 for name in files:
949 if isinstance(name, str):
950 short_name = name.split("/")[-1]
951 else:
952 short_name, name = name
953 if name in res:
954 raise KeyError(f"Dual definition for template {name}")
955 res[short_name] = cls(name)
957 return res
960"""
961Named tuple representing a placeholder value with three fields.
963:param key: string with placeholder key in pipeline.
964:param index: integer with the index (or multiple integers with the indices if `no_iter` is set for this placeholder) of the placeholder values in list of all possible placeholder values.
965:param value: actual placeholder value (or sequence of values if `no_iter is set for this placeholder)
966"""
967PlaceholderValue = namedtuple("PlaceholderValue", ["key", "index", "value"])
970def _single_job_kwargs(
971 wrapper_kwargs,
972 single_tree: FileTree,
973 full_tree: FileTree,
974 all_targets,
975 as_path=False,
976 is_submit_param=False,
977 mark_output_intermediate=False,
978):
979 """Create the keywords, input, and output filenames for a single job."""
980 from .job import get_target
982 final_dict = {}
983 input_filenames = []
984 output_filenames = []
985 optional_filenames = []
986 for kwarg_key, value in wrapper_kwargs.items():
987 if isinstance(value, Template):
988 key = kwarg_key if value.key is None else value.key
990 use_glob = "*" in key or "?" in key
991 iter_keys = (
992 fnmatch.filter(single_tree.template_keys(), key) if use_glob else [key]
993 )
994 res = {}
995 for proc_key in iter_keys:
996 try:
997 proc_res = single_tree.get(proc_key)
998 is_xarray = False
999 except KeyError:
1000 proc_res = single_tree.get_mult(proc_key)
1001 is_xarray = True
1002 if is_xarray:
1003 if is_submit_param:
1004 raise ValueError(
1005 f"Submit parameter {kwarg_key} is set to the template {key}, which has more than one possible value"
1006 )
1007 filenames = [
1008 get_target(fn, all_targets, proc_key)
1009 for fn in proc_res.data.flatten()
1010 ]
1011 if (
1012 value.output and mark_output_intermediate
1013 ) or value.intermediate:
1014 for fn in filenames:
1015 fn.mark_intermediate()
1016 if value.input:
1017 input_filenames.extend(filenames)
1018 if value.output:
1019 output_filenames.extend(filenames)
1020 if value.optional:
1021 optional_filenames.extend(filenames)
1022 proc_res = xarray.apply_ufunc(
1023 Path if as_path else str, proc_res, vectorize=True
1024 )
1025 else:
1026 filename = get_target(proc_res, all_targets, proc_key)
1027 if (
1028 value.output and mark_output_intermediate
1029 ) or value.intermediate:
1030 filename.mark_intermediate()
1031 if value.input:
1032 input_filenames.append(filename)
1033 if value.output:
1034 output_filenames.append(filename)
1035 if value.optional:
1036 optional_filenames.append(filename)
1037 proc_res = (Path if as_path else str)(proc_res)
1038 res[proc_key] = proc_res
1039 if not use_glob:
1040 res = res[key]
1041 elif isinstance(value, PlaceHolder):
1042 key = single_tree.placeholders.find_key(
1043 kwarg_key if value.key is None else value.key
1044 )
1045 plac_value = single_tree.placeholders[key]
1046 ref_values = (
1047 [full_tree.placeholders[key]]
1048 if is_singular(full_tree.placeholders[key])
1049 else list(full_tree.placeholders[key])
1050 )
1052 if is_singular(plac_value):
1053 res = PlaceholderValue(key, ref_values.index(plac_value), plac_value)
1054 if is_submit_param:
1055 res = res.value
1056 else:
1057 res = PlaceholderValue(
1058 key,
1059 tuple(ref_values.index(v) for v in plac_value),
1060 tuple(plac_value),
1061 )
1062 if is_submit_param:
1063 raise ValueError(
1064 f"Submit parameter {kwarg_key} is set to the placeholder {key}, which has more than one possible value"
1065 )
1066 else:
1067 res = value
1068 final_dict[kwarg_key] = res
1069 return final_dict, input_filenames, output_filenames, optional_filenames
1072def check_submit_parameters(submit_params: Dict):
1073 """
1074 Check that the submit parameters are actually valid.
1076 Raises a ValueError if there are any submit parameters set not expected by fsl_sub.
1077 """
1078 signature = inspect.signature(submit)
1079 unrecognised = set(submit_params.keys()).difference(signature.parameters.keys())
1080 if len(unrecognised) > 0:
1081 raise ValueError(f"Unrecognised fsl_sub submit keywords: {unrecognised}")
1082 if "jobhold" in submit_params:
1083 raise ValueError(
1084 "Job-holds are managed by the fsl-pipe and cannot be set by the pipeline definition."
1085 )
1088def _update_key(key, sub_tree, other_mappings):
1089 """Update template key with sub_tree precursor, unless listed in other_mappings."""
1090 if key in other_mappings:
1091 return other_mappings[key]
1092 elif sub_tree is not None:
1093 return sub_tree + "/" + key
1094 else:
1095 return key
1098def _update_kwargs(input_dict, sub_tree, other_mappings):
1099 """Update placeholders and templates in input_dict with sub_tree precursor, unless listed in other_mappings."""
1100 kwargs = {}
1101 for key, value in input_dict.items():
1102 kwargs[key] = value
1103 if isinstance(value, PlaceHolder) or isinstance(value, Template):
1104 use_key = value.key if value.key is not None else key
1105 kwargs[key] = value(_update_key(use_key, sub_tree, other_mappings))
1106 return kwargs
1109pipe = Pipeline()