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['::']


[docs]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)
[docs] 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
[docs]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)