Source code for nipype.interfaces.base.core

# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
# vi: set ft=python sts=4 ts=4 sw=4 et:
Nipype interfaces core

Defines the ``Interface`` API and the body of the
most basic interfaces.
The I/O specifications corresponding to these base
interfaces are found in the ``specs`` module.

import os
import subprocess as sp
import shlex
import simplejson as json
from traits.trait_errors import TraitError

from ... import config, logging, LooseVersion
from ...utils.provenance import write_provenance
from ...utils.misc import str2bool
from ...utils.filemanip import (
from ...utils.subprocess import run_command

from ...external.due import due

from .traits_extension import traits, isdefined, Undefined
from .specs import (
from .support import (

iflogger = logging.getLogger("nipype.interface")

__docformat__ = "restructuredtext"

[docs]class Interface(object): """This is an abstract definition for Interface objects. It provides no functionality. It defines the necessary attributes and methods all Interface objects should have. """ input_spec = None """ The specification of the input, defined by a :py:class:`~traits.has_traits.HasTraits` class. """ output_spec = None """ The specification of the output, defined by a :py:class:`~traits.has_traits.HasTraits` class. """ _can_resume = False # See property below _always_run = False # See property below @property def can_resume(self): """Defines if the interface can reuse partial results after interruption. Only applies to interfaces being run within a workflow context.""" return self._can_resume @property def always_run(self): """Should the interface be always run even if the inputs were not changed? Only applies to interfaces being run within a workflow context.""" return self._always_run @property def version(self): """interfaces should implement a version property""" raise NotImplementedError @classmethod def _outputs(cls): """Initializes outputs""" raise NotImplementedError
[docs] @classmethod def help(cls, returnhelp=False): """Prints class help""" allhelp = format_help(cls) if returnhelp: return allhelp print(allhelp) return None # R1710
[docs] def __init__(self): """Subclasses must implement __init__""" raise NotImplementedError
[docs] def run(self): """Execute the command.""" raise NotImplementedError
[docs] def aggregate_outputs(self, runtime=None, needed_outputs=None): """Called to populate outputs""" raise NotImplementedError
def _list_outputs(self): """List expected outputs""" raise NotImplementedError @classmethod def _get_filecopy_info(cls): """Provides information about file inputs to copy or link to cwd. Necessary for pipeline operation """ iflogger.warning( "_get_filecopy_info member of Interface was deprecated " "in nipype-1.1.6 and will be removed in 1.2.0" ) return get_filecopy_info(cls)
class BaseInterface(Interface): """Implement common interface functionality. * Initializes inputs/outputs from input_spec/output_spec * Provides help based on input_spec and output_spec * Checks for mandatory inputs before running an interface * Runs an interface and returns results * Determines which inputs should be copied or linked to cwd This class does not implement aggregate_outputs, input_spec or output_spec. These should be defined by derived classes. This class cannot be instantiated. Attributes ---------- input_spec: :obj:`nipype.interfaces.base.specs.TraitedSpec` points to the traited class for the inputs output_spec: :obj:`nipype.interfaces.base.specs.TraitedSpec` points to the traited class for the outputs _redirect_x: bool should be set to ``True`` when the interface requires connecting to a ``$DISPLAY`` (default is ``False``). resource_monitor: bool If ``False``, prevents resource-monitoring this interface If ``True`` monitoring will be enabled IFF the general Nipype config is set on (``resource_monitor = true``). """ input_spec = BaseInterfaceInputSpec _version = None _additional_metadata = [] _redirect_x = False _references = [] resource_monitor = True # Enabled for this interface IFF enabled in the config _etelemetry_version_data = None def __init__( self, from_file=None, resource_monitor=None, ignore_exception=False, **inputs ): if ( config.getboolean("execution", "check_version") and "NIPYPE_NO_ET" not in os.environ ): from ... import check_latest_version if BaseInterface._etelemetry_version_data is None: BaseInterface._etelemetry_version_data = check_latest_version() or "n/a" if not self.input_spec: raise Exception("No input_spec in class: %s" % self.__class__.__name__) # Create input spec, disable any defaults that are unavailable due to # version, and then apply the inputs that were passed. self.inputs = self.input_spec() unavailable_traits = self._check_version_requirements( self.inputs, permissive=True ) if unavailable_traits: self.inputs.trait_set(**{k: Undefined for k in unavailable_traits}) self.inputs.trait_set(**inputs) self.ignore_exception = ignore_exception if resource_monitor is not None: self.resource_monitor = resource_monitor if from_file is not None: self.load_inputs_from_json(from_file, overwrite=True) for name, value in list(inputs.items()): setattr(self.inputs, name, value) def _outputs(self): """Returns a bunch containing output fields for the class""" outputs = None if self.output_spec: outputs = self.output_spec() return outputs def _check_requires(self, spec, name, value): """check if required inputs are satisfied""" if spec.requires: values = [ not isdefined(getattr(self.inputs, field)) for field in spec.requires ] if any(values) and isdefined(value): if len(values) > 1: fmt = ( "%s requires values for inputs %s because '%s' is set. " "For a list of required inputs, see" ) else: fmt = ( "%s requires a value for input %s because '%s' is set. " "For a list of required inputs, see" ) msg = fmt % ( self.__class__.__name__, ", ".join("'%s'" % req for req in spec.requires), name, self.__class__.__name__, ) raise ValueError(msg) def _check_xor(self, spec, name, value): """check if mutually exclusive inputs are satisfied""" if spec.xor: values = [isdefined(getattr(self.inputs, field)) for field in spec.xor] if not any(values) and not isdefined(value): msg = ( "%s requires a value for one of the inputs '%s'. " "For a list of required inputs, see" % ( self.__class__.__name__, ", ".join(spec.xor), self.__class__.__name__, ) ) raise ValueError(msg) def _check_mandatory_inputs(self): """Raises an exception if a mandatory input is Undefined""" for name, spec in list(self.inputs.traits(mandatory=True).items()): value = getattr(self.inputs, name) self._check_xor(spec, name, value) if not isdefined(value) and spec.xor is None: msg = ( "%s requires a value for input '%s'. " "For a list of required inputs, see" % (self.__class__.__name__, name, self.__class__.__name__) ) raise ValueError(msg) if isdefined(value): self._check_requires(spec, name, value) for name, spec in list( self.inputs.traits(mandatory=None, transient=None).items() ): self._check_requires(spec, name, getattr(self.inputs, name)) def _check_version_requirements(self, trait_object, permissive=False): """Raises an exception on version mismatch Set the ``permissive`` attribute to True to suppress warnings and exceptions. This is currently only used in __init__ to silently identify unavailable traits. """ unavailable_traits = [] # check minimum version check = dict(min_ver=lambda t: t is not None) names = trait_object.trait_names(**check) if names and self.version: version = LooseVersion(str(self.version)) for name in names: min_ver = LooseVersion(str(trait_object.traits()[name].min_ver)) try: too_old = min_ver > version except TypeError as err: msg = ( f"Nipype cannot validate the package version {version!r} for " f"{self.__class__.__name__}. Trait {name} requires version >={min_ver}." ) if not permissive: iflogger.warning(f"{msg}. Please verify validity.") if config.getboolean("execution", "stop_on_unknown_version"): raise ValueError(msg) from err continue if too_old: unavailable_traits.append(name) if not isdefined(getattr(trait_object, name)): continue if not permissive: raise Exception( "Trait %s (%s) (version %s < required %s)" % (name, self.__class__.__name__, version, min_ver) ) # check maximum version check = dict(max_ver=lambda t: t is not None) names = trait_object.trait_names(**check) if names and self.version: version = LooseVersion(str(self.version)) for name in names: max_ver = LooseVersion(str(trait_object.traits()[name].max_ver)) try: too_new = max_ver < version except TypeError as err: msg = ( f"Nipype cannot validate the package version {version!r} for " f"{self.__class__.__name__}. Trait {name} requires version <={max_ver}." ) if not permissive: iflogger.warning(f"{msg}. Please verify validity.") if config.getboolean("execution", "stop_on_unknown_version"): raise ValueError(msg) from err continue if too_new: unavailable_traits.append(name) if not isdefined(getattr(trait_object, name)): continue if not permissive: raise Exception( "Trait %s (%s) (version %s > required %s)" % (name, self.__class__.__name__, version, max_ver) ) return unavailable_traits def _run_interface(self, runtime): """Core function that executes interface""" raise NotImplementedError def _duecredit_cite(self): """Add the interface references to the duecredit citations""" for r in self._references: r["path"] = self.__module__ due.cite(**r) def run(self, cwd=None, ignore_exception=None, **inputs): """Execute this interface. This interface will not raise an exception if runtime.returncode is non-zero. Parameters ---------- cwd : specify a folder where the interface should be run inputs : allows the interface settings to be updated Returns ------- results : :obj:`` A copy of the instance that was executed, provenance information and, if successful, results """ rtc = RuntimeContext( resource_monitor=config.resource_monitor and self.resource_monitor, ignore_exception=ignore_exception if ignore_exception is not None else self.ignore_exception, ) with indirectory(cwd or os.getcwd()): self.inputs.trait_set(**inputs) self._check_mandatory_inputs() self._check_version_requirements(self.inputs) with rtc(self, cwd=cwd, redirect_x=self._redirect_x) as runtime: # Grab inputs now, as they should not change during execution inputs = self.inputs.get_traitsfree() outputs = None # Run interface runtime = self._pre_run_hook(runtime) runtime = self._run_interface(runtime) runtime = self._post_run_hook(runtime) # Collect outputs outputs = self.aggregate_outputs(runtime) results = InterfaceResult( self.__class__, rtc.runtime, inputs=inputs, outputs=outputs, provenance=None, ) # Add provenance (if required) if str2bool(config.get("execution", "write_provenance", "false")): # Provenance will only throw a warning if something went wrong results.provenance = write_provenance(results) self._duecredit_cite() return results def _list_outputs(self): """List the expected outputs""" if self.output_spec: raise NotImplementedError else: return None def aggregate_outputs(self, runtime=None, needed_outputs=None): """Collate expected outputs and apply output traits validation.""" outputs = self._outputs() # Generate an empty output spec object predicted_outputs = self._list_outputs() # Predictions from _list_outputs if not predicted_outputs: return outputs # Precalculate the list of output trait names that should be aggregated aggregate_names = set(predicted_outputs) if needed_outputs is not None: aggregate_names = set(needed_outputs).intersection(aggregate_names) if aggregate_names: # Make sure outputs are compatible _na_outputs = self._check_version_requirements(outputs) na_names = aggregate_names.intersection(_na_outputs) if na_names: # XXX Change to TypeError in Nipype 2.0 raise KeyError( """\ Output trait(s) %s not available in version %s of interface %s.\ """ % (", ".join(na_names), self.version, self.__class__.__name__) ) for key in aggregate_names: # Final aggregation val = predicted_outputs[key] try: setattr(outputs, key, val) except TraitError as error: if "an existing" in getattr(error, "info", "default"): msg = ( "No such file or directory '%s' for output '%s' of a %s interface" % (val, key, self.__class__.__name__) ) raise FileNotFoundError(msg) raise error return outputs @property def version(self): if self._version is None: if str2bool(config.get("execution", "stop_on_unknown_version")): raise ValueError( "Interface %s has no version information" % self.__class__.__name__ ) return self._version def load_inputs_from_json(self, json_file, overwrite=True): """ A convenient way to load pre-set inputs from a JSON file. """ with open(json_file) as fhandle: inputs_dict = json.load(fhandle) def_inputs = [] if not overwrite: def_inputs = list(self.inputs.get_traitsfree().keys()) new_inputs = list(set(list(inputs_dict.keys())) - set(def_inputs)) for key in new_inputs: if hasattr(self.inputs, key): setattr(self.inputs, key, inputs_dict[key]) def save_inputs_to_json(self, json_file): """ A convenient way to save current inputs to a JSON file. """ inputs = self.inputs.get_traitsfree() iflogger.debug("saving inputs %s", inputs) with open(json_file, "w") as fhandle: json.dump(inputs, fhandle, indent=4, ensure_ascii=False) def _pre_run_hook(self, runtime): """ Perform any pre-_run_interface() processing Subclasses may override this function to modify ``runtime`` object or interface state MUST return runtime object """ return runtime def _post_run_hook(self, runtime): """ Perform any post-_run_interface() processing Subclasses may override this function to modify ``runtime`` object or interface state MUST return runtime object """ return runtime class SimpleInterface(BaseInterface): """An interface pattern that allows outputs to be set in a dictionary called ``_results`` that is automatically interpreted by ``_list_outputs()`` to find the outputs. When implementing ``_run_interface``, set outputs with:: self._results[out_name] = out_value This can be a way to upgrade a ``Function`` interface to do type checking. Examples -------- >>> from nipype.interfaces.base import ( ... SimpleInterface, BaseInterfaceInputSpec, TraitedSpec) >>> def double(x): ... return 2 * x ... >>> class DoubleInputSpec(BaseInterfaceInputSpec): ... x = traits.Float(mandatory=True) ... >>> class DoubleOutputSpec(TraitedSpec): ... doubled = traits.Float() ... >>> class Double(SimpleInterface): ... input_spec = DoubleInputSpec ... output_spec = DoubleOutputSpec ... ... def _run_interface(self, runtime): ... self._results['doubled'] = double(self.inputs.x) ... return runtime >>> dbl = Double() >>> dbl.inputs.x = 2 >>> 4.0 """ def __init__(self, from_file=None, resource_monitor=None, **inputs): super(SimpleInterface, self).__init__( from_file=from_file, resource_monitor=resource_monitor, **inputs ) self._results = {} def _list_outputs(self): return self._results class CommandLine(BaseInterface): """Implements functionality to interact with command line programs class must be instantiated with a command argument Parameters ---------- command : str define base immutable `command` you wish to run args : str, optional optional arguments passed to base `command` Examples -------- >>> import pprint >>> from nipype.interfaces.base import CommandLine >>> cli = CommandLine(command='ls', environ={'DISPLAY': ':1'}) >>> cli.inputs.args = '-al' >>> cli.cmdline 'ls -al' >>> # Use get_traitsfree() to check all inputs set >>> pprint.pprint(cli.inputs.get_traitsfree()) # doctest: {'args': '-al', 'environ': {'DISPLAY': ':1'}} >>> cli.inputs.get_hashval()[0][0] ('args', '-al') >>> cli.inputs.get_hashval()[1] '11c37f97649cd61627f4afe5136af8c0' """ input_spec = CommandLineInputSpec _cmd_prefix = "" _cmd = None _version = None _terminal_output = "stream" _write_cmdline = False @classmethod def set_default_terminal_output(cls, output_type): """Set the default terminal output for CommandLine Interfaces. This method is used to set default terminal output for CommandLine Interfaces. However, setting this will not update the output type for any existing instances. For these, assign the <instance>.terminal_output. """ if output_type in VALID_TERMINAL_OUTPUT: cls._terminal_output = output_type else: raise AttributeError("Invalid terminal output_type: %s" % output_type) def __init__( self, command=None, terminal_output=None, write_cmdline=False, **inputs ): super(CommandLine, self).__init__(**inputs) self._environ = None # Set command. Input argument takes precedence self._cmd = command or getattr(self, "_cmd", None) # Store dependencies in runtime object self._ldd = str2bool(config.get("execution", "get_linked_libs", "true")) if self._cmd is None: raise Exception("Missing command") if terminal_output is not None: self.terminal_output = terminal_output self._write_cmdline = write_cmdline @property def cmd(self): """sets base command, immutable""" if not self._cmd: raise NotImplementedError( "CommandLineInterface should wrap an executable, but " "none has been set." ) return self._cmd @property def cmdline(self): """`command` plus any arguments (args) validates arguments and generates command line""" self._check_mandatory_inputs() allargs = [self._cmd_prefix + self.cmd] + self._parse_inputs() return " ".join(allargs) @property def terminal_output(self): return self._terminal_output @terminal_output.setter def terminal_output(self, value): if value not in VALID_TERMINAL_OUTPUT: raise RuntimeError( 'Setting invalid value "%s" for terminal_output. Valid values are ' "%s." % (value, ", ".join(['"%s"' % v for v in VALID_TERMINAL_OUTPUT])) ) self._terminal_output = value @property def write_cmdline(self): return self._write_cmdline @write_cmdline.setter def write_cmdline(self, value): self._write_cmdline = value is True def raise_exception(self, runtime): raise RuntimeError( ( "Command:\n{cmdline}\nStandard output:\n{stdout}\n" "Standard error:\n{stderr}\nReturn code: {returncode}" ).format(**runtime.dictcopy()) ) def _get_environ(self): return getattr(self.inputs, "environ", {}) def version_from_command(self, flag="-v", cmd=None): iflogger.warning( "version_from_command member of CommandLine was " "Deprecated in nipype-1.0.0 and deleted in 1.1.0" ) if cmd is None: cmd = self.cmd.split()[0] env = dict(os.environ) if which(cmd, env=env): out_environ = self._get_environ() env.update(out_environ) proc = sp.Popen( " ".join((cmd, flag)), shell=True, env=canonicalize_env(env), stdout=sp.PIPE, stderr=sp.PIPE, ) o, e = proc.communicate() return o def _run_interface(self, runtime, correct_return_codes=(0,)): """Execute command via subprocess Parameters ---------- runtime : passed by the run function Returns ------- runtime : updated runtime information adds stdout, stderr, merged, cmdline, dependencies, command_path """ out_environ = self._get_environ() # Initialize runtime Bunch try: runtime.cmdline = self.cmdline except Exception as exc: raise RuntimeError( "Error raised when interpolating the command line" ) from exc runtime.stdout = None runtime.stderr = None runtime.cmdline = self.cmdline runtime.environ.update(out_environ) runtime.success_codes = correct_return_codes # which $cmd executable_name = shlex.split(self._cmd_prefix + self.cmd)[0] cmd_path = which(executable_name, env=runtime.environ) if cmd_path is None: raise IOError( 'No command "%s" found on host %s. Please check that the ' "corresponding package is installed." % (executable_name, runtime.hostname) ) runtime.command_path = cmd_path runtime.dependencies = ( get_dependencies(executable_name, runtime.environ) if self._ldd else "<skipped>" ) runtime = run_command( runtime, output=self.terminal_output, write_cmdline=self.write_cmdline, ) return runtime def _format_arg(self, name, trait_spec, value): """A helper function for _parse_inputs Formats a trait containing argstr metadata """ argstr = trait_spec.argstr iflogger.debug("%s_%s", name, value) if trait_spec.is_trait_type(traits.Bool) and "%" not in argstr: # Boolean options have no format string. Just append options if True. return argstr if value else None # traits.Either turns into traits.TraitCompound and does not have any # inner_traits elif trait_spec.is_trait_type(traits.List) or ( trait_spec.is_trait_type(traits.TraitCompound) and isinstance(value, list) ): # This is a bit simple-minded at present, and should be # construed as the default. If more sophisticated behavior # is needed, it can be accomplished with metadata (e.g. # format string for list member str'ification, specifying # the separator, etc.) # Depending on whether we stick with traitlets, and whether or # not we beef up traitlets.List, we may want to put some # type-checking code here as well sep = trait_spec.sep if trait_spec.sep is not None else " " if argstr.endswith("..."): # repeatable option # --id %d... will expand to # --id 1 --id 2 --id 3 etc.,. argstr = argstr.replace("...", "") return sep.join([argstr % elt for elt in value]) else: return argstr % sep.join(str(elt) for elt in value) else: # Append options using format string. return argstr % value def _filename_from_source(self, name, chain=None): if chain is None: chain = [] trait_spec = self.inputs.trait(name) retval = getattr(self.inputs, name) source_ext = None if not isdefined(retval) or "%s" in retval: if not trait_spec.name_source: return retval # Do not generate filename when excluded by other inputs if any( isdefined(getattr(self.inputs, field)) for field in trait_spec.xor or () ): return retval # Do not generate filename when required fields are missing if not all( isdefined(getattr(self.inputs, field)) for field in trait_spec.requires or () ): return retval if isdefined(retval) and "%s" in retval: name_template = retval else: name_template = trait_spec.name_template if not name_template: name_template = "%s_generated" ns = trait_spec.name_source while isinstance(ns, (list, tuple)): if len(ns) > 1: iflogger.warning("Only one name_source per trait is allowed") ns = ns[0] if not isinstance(ns, (str, bytes)): raise ValueError( "name_source of '{}' trait should be an input trait " "name, but a type {} object was found".format(name, type(ns)) ) if isdefined(getattr(self.inputs, ns)): name_source = ns source = getattr(self.inputs, name_source) while isinstance(source, list): source = source[0] # special treatment for files try: _, base, source_ext = split_filename(source) except (AttributeError, TypeError): base = source else: if name in chain: raise NipypeInterfaceError("Mutually pointing name_sources") chain.append(name) base = self._filename_from_source(ns, chain) if isdefined(base): _, _, source_ext = split_filename(base) else: # Do not generate filename when required fields are missing return retval chain = None retval = name_template % base _, _, ext = split_filename(retval) if trait_spec.keep_extension and (ext or source_ext): if (ext is None or not ext) and source_ext: retval = retval + source_ext else: retval = self._overload_extension(retval, name) return retval def _gen_filename(self, name): raise NotImplementedError def _overload_extension(self, value, name=None): return value def _list_outputs(self): metadata = dict(name_source=lambda t: t is not None) traits = self.inputs.traits(**metadata) if traits: outputs = self.output_spec().trait_get() for name, trait_spec in list(traits.items()): out_name = name if trait_spec.output_name is not None: out_name = trait_spec.output_name fname = self._filename_from_source(name) if isdefined(fname): outputs[out_name] = os.path.abspath(fname) return outputs def _parse_inputs(self, skip=None): """Parse all inputs using the ``argstr`` format string in the Trait. Any inputs that are assigned (not the default_value) are formatted to be added to the command line. Returns ------- all_args : list A list of all inputs formatted for the command line. """ all_args = [] initial_args = {} final_args = {} metadata = dict(argstr=lambda t: t is not None) for name, spec in sorted(self.inputs.traits(**metadata).items()): if skip and name in skip: continue value = getattr(self.inputs, name) if spec.name_source: value = self._filename_from_source(name) elif spec.genfile: if not isdefined(value) or value is None: value = self._gen_filename(name) if not isdefined(value): continue try: arg = self._format_arg(name, spec, value) except Exception as exc: raise ValueError( f"Error formatting command line argument '{name}' with value '{value}'" ) from exc if arg is None: continue pos = spec.position if pos is not None: if int(pos) >= 0: initial_args[pos] = arg else: final_args[pos] = arg else: all_args.append(arg) first_args = [el for _, el in sorted(initial_args.items())] last_args = [el for _, el in sorted(final_args.items())] return first_args + all_args + last_args class StdOutCommandLine(CommandLine): input_spec = StdOutCommandLineInputSpec def _gen_filename(self, name): return self._gen_outfilename() if name == "out_file" else None def _gen_outfilename(self): raise NotImplementedError class MpiCommandLine(CommandLine): """Implements functionality to interact with command line programs that can be run with MPI (i.e. using 'mpiexec'). Examples -------- >>> from nipype.interfaces.base import MpiCommandLine >>> mpi_cli = MpiCommandLine(command='my_mpi_prog') >>> mpi_cli.inputs.args = '-v' >>> mpi_cli.cmdline 'my_mpi_prog -v' >>> mpi_cli.inputs.use_mpi = True >>> mpi_cli.inputs.n_procs = 8 >>> mpi_cli.cmdline 'mpiexec -n 8 my_mpi_prog -v' """ input_spec = MpiCommandLineInputSpec @property def cmdline(self): """Adds 'mpiexec' to beginning of command""" result = [] if self.inputs.use_mpi: result.append("mpiexec") if self.inputs.n_procs: result.append("-n %d" % self.inputs.n_procs) result.append(super(MpiCommandLine, self).cmdline) return " ".join(result) class SEMLikeCommandLine(CommandLine): """In SEM derived interface all outputs have corresponding inputs. However, some SEM commands create outputs that are not defined in the XML. In those cases one has to create a subclass of the autogenerated one and overload the _list_outputs method. _outputs_from_inputs should still be used but only for the reduced (by excluding those that do not have corresponding inputs list of outputs. """ def _list_outputs(self): outputs = self.output_spec().trait_get() return self._outputs_from_inputs(outputs) def _outputs_from_inputs(self, outputs): for name in list(outputs.keys()): corresponding_input = getattr(self.inputs, name) if isdefined(corresponding_input): if isinstance(corresponding_input, bool) and corresponding_input: outputs[name] = os.path.abspath(self._outputs_filenames[name]) else: if isinstance(corresponding_input, list): outputs[name] = [ os.path.abspath(inp) for inp in corresponding_input ] else: outputs[name] = os.path.abspath(corresponding_input) return outputs def _format_arg(self, name, spec, value): if name in list(self._outputs_filenames.keys()): if isinstance(value, bool): if value: value = os.path.abspath(self._outputs_filenames[name]) else: return "" return super(SEMLikeCommandLine, self)._format_arg(name, spec, value) class LibraryBaseInterface(BaseInterface): _pkg = None imports = () def __init__(self, check_import=True, *args, **kwargs): super(LibraryBaseInterface, self).__init__(*args, **kwargs) if check_import: import pkgutil failed_imports = [] for pkg in (self._pkg,) + tuple(self.imports): if pkgutil.find_loader(pkg) is None: failed_imports.append(pkg) if failed_imports: iflogger.warning( "Unable to import %s; %s interface may fail to " "run", failed_imports, self.__class__.__name__, ) @property def version(self): if self._version is None: import importlib try: self._version = importlib.import_module(self._pkg).__version__ except (ImportError, AttributeError): pass return super(LibraryBaseInterface, self).version class PackageInfo(object): _version = None version_cmd = None version_file = None @classmethod def version(klass): if klass._version is None: if klass.version_cmd is not None: try: clout = CommandLine( command=klass.version_cmd, resource_monitor=False, terminal_output="allatonce", ).run() except IOError: return None raw_info = clout.runtime.stdout elif klass.version_file is not None: try: with open(klass.version_file, "rt") as fobj: raw_info = except OSError: return None else: return None klass._version = klass.parse_version(raw_info) return klass._version @staticmethod def parse_version(raw_info): raise NotImplementedError