base.py

# -*- 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:
"""The spm module provides basic functions for interfacing with SPM  tools.

In order to use the standalone MCR version of spm, you need to ensure that
the following commands are executed at the beginning of your script::

   from nipype.interfaces import spm
   matlab_cmd = '/path/to/run_spm8.sh /path/to/Compiler_Runtime/v713/ script'
   spm.SPMCommand.set_mlab_paths(matlab_cmd=matlab_cmd, use_mcr=True)

you can test by calling::

   spm.SPMCommand().version
"""
# Standard library imports
import os
from copy import deepcopy

# Third-party imports
from nibabel import load
import numpy as np

# Local imports
from ... import logging
from ...utils import spm_docs as sd
from ..base import (
    BaseInterface,
    traits,
    isdefined,
    InputMultiPath,
    BaseInterfaceInputSpec,
    Directory,
    Undefined,
    ImageFile,
    PackageInfo,
)
from ..base.traits_extension import NoDefaultSpecified
from ..matlab import MatlabCommand
from ...external.due import due, Doi, BibTeX

__docformat__ = "restructuredtext"
logger = logging.getLogger("nipype.interface")


def func_is_3d(in_file):
    """Checks if input functional files are 3d."""

    if isinstance(in_file, list):
        return func_is_3d(in_file[0])
    else:
        img = load(in_file)
        shape = img.shape
        if len(shape) == 3 or (len(shape) == 4 and shape[3] == 1):
            return True
        else:
            return False


def get_first_3dfile(in_files):
    if not func_is_3d(in_files):
        return None
    if isinstance(in_files[0], list):
        return in_files[0]
    return in_files


def scans_for_fname(fname):
    """Reads a nifti file and converts it to a numpy array storing
    individual nifti volumes.

    Opens images so will fail if they are not found.

    """
    if isinstance(fname, list):
        scans = np.zeros((len(fname),), dtype=object)
        for sno, f in enumerate(fname):
            scans[sno] = "%s,1" % f
        return scans
    img = load(fname)
    if len(img.shape) == 3:
        return np.array(("%s,1" % fname,), dtype=object)
    else:
        n_scans = img.shape[3]
        scans = np.zeros((n_scans,), dtype=object)
        for sno in range(n_scans):
            scans[sno] = "%s,%d" % (fname, sno + 1)
        return scans


def scans_for_fnames(fnames, keep4d=False, separate_sessions=False):
    """Converts a list of files to a concatenated numpy array for each
    volume.

    keep4d : boolean
        keeps the entries of the numpy array as 4d files instead of
        extracting the individual volumes.
    separate_sessions: boolean
        if 4d nifti files are being used, then separate_sessions
        ensures a cell array per session is created in the structure.

    """
    flist = None
    if not isinstance(fnames[0], list):
        if func_is_3d(fnames[0]):
            fnames = [fnames]
    if separate_sessions or keep4d:
        flist = np.zeros((len(fnames),), dtype=object)
    for i, f in enumerate(fnames):
        if separate_sessions:
            if keep4d:
                if isinstance(f, list):
                    flist[i] = np.array(f, dtype=object)
                else:
                    flist[i] = np.array([f], dtype=object)
            else:
                flist[i] = scans_for_fname(f)
        else:
            if keep4d:
                flist[i] = f
            else:
                scans = scans_for_fname(f)
                if flist is None:
                    flist = scans
                else:
                    flist = np.concatenate((flist, scans))
    return flist


