Merge pull request #3472 from ghisvail/enh/gunzip-compression

ENH: Enable compression with Gunzip interface

Author: effigies

nipype/algorithms/misc.py

@@ -271,43 +271,58 @@ def _list_outputs(self):
         return outputs
 
 
-class GunzipInputSpec(BaseInterfaceInputSpec):
-    in_file = File(exists=True, mandatory=True)
+class GzipInputSpec(TraitedSpec):
+    in_file = File(exists=True, mandatory=True, desc="file to (de)compress")
+    mode = traits.Enum(
+        "compress", "decompress", usedefault=True, desc="compress or decompress"
+    )
 
 
-class GunzipOutputSpec(TraitedSpec):
-    out_file = File(exists=True)
+class GzipOutputSpec(TraitedSpec):
+    out_file = File()
 
 
-class Gunzip(BaseInterface):
-    """Gunzip wrapper
+class Gzip(BaseInterface):
+    """Gzip wrapper
 
-    >>> from nipype.algorithms.misc import Gunzip
-    >>> gunzip = Gunzip(in_file='tpms_msk.nii.gz')
-    >>> res = gunzip.run()
+    >>> from nipype.algorithms.misc import Gzip
+    >>> gzip = Gzip(in_file='tpms_msk.nii.gz', mode="decompress")
+    >>> res = gzip.run()
     >>> res.outputs.out_file  # doctest: +ELLIPSIS
     '.../tpms_msk.nii'
 
+    >>> gzip = Gzip(in_file='tpms_msk.nii')
+    >>> res = gzip.run()
+    >>> res.outputs.out_file  # doctest: +ELLIPSIS
+    '.../tpms_msk.nii.gz'
+
     .. testcleanup::
 
     >>> os.unlink('tpms_msk.nii')
     """
 
-    input_spec = GunzipInputSpec
-    output_spec = GunzipOutputSpec
+    input_spec = GzipInputSpec
+    output_spec = GzipOutputSpec
 
     def _gen_output_file_name(self):
         _, base, ext = split_filename(self.inputs.in_file)
-        if ext[-3:].lower() == ".gz":
+        if self.inputs.mode == "decompress" and ext[-3:].lower() == ".gz":
             ext = ext[:-3]
+        elif self.inputs.mode == "compress":
+            ext = f"{ext}.gz"
         return os.path.abspath(base + ext)
 
     def _run_interface(self, runtime):
         import gzip
         import shutil
 
-        with gzip.open(self.inputs.in_file, "rb") as in_file:
-            with open(self._gen_output_file_name(), "wb") as out_file:
+        if self.inputs.mode == "compress":
+            open_input, open_output = open, gzip.open
+        else:
+            open_input, open_output = gzip.open, open
+
+        with open_input(self.inputs.in_file, "rb") as in_file:
+            with open_output(self._gen_output_file_name(), "wb") as out_file:
                 shutil.copyfileobj(in_file, out_file)
         return runtime
 
@@ -317,6 +332,27 @@ def _list_outputs(self):
         return outputs
 
 
+class GunzipInputSpec(GzipInputSpec):
+    mode = traits.Enum("decompress", usedefault=True, desc="decompress or compress")
+
+
+class Gunzip(Gzip):
+    """Gunzip wrapper
+
+    >>> from nipype.algorithms.misc import Gunzip
+    >>> gunzip = Gunzip(in_file='tpms_msk.nii.gz')
+    >>> res = gunzip.run()
+    >>> res.outputs.out_file  # doctest: +ELLIPSIS
+    '.../tpms_msk.nii'
+
+    .. testcleanup::
+
+    >>> os.unlink('tpms_msk.nii')
+    """
+
+    input_spec = GunzipInputSpec
+
+
 def replaceext(in_list, ext):
     out_list = list()
     for filename in in_list:

nipype/algorithms/tests/test_auto_Gunzip.py

@@ -8,6 +8,9 @@ def test_Gunzip_inputs():
             extensions=None,
             mandatory=True,
         ),
+        mode=dict(
+            usedefault=True,
+        ),
     )
     inputs = Gunzip.input_spec()
 

nipype/algorithms/tests/test_auto_Gzip.py

@@ -0,0 +1,32 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from ..misc import Gzip
+
+
+def test_Gzip_inputs():
+    input_map = dict(
+        in_file=dict(
+            extensions=None,
+            mandatory=True,
+        ),
+        mode=dict(
+            usedefault=True,
+        ),
+    )
+    inputs = Gzip.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_Gzip_outputs():
+    output_map = dict(
+        out_file=dict(
+            extensions=None,
+        ),
+    )
+    outputs = Gzip.output_spec()
+
+    for key, metadata in list(output_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(outputs.traits()[key], metakey) == value