Source code for rpy2.robjects.robject

import abc
import os
import typing
import warnings
import weakref
import rpy2.rinterface
import rpy2.rinterface_lib.callbacks

from rpy2.robjects import conversion

rpy2.rinterface.initr_simple()


def _add_warn_reticulate_hook():
    msg = """
    WARNING: The R package "reticulate" only fixed recently
    an issue that caused a segfault when used with rpy2:
    https://github.com/rstudio/reticulate/pull/1188
    Make sure that you use a version of that package that includes
    the fix.
    """
    rpy2.rinterface.evalr(f"""
    setHook(packageEvent("reticulate", "onLoad"),
            function(...) cat({repr(msg)}))
    """)


_add_warn_reticulate_hook()


class RSlots(object):
    """ Attributes of an R object as a Python mapping.

    R objects can have attributes (slots) that are identified
    by a string key (a name) and that can have any R object
    as the associated value. This class represents a view
    of those attributes that is a Python mapping.

    The proxy to the underlying "parent" R object is held as a
    weak reference. The attributes are therefore not protected
    from garbage collection unless bound to a Python symbol or
    in an other container.
    """

    __slots__ = ['_robj', ]

    def __init__(self, robj):
        self._robj = weakref.proxy(robj)

    def __getitem__(self, key: str):
        value = self._robj.do_slot(key)
        return conversion.get_conversion().rpy2py(value)

    def __setitem__(self, key: str, value):
        rpy2_value = conversion.get_conversion().py2rpy(value)
        self._robj.do_slot_assign(key, rpy2_value)

    def __len__(self):
        return len(self._robj.list_attrs())

    def keys(self):
        for k in self._robj.list_attrs():
            yield k

    __iter__ = keys

    def items(self):
        for k in self._robj.list_attrs():
            v = self[k]
            yield (k, v)

    def values(self):
        for k in self._robj.list_attrs():
            v = self[k]
            yield v


_get_exported_value = rpy2.rinterface.baseenv['::']


class RObjectMixin(abc.ABC):
    """ Abstract class to provide methods common to all RObject instances. """

    __rname__: typing.Optional[str] = None

    __tempfile = rpy2.rinterface.baseenv.find("tempfile")
    __file = rpy2.rinterface.baseenv.find("file")
    __fifo = rpy2.rinterface.baseenv.find("fifo")
    __sink = rpy2.rinterface.baseenv.find("sink")
    __close = rpy2.rinterface.baseenv.find("close")
    __readlines = rpy2.rinterface.baseenv.find("readLines")
    __unlink = rpy2.rinterface.baseenv.find("unlink")
    __show = _get_exported_value('methods', 'show')
    __print = _get_exported_value('base', 'print')

    __slots = None

    @property
    def slots(self):
        """ Attributes of the underlying R object as a Python mapping.

        The attributes can accessed and assigned by name (as if they
        were in a Python `dict`)."""
        if self.__slots is None:
            self.__slots = RSlots(self)
        return self.__slots

    def __str__(self):
        s = []

        with (rpy2.rinterface_lib
              .callbacks.obj_in_module(rpy2.rinterface_lib.callbacks,
                                       'consolewrite_print', s.append)):
            try:
                self.__show(self)
                # There can be situation where an invalid call to R`s
                # show is made. Possibly some form of signature overriding
                # that goes through in R through dispatch (although it
                # should not?). In that case this is an problem upstream
                # and this try/except is a workaround until it gets fixed.
                # (issue #908).
            except rpy2.rinterface.embedded.RRuntimeError as rre:
                warnings.warn(f'Invalid call to "show()" in R: {rre}')
                self.__print(self)
        s = str.join('', s)
        return s

    def __getstate__(self, ):
        return (super().__getstate__(), self.__dict__.copy())

    def __setstate__(self, state):
        rds, __dict__ = state
        super().__setstate__(rds)
        self.__dict__.update(__dict__)

    def __repr__(self):
        res = [super(RObjectMixin, self).__repr__()]
        try:
            res.append(
                'R classes: {}'
                .format(tuple(self.rclass))
            )
        except Exception:
            res.append('Unable to fetch R classes.')
        return os.linesep.join(res)

    def r_repr(self):
        """ String representation for an object that can be
        directly evaluated as R code.
        """
        return repr_robject(self, linesep='\n')

    @property
    def rclass(self):
        """
        R class for the object, stored as an R string vector.

        When setting the rclass, the new value will be:

        - wrapped in a Python tuple if a string (the R class
          is a vector of strings, and this is made for convenience)
        - wrapped in a StrSexpVector

        Note that when setting the class R may make a copy of
        the whole object (R is mostly a functional language).
        If this must be avoided, and if the number of parent
        classes before and after the change are compatible,
        the class name can be changed in-place by replacing
        vector elements.
        """

        try:
            res = super(RObjectMixin, self).rclass
            res = rpy2.rinterface.sexp.rclass_get(self.__sexp__)
            return res
        except rpy2.rinterface._rinterface.embedded.RRuntimeError as rre:
            if self.typeof == rpy2.rinterface.RTYPES.SYMSXP:
                # Unevaluated expression: has no class.
                return (None, )
            else:
                raise rre

    @rclass.setter
    def rclass(self, value):
        if isinstance(value, str):
            value = (value, )
        new_cls = rpy2.rinterface.StrSexpVector(value)
        rpy2.rinterface.sexp.rclass_set(self.__sexp__, new_cls)


def repr_robject(o, linesep=os.linesep):
    s = rpy2.rinterface.baseenv.find("deparse")(o)
    s = str.join(linesep, s)
    return s


class RObject(RObjectMixin, rpy2.rinterface.Sexp):
    """ Base class for all non-vector R objects. """

    def __setattr__(self, name, value):
        if name == '_sexp':
            if not isinstance(value, rpy2.rinterface.Sexp):
                raise ValueError(
                    '_attr must contain an object '
                    'that inherits from rpy2.rinterface.Sexp '
                    '(not from %s)' % type(value)
                )
        super(RObject, self).__setattr__(name, value)