ENH: Move compression to separate Gzip interface

ghisvail · ghisvail · commit e9da30c65c66 · 2022-05-16T18:55:50.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::
 
@@ -304,27 +299,17 @@ class Gunzip(BaseInterface):
     def _gen_output_file_name(self):
         _, base, ext = split_filename(self.inputs.in_file)
 
-        if self.inputs.compress:
-            ext += ".gz"
-        else:
-            if ext[-3:].lower() == ".gz":
-                ext = ext[:-3]
-
+        if ext[-3:].lower() == ".gz":
+            ext = ext[:-3]
         return os.path.abspath(base + ext)
 
     def _run_interface(self, runtime):
         import gzip
         import shutil
 
-        if self.inputs.compress:
-            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)
-
+        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):
@@ -333,6 +318,52 @@ def _list_outputs(self):
         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'
+
+    .. 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:
+            _, base, ext = split_filename(self.inputs.in_file)
+            filename = os.path.abspath(base + ext + ".gz")
+
+        return filename
+
+    def _run_interface(self, runtime):
+        import gzip
+        import shutil
+
+        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)
+
+        return runtime
+
+
 def replaceext(in_list, ext):
     out_list = list()
     for filename in in_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
