Merge pull request nipreps#472 from oesteban/maint/cleanup-and-coverage

MAINT: Miscellaneous housekeeping

Author: oesteban

docs/api.rst

@@ -7,7 +7,6 @@ Information on specific functions, classes, and methods.
    :glob:
 
    api/niworkflows.anat
-   api/niworkflows.common
    api/niworkflows.dwi
    api/niworkflows.engine
    api/niworkflows.func

niworkflows/common/__init__.py

@@ -0,0 +1,40 @@
+"""Test the LiterateWorkflow."""
+from nipype.pipeline.engine import Node
+from nipype.interfaces import afni, utility as niu
+from ..workflows import LiterateWorkflow as Workflow
+
+
+def _reorient_wf(name='ReorientWorkflow'):
+    """A workflow to reorient images to 'RPI' orientation."""
+    workflow = Workflow(name=name)
+    workflow.__desc__ = "Inner workflow. "
+    inputnode = Node(niu.IdentityInterface(fields=['in_file']),
+                     name='inputnode')
+    outputnode = Node(niu.IdentityInterface(
+        fields=['out_file']), name='outputnode')
+    deoblique = Node(afni.Refit(deoblique=True), name='deoblique')
+    reorient = Node(afni.Resample(
+        orientation='RPI', outputtype='NIFTI_GZ'), name='reorient')
+    workflow.connect([
+        (inputnode, deoblique, [('in_file', 'in_file')]),
+        (deoblique, reorient, [('out_file', 'in_file')]),
+        (reorient, outputnode, [('out_file', 'out_file')])
+    ])
+    return workflow
+
+
+def test_boilerplate():
+    """Check the boilerplate is generated."""
+    workflow = Workflow(name='test')
+    workflow.__desc__ = "Outer workflow. "
+    workflow.__postdesc__ = "Outer workflow (postdesc)."
+
+    inputnode = Node(niu.IdentityInterface(fields=['in_file']),
+                     name='inputnode')
+    inner = _reorient_wf()
+
+    workflow.connect([
+        (inputnode, inner, [('in_file', 'inputnode.in_file')]),
+    ])
+
+    assert workflow.visit_desc() == "Outer workflow. Inner workflow. Outer workflow (postdesc)."

niworkflows/common/orient.py

@@ -1,9 +1,7 @@
-# -*- coding: utf-8 -*-
 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
 # vi: set ft=python sts=4 ts=4 sw=4 et:
 """
-Supercharging Nipype's workflow engine
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Supercharging Nipype's workflow engine.
 
 Add special features to the Nipype's vanilla workflows
 """
@@ -14,23 +12,23 @@ class LiterateWorkflow(pe.Workflow):
     """Controls the setup and execution of a pipeline of processes."""
 
     def __init__(self, name, base_dir=None):
-        """Create a workflow object.
+        """
+        Create a workflow object.
+
         Parameters
         ----------
-        name : alphanumeric string
+        name : alphanumeric :obj:`str`
             unique identifier for the workflow
-        base_dir : string, optional
+        base_dir : :obj:`str`, optional
             path to workflow storage
