Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def process_revents() -> None:
"""Process R events.
Calling this function a regular interval can help ensure that
R is processing input events (e.g., resizing of a window with
graphics)."""
openrlib.rlib.rpy2_runHandlers(openrlib.rlib.R_InputHandlers)
def protect(self, cdata):
"""Pass-through function that adds the R object to the short-term
stack of objects protected from garbase collection."""
cdata = openrlib.rlib.Rf_protect(cdata)
self._counter += 1
return cdata
def build_rcall(rfunction,
args=[],
kwargs=[]):
rlib = openrlib.rlib
with memorymanagement.rmemory() as rmemory:
rcall = rmemory.protect(
rlib.Rf_allocList(len(args)+len(kwargs)+1)
)
_SET_TYPEOF(rcall, rlib.LANGSXP)
rlib.SETCAR(rcall, rfunction)
item = rlib.CDR(rcall)
for val in args:
cdata = rmemory.protect(conversion._get_cdata(val))
rlib.SETCAR(item, cdata)
item = rlib.CDR(item)
for key, val in kwargs:
if key is not None:
_assert_valid_slotname(key)
rlib.SET_TAG(
item,
def nchar(self, what: NCHAR_TYPE = NCHAR_TYPE.Bytes) -> int:
# TODO: nchar_type is not parsed properly by cffi ?
return openrlib.rlib.R_nchar(self.__sexp__._cdata,
what.value,
openrlib.rlib.FALSE,
openrlib.rlib.FALSE,
self._NCHAR_MSG)
def rclass_get(scaps: _rinterface.CapsuleBase) -> StrSexpVector:
rlib = openrlib.rlib
with memorymanagement.rmemory() as rmemory:
classes = rmemory.protect(
rlib.Rf_getAttrib(scaps._cdata,
rlib.R_ClassSymbol))
if rlib.Rf_length(classes) == 0:
dim = rmemory.protect(
rlib.Rf_getAttrib(scaps._cdata,
rlib.R_DimSymbol))
ndim = rlib.Rf_length(dim)
if ndim > 0:
if ndim == 2:
classname = 'matrix'
else:
classname = 'array'
else:
typeof = RTYPES(scaps.typeof)
def make_extptr(obj, tag, protected):
if protected is None:
cdata_protected = openrlib.rlib.R_NilValue
else:
try:
cdata_protected = protected.__sexp__._cdata
except AttributeError:
raise TypeError('Argument protected must inherit from %s' %
type(Sexp))
ptr = _rinterface.ffi.new_handle(obj)
with memorymanagement.rmemory() as rmemory:
cdata = rmemory.protect(
openrlib.rlib.R_MakeExternalPtr(
ptr,
tag,
cdata_protected))
openrlib.rlib.R_RegisterCFinalizer(
cdata,
(_rinterface._capsule_finalizer_c
if _rinterface._capsule_finalizer_c
else _rinterface._capsule_finalizer))
res = _rinterface.SexpCapsuleWithPassenger(cdata, obj, ptr)
return res
def _string_setitem(cdata: FFI.CData, i: int, value_b) -> None:
rlib = openrlib.rlib
rlib.SET_STRING_ELT(
cdata, i, rlib.Rf_mkCharCE(value_b, conversion._CE_DEFAULT_VALUE)
)
def __new__(cls, *args, **kwargs):
embedded.assert_isready()
return super().__new__(cls, openrlib.rlib.R_NaInt)
def __setitem__(self, i: int, value: sexp.SupportsSEXP) -> None:
cdata = self.__sexp__._cdata
i_c = _rinterface._python_index_to_c(cdata, i)
openrlib.rlib.SETCAR(
openrlib.rlib.Rf_nthcdr(cdata, i_c),
value.__sexp__._cdata
)
def __init__(self):
if embedded.isready():
tmp = Sexp(
_rinterface.UnmanagedSexpCapsule(
openrlib.rlib.R_NilValue
)
)
else:
tmp = Sexp(_rinterface.UninitializedRCapsule(RTYPES.NILSXP.value))
super().__init__(tmp)