ENH: Interface for R

nipype/interfaces/r.py

@@ -0,0 +1,117 @@
+# -*- 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:
+"""Interfaces to run R scripts."""
+import os
+from shutil import which
+
+from .. import config
+from .base import (
+    CommandLineInputSpec,
+    InputMultiPath,
+    isdefined,
+    CommandLine,
+    traits,
+    File,
+    Directory,
+)
+
+
+def get_r_command():
+    if "NIPYPE_NO_R" in os.environ:
+        return None
+    r_cmd = os.getenv("RCMD", default="R")
+
+    return r_cmd if which(r_cmd) else None
+
+
+no_r = get_r_command() is None
+
+
+class RInputSpec(CommandLineInputSpec):
+    """Basic expected inputs to R interface"""
+
+    script = traits.Str(
+        argstr='-e "%s"', desc="R code to run", mandatory=True, position=-1
+    )
+    # non-commandline options
+    rfile = traits.Bool(True, desc="Run R using R script", usedefault=True)
+    script_file = File(
+        "pyscript.R", usedefault=True, desc="Name of file to write R code to"
+    )
+
+
+class RCommand(CommandLine):
+    """Interface that runs R code
+
+    >>> import nipype.interfaces.r as r
+    >>> r = r.RCommand(rfile=False) # doctest: +SKIP
+    >>> r.inputs.script = "Sys.getenv('USER')" # doctest: +SKIP
+    >>> out = r.run()  # doctest: +SKIP
+    """
+
+    _cmd = get_r_command()
+    input_spec = RInputSpec
+
+    def __init__(self, r_cmd=None, **inputs):
+        """initializes interface to r
+        (default 'R')
+        """
+        super(RCommand, self).__init__(**inputs)
+        if r_cmd and isdefined(r_cmd):
+            self._cmd = r_cmd
+
+        # For r commands force all output to be returned since r
+        # does not have a clean way of notifying an error
+        self.terminal_output = "allatonce"
+
+    def set_default_r_cmd(self, r_cmd):
+        """Set the default R command line for R classes.
+
+        This method is used to set values for all R
+        subclasses.
+        """
+        self._cmd = r_cmd
+
+    def set_default_rfile(self, rfile):
+        """Set the default R script file format for R classes.
+
+        This method is used to set values for all R
+        subclasses.
+        """
+        self._rfile = rfile
+
+    def _run_interface(self, runtime):
+        self.terminal_output = "allatonce"
+        runtime = super(RCommand, self)._run_interface(runtime)
+        if "R code threw an exception" in runtime.stderr:
+            self.raise_exception(runtime)
+        return runtime
+
+    def _format_arg(self, name, trait_spec, value):
+        if name in ["script"]:
+            argstr = trait_spec.argstr
+            return self._gen_r_command(argstr, value)
+        return super(RCommand, self)._format_arg(name, trait_spec, value)
+
+    def _gen_r_command(self, argstr, script_lines):
+        """Generates commands and, if rfile specified, writes it to disk."""
+        if not self.inputs.rfile:
+            # replace newlines with ;, strip comments
+            script = "; ".join(
+                [
+                    line
+                    for line in script_lines.split("\n")
+                    if not line.strip().startswith("#")
+                ]
+            )
+            # escape " and $
+            script = script.replace('"', '\\"')
+            script = script.replace("$", "\\$")
+        else:
+            script_path = os.path.join(os.getcwd(), self.inputs.script_file)
+            with open(script_path, "wt") as rfile:
+                rfile.write(script_lines)
+            script = "source('%s')" % script_path
+
+        return argstr % script

nipype/interfaces/tests/test_auto_RCommand.py

@@ -0,0 +1,31 @@
+# AUTO-GENERATED by tools/checkspecs.py - DO NOT EDIT
+from ..r import RCommand
+
+
+def test_RCommand_inputs():
+    input_map = dict(
+        args=dict(
+            argstr="%s",
+        ),
+        environ=dict(
+            nohash=True,
+            usedefault=True,
+        ),
+        rfile=dict(
+            usedefault=True,
+        ),
+        script=dict(
+            argstr='-e "%s"',
+            mandatory=True,
+            position=-1,
+        ),
+        script_file=dict(
+            extensions=None,
+            usedefault=True,
+        ),
+    )
+    inputs = RCommand.input_spec()
+
+    for key, metadata in list(input_map.items()):
+        for metakey, value in list(metadata.items()):
+            assert getattr(inputs.traits()[key], metakey) == value

nipype/interfaces/tests/test_r.py

@@ -0,0 +1,62 @@
+# -*- 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:
+import os
+
+import pytest
+from nipype.interfaces import r
+
+no_r = r.no_r
+
+
+@pytest.mark.skipif(no_r, reason="R is not available")
+def test_cmdline(tmp_path):
+    default_script_file = str(tmp_path / "testscript")
+    ri = r.RCommand(script="1 + 1", script_file=default_script_file, rfile=False)
+    r_cmd = r.get_r_command()
+
+    assert ri.cmdline == r_cmd + (' -e "1 + 1"')
+
+    assert ri.inputs.script == "1 + 1"
+    assert ri.inputs.script_file == default_script_file
+    assert not os.path.exists(ri.inputs.script_file), "scriptfile should not exist"
+    assert not os.path.exists(
+        default_script_file
+    ), "default scriptfile should not exist."
+
+
+@pytest.mark.skipif(no_r, reason="R is not available")
+def test_run_interface(tmpdir):
+    cwd = tmpdir.chdir()
+    default_script_file = r.RInputSpec().script_file
+
+    rc = r.RCommand(r_cmd="foo_m")
+    assert not os.path.exists(default_script_file), "scriptfile should not exist 1."
+    with pytest.raises(ValueError):
+        rc.run()  # script is mandatory
+    assert not os.path.exists(default_script_file), "scriptfile should not exist 2."
+    if os.path.exists(default_script_file):  # cleanup
+        os.remove(default_script_file)
+
+    rc.inputs.script = "a=1;"
+    assert not os.path.exists(default_script_file), "scriptfile should not exist 3."
+    with pytest.raises(IOError):
+        rc.run()  # foo_m is not an executable
+    assert os.path.exists(default_script_file), "scriptfile should exist 3."
+    if os.path.exists(default_script_file):  # cleanup
+        os.remove(default_script_file)
+    cwd.chdir()
+
+
+@pytest.mark.skipif(no_r, reason="R is not available")
+def test_set_rcmd(tmpdir):
+    cwd = tmpdir.chdir()
+    default_script_file = r.RInputSpec().script_file
+
+    ri = r.RCommand()
+    _default_r_cmd = ri._cmd
+    ri.set_default_r_cmd("foo")
+    assert not os.path.exists(default_script_file), "scriptfile should not exist."
+    assert ri._cmd == "foo"
+    ri.set_default_r_cmd(_default_r_cmd)
+    cwd.chdir()