class Info(PackageInfo):
    """Handles SPM version information

    If you use `SPMCommand.set_mlab_paths` to set alternate entries for
    matlab_cmd, paths, and use_mcr, then you will need to use the same entries
    to any call in the Info class to maintain memoization. Otherwise, it will
    default to the parameters in the `getinfo` function below.
    """

    _path = None
    _name = None
    _command = None
    _paths = None
    _version = None

    @classmethod
    def path(klass, matlab_cmd=None, paths=None, use_mcr=None):
        klass.getinfo(matlab_cmd, paths, use_mcr)
        return klass._path

    @classmethod
    def version(klass, matlab_cmd=None, paths=None, use_mcr=None):
        klass.getinfo(matlab_cmd, paths, use_mcr)
        return klass._version

    @classmethod
    def name(klass, matlab_cmd=None, paths=None, use_mcr=None):
        klass.getinfo(matlab_cmd, paths, use_mcr)
        return klass._name

    @classmethod
    def getinfo(klass, matlab_cmd=None, paths=None, use_mcr=None):
        """
        Returns the path to the SPM directory in the Matlab path
        If path not found, returns None.

        Parameters
        ----------
        matlab_cmd: str
            Sets the default matlab command. If None, the value of the
            environment variable SPMMCRCMD will be used if set and use_mcr
            is True or the environment variable FORCE_SPMMCR is set.
            If one of FORCE_SPMMCR or SPMMCRCMD is not set, the existence
            of the environment variable MATLABCMD is checked and its value
            is used as the matlab command if possible.
            If none of the above was successful, the fallback value of
            'matlab -nodesktop -nosplash' will be used.
        paths : str
            Add paths to matlab session
        use_mcr : bool
            Whether to use the MATLAB Common Runtime. In this case, the
            matlab_cmd is expected to be a valid MCR call.

        Returns
        -------
        spm_path : string representing path to SPM directory

            returns None of path not found
        """

        use_mcr = use_mcr or "FORCE_SPMMCR" in os.environ
        matlab_cmd = matlab_cmd or (
            (use_mcr and os.getenv("SPMMCRCMD"))
            or os.getenv("MATLABCMD", "matlab -nodesktop -nosplash")
        )

        if (
            klass._name
            and klass._path
            and klass._version
            and klass._command == matlab_cmd
            and klass._paths == paths
        ):

            return {"name": klass._name, "path": klass._path, "release": klass._version}
        logger.debug("matlab command or path has changed. recomputing version.")
        mlab = MatlabCommand(matlab_cmd=matlab_cmd, resource_monitor=False)
        mlab.inputs.mfile = False
        if paths:
            mlab.inputs.paths = paths
        if use_mcr:
            mlab.inputs.nodesktop = Undefined
            mlab.inputs.nosplash = Undefined
            mlab.inputs.single_comp_thread = Undefined
            mlab.inputs.mfile = True
            mlab.inputs.uses_mcr = True
        mlab.inputs.script = """
if isempty(which('spm')),
throw(MException('SPMCheck:NotFound','SPM not in matlab path'));
end;
spm_path = spm('dir');
[name, version] = spm('ver');
fprintf(1, 'NIPYPE path:%s|name:%s|release:%s', spm_path, name, version);
exit;
        """
        try:
            out = mlab.run()
        except (IOError, RuntimeError) as e:
            # if no Matlab at all -- exception could be raised
            # No Matlab -- no spm
            logger.debug("%s", e)
            klass._version = None
            klass._path = None
            klass._name = None
            klass._command = matlab_cmd
            klass._paths = paths
            return None

        out = sd._strip_header(out.runtime.stdout)
        out_dict = {}
        for part in out.split("|"):
            key, val = part.split(":")
            out_dict[key] = val

        klass._version = out_dict["release"]
        klass._path = out_dict["path"]
        klass._name = out_dict["name"]
        klass._command = matlab_cmd
        klass._paths = paths
        return out_dict


def no_spm():
    """Checks if SPM is NOT installed
    used with pytest.mark.skipif decorator to skip tests
    that will fail if spm is not installed"""

    if "NIPYPE_NO_MATLAB" in os.environ or Info.version() is None:
        return True
    else:
        return False


