Merge pull request #3054 from stilley2/feature/Rename_copy_trait

ENH: Add "ExportFile" interface as simple alternative to "DataSink"

Author: effigies

nipype/interfaces/io.py

@@ -32,12 +32,13 @@
 from .. import config, logging
 from ..utils.filemanip import (
     copyfile, simplify_list, ensure_list,
-    get_related_files)
+    get_related_files, split_filename,
+    FileExistsError)
 from ..utils.misc import human_order_sorted, str2bool
 from .base import (
     TraitedSpec, traits, Str, File, Directory, BaseInterface, InputMultiPath,
     isdefined, OutputMultiPath, DynamicTraitedSpec, Undefined, BaseInterfaceInputSpec,
-    LibraryBaseInterface)
+    LibraryBaseInterface, SimpleInterface)
 
 iflogger = logging.getLogger('nipype.interface')
 
@@ -2863,3 +2864,51 @@ def _list_outputs(self):
 
     def _add_output_traits(self, base):
         return add_traits(base, list(self.inputs.output_query.keys()))
+
+
+class ExportFileInputSpec(BaseInterfaceInputSpec):
+    in_file = File(exists=True, mandatory=True, desc='Input file name')
+    out_file = File(mandatory=True, desc='Output file name')
+    check_extension = traits.Bool(True, desc='Ensure that the input and output file extensions match')
+    clobber = traits.Bool(desc='Permit overwriting existing files')
+
+
+class ExportFileOutputSpec(TraitedSpec):
+    out_file = File(exists=True, desc='Output file name')
+
+
+class ExportFile(SimpleInterface):
+    """ Export a file to an absolute path
+
+    This interface copies an input file to a named output file.
+    This is useful to save individual files to a specific location,
+    instead of more flexible interfaces like DataSink.
+
+    Examples
+    --------
+
+    >>> from nipype.interfaces.io import ExportFile
+    >>> import os.path as op
+    >>> ef = ExportFile()
+    >>> ef.inputs.in_file = "T1.nii.gz"
+    >>> os.mkdir("output_folder")
+    >>> ef.inputs.out_file = op.abspath("output_folder/sub1_out.nii.gz")
+    >>> res = ef.run()
+    >>> os.path.exists(res.outputs.out_file)
+    True
+
+    """
+    input_spec = ExportFileInputSpec
+    output_spec = ExportFileOutputSpec
+
+    def _run_interface(self, runtime):
+        if not self.inputs.clobber and op.exists(self.inputs.out_file):
+            raise FileExistsError(self.inputs.out_file)
+        if not op.isabs(self.inputs.out_file):
+            raise ValueError('Out_file must be an absolute path.')
+        if (self.inputs.check_extension and
+                split_filename(self.inputs.in_file)[2] != split_filename(self.inputs.out_file)[2]):
+            raise RuntimeError('%s and %s have different extensions' % (self.inputs.in_file, self.inputs.out_file))
+        shutil.copy(str(self.inputs.in_file), str(self.inputs.out_file))
+        self._results['out_file'] = self.inputs.out_file
+        return runtime

nipype/interfaces/tests/test_auto_ExportFile.py

@@ -0,0 +1,30 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from __future__ import unicode_literals
+from ..io import ExportFile
+
+
+def test_ExportFile_inputs():
+    input_map = dict(
+        check_extension=dict(),
+        clobber=dict(),
+        in_file=dict(
+            extensions=None,
+            mandatory=True,
+        ),
+        out_file=dict(
+            extensions=None,
+            mandatory=True,
+        ),
+    )
+    inputs = ExportFile.input_spec()
+
+    for key, metadata in list(input_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(inputs.traits()[key], metakey) == value
+def test_ExportFile_outputs():
+    output_map = dict(out_file=dict(extensions=None, ), )
+    outputs = ExportFile.output_spec()
+
+    for key, metadata in list(output_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(outputs.traits()[key], metakey) == value

nipype/interfaces/tests/test_io.py

@@ -20,7 +20,7 @@
 import nipype.interfaces.io as nio
 from nipype.interfaces.base.traits_extension import isdefined
 from nipype.interfaces.base import Undefined, TraitError
-from nipype.utils.filemanip import dist_is_editable
+from nipype.utils.filemanip import dist_is_editable, FileExistsError
 
 # Check for boto
 noboto = False
@@ -680,3 +680,26 @@ def _mock_get_ssh_client(self):
             .check(file=True, exists=True)) # exists?
 
     old_cwd.chdir()
+
+
+def test_ExportFile(tmp_path):
+    testin = tmp_path / 'in.txt'
+    testin.write_text('test string')
+    i = nio.ExportFile()
+    i.inputs.in_file = str(testin)
+    i.inputs.out_file = str(tmp_path / 'out.tsv')
+    i.inputs.check_extension = True
+    with pytest.raises(RuntimeError):
+        i.run()
+    i.inputs.check_extension = False
+    i.run()
+    assert (tmp_path / 'out.tsv').read_text() == 'test string'
+    i.inputs.out_file = str(tmp_path / 'out.txt')
+    i.inputs.check_extension = True
+    i.run()
+    assert (tmp_path / 'out.txt').read_text() == 'test string'
+    with pytest.raises(FileExistsError):
+        i.run()
+    i.inputs.clobber = True
+    i.run()
+    assert (tmp_path / 'out.txt').read_text() == 'test string'

nipype/utils/filemanip.py

@@ -41,7 +41,7 @@
 PY3 = sys.version_info[0] >= 3
 
 try:
-    from builtins import FileNotFoundError
+    from builtins import FileNotFoundError, FileExistsError
 except ImportError:  # PY27
     class FileNotFoundError(OSError):  # noqa
         """Defines the exception for Python 2."""
@@ -51,6 +51,14 @@ def __init__(self, path):
             super(FileNotFoundError, self).__init__(
                 2, 'No such file or directory', '%s' % path)
 
+    class FileExistsError(OSError):  # noqa
+        """Defines the exception for Python 2."""
+
+        def __init__(self, path):
+            """Initialize the exception."""
+            super(FileExistsError, self).__init__(
+                17, 'File or directory exists', '%s' % path)
+
 
 USING_PATHLIB2 = False
 try: