Merge pull request #1 from effigies/enh/gunzip-compression

RF: Unify Gzip/Gunzip, make Gzip the base class

Author: ghisvail

nipype/algorithms/misc.py

@@ -271,43 +271,57 @@ 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,51 +331,26 @@ def _list_outputs(self):
         return outputs
 
 
-class GzipInputSpec(GunzipInputSpec):
-    mode = traits.Enum("compress", "decompress", usedefault=True)
+class GunzipInputSpec(GzipInputSpec):
+    mode = traits.Enum("decompress", usedefault=True,
+                       desc="decompress or compress")
 
 
-class Gzip(Gunzip):
-    """Gzip wrapper supporting both compression and decompression.
+class Gunzip(Gzip):
+    """Gunzip wrapper
 
-    >>> from nipype.algorithms.misc import Gzip
-    >>> gzip = Gzip(in_file='tpms_msk.nii.gz', mode="decompress")
-    >>> res = gzip.run()
+    >>> 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'
-    >>> 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
+    input_spec = GunzipInputSpec
 
 
 def replaceext(in_list, ext):