class SPMCommandInputSpec(BaseInterfaceInputSpec):
    matlab_cmd = traits.Str(desc="matlab command to use")
    paths = InputMultiPath(Directory(), desc="Paths to add to matlabpath")
    mfile = traits.Bool(True, desc="Run m-code using m-file", usedefault=True)
    use_mcr = traits.Bool(desc="Run m-code using SPM MCR")
    use_v8struct = traits.Bool(
        True,
        min_ver="8",
        usedefault=True,
        desc=("Generate SPM8 and higher " "compatible jobs"),
    )


class SPMCommand(BaseInterface):
    """Extends `BaseInterface` class to implement SPM specific interfaces.

    WARNING: Pseudo prototype class, meant to be subclassed
    """

    input_spec = SPMCommandInputSpec
    _additional_metadata = ["field"]

    _jobtype = "basetype"
    _jobname = "basename"

    _matlab_cmd = None
    _paths = None
    _use_mcr = None

    _references = [
        {
            "entry": BibTeX(
                "@book{FrackowiakFristonFrithDolanMazziotta1997,"
                "author={R.S.J. Frackowiak, K.J. Friston, C.D. Frith, R.J. Dolan, and J.C. Mazziotta},"
                "title={Human Brain Function},"
                "publisher={Academic Press USA},"
                "year={1997},"
                "}"
            ),
            "description": "The fundamental text on Statistical Parametric Mapping (SPM)",
            # 'path': "nipype.interfaces.spm",
            "tags": ["implementation"],
        }
    ]

    def __init__(self, **inputs):
        super(SPMCommand, self).__init__(**inputs)
        self.inputs.on_trait_change(
            self._matlab_cmd_update, ["matlab_cmd", "mfile", "paths", "use_mcr"]
        )
        self._find_mlab_cmd_defaults()
        self._check_mlab_inputs()
        self._matlab_cmd_update()

    @classmethod
    def set_mlab_paths(cls, matlab_cmd=None, paths=None, use_mcr=None):
        cls._matlab_cmd = matlab_cmd
        cls._paths = paths
        cls._use_mcr = use_mcr
        info_dict = Info.getinfo(matlab_cmd=matlab_cmd, paths=paths, use_mcr=use_mcr)

    def _find_mlab_cmd_defaults(self):
        # check if the user has set environment variables to enforce
        # the standalone (MCR) version of SPM
        if self._use_mcr or "FORCE_SPMMCR" in os.environ:
            self._use_mcr = True
            if self._matlab_cmd is None:
                try:
                    self._matlab_cmd = os.environ["SPMMCRCMD"]
                except KeyError:
                    pass

    def _matlab_cmd_update(self):
        # MatlabCommand has to be created here,
        # because matlab_cmd is not a proper input
        # and can be set only during init
        self.mlab = MatlabCommand(
            matlab_cmd=self.inputs.matlab_cmd,
            mfile=self.inputs.mfile,
            paths=self.inputs.paths,
            resource_monitor=False,
        )
        self.mlab.inputs.script_file = (
            "pyscript_%s.m" % self.__class__.__name__.split(".")[-1].lower()
        )
        if isdefined(self.inputs.use_mcr) and self.inputs.use_mcr:
            self.mlab.inputs.nodesktop = Undefined
            self.mlab.inputs.nosplash = Undefined
            self.mlab.inputs.single_comp_thread = Undefined
            self.mlab.inputs.uses_mcr = True
            self.mlab.inputs.mfile = True

    @property
    def version(self):
        info_dict = Info.getinfo(
            matlab_cmd=self.inputs.matlab_cmd,
            paths=self.inputs.paths,
            use_mcr=self.inputs.use_mcr,
        )
        if info_dict:
            return "%s.%s" % (info_dict["name"].split("SPM")[-1], info_dict["release"])

    @property
    def jobtype(self):
        return self._jobtype

    @property
    def jobname(self):
        return self._jobname

    def _check_mlab_inputs(self):
        if not isdefined(self.inputs.matlab_cmd) and self._matlab_cmd:
            self.inputs.matlab_cmd = self._matlab_cmd
        if not isdefined(self.inputs.paths) and self._paths:
            self.inputs.paths = self._paths
        if not isdefined(self.inputs.use_mcr) and self._use_mcr:
            self.inputs.use_mcr = self._use_mcr

    def _run_interface(self, runtime):
        """Executes the SPM function using MATLAB."""
        self.mlab.inputs.script = self._make_matlab_command(
            deepcopy(self._parse_inputs())
        )
        results = self.mlab.run()
        runtime.returncode = results.runtime.returncode
        if self.mlab.inputs.uses_mcr:
            if "Skipped" in results.runtime.stdout:
                self.raise_exception(runtime)
        runtime.stdout = results.runtime.stdout
        runtime.stderr = results.runtime.stderr
        runtime.merged = results.runtime.merged
        return runtime

    def _list_outputs(self):
        """Determine the expected outputs based on inputs."""

        raise NotImplementedError

    def _format_arg(self, opt, spec, val):
        """Convert input to appropriate format for SPM."""
        if spec.is_trait_type(traits.Bool):
            return int(val)
        elif spec.is_trait_type(traits.Tuple):
            return list(val)
        else:
            return val

    def _parse_inputs(self, skip=()):
        spmdict = {}
        metadata = dict(field=lambda t: t is not None)
        for name, spec in list(self.inputs.traits(**metadata).items()):
            if skip and name in skip:
                continue
            value = getattr(self.inputs, name)
            if not isdefined(value):
                continue
            field = spec.field
            if "." in field:
                fields = field.split(".")
                dictref = spmdict
                for f in fields[:-1]:
                    if f not in list(dictref.keys()):
                        dictref[f] = {}
                    dictref = dictref[f]
                dictref[fields[-1]] = self._format_arg(name, spec, value)
            else:
                spmdict[field] = self._format_arg(name, spec, value)
        return [spmdict]

    def _reformat_dict_for_savemat(self, contents):
        """Encloses a dict representation within hierarchical lists.

        In order to create an appropriate SPM job structure, a Python
        dict storing the job needs to be modified so that each dict
        embedded in dict needs to be enclosed as a list element.

        Examples
        --------
        >>> a = SPMCommand()._reformat_dict_for_savemat(dict(a=1,
        ...                                                  b=dict(c=2, d=3)))
        >>> a == [{'a': 1, 'b': [{'c': 2, 'd': 3}]}]
        True

        """
        newdict = {}
        try:
            for key, value in list(contents.items()):
                if isinstance(value, dict):
                    if value:
                        newdict[key] = self._reformat_dict_for_savemat(value)
                    # if value is None, skip
                else:
                    newdict[key] = value

            return [newdict]
        except TypeError:
            print("Requires dict input")

    def _generate_job(self, prefix="", contents=None):
        """Recursive function to generate spm job specification as a string

        Parameters
        ----------
        prefix : string
            A string that needs to get
        contents : dict
            A non-tuple Python structure containing spm job
            information gets converted to an appropriate sequence of
            matlab commands.

        """
        jobstring = ""
        if contents is None:
            return jobstring
        if isinstance(contents, list):
            for i, value in enumerate(contents):
                if prefix.endswith(")"):
                    newprefix = "%s,%d)" % (prefix[:-1], i + 1)
                else:
                    newprefix = "%s(%d)" % (prefix, i + 1)
                jobstring += self._generate_job(newprefix, value)
            return jobstring
        if isinstance(contents, dict):
            for key, value in list(contents.items()):
                newprefix = "%s.%s" % (prefix, key)
                jobstring += self._generate_job(newprefix, value)
            return jobstring
        if isinstance(contents, np.ndarray):
            if contents.dtype == np.dtype(object):
                if prefix:
                    jobstring += "%s = {...\n" % (prefix)
                else:
                    jobstring += "{...\n"
                for i, val in enumerate(contents):
                    if isinstance(val, np.ndarray):
                        jobstring += self._generate_job(prefix=None, contents=val)
                    elif isinstance(val, list):
                        items_format = []
                        for el in val:
                            items_format += [
                                "{}" if not isinstance(el, (str, bytes)) else "'{}'"
                            ]
                        val_format = ", ".join(items_format).format
                        jobstring += "[{}];...\n".format(val_format(*val))
                    elif isinstance(val, (str, bytes)):
                        jobstring += "'{}';...\n".format(val)
                    else:
                        jobstring += "%s;...\n" % str(val)
                jobstring += "};\n"
            else:
                for i, val in enumerate(contents):
                    for field in val.dtype.fields:
                        if prefix:
                            newprefix = "%s(%d).%s" % (prefix, i + 1, field)
                        else:
                            newprefix = "(%d).%s" % (i + 1, field)
                        jobstring += self._generate_job(newprefix, val[field])
            return jobstring
        if isinstance(contents, (str, bytes)):
            jobstring += "%s = '%s';\n" % (prefix, contents)
            return jobstring
        jobstring += "%s = %s;\n" % (prefix, str(contents))
        return jobstring

    def _make_matlab_command(self, contents, postscript=None):
        """Generates a mfile to build job structure
        Parameters
        ----------

        contents : list
            a list of dicts generated by _parse_inputs
            in each subclass

        cwd : string
            default os.getcwd()

        Returns
        -------
        mscript : string
            contents of a script called by matlab

        """
        cwd = os.getcwd()
        mscript = """
        %% Generated by nipype.interfaces.spm
        if isempty(which('spm')),
             throw(MException('SPMCheck:NotFound', 'SPM not in matlab path'));
        end
        [name, version] = spm('ver');
        fprintf('SPM version: %s Release: %s\\n',name, version);
        fprintf('SPM path: %s\\n', which('spm'));
        spm('Defaults','fMRI');

        if strcmp(name, 'SPM8') || strcmp(name(1:5), 'SPM12'),
           spm_jobman('initcfg');
           spm_get_defaults('cmdline', 1);
        end\n
        """
        if self.mlab.inputs.mfile:
            if isdefined(self.inputs.use_v8struct) and self.inputs.use_v8struct:
                mscript += self._generate_job(
                    "jobs{1}.spm.%s.%s" % (self.jobtype, self.jobname), contents[0]
                )
            else:
                if self.jobname in [
                    "st",
                    "smooth",
                    "preproc",
                    "preproc8",
                    "fmri_spec",
                    "fmri_est",
                    "factorial_design",
                    "defs",
                ]:
                    # parentheses
                    mscript += self._generate_job(
                        "jobs{1}.%s{1}.%s(1)" % (self.jobtype, self.jobname),
                        contents[0],
                    )
                else:
                    # curly brackets
                    mscript += self._generate_job(
                        "jobs{1}.%s{1}.%s{1}" % (self.jobtype, self.jobname),
                        contents[0],
                    )
        else:
            from scipy.io import savemat

            jobdef = {
                "jobs": [
                    {
                        self.jobtype: [
                            {self.jobname: self.reformat_dict_for_savemat(contents[0])}
                        ]
                    }
                ]
            }
            savemat(os.path.join(cwd, "pyjobs_%s.mat" % self.jobname), jobdef)
            mscript += "load pyjobs_%s;\n\n" % self.jobname
        mscript += """
        spm_jobman(\'run\', jobs);\n
        """
        if self.inputs.use_mcr:
            mscript += """
        if strcmp(name, 'SPM8') || strcmp(name(1:5), 'SPM12'),
            close(\'all\', \'force\');
        end;
            """
        if postscript is not None:
            mscript += postscript
        return mscript


class ImageFileSPM(ImageFile):
    """Defines a trait whose value must be a NIfTI file."""

    def __init__(
        self, value=NoDefaultSpecified, exists=False, resolve=False, **metadata
    ):
        """Create an ImageFileSPM trait."""
        super(ImageFileSPM, self).__init__(
            value=value,
            exists=exists,
            types=["nifti1", "nifti2"],
            allow_compressed=False,
            resolve=resolve,
            **metadata
        )