+
         """
         super(LiterateWorkflow, self).__init__(name, base_dir)
         self.__desc__ = None
         self.__postdesc__ = None
 
     def visit_desc(self):
-        """
-        Builds a citation boilerplate by visiting all workflows
-        appending their ``__desc__`` field
-        """
+        """Build a citation boilerplate by visiting all workflows."""
         desc = []
 
         if self.__desc__:

niworkflows/engine/tests/__init__.py

@@ -4,8 +4,6 @@
 import os
 from mimetypes import guess_type
 from tempfile import TemporaryDirectory
-import numpy as np
-import nibabel as nb
 
 from nipype import logging
 from nipype.utils.filemanip import fname_presuffix
@@ -142,68 +140,6 @@ def _run_interface(self, runtime):
         return runtime
 
 
-class _FUGUEvsm2ANTSwarpInputSpec(BaseInterfaceInputSpec):
-    in_file = File(exists=True, mandatory=True,
-                   desc='input displacements field map')
-    pe_dir = traits.Enum('i', 'i-', 'j', 'j-', 'k', 'k-',
-                         desc='phase-encoding axis')
-
-
-class _FUGUEvsm2ANTSwarpOutputSpec(TraitedSpec):
-    out_file = File(desc='the output warp field')
-
-
-class FUGUEvsm2ANTSwarp(SimpleInterface):
-    """Convert a voxel-shift-map to ants warp."""
-    input_spec = _FUGUEvsm2ANTSwarpInputSpec
-    output_spec = _FUGUEvsm2ANTSwarpOutputSpec
-
-    def _run_interface(self, runtime):
-
-        nii = nb.load(self.inputs.in_file)
-
-        phaseEncDim = {'i': 0, 'j': 1, 'k': 2}[self.inputs.pe_dir[0]]
-
-        if len(self.inputs.pe_dir) == 2:
-            phaseEncSign = 1.0
-        else:
-            phaseEncSign = -1.0
-
-        # Fix header
-        hdr = nii.header.copy()
-        hdr.set_data_dtype(np.dtype('<f4'))
-        hdr.set_intent('vector', (), '')
-
-        # Get data, convert to mm
-        data = nii.get_fdata()
-
-        aff = np.diag([1.0, 1.0, -1.0])
-        if np.linalg.det(aff) < 0 and phaseEncDim != 0:
-            # Reverse direction since ITK is LPS
-            aff *= -1.0
-
-        aff = aff.dot(nii.affine[:3, :3])
-
-        data *= phaseEncSign * nii.header.get_zooms()[phaseEncDim]
-
-        # Add missing dimensions
-        zeros = np.zeros_like(data)
-        field = [zeros, zeros]
-        field.insert(phaseEncDim, data)
-        field = np.stack(field, -1)
-        # Add empty axis
-        field = field[:, :, :, np.newaxis, :]
-
-        # Write out
-        self._results['out_file'] = fname_presuffix(
-            self.inputs.in_file, suffix='_antswarp', newpath=runtime.cwd)
-        nb.Nifti1Image(
-            field.astype(np.dtype('<f4')), nii.affine, hdr).to_filename(
-                self._results['out_file'])
-
-        return runtime
-
-
 def _mat2itk(args):
     from nipype.interfaces.c3 import C3dAffineTool
     from nipype.utils.filemanip import fname_presuffix

niworkflows/engine/tests/test_workflows.py

@@ -0,0 +1,12 @@
+"""KeySelect tests."""
+import pytest
+from ..utility import KeySelect
+
+
+def test_KeySelect():
+    """Test KeySelect."""
+    with pytest.raises(ValueError):
+        KeySelect(fields='field1', keys=['a', 'b', 'c', 'a'])
+
+    with pytest.raises(ValueError):
+        KeySelect(fields=[])

niworkflows/engine/workflows.py

@@ -89,6 +89,36 @@ class KeySelect(BaseInterface):
     output_spec = _KeySelectOutputSpec
 
     def __init__(self, keys=None, fields=None, **inputs):
+        """
+        Instantiate a KeySelect utility interface.
+
+        Examples
+        --------
+        >>> ks = KeySelect(fields='field1')
+        >>> ks.inputs
+        <BLANKLINE>
+        field1 = <undefined>
+        key = <undefined>
+        keys = <undefined>
+        <BLANKLINE>
+
+        >>> ks = KeySelect(fields='field1', field1=['a', 'b'])
+        >>> ks.inputs
+        <BLANKLINE>
+        field1 = ['a', 'b']
+        key = <undefined>
+        keys = <undefined>
+        <BLANKLINE>
+
+        >>> ks = KeySelect()
+        Traceback (most recent call last):
+        ValueError: A list or multiplexed...
+
+        >>> ks = KeySelect(fields='key')
+        Traceback (most recent call last):
+        ValueError: Some fields are invalid...
+
+        """
         # Call constructor
         super(KeySelect, self).__init__(**inputs)
 
@@ -151,6 +181,5 @@ def _list_outputs(self):
 
     def _outputs(self):
         base = super(KeySelect, self)._outputs()
-        if self._fields:
-            base = add_traits(base, self._fields)
+        base = add_traits(base, self._fields)
         return base

niworkflows/interfaces/itk.py

@@ -1,11 +1,8 @@
-# -*- coding: utf-8 -*-
 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
 # vi: set ft=python sts=4 ts=4 sw=4 et:
 """Helper tools for visualization purposes"""
-from __future__ import absolute_import, division, print_function, unicode_literals
-
-import os
 import os.path as op
+from shutil import which
 import subprocess
 import base64
 import re
@@ -22,83 +19,25 @@
 from svgutils.transform import SVGFigure
 from seaborn import color_palette
 
-from .. import NIWORKFLOWS_LOG
 from nipype.utils import filemanip
-
-try:
-    from shutil import which
-except ImportError:
-
-    def which(cmd):
-        """
-        A homemade which command
-
-        >>> from niworkflows.viz.utils import which
-        >>> which('ls')
-        True
-        >>> which('madeoutcommand')
-        False
-
-        """
-
-        try:
-            subprocess.run([cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                           close_fds=True)
-        except OSError as e:
-            from errno import ENOENT
-            if e.errno == ENOENT:
-                return False
-            raise e
-        return True
+from .. import NIWORKFLOWS_LOG
 
 
 SVGNS = "http://www.w3.org/2000/svg"
 PY3 = version_info[0] > 2
 
-# Patch subprocess in python 2
-if not hasattr(subprocess, 'DEVNULL'):
-    setattr(subprocess, 'DEVNULL', -3)
-
-if not hasattr(subprocess, 'run'):
-    def _run(args, input=None, stdout=None, stderr=None, shell=False, check=False,
-             close_fds=False):
-        from collections import namedtuple
-
-        devnull = open(os.devnull, 'r+')
-        stdin = subprocess.PIPE if input is not None else None
 
-        if stdout == subprocess.DEVNULL:
-            stdout = devnull
-
-        if stderr == subprocess.DEVNULL:
-            stderr = devnull
-
-        proc = subprocess.Popen(args, stdout=stdout, shell=shell, stdin=stdin,
-                                close_fds=close_fds)
-        result = namedtuple('CompletedProcess', 'stdout stderr')
-        res = result(*proc.communicate(input=input))
-
-        devnull.close()
-
-        if check and proc.returncode != 0:
-            raise subprocess.CalledProcessError(proc.returncode, args)
-
-        return res
-    setattr(subprocess, 'run', _run)
-
-
-def robust_set_limits(data, plot_params):
+def robust_set_limits(data, plot_params, percentiles=(15, 99.8)):
+    """Set (vmax, vmin) based on percentiles of the data."""
     plot_params['vmin'] = plot_params.get(
-        'vmin', np.percentile(data, 15))
+        'vmin', np.percentile(data, percentiles[0]))
     plot_params['vmax'] = plot_params.get(
-        'vmax', np.percentile(data, 99.8))
+        'vmax', np.percentile(data, percentiles[1]))
     return plot_params
 
 
 def svg_compress(image, compress='auto'):
-    ''' takes an image as created by nilearn.plotting and returns a blob svg.
-    Performs compression (can be disabled). A bit hacky. '''
-
+    """Generate a blob SVG from a matplotlib figure, may perform compression."""
     # Check availability of svgo and cwebp
     has_compress = all((which('svgo'), which('cwebp')))
     if compress is True and not has_compress:
@@ -161,9 +100,7 @@ def svg_compress(image, compress='auto'):
 
 
 def svg2str(display_object, dpi=300):
-    """
-    Serializes a nilearn display object as a string
-    """
+    """Serialize a nilearn display object to string."""
     from io import StringIO
     image_buf = StringIO()
     display_object.frame_axes.figure.savefig(
@@ -173,9 +110,7 @@ def svg2str(display_object, dpi=300):
 
 
 def extract_svg(display_object, dpi=300, compress='auto'):
-    """
-    Removes the preamble of the svg files generated with nilearn
-    """
+    """Remove the preamble of the svg files generated with nilearn."""
     image_svg = svg2str(display_object, dpi)
     if compress is True or compress == 'auto':
         image_svg = svg_compress(image_svg, compress)
@@ -197,7 +132,7 @@ def extract_svg(display_object, dpi=300, compress='auto'):
 
 
 def cuts_from_bbox(mask_nii, cuts=3):
-    """Finds equi-spaced cuts for presenting images"""
+    """Find equi-spaced cuts for presenting images."""
     from nibabel.affines import apply_affine
 
     mask_data = np.asanyarray(mask_nii.dataobj) > 0.0