# -*- coding: utf-8 -*- # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*- # vi: set ft=python sts=4 ts=4 sw=4 et: """Defines functionality for pipelined execution of interfaces The `Node` class provides core functionality for batch processing. """ from collections import OrderedDict, defaultdict import os import os.path as op from pathlib import Path import shutil import socket from copy import deepcopy from glob import glob from logging import INFO from tempfile import mkdtemp from ... import config, logging from ...utils.misc import flatten, unflatten, str2bool, dict_diff from ...utils.filemanip import ( md5, ensure_list, simplify_list, copyfiles, fnames_presuffix, loadpkl, split_filename, load_json, emptydirs, savepkl, silentrm, ) from ...interfaces.base import ( traits, InputMultiPath, CommandLine, Undefined, DynamicTraitedSpec, Bunch, InterfaceResult, Interface, isdefined, ) from ...interfaces.base.specs import get_filecopy_info from .utils import ( _parameterization_dir, save_hashfile as _save_hashfile, load_resultfile as _load_resultfile, save_resultfile as _save_resultfile, nodelist_runner as _node_runner, strip_temp as _strip_temp, write_node_report, clean_working_directory, merge_dict, evaluate_connect_function, ) from .base import EngineBase logger = logging.getLogger("nipype.workflow") class NodeExecutionError(RuntimeError): """A nipype-specific name for exceptions when executing a Node.""" class Node(EngineBase): """ Wraps interface objects for use in pipeline A Node creates a sandbox-like directory for executing the underlying interface. It will copy or link inputs into this directory to ensure that input data are not overwritten. A hash of the input state is used to determine if the Node inputs have changed and whether the node needs to be re-executed. Examples -------- >>> from nipype import Node >>> from nipype.interfaces import spm >>> realign = Node(spm.Realign(), 'realign') >>> realign.inputs.in_files = 'functional.nii' >>> realign.inputs.register_to_mean = True >>> realign.run() # doctest: +SKIP """ def __init__( self, interface, name, iterables=None, itersource=None, synchronize=False, overwrite=None, needed_outputs=None, run_without_submitting=False, n_procs=None, mem_gb=0.20, **kwargs, ): """ Parameters ---------- interface : interface object node specific interface (fsl.Bet(), spm.Coregister()) name : alphanumeric string node specific name iterables : generator Input field and list to iterate using the pipeline engine for example to iterate over different frac values in fsl.Bet() for a single field the input can be a tuple, otherwise a list of tuples :: node.iterables = ('frac',[0.5,0.6,0.7]) node.iterables = [('fwhm',[2,4]),('fieldx',[0.5,0.6,0.7])] If this node has an itersource, then the iterables values is a dictionary which maps an iterable source field value to the target iterables field values, e.g.: :: inputspec.iterables = ('images',['img1.nii', 'img2.nii']]) node.itersource = ('inputspec', ['frac']) node.iterables = ('frac', {'img1.nii': [0.5, 0.6], 'img2.nii': [0.6, 0.7]}) If this node's synchronize flag is set, then an alternate form of the iterables is a [fields, values] list, where fields is the list of iterated fields and values is the list of value tuples for the given fields, e.g.: :: node.synchronize = True node.iterables = [('frac', 'threshold'), [(0.5, True), (0.6, False)]] itersource: tuple The (name, fields) iterables source which specifies the name of the predecessor iterable node and the input fields to use from that source node. The output field values comprise the key to the iterables parameter value mapping dictionary. synchronize: boolean Flag indicating whether iterables are synchronized. If the iterables are synchronized, then this iterable node is expanded once per iteration over all of the iterables values. Otherwise, this iterable node is expanded once per each permutation of the iterables values. overwrite : Boolean Whether to overwrite contents of output directory if it already exists. If directory exists and hash matches it assumes that process has been executed needed_outputs : list of output_names Force the node to keep only specific outputs. By default all outputs are kept. Setting this attribute will delete any output files and directories from the node's working directory that are not part of the `needed_outputs`. run_without_submitting : boolean Run the node without submitting to a job engine or to a multiprocessing pool """ # Make sure an interface is set, and that it is an Interface if interface is None: raise IOError("Interface must be provided") if not isinstance(interface, Interface): raise IOError("interface must be an instance of an Interface") super(Node, self).__init__(name, kwargs.get("base_dir")) self._interface = interface self._hierarchy = None self._got_inputs = False self._originputs = None self._output_dir = None self.iterables = iterables self.synchronize = synchronize self.itersource = itersource self.overwrite = overwrite self.parameterization = [] self.input_source = {} self.plugin_args = {} self.run_without_submitting = run_without_submitting self._mem_gb = mem_gb self._n_procs = n_procs # Downstream n_procs if hasattr(self._interface.inputs, "num_threads") and self._n_procs is not None: self._interface.inputs.num_threads = self._n_procs # Initialize needed_outputs and hashes self._hashvalue = None self._hashed_inputs = None self._needed_outputs = [] self.needed_outputs = needed_outputs self.config = None if hasattr(self._interface, "write_cmdline"): self._interface.write_cmdline = True @property def interface(self): """Return the underlying interface object""" return self._interface @property def result(self): """Get result from result file (do not hold it in memory)""" return _load_resultfile( op.join(self.output_dir(), "result_%s.pklz" % self.name) ) @property def inputs(self): """Return the inputs of the underlying interface""" return self._interface.inputs @property def outputs(self): """Return the output fields of the underlying interface""" return self._interface._outputs() @property def needed_outputs(self): return self._needed_outputs @needed_outputs.setter def needed_outputs(self, new_outputs): """Needed outputs changes the hash, refresh if changed""" new_outputs = sorted(list(set(new_outputs or []))) if new_outputs != self._needed_outputs: # Reset hash self._hashvalue = None self._hashed_inputs = None self._needed_outputs = new_outputs @property def mem_gb(self): """Get estimated memory (GB)""" if hasattr(self._interface, "estimated_memory_gb"): self._mem_gb = self._interface.estimated_memory_gb logger.warning( 'Setting "estimated_memory_gb" on Interfaces has been ' "deprecated as of nipype 1.0, please use Node.mem_gb." ) return self._mem_gb @property def n_procs(self): """Get the estimated number of processes/threads""" if self._n_procs is not None: return self._n_procs if hasattr(self._interface.inputs, "num_threads") and isdefined( self._interface.inputs.num_threads ): return self._interface.inputs.num_threads return 1 @n_procs.setter def n_procs(self, value): """Set an estimated number of processes/threads""" self._n_procs = value # Overwrite interface's dynamic input of num_threads if hasattr(self._interface.inputs, "num_threads"): self._interface.inputs.num_threads = self._n_procs def output_dir(self): """Return the location of the output directory for the node""" # Output dir is cached if self._output_dir: return self._output_dir # Calculate & cache otherwise if self.base_dir is None: self.base_dir = mkdtemp() outputdir = self.base_dir if self._hierarchy: outputdir = op.join(outputdir, *self._hierarchy.split(".")) if self.parameterization: params_str = ["{}".format(p) for p in self.parameterization] if not str2bool(self.config["execution"]["parameterize_dirs"]): params_str = [_parameterization_dir(p) for p in params_str] outputdir = op.join(outputdir, *params_str) self._output_dir = op.realpath(op.join(outputdir, self.name)) return self._output_dir def set_input(self, parameter, val): """Set interface input value""" logger.debug( "[Node] %s - setting input %s = %s", self.name, parameter, str(val) ) setattr(self.inputs, parameter, deepcopy(val)) def get_output(self, parameter): """Retrieve a particular output of the node""" return getattr(self.result.outputs, parameter, None) def help(self): """Print interface help""" self._interface.help() def is_cached(self, rm_outdated=False): """ Check if the interface has been run previously, and whether cached results are up-to-date. """ outdir = self.output_dir() # The output folder does not exist: not cached if not op.exists(outdir) or not op.exists( op.join(outdir, "result_%s.pklz" % self.name) ): logger.debug('[Node] Not cached "%s".', outdir) return False, False # Check if there are hashfiles globhashes = glob(op.join(outdir, "_0x*.json")) unfinished = [path for path in globhashes if path.endswith("_unfinished.json")] hashfiles = list(set(globhashes) - set(unfinished)) # Update hash hashed_inputs, hashvalue = self._get_hashval() hashfile = op.join(outdir, "_0x%s.json" % hashvalue) logger.debug( "[Node] Hashes: %s, %s, %s, %s", hashed_inputs, hashvalue, hashfile, hashfiles, ) cached = hashfile in hashfiles # No previous hashfiles found, we're all set. if cached and len(hashfiles) == 1: assert hashfile == hashfiles[0] logger.debug('[Node] Up-to-date cache found for "%s".', self.fullname) return True, True # Cached and updated if len(hashfiles) > 1: if cached: hashfiles.remove(hashfile) # Do not clean up the node, if cached logger.warning( "[Node] Found %d previous hashfiles indicating that the working " 'directory of node "%s" is stale, deleting old hashfiles.', len(hashfiles), self.fullname, ) for rmfile in hashfiles: os.remove(rmfile) hashfiles = [hashfile] if cached else [] if not hashfiles: logger.debug('[Node] No hashfiles found in "%s".', outdir) assert not cached return False, False # At this point only one hashfile is in the folder # and we directly check whether it is updated updated = hashfile == hashfiles[0] if not updated: # Report differences depending on log verbosity cached = True logger.info('[Node] Outdated cache found for "%s".', self.fullname) # If logging is more verbose than INFO (20), print diff between hashes loglevel = logger.getEffectiveLevel() if loglevel < INFO: # Lazy logging: only < INFO exp_hash_file_base = split_filename(hashfiles[0])[1] exp_hash = exp_hash_file_base[len("_0x") :] logger.log( loglevel, "[Node] Old/new hashes = %s/%s", exp_hash, hashvalue ) try: prev_inputs = load_json(hashfiles[0]) except Exception: pass else: logger.log(loglevel, dict_diff(prev_inputs, hashed_inputs, 10)) if rm_outdated: os.remove(hashfiles[0]) assert cached # At this point, node is cached (may not be up-to-date) return cached, updated def hash_exists(self, updatehash=False): """ Decorate the new `is_cached` method with hash updating to maintain backwards compatibility. """ # Get a dictionary with hashed filenames and a hashvalue # of the dictionary itself. cached, updated = self.is_cached(rm_outdated=True) outdir = self.output_dir() hashfile = op.join(outdir, "_0x%s.json" % self._hashvalue) if updated: return True, self._hashvalue, hashfile, self._hashed_inputs # Update only possible if it exists if cached and updatehash: logger.debug("[Node] Updating hash: %s", self._hashvalue) _save_hashfile(hashfile, self._hashed_inputs) return cached, self._hashvalue, hashfile, self._hashed_inputs def run(self, updatehash=False): """ Execute the node in its directory. Parameters ---------- updatehash: boolean When the hash stored in the output directory as a result of a previous run does not match that calculated for this execution, updatehash=True only updates the hash without re-running. """ if self.config is None: self.config = {} self.config = merge_dict(deepcopy(config._sections), self.config) outdir = self.output_dir() force_run = self.overwrite or ( self.overwrite is None and self._interface.always_run ) # Check hash, check whether run should be enforced if not isinstance(self, MapNode): logger.info(f'[Node] Setting-up "{self.fullname}" in "{outdir}".') cached, updated = self.is_cached() # If the node is cached, check on pklz files and finish if not force_run and (updated or (not updated and updatehash)): logger.debug("Only updating node hashes or skipping execution") inputs_file = op.join(outdir, "_inputs.pklz") if not op.exists(inputs_file): logger.debug("Creating inputs file %s", inputs_file) savepkl(inputs_file, self.inputs.get_traitsfree()) node_file = op.join(outdir, "_node.pklz") if not op.exists(node_file): logger.debug("Creating node file %s", node_file) savepkl(node_file, self) result = self._run_interface( execute=False, updatehash=updatehash and not updated ) logger.info( '[Node] "%s" found cached%s.', self.fullname, " (and hash updated)" * (updatehash and not updated), ) return result if cached and updated and not isinstance(self, MapNode): logger.debug('[Node] Rerunning cached, up-to-date node "%s"', self.fullname) if not force_run and str2bool( self.config["execution"]["stop_on_first_rerun"] ): raise Exception( 'Cannot rerun when "stop_on_first_rerun" is set to True' ) # Remove any hashfile that exists at this point (re)running. if cached: for outdatedhash in glob(op.join(self.output_dir(), "_0x*.json")): os.remove(outdatedhash) # _get_hashval needs to be called before running. When there is a valid (or seemingly # valid cache), the is_cached() member updates the hashval via _get_hashval. # However, if this node's folder doesn't exist or the result file is not found, then # the hashval needs to be generated here. See #3026 for a larger context. self._get_hashval() # Hashfile while running hashfile_unfinished = op.join(outdir, "_0x%s_unfinished.json" % self._hashvalue) # Delete directory contents if this is not a MapNode or can't resume can_resume = not (self._interface.can_resume and op.isfile(hashfile_unfinished)) if can_resume and not isinstance(self, MapNode): emptydirs(outdir, noexist_ok=True) else: logger.debug( "[%sNode] Resume - hashfile=%s", "Map" * int(isinstance(self, MapNode)), hashfile_unfinished, ) if isinstance(self, MapNode): # remove old json files for filename in glob(op.join(outdir, "_0x*.json")): os.remove(filename) # Make sure outdir is created os.makedirs(outdir, exist_ok=True) # Store runtime-hashfile, pre-execution report, the node and the inputs set. _save_hashfile(hashfile_unfinished, self._hashed_inputs) write_node_report(self, is_mapnode=isinstance(self, MapNode)) savepkl(op.join(outdir, "_node.pklz"), self) savepkl(op.join(outdir, "_inputs.pklz"), self.inputs.get_traitsfree()) try: result = self._run_interface(execute=True) except Exception: logger.warning('[Node] Error on "%s" (%s)', self.fullname, outdir) # Tear-up after error if not silentrm(hashfile_unfinished): logger.warning( """\ Interface finished unexpectedly and the corresponding unfinished hashfile %s \ does not exist. Another nipype instance may be running against the same work \ directory. Please ensure no other concurrent workflows are racing""", hashfile_unfinished, ) raise # Tear-up after success shutil.move(hashfile_unfinished, hashfile_unfinished.replace("_unfinished", "")) write_node_report(self, result=result, is_mapnode=isinstance(self, MapNode)) return result def _get_hashval(self): """Return a hash of the input state""" self._get_inputs() if self._hashvalue is None and self._hashed_inputs is None: self._hashed_inputs, self._hashvalue = self.inputs.get_hashval( hash_method=self.config["execution"]["hash_method"] ) rm_extra = self.config["execution"]["remove_unnecessary_outputs"] if str2bool(rm_extra) and self.needed_outputs: hashobject = md5() hashobject.update(self._hashvalue.encode()) hashobject.update(str(self.needed_outputs).encode()) self._hashvalue = hashobject.hexdigest() self._hashed_inputs.append(("needed_outputs", self.needed_outputs)) return self._hashed_inputs, self._hashvalue def _get_inputs(self): """ Retrieve inputs from pointers to results files. This mechanism can be easily extended/replaced to retrieve data from other data sources (e.g., XNAT, HTTP, etc.,.) """ if self._got_inputs: # Inputs cached return if not self.input_source: # No previous nodes self._got_inputs = True return prev_results = defaultdict(list) for key, info in list(self.input_source.items()): prev_results[info[0]].append((key, info[1])) logger.debug( '[Node] Setting %d connected inputs of node "%s" from %d previous nodes.', len(self.input_source), self.name, len(prev_results), ) for results_fname, connections in list(prev_results.items()): outputs = None try: outputs = _load_resultfile(results_fname).outputs except AttributeError as e: logger.critical("%s", e) if outputs is None: raise NodeExecutionError( """\ Error populating the inputs of node "%s": the results file of the source node \ (%s) does not contain any outputs.""" % (self.name, results_fname) ) for key, conn in connections: output_value = Undefined if isinstance(conn, tuple): value = getattr(outputs, conn[0]) if isdefined(value): output_value = evaluate_connect_function( conn[1], conn[2], value ) else: output_name = conn try: output_value = outputs.trait_get()[output_name] except AttributeError: output_value = outputs.dictcopy()[output_name] logger.debug("output: %s", output_name) try: self.set_input(key, deepcopy(output_value)) except traits.TraitError as e: msg = ( e.args[0], "", "Error setting node input:", "Node: %s" % self.name, "input: %s" % key, "results_file: %s" % results_fname, "value: %s" % str(output_value), ) e.args = ("\n".join(msg),) raise # Successfully set inputs self._got_inputs = True def _update_hash(self): for outdatedhash in glob(op.join(self.output_dir(), "_0x*.json")): os.remove(outdatedhash) _save_hashfile(self._hashvalue, self._hashed_inputs) def _run_interface(self, execute=True, updatehash=False): if updatehash: self._update_hash() return self._load_results() return self._run_command(execute) def _load_results(self): cwd = self.output_dir() try: result = _load_resultfile(op.join(cwd, "result_%s.pklz" % self.name)) except (traits.TraitError, EOFError): logger.debug("Error populating inputs/outputs, (re)aggregating results...") except (AttributeError, ImportError) as err: logger.debug( "attribute error: %s probably using " "different trait pickled file", str(err), ) old_inputs = loadpkl(op.join(cwd, "_inputs.pklz")) self.inputs.trait_set(**old_inputs) else: return result # try aggregating first if not isinstance(self, MapNode): self._copyfiles_to_wd(linksonly=True) aggouts = self._interface.aggregate_outputs( needed_outputs=self.needed_outputs ) runtime = Bunch( cwd=cwd, returncode=0, environ=dict(os.environ), hostname=socket.gethostname(), ) result = InterfaceResult( interface=self._interface.__class__, runtime=runtime, inputs=self._interface.inputs.get_traitsfree(), outputs=aggouts, ) _save_resultfile( result, cwd, self.name, rebase=str2bool(self.config["execution"]["use_relative_paths"]), ) else: logger.debug("aggregating mapnode results") result = self._run_interface() return result def _run_command(self, execute, copyfiles=True): if not execute: try: result = self._load_results() except (FileNotFoundError, AttributeError): # if aggregation does not work, rerun the node logger.info( "[Node] Some of the outputs were not found: " "rerunning node." ) copyfiles = False # OE: this was like this before, execute = True # I'll keep them for safety else: logger.info( '[Node] Cached "%s" - collecting precomputed outputs', self.fullname ) return result outdir = Path(self.output_dir()) if copyfiles: self._originputs = deepcopy(self._interface.inputs) self._copyfiles_to_wd(execute=execute) # Run command: either execute is true or load_results failed. logger.info( f'[Node] Executing "{self.name}" <{self._interface.__module__}' f".{self._interface.__class__.__name__}>" ) # Invoke core run method of the interface ignoring exceptions result = self._interface.run(cwd=outdir, ignore_exception=True) logger.info( f'[Node] Finished "{self.name}", elapsed time {result.runtime.duration}s.' ) exc_tb = getattr(result.runtime, "traceback", None) if not exc_tb: # Clean working directory if no errors dirs2keep = None if isinstance(self, MapNode): dirs2keep = [op.join(outdir, "mapflow")] result.outputs = clean_working_directory( result.outputs, outdir, self._interface.inputs, self.needed_outputs, self.config, dirs2keep=dirs2keep, ) # Store results file under all circumstances _save_resultfile( result, outdir, self.name, rebase=str2bool(self.config["execution"]["use_relative_paths"]), ) if exc_tb: raise NodeExecutionError( f"Exception raised while executing Node {self.name}.\n\n{result.runtime.traceback}" ) return result def _copyfiles_to_wd(self, execute=True, linksonly=False): """copy files over and change the inputs""" filecopy_info = get_filecopy_info(self.interface) if not filecopy_info: # Nothing to be done return logger.debug( "copying files to wd [execute=%s, linksonly=%s]", execute, linksonly ) outdir = self.output_dir() if execute and linksonly: olddir = outdir outdir = op.join(outdir, "_tempinput") os.makedirs(outdir, exist_ok=True) for info in filecopy_info: files = self.inputs.trait_get().get(info["key"]) if not isdefined(files) or not files: continue infiles = ensure_list(files) if execute: if linksonly: if not info["copy"]: newfiles = copyfiles( infiles, [outdir], copy=info["copy"], create_new=True ) else: newfiles = fnames_presuffix(infiles, newpath=outdir) newfiles = _strip_temp( newfiles, op.abspath(olddir).split(op.sep)[-1] ) else: newfiles = copyfiles( infiles, [outdir], copy=info["copy"], create_new=True ) else: newfiles = fnames_presuffix(infiles, newpath=outdir) if not isinstance(files, list): newfiles = simplify_list(newfiles) setattr(self.inputs, info["key"], newfiles) if execute and linksonly: emptydirs(outdir, noexist_ok=True) def update(self, **opts): """Update inputs""" self.inputs.update(**opts) class JoinNode(Node): """Wraps interface objects that join inputs into a list. Examples -------- >>> import nipype.pipeline.engine as pe >>> from nipype import Node, JoinNode, Workflow >>> from nipype.interfaces.utility import IdentityInterface >>> from nipype.interfaces import (ants, dcm2nii, fsl) >>> wf = Workflow(name='preprocess') >>> inputspec = Node(IdentityInterface(fields=['image']), ... name='inputspec') >>> inputspec.iterables = [('image', ... ['img1.nii', 'img2.nii', 'img3.nii'])] >>> img2flt = Node(fsl.ImageMaths(out_data_type='float'), ... name='img2flt') >>> wf.connect(inputspec, 'image', img2flt, 'in_file') >>> average = JoinNode(ants.AverageImages(), joinsource='inputspec', ... joinfield='images', name='average') >>> wf.connect(img2flt, 'out_file', average, 'images') >>> realign = Node(fsl.FLIRT(), name='realign') >>> wf.connect(img2flt, 'out_file', realign, 'in_file') >>> wf.connect(average, 'output_average_image', realign, 'reference') >>> strip = Node(fsl.BET(), name='strip') >>> wf.connect(realign, 'out_file', strip, 'in_file') """ def __init__( self, interface, name, joinsource, joinfield=None, unique=False, **kwargs ): """ Parameters ---------- interface : interface object node specific interface (fsl.Bet(), spm.Coregister()) name : alphanumeric string node specific name joinsource : node name name of the join predecessor iterable node joinfield : string or list of strings name(s) of list input fields that will be aggregated. The default is all of the join node input fields. unique : flag indicating whether to ignore duplicate input values See Node docstring for additional keyword arguments. """ super(JoinNode, self).__init__(interface, name, **kwargs) self._joinsource = None # The member should be defined self.joinsource = joinsource # Let the setter do the job """the join predecessor iterable node""" if not joinfield: # default is the interface fields joinfield = self._interface.inputs.copyable_trait_names() elif isinstance(joinfield, (str, bytes)): joinfield = [joinfield] self.joinfield = joinfield """the fields to join""" self._inputs = self._override_join_traits( self._interface.inputs, self.joinfield ) """the override inputs""" self._unique = unique """flag indicating whether to ignore duplicate input values""" self._next_slot_index = 0 """the joinfield index assigned to an iterated input""" @property def joinsource(self): return self._joinsource @joinsource.setter def joinsource(self, value): """Set the joinsource property. If the given value is a Node, then the joinsource is set to the node name. """ if isinstance(value, Node): value = value.name self._joinsource = value @property def inputs(self): """The JoinNode inputs include the join field overrides.""" return self._inputs def _add_join_item_fields(self): """Add new join item fields assigned to the next iterated input This method is intended solely for workflow graph expansion. Examples -------- >>> from nipype.interfaces.utility import IdentityInterface >>> import nipype.pipeline.engine as pe >>> from nipype import Node, JoinNode, Workflow >>> inputspec = Node(IdentityInterface(fields=['image']), ... name='inputspec'), >>> join = JoinNode(IdentityInterface(fields=['images', 'mask']), ... joinsource='inputspec', joinfield='images', name='join') >>> join._add_join_item_fields() {'images': 'imagesJ1'} Return the {base field: slot field} dictionary """ # create the new join item fields idx = self._next_slot_index newfields = dict( [(field, self._add_join_item_field(field, idx)) for field in self.joinfield] ) # increment the join slot index logger.debug("Added the %s join item fields %s.", self, newfields) self._next_slot_index += 1 return newfields def _add_join_item_field(self, field, index): """Add new join item fields qualified by the given index Return the new field name """ # the new field name name = "%sJ%d" % (field, index + 1) # make a copy of the join trait trait = self._inputs.trait(field, False, True) # add the join item trait to the override traits self._inputs.add_trait(name, trait) return name def _override_join_traits(self, basetraits, fields): """Convert the given join fields to accept an input that is a list item rather than a list. Non-join fields delegate to the interface traits. Return the override DynamicTraitedSpec """ dyntraits = DynamicTraitedSpec() if fields is None: fields = basetraits.copyable_trait_names() else: # validate the fields for field in fields: if not basetraits.trait(field): raise ValueError( "The JoinNode %s does not have a field" " named %s" % (self.name, field) ) for name, trait in list(basetraits.items()): # if a join field has a single inner trait, then the item # trait is that inner trait. Otherwise, the item trait is # a new Any trait. if name in fields and len(trait.inner_traits) == 1: item_trait = trait.inner_traits[0] dyntraits.add_trait(name, item_trait) setattr(dyntraits, name, Undefined) logger.debug( "Converted the join node %s field %s trait type from %s to %s", self, name, trait.trait_type.info(), item_trait.info(), ) else: dyntraits.add_trait(name, traits.Any) setattr(dyntraits, name, Undefined) return dyntraits def _run_command(self, execute, copyfiles=True): """Collates the join inputs prior to delegating to the superclass.""" self._collate_join_field_inputs() return super(JoinNode, self)._run_command(execute, copyfiles) def _collate_join_field_inputs(self): """ Collects each override join item field into the interface join field input.""" for field in self.inputs.copyable_trait_names(): if field in self.joinfield: # collate the join field val = self._collate_input_value(field) try: setattr(self._interface.inputs, field, val) except Exception as e: raise ValueError( ">>JN %s %s %s %s %s: %s" % ( self, field, val, self.inputs.copyable_trait_names(), self.joinfield, e, ) ) elif hasattr(self._interface.inputs, field): # copy the non-join field val = getattr(self._inputs, field) if isdefined(val): setattr(self._interface.inputs, field, val) logger.debug( "Collated %d inputs into the %s node join fields", self._next_slot_index, self, ) def _collate_input_value(self, field): """ Collects the join item field values into a list or set value for the given field, as follows: - If the field trait is a Set, then the values are collected into a set. - Otherwise, the values are collected into a list which preserves the iterables order. If the ``unique`` flag is set, then duplicate values are removed but the iterables order is preserved. """ val = [self._slot_value(field, idx) for idx in range(self._next_slot_index)] basetrait = self._interface.inputs.trait(field) if isinstance(basetrait.trait_type, traits.Set): return set(val) if self._unique: return list(OrderedDict.fromkeys(val)) return val def _slot_value(self, field, index): slot_field = "%sJ%d" % (field, index + 1) try: return getattr(self._inputs, slot_field) except AttributeError as e: raise AttributeError( "The join node %s does not have a slot field %s" " to hold the %s value at index %d: %s" % (self, slot_field, field, index, e) ) class MapNode(Node): """Wraps interface objects that need to be iterated on a list of inputs. Examples -------- >>> from nipype import MapNode >>> from nipype.interfaces import fsl >>> realign = MapNode(fsl.MCFLIRT(), 'in_file', 'realign') >>> realign.inputs.in_file = ['functional.nii', ... 'functional2.nii', ... 'functional3.nii'] >>> realign.run() # doctest: +SKIP """ def __init__( self, interface, iterfield, name, serial=False, nested=False, **kwargs ): """ Parameters ---------- interface : interface object node specific interface (fsl.Bet(), spm.Coregister()) iterfield : string or list of strings name(s) of input fields that will receive a list of whatever kind of input they take. the node will be run separately for each value in these lists. for more than one input, the values are paired (i.e. it does not compute a combinatorial product). name : alphanumeric string node specific name serial : boolean flag to enforce executing the jobs of the mapnode in a serial manner rather than parallel nested : boolean support for nested lists. If set, the input list will be flattened before running and the nested list structure of the outputs will be resored. See Node docstring for additional keyword arguments. """ super(MapNode, self).__init__(interface, name, **kwargs) if isinstance(iterfield, (str, bytes)): iterfield = [iterfield] self.iterfield = iterfield self.nested = nested self._inputs = self._create_dynamic_traits( self._interface.inputs, fields=self.iterfield ) self._inputs.on_trait_change(self._set_mapnode_input) self._got_inputs = False self._serial = serial def _create_dynamic_traits(self, basetraits, fields=None, nitems=None): """Convert specific fields of a trait to accept multiple inputs""" output = DynamicTraitedSpec() if fields is None: fields = basetraits.copyable_trait_names() for name, spec in list(basetraits.items()): if name in fields and ((nitems is None) or (nitems > 1)): logger.debug("adding multipath trait: %s", name) if self.nested: output.add_trait(name, InputMultiPath(traits.Any())) else: output.add_trait(name, InputMultiPath(spec.trait_type)) else: output.add_trait(name, traits.Trait(spec)) setattr(output, name, Undefined) value = getattr(basetraits, name) if isdefined(value): setattr(output, name, value) value = getattr(output, name) return output def set_input(self, parameter, val): """ Set interface input value or nodewrapper attribute Priority goes to interface. """ logger.debug( "setting nodelevel(%s) input %s = %s", str(self), parameter, str(val) ) self._set_mapnode_input(parameter, deepcopy(val)) def _set_mapnode_input(self, name, newvalue): logger.debug( "setting mapnode(%s) input: %s -> %s", str(self), name, str(newvalue) ) if name in self.iterfield: setattr(self._inputs, name, newvalue) else: setattr(self._interface.inputs, name, newvalue) def _get_hashval(self): """Compute hash including iterfield lists.""" self._get_inputs() if self._hashvalue is not None and self._hashed_inputs is not None: return self._hashed_inputs, self._hashvalue self._check_iterfield() hashinputs = deepcopy(self._interface.inputs) for name in self.iterfield: hashinputs.remove_trait(name) hashinputs.add_trait( name, InputMultiPath(self._interface.inputs.traits()[name].trait_type) ) logger.debug("setting hashinput %s-> %s", name, getattr(self._inputs, name)) if self.nested: setattr(hashinputs, name, flatten(getattr(self._inputs, name))) else: setattr(hashinputs, name, getattr(self._inputs, name)) hashed_inputs, hashvalue = hashinputs.get_hashval( hash_method=self.config["execution"]["hash_method"] ) rm_extra = self.config["execution"]["remove_unnecessary_outputs"] if str2bool(rm_extra) and self.needed_outputs: hashobject = md5() hashobject.update(hashvalue.encode()) sorted_outputs = sorted(self.needed_outputs) hashobject.update(str(sorted_outputs).encode()) hashvalue = hashobject.hexdigest() hashed_inputs.append(("needed_outputs", sorted_outputs)) self._hashed_inputs, self._hashvalue = hashed_inputs, hashvalue return self._hashed_inputs, self._hashvalue @property def inputs(self): return self._inputs @property def outputs(self): if self._interface._outputs(): return Bunch(self._interface._outputs().trait_get()) def _make_nodes(self, cwd=None): if cwd is None: cwd = self.output_dir() if self.nested: nitems = len(flatten(ensure_list(getattr(self.inputs, self.iterfield[0])))) else: nitems = len(ensure_list(getattr(self.inputs, self.iterfield[0]))) for i in range(nitems): nodename = "_%s%d" % (self.name, i) node = Node( deepcopy(self._interface), n_procs=self._n_procs, mem_gb=self._mem_gb, overwrite=self.overwrite, needed_outputs=self.needed_outputs, run_without_submitting=self.run_without_submitting, base_dir=op.join(cwd, "mapflow"), name=nodename, ) node.plugin_args = self.plugin_args node.interface.inputs.trait_set( **deepcopy(self._interface.inputs.trait_get()) ) node.interface.resource_monitor = self._interface.resource_monitor for field in self.iterfield: if self.nested: fieldvals = flatten(ensure_list(getattr(self.inputs, field))) else: fieldvals = ensure_list(getattr(self.inputs, field)) logger.debug("setting input %d %s %s", i, field, fieldvals[i]) setattr(node.inputs, field, fieldvals[i]) node.config = self.config yield i, node def _collate_results(self, nodes): finalresult = InterfaceResult( interface=[], runtime=[], provenance=[], inputs=[], outputs=self.outputs ) returncode = [] for i, nresult, err in nodes: finalresult.runtime.insert(i, None) returncode.insert(i, err) if nresult: if hasattr(nresult, "runtime"): finalresult.interface.insert(i, nresult.interface) finalresult.inputs.insert(i, nresult.inputs) finalresult.runtime[i] = nresult.runtime if hasattr(nresult, "provenance"): finalresult.provenance.insert(i, nresult.provenance) if self.outputs: for key, _ in list(self.outputs.items()): rm_extra = self.config["execution"]["remove_unnecessary_outputs"] if str2bool(rm_extra) and self.needed_outputs: if key not in self.needed_outputs: continue values = getattr(finalresult.outputs, key) if not isdefined(values): values = [] if nresult and nresult.outputs: values.insert(i, nresult.outputs.trait_get()[key]) else: values.insert(i, None) defined_vals = [isdefined(val) for val in values] if any(defined_vals) and finalresult.outputs: setattr(finalresult.outputs, key, values) if self.nested: for key, _ in list(self.outputs.items()): values = getattr(finalresult.outputs, key) if isdefined(values): values = unflatten( values, ensure_list(getattr(self.inputs, self.iterfield[0])) ) setattr(finalresult.outputs, key, values) if returncode and any([code is not None for code in returncode]): msg = [] for i, code in enumerate(returncode): if code is not None: msg += ["Subnode %d failed" % i] msg += ["Error: %s" % str(code)] raise NodeExecutionError( "Subnodes of node: %s failed:\n%s" % (self.name, "\n".join(msg)) ) return finalresult def get_subnodes(self): """Generate subnodes of a mapnode and write pre-execution report""" self._get_inputs() self._check_iterfield() write_node_report(self, result=None, is_mapnode=True) return [node for _, node in self._make_nodes()] def num_subnodes(self): """Get the number of subnodes to iterate in this MapNode""" self._get_inputs() self._check_iterfield() if self._serial: return 1 if self.nested: return len(ensure_list(flatten(getattr(self.inputs, self.iterfield[0])))) return len(ensure_list(getattr(self.inputs, self.iterfield[0]))) def _get_inputs(self): old_inputs = self._inputs.trait_get() self._inputs = self._create_dynamic_traits( self._interface.inputs, fields=self.iterfield ) self._inputs.trait_set(**old_inputs) super(MapNode, self)._get_inputs() def _check_iterfield(self): """Checks iterfield * iterfield must be in inputs * number of elements must match across iterfield """ for iterfield in self.iterfield: if not isdefined(getattr(self.inputs, iterfield)): raise ValueError( ("Input %s was not set but it is listed " "in iterfields.") % iterfield ) if len(self.iterfield) > 1: first_len = len(ensure_list(getattr(self.inputs, self.iterfield[0]))) for iterfield in self.iterfield[1:]: if first_len != len(ensure_list(getattr(self.inputs, iterfield))): raise ValueError( ( "All iterfields of a MapNode have to " "have the same length. %s" ) % str(self.inputs) ) def _run_interface(self, execute=True, updatehash=False): """Run the mapnode interface This is primarily intended for serial execution of mapnode. A parallel execution requires creation of new nodes that can be spawned """ self._check_iterfield() cwd = self.output_dir() if not execute: return self._load_results() # Set up mapnode folder names if self.nested: nitems = len(ensure_list(flatten(getattr(self.inputs, self.iterfield[0])))) else: nitems = len(ensure_list(getattr(self.inputs, self.iterfield[0]))) nnametpl = "_%s{}" % self.name nodenames = [nnametpl.format(i) for i in range(nitems)] # Run mapnode outdir = self.output_dir() result = InterfaceResult( interface=self._interface.__class__, runtime=Bunch( cwd=outdir, returncode=1, environ=dict(os.environ), hostname=socket.gethostname(), ), inputs=self._interface.inputs.get_traitsfree(), ) try: result = self._collate_results( _node_runner( self._make_nodes(cwd), updatehash=updatehash, stop_first=str2bool( self.config["execution"]["stop_on_first_crash"] ), ) ) except Exception as msg: result.runtime.stderr = "%s\n\n%s".format( getattr(result.runtime, "stderr", ""), msg ) _save_resultfile( result, outdir, self.name, rebase=str2bool(self.config["execution"]["use_relative_paths"]), ) raise # And store results _save_resultfile(result, cwd, self.name, rebase=False) # remove any node directories no longer required dirs2remove = [] for path in glob(op.join(cwd, "mapflow", "*")): if op.isdir(path): if path.split(op.sep)[-1] not in nodenames: dirs2remove.append(path) for path in dirs2remove: logger.debug('[MapNode] Removing folder "%s".', path) shutil.rmtree(path) return result