Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test__sexp__set():
x = rinterface.IntSexpVector([1, 2, 3])
x_s = x.__sexp__
x_rid = x.rid
# The Python reference count of the capsule is incremented,
# not the rpy2 reference count
assert x.__sexp_refcount__ == 1
y = rinterface.IntSexpVector([4, 5, 6])
y_count = y.__sexp_refcount__
y_rid = y.rid
assert y_count == 1
assert x_rid in [elt[0] for elt in rinterface._rinterface.protected_rids()]
x.__sexp__ = y.__sexp__
# x_s is still holding a refcount to the capsule
assert x_rid in [elt[0] for elt in rinterface._rinterface.protected_rids()]
# when gone, the capsule will be collected and the id no longer preserved
def test_get_integer_elt_fallback():
rpy2.rinterface.initr()
v = rpy2.rinterface.IntSexpVector([1, 2, 3])
assert (openrlib.INTEGER_ELT(v.__sexp__._cdata, 1)
==
openrlib._get_integer_elt_fallback(v.__sexp__._cdata, 1))
def test_call_OrdDictEnv():
ad = rlc.OrdDict(((None, rinterface.parse('sum(x)')), ))
env_a = rinterface.baseenv['new.env']()
env_a['x'] = rinterface.IntSexpVector([1, 2, 3])
sum_a = rinterface.baseenv['eval'].rcall(tuple(ad.items()), env_a)
assert 6 == sum_a[0]
env_b = rinterface.baseenv['new.env']()
env_b['x'] = rinterface.IntSexpVector([4, 5, 6])
sum_b = rinterface.baseenv['eval'].rcall(tuple(ad.items()), env_b)
assert 15 == sum_b[0]
def test_getitem():
seq = (ri.FloatSexpVector([1.0]),
ri.IntSexpVector([2, 3]),
ri.StrSexpVector(['foo', 'bar']))
vec = ri.ListSexpVector(seq)
utils.assert_equal_sequence(vec[1], ri.IntSexpVector([2, 3]))
with pytest.raises(TypeError):
vec[(2, 3)]
def test_getstrides():
v = rinterface.IntSexpVector([1, 2, 3])
assert bufferprotocol.getstrides(v.__sexp__._cdata, [3], 8) == (8, )
m = rinterface.baseenv.find('matrix')(nrow=2, ncol=3)
shape = (2, 3)
sizeof = 8
# 1, 3, 5
# 2, 4, 6
expected = (sizeof, shape[0] * sizeof)
assert (bufferprotocol
.getstrides(m.__sexp__._cdata, shape, sizeof)) == expected
cs_r[str(skeleton_id)] = read_neuron_local(str(skeleton_id),
robjects.ListVector(skeleton_envelope), project_id)
# Make sure all temporary R values are garbage collected. With many
# skeletons, this can otherwise become a memory problem quickly (the
# Python GC doesn't now about the R memory).
del r_nodes, r_connectors, r_tags, skeleton_data
# Explicitly garbage collect after each skeleton is loaded.
gc.collect()
if progress:
bar.finish()
objects = concat_neurons_local(
rinterface.IntSexpVector(skeleton_ids),
robjects.ListVector(cs_r), **{
'.progress': 'text' if progress else 'none',
'OmitFailures': omit_failures
})
print("Converted {}/{} neurons".format(len(objects), len(skeleton_ids)))
del(cs_r)
gc.collect()
return objects
def dotprops_for_skeletons(project_id, skeleton_ids, omit_failures=False,
scale=None, conn=None, progress=False):
"""Get the R dotprops data structure for a set of skeleton IDs.
If is true, those skeletons will be requested through HTTP.
"""
if conn:
rcatmaid = importr('catmaid')
objects = rcatmaid.read_neurons_catmaid(rinterface.IntSexpVector(skeleton_ids), **{
'conn': conn,
'.progress': 'text' if progress else 'none',
'OmitFailures': omit_failures,
})
return objects * scale if scale else objects
read_neuron_local = robjects.r('''
somapos.catmaidneuron <- function(x, swc=x$d, tags=x$tags, skid=NULL, ...) {
# Find soma position, based on plausible tags
soma_tags<-grep("(cell body|soma)", ignore.case = T, names(tags), value = T)
# soma is the preferred tag - use this for preference if it exists
if(any(soma_tags=="soma")) soma_tags="soma"
soma_id=unlist(unique(tags[soma_tags]))
soma_id_in_neuron=intersect(soma_id, swc$PointNo)
def npint_py2rpy(obj):
return rinterface.IntSexpVector([obj, ])
('z', rinterface.FloatSexpVector, robjects.NA_Real)
]
connectors = [(k,[]) for k,_,_ in connector_cols] # type: List
for rn in raw_connectors:
for n, kv in enumerate(connector_cols):
val = rn[n]
if val is None:
val = kv[2]
connectors[n][1].append(val)
r_connectors = [(kv[0], connector_cols[n][1](kv[1]))
for n, kv in enumerate(connectors)]
# Tags in Rpy2 format
r_tags = {}
for tag, node_ids in raw_tags.items():
r_tags[tag] = rinterface.IntSexpVector(node_ids)
# Construct output similar to rcatmaid's request response parsing function.
skeleton_data = robjects.ListVector({
'nodes': robjects.DataFrame(rlc.OrdDict(r_nodes)),
'connectors': robjects.DataFrame(rlc.OrdDict(r_connectors)),
'tags': robjects.ListVector(r_tags),
})
skeleton_envelope = {}
skeleton_envelope[str(skeleton_id)] = skeleton_data
cs_r[str(skeleton_id)] = read_neuron_local(str(skeleton_id),
robjects.ListVector(skeleton_envelope), project_id)
# Make sure all temporary R values are garbage collected. With many
# skeletons, this can otherwise become a memory problem quickly (the
# Python GC doesn't now about the R memory).