ENH: Move compression to separate Gzip interface

ghisvail · ghisvail · commit 735097077345 · 2022-05-16T19:02:40.000+02:00
Rationales: Keep the Gunzip interface consistent with the behavior of `gunzip`
on the CLI, which is basically `gzip` with decompression mode set nowadays.
diff --git a/nipype/algorithms/misc.py b/nipype/algorithms/misc.py
@@ -273,7 +273,6 @@ def _list_outputs(self):
 
 class GunzipInputSpec(BaseInterfaceInputSpec):
     in_file = File(exists=True, mandatory=True)
-    compress = traits.Bool(default_value=False)
 
 
 class GunzipOutputSpec(TraitedSpec):
@@ -288,10 +287,6 @@ class Gunzip(BaseInterface):
     >>> res = gunzip.run()
     >>> res.outputs.out_file  # doctest: +ELLIPSIS
     '.../tpms_msk.nii'
-    >>> gunzip = Gunzip(in_file='tpms_msk.nii', compress=True)
-    >>> res = gunzip.run()
-    >>> res.outputs.out_file  # doctest: +ELLIPSIS
-    '.../tpms_msk.nii.gz'
 
     .. testcleanup::
 
@@ -303,35 +298,71 @@ class Gunzip(BaseInterface):
 
     def _gen_output_file_name(self):
         _, base, ext = split_filename(self.inputs.in_file)
+        if ext[-3:].lower() == ".gz":
+            ext = ext[:-3]
+        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:
+                shutil.copyfileobj(in_file, out_file)
+        return runtime
+
+    def _list_outputs(self):
+        outputs = self._outputs().get()
+        outputs["out_file"] = self._gen_output_file_name()
+        return outputs
+
+
+class GzipInputSpec(GunzipInputSpec):
+    mode = traits.Enum("compress", "decompress", usedefault=True)
+
+
+class Gzip(Gunzip):
+    """Gzip wrapper supporting both compression and decompression.
+
+    >>> 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'
 
-        if self.inputs.compress:
-            ext += ".gz"
+    .. testcleanup::
+
+    >>> os.unlink('tpms_msk.nii')
+    """
+
+    input_spec = GzipInputSpec
+
+    def _gen_output_file_name(self):
+        if mode == "decompress":
+            filename = super()._gen_output_file_name()
         else:
-            if ext[-3:].lower() == ".gz":
-                ext = ext[:-3]
+            _, base, ext = split_filename(self.inputs.in_file)
+            filename = os.path.abspath(base + ext + ".gz")
 
-        return os.path.abspath(base + ext)
+        return filename
 
     def _run_interface(self, runtime):
         import gzip
         import shutil
 
-        if self.inputs.compress:
+        if mode == "decompress":
+            runtime = super()._run_interface(runtime)
+        else:
             with open(self.inputs.in_file, "rb") as in_file:
                 with gzip.open(self._gen_output_file_name(), "wb") as out_file:
                     shutil.copyfileobj(in_file, out_file)
-        else:
-            with gzip.open(self.inputs.in_file, "rb") as in_file:
-                with open(self._gen_output_file_name(), "wb") as out_file:
-                    shutil.copyfileobj(in_file, out_file)
 
         return runtime
 
-    def _list_outputs(self):
-        outputs = self._outputs().get()
-        outputs["out_file"] = self._gen_output_file_name()
-        return outputs
-
 
 def replaceext(in_list, ext):
     out_list = list()
diff --git a/nipype/algorithms/tests/test_auto_Gunzip.py b/nipype/algorithms/tests/test_auto_Gunzip.py
@@ -4,7 +4,6 @@
 
 def test_Gunzip_inputs():
     input_map = dict(
-        compress=dict(),
         in_file=dict(
             extensions=None,
             mandatory=True,
diff --git a/nipype/algorithms/tests/test_auto_Gzip.py b/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
