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

nipype/interfaces/io.py

@@ -28,6 +28,7 @@
 import tempfile
 from os.path import join, dirname
 from warnings import warn
+import errno
 
 from .. import config, logging
 from ..utils.filemanip import (
@@ -2863,3 +2864,33 @@ 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, desc='Input file name')
+    out_file = File(exists=False, desc='Output file name')
+    check_extension = traits.Bool(False, desc='Ensure that the input and output file extensions match')
+    clobber = traits.Bool(False, desc='Permit overwriting existing files')
+
+
+class ExportFileOutputSpec(TraitedSpec):
+    out_file = File(exists=True, desc='Output file name')
+
+
+class ExportFile(BaseInterface):
+    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(errno.EEXIST, f'File {self.inputs.out_file} exists')
+        if (self.inputs.check_extension and
+                op.splitext(self.inputs.in_file)[1] != op.splitext(self.inputs.out_file)[1]):
+            raise RuntimeError(f'{self.inputs.in_file} and {self.inputs.out_file} have different extensions')
+        shutil.copy(str(self.inputs.in_file), str(self.inputs.out_file))
+        return runtime
+
+    def _list_outputs(self):
+        outputs = self.output_spec().get()
+        outputs['out_file'] = self.inputs.out_file
+        return outputs

nipype/interfaces/tests/test_io.py

@@ -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 = testin
+    i.inputs.out_file = 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 = 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'