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)