Python operator 模块，sub() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用operator.sub()。
def _clean_names(cls):
'''
Remove indices from non-repeated peeker names that don't need them.
When created, all peekers get an index appended to their name to
disambiguate any repeated names. If the name isn't actually repeated,
then the index is removed.
'''
index_re = '\[\d+\]$'
for name, peeker in cls._peekers.items():
if not peeker.name_dup:
new_name = re.sub(index_re, '', name)
if new_name != name:
peeker.trace.name = new_name
cls._peekers[new_name] = cls._peekers.pop(name)
def _test_quantity_add_sub(self, unit, func):
x = self.Q_(unit, 'centimeter')
y = self.Q_(unit, 'inch')
z = self.Q_(unit, 'second')
a = self.Q_(unit, None)
func(op.add, x, x, self.Q_(unit + unit, 'centimeter'))
func(op.add, x, y, self.Q_(unit + 2.54 * unit, 'centimeter'))
func(op.add, y, x, self.Q_(unit + unit / (2.54 * unit), 'inch'))
func(op.add, a, unit, self.Q_(unit + unit, None))
self.assertRaises(DimensionalityError, op.add, 10, x)
self.assertRaises(DimensionalityError, op.add, x, 10)
self.assertRaises(DimensionalityError, op.add, x, z)
func(op.sub, x, x, self.Q_(unit - unit, 'centimeter'))
func(op.sub, x, y, self.Q_(unit - 2.54 * unit, 'centimeter'))
func(op.sub, y, x, self.Q_(unit - unit / (2.54 * unit), 'inch'))
func(op.sub, a, unit, self.Q_(unit - unit, None))
self.assertRaises(DimensionalityError, op.sub, 10, x)
self.assertRaises(DimensionalityError, op.sub, x, 10)
self.assertRaises(DimensionalityError, op.sub, x, z)
def string_preprocessor(input_string):
input_string = input_string.replace(",", "")
input_string = input_string.replace(" per ", "/")
for a, b in _subs_re:
input_string = a.sub(b, input_string)
# Replace pretty format characters
for pretty_exp in _pretty_exp_re.findall(input_string):
exp = '**' + pretty_exp.translate(_pretty_table)
input_string = input_string.replace(pretty_exp, exp)
input_string = input_string.translate(_pretty_table)
# Handle caret exponentiation
input_string = input_string.replace("^", "**")
return input_string
def test_safe_binop():
# Test checked arithmetic routines
ops = [
(operator.add, 1),
(operator.sub, 2),
(operator.mul, 3)
]
with exc_iter(ops, INT64_VALUES, INT64_VALUES) as it:
for xop, a, b in it:
pyop, op = xop
c = pyop(a, b)
if not (INT64_MIN <= c <= INT64_MAX):
assert_raises(OverflowError, mt.extint_safe_binop, a, b, op)
else:
d = mt.extint_safe_binop(a, b, op)
if c != d:
# assert_equal is slow
assert_equal(d, c)
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def countByValueAndWindow(self, windowDuration, slideDuration, numPartitions=None):
"""
Return a new DStream in which each RDD contains the count of distinct elements in
RDDs in a sliding window over this DStream.
@param windowDuration: width of the window; must be a multiple of this DStream's
batching interval
@param slideDuration: sliding interval of the window (i.e., the interval after which
the new DStream will generate RDDs); must be a multiple of this
DStream's batching interval
@param numPartitions: number of partitions of each RDD in the new DStream.
"""
keyed = self.map(lambda x: (x, 1))
counted = keyed.reduceByKeyAndWindow(operator.add, operator.sub,
windowDuration, slideDuration, numPartitions)
return counted.filter(lambda kv: kv[1] > 0)
def operate(self, left, right, operation):
""" Do operation on colors
args:
left (str): left side
right (str): right side
operation (str): Operation
returns:
str
"""
operation = {
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'/': operator.truediv
}.get(operation)
return operation(left, right)
def fetch_span(width_span, height_span, step=100):
width_diff = sub(*reversed(width_span))
step_count = width_diff // step
if step_count <= 1:
width = min(*width_span)
else:
width = width_span[0] + randint(0, step_count) * step
height_diff = sub(*reversed(height_span))
step_count = height_diff // step
if step_count <= 1:
height = min(*height_span)
else:
height = height_span[0] + randint(0, step_count) * step
return width, height
def test_safe_binop():
# Test checked arithmetic routines
ops = [
(operator.add, 1),
(operator.sub, 2),
(operator.mul, 3)
]
with exc_iter(ops, INT64_VALUES, INT64_VALUES) as it:
for xop, a, b in it:
pyop, op = xop
c = pyop(a, b)
if not (INT64_MIN <= c <= INT64_MAX):
assert_raises(OverflowError, mt.extint_safe_binop, a, b, op)
else:
d = mt.extint_safe_binop(a, b, op)
if c != d:
# assert_equal is slow
assert_equal(d, c)
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def select(self, id, start, stop, rate=False, maxlen=float('inf'), fixed=0):
"""Return points for a series within inclusive interval of most granular samples.
Optionally derive the rate of change of points.
Optionally limit number of points by increasing sample resolution.
Optionally return fixed intervals with padding and arbitrary resolution.
"""
minstep = total_seconds(stop - start) / maxlen
for index, model in enumerate(self):
if start >= model.start(id) and model.step >= minstep:
break
points = model.select(id, dt__gte=start, dt__lt=stop)
points = list(points if index else model.reduce(points))
if rate:
points = map(operator.sub, points[1:], points[:-1])
if fixed:
step = (stop - start) / fixed
intervals = [Point(start + step * index, 0.0, 0) for index in range(fixed)]
for point in points:
intervals[int(total_seconds(point.dt - start) / total_seconds(step))] += point
points = intervals
return points
def install_subnetwork(self, owner_object, node_tree_name, initial_status):
#transform the tree name into a NL module name
valid_characters = "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ"
stripped_name = "".join([c for c in node_tree_name if c in valid_characters])
if stripped_name in owner_object:
print("Network {} already installed for {}".format(stripped_name, owner_object.name))
if(initial_status is True): owner_object[node_tree_name].stopped = False
else:
print("Installing sub network...")
initial_status_key = 'NL_{}_initial_status'.format(node_tree_name)
owner_object[initial_status_key] = initial_status
module_name = 'bgelogic.NL{}'.format(stripped_name)
module = load_user_module(module_name)
module._initialize(owner_object)
subnetwork = owner_object[node_tree_name]
self.sub_networks.append(subnetwork)
def __init__(self, code, objects=None):
self._OPERATORS = [
('|', operator.or_),
('^', operator.xor),
('&', operator.and_),
('>>', operator.rshift),
('<<', operator.lshift),
('-', operator.sub),
('+', operator.add),
('%', operator.mod),
('/', operator.truediv),
('*', operator.mul),
]
self._ASSIGN_OPERATORS = [(op + '=', opfunc)
for op, opfunc in self._OPERATORS]
self._ASSIGN_OPERATORS.append(('=', lambda cur, right: right))
self._VARNAME_PATTERN = r'[a-zA-Z_$][a-zA-Z_$0-9]*'
if objects is None:
objects = {}
self.code = code
self._functions = {}
self._objects = objects
def number_of_args(fn):
"""Return the number of positional arguments for a function, or None if the number is variable.
Looks inside any decorated functions."""
try:
if hasattr(fn, '__wrapped__'):
return number_of_args(fn.__wrapped__)
if any(p.kind == p.VAR_POSITIONAL for p in signature(fn).parameters.values()):
return None
else:
return sum(p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) for p in signature(fn).parameters.values())
except ValueError:
# signatures don't work for built-in operators, so check for a few explicitly
UNARY_OPS = [len, op.not_, op.truth, op.abs, op.index, op.inv, op.invert, op.neg, op.pos]
BINARY_OPS = [op.lt, op.le, op.gt, op.ge, op.eq, op.ne, op.is_, op.is_not, op.add, op.and_, op.floordiv, op.lshift, op.mod, op.mul, op.or_, op.pow, op.rshift, op.sub, op.truediv, op.xor, op.concat, op.contains, op.countOf, op.delitem, op.getitem, op.indexOf]
TERNARY_OPS = [op.setitem]
if fn in UNARY_OPS:
return 1
elif fn in BINARY_OPS:
return 2
elif fn in TERNARY_OPS:
return 3
else:
raise NotImplementedError("Bult-in operator {} not supported".format(fn))
def focus_selectable(self, forward=True):
"""Change focus to next selectable widget
forward: True to select next widget, False to select previous widget
Returns True if focus was changed, False otherwise.
"""
op = operator.add if forward else operator.sub
max_pos = len(self._main.contents)-1
new_pos = None
pos = self.focus_position
while 0 < pos < max_pos:
pos = op(pos, 1)
item = self._get_item_by_position(pos, visible=True)
if item is not None and item['widget'].selectable():
new_pos = pos
break
if new_pos is not None:
self.focus_position = new_pos
return True
return False
def filter(self, result, from_addr, to_addr, distance):
filtered_result = {}
for key in result:
filtered_list = result[key]
if from_addr:
filtered_list = [item for item in filtered_list if self.base_address + item[0] >= from_addr]
if to_addr:
filtered_list = [item for item in filtered_list if self.base_address + item[0] <= to_addr]
if distance:
if len(filtered_list) < 2:
filtered_list = []
else:
offsets_a = [item[0] for item in filtered_list]
offsets_b = offsets_a[1:] + [0]
api_distances = list(map(operator.sub, offsets_b, offsets_a))
distance_filtered = []
for index, api_distance in enumerate(api_distances[:-1]):
if api_distance <= distance:
if filtered_list[index] not in distance_filtered:
distance_filtered.append(filtered_list[index])
if filtered_list[index + 1] not in distance_filtered:
distance_filtered.append(filtered_list[index + 1])
filtered_list = distance_filtered
filtered_result[key] = filtered_list
return filtered_result
def run_single_regret(bandit_list,bandit_params,plays):
sum_probs_chosen=0
opt=np.zeros(plays)
chosen=np.zeros(plays)
bandit_probs = [x.get_prob() for x in bandit_list]
opt_solution = max(bandit_probs)
for i in range(0,plays):
index = sample_distributions_and_choose(bandit_params)
sum_probs_chosen+=bandit_probs[index]
if(bandit_list[index].pull_handle()):
bandit_params[index]=\
(bandit_params[index][0]+1,bandit_params[index][1])
else:
bandit_params[index]=\
(bandit_params[index][0],bandit_params[index][1]+1)
opt[i] = (i+1)*opt_solution
chosen[i] = sum_probs_chosen
regret_total = map(sub,opt,chosen)
return regret_total
#7.9
#Plot params beforehand
def test_binops(self):
ops = [operator.add, operator.sub, operator.mul, operator.floordiv,
operator.truediv, pow]
scalars = [-1, 1, 2]
idxs = [RangeIndex(0, 10, 1), RangeIndex(0, 20, 2),
RangeIndex(-10, 10, 2), RangeIndex(5, -5, -1)]
for op in ops:
for a, b in combinations(idxs, 2):
result = op(a, b)
expected = op(Int64Index(a), Int64Index(b))
tm.assert_index_equal(result, expected)
for idx in idxs:
for scalar in scalars:
result = op(idx, scalar)
expected = op(Int64Index(idx), scalar)
tm.assert_index_equal(result, expected)
def test_explicit_conversions(self):
# GH 8608
# add/sub are overriden explicity for Float/Int Index
idx = RangeIndex(5)
# float conversions
arr = np.arange(5, dtype='int64') * 3.2
expected = Float64Index(arr)
fidx = idx * 3.2
tm.assert_index_equal(fidx, expected)
fidx = 3.2 * idx
tm.assert_index_equal(fidx, expected)
# interops with numpy arrays
expected = Float64Index(arr)
a = np.zeros(5, dtype='float64')
result = fidx - a
tm.assert_index_equal(result, expected)
expected = Float64Index(-arr)
a = np.zeros(5, dtype='float64')
result = a - fidx
tm.assert_index_equal(result, expected)
def test_arith(self):
self._test_op(self.panel, operator.add)
self._test_op(self.panel, operator.sub)
self._test_op(self.panel, operator.mul)
self._test_op(self.panel, operator.truediv)
self._test_op(self.panel, operator.floordiv)
self._test_op(self.panel, operator.pow)
self._test_op(self.panel, lambda x, y: y + x)
self._test_op(self.panel, lambda x, y: y - x)
self._test_op(self.panel, lambda x, y: y * x)
self._test_op(self.panel, lambda x, y: y / x)
self._test_op(self.panel, lambda x, y: y ** x)
self._test_op(self.panel, lambda x, y: x + y) # panel + 1
self._test_op(self.panel, lambda x, y: x - y) # panel - 1
self._test_op(self.panel, lambda x, y: x * y) # panel * 1
self._test_op(self.panel, lambda x, y: x / y) # panel / 1
self._test_op(self.panel, lambda x, y: x ** y) # panel ** 1
self.assertRaises(Exception, self.panel.__add__, self.panel['ItemA'])
def test_arith_flex_panel(self):
ops = ['add', 'sub', 'mul', 'div', 'truediv', 'pow', 'floordiv', 'mod']
if not compat.PY3:
aliases = {}
else:
aliases = {'div': 'truediv'}
self.panel = self.panel.to_panel()
for n in [np.random.randint(-50, -1), np.random.randint(1, 50), 0]:
for op in ops:
alias = aliases.get(op, op)
f = getattr(operator, alias)
exp = f(self.panel, n)
result = getattr(self.panel, op)(n)
assert_panel_equal(result, exp, check_panel_type=True)
# rops
r_f = lambda x, y: f(y, x)
exp = r_f(self.panel, n)
result = getattr(self.panel, 'r' + op)(n)
assert_panel_equal(result, exp)
def test_operators_none_as_na(self):
df = DataFrame({"col1": [2, 5.0, 123, None],
"col2": [1, 2, 3, 4]}, dtype=object)
ops = [operator.add, operator.sub, operator.mul, operator.truediv]
# since filling converts dtypes from object, changed expected to be
# object
for op in ops:
filled = df.fillna(np.nan)
result = op(df, 3)
expected = op(filled, 3).astype(object)
expected[com.isnull(expected)] = None
assert_frame_equal(result, expected)
result = op(df, df)
expected = op(filled, filled).astype(object)
expected[com.isnull(expected)] = None
assert_frame_equal(result, expected)
result = op(df, df.fillna(7))
assert_frame_equal(result, expected)
result = op(df.fillna(7), df)
assert_frame_equal(result, expected, check_dtype=False)
def test_arith_getitem_commute(self):
df = DataFrame({'A': [1.1, 3.3], 'B': [2.5, -3.9]})
self._test_op(df, operator.add)
self._test_op(df, operator.sub)
self._test_op(df, operator.mul)
self._test_op(df, operator.truediv)
self._test_op(df, operator.floordiv)
self._test_op(df, operator.pow)
self._test_op(df, lambda x, y: y + x)
self._test_op(df, lambda x, y: y - x)
self._test_op(df, lambda x, y: y * x)
self._test_op(df, lambda x, y: y / x)
self._test_op(df, lambda x, y: y ** x)
self._test_op(df, lambda x, y: x + y)
self._test_op(df, lambda x, y: x - y)
self._test_op(df, lambda x, y: x * y)
self._test_op(df, lambda x, y: x / y)
self._test_op(df, lambda x, y: x ** y)
def test_binary_operators(self):
# skipping for now #####
raise nose.SkipTest("skipping sparse binary operators test")
def _check_inplace_op(iop, op):
tmp = self.bseries.copy()
expected = op(tmp, self.bseries)
iop(tmp, self.bseries)
assert_sp_series_equal(tmp, expected)
inplace_ops = ['add', 'sub', 'mul', 'truediv', 'floordiv', 'pow']
for op in inplace_ops:
_check_inplace_op(getattr(operator, "i%s" % op),
getattr(operator, op))
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def sample_environment():
environment = Environment.create_standard()
extras = {
'variable': 20,
'abc': None,
'def': None,
'add': operator.add,
'subtract': operator.sub,
'multiply': lambda x, y, z: x * y * z,
'return10': lambda: 10,
'x': 10,
'y': 20,
'z': 30,
'foo': 100,
}
environment.define_batch_of(extras.items())
return environment
def get_default_operators():
""" generate a mapping of default operators allowed for evaluation """
return {
'u-': Func(1, operator.neg), # unary negation
'u%': Func(1, lambda a: a / Decimal(100)), # unary percentage
'&': Func(2, operator.concat),
'^': Func(2, operator.pow),
'+': Func(2, op_add),
'-': Func(2, operator.sub),
'/': Func(2, operator.truediv),
'*': Func(2, operator.mul),
'=': Func(2, operator.eq),
'<>': Func(2, lambda a, b: not operator.eq(a, b)),
'>': Func(2, operator.gt),
'<': Func(2, operator.lt),
'>=': Func(2, operator.ge),
'<=': Func(2, operator.le),
}
def diffWaysToCompute(self, s):
"""
:type s: str
:rtype: List[int]
"""
tokens = re.split('(\D)', s) # magical regular expression
nums = map(int, tokens[::2]) # nums: an array, ops: a method (operator.xxx)
ops = map({'+': operator.add, '-': operator.sub, '*': operator.mul}.get, tokens[1::2])
def build(lo, hi): # top-down method
if lo == hi:
return [nums[lo]]
return [ops[i](a, b)
for i in xrange(lo, hi)
for a in build(lo, i)
for b in build(i + 1, hi)]
return build(0, len(nums)-1)
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def countByValueAndWindow(self, windowDuration, slideDuration, numPartitions=None):
"""
Return a new DStream in which each RDD contains the count of distinct elements in
RDDs in a sliding window over this DStream.
@param windowDuration: width of the window; must be a multiple of this DStream's
batching interval
@param slideDuration: sliding interval of the window (i.e., the interval after which
the new DStream will generate RDDs); must be a multiple of this
DStream's batching interval
@param numPartitions: number of partitions of each RDD in the new DStream.
"""
keyed = self.map(lambda x: (x, 1))
counted = keyed.reduceByKeyAndWindow(operator.add, operator.sub,
windowDuration, slideDuration, numPartitions)
return counted.filter(lambda kv: kv[1] > 0).count()
def subtract(minuend, subtrahend):
"""Subtracts two numbers.
Args:
minuend (int/float): Value passed in by the user.
subtrahend (int/float): Value passed in by the user.
Returns:
int/float: Result of the difference from the given values.
Example:
>>> subtract(10, 5)
5
>>> subtract(-10, 4)
-14
>>> subtract(2, 0.5)
1.5
.. versionadded:: 4.0.0
"""
return call_math_operator(minuend, subtrahend, operator.sub, 0)
def test_safe_binop():
# Test checked arithmetic routines
ops = [
(operator.add, 1),
(operator.sub, 2),
(operator.mul, 3)
]
with exc_iter(ops, INT64_VALUES, INT64_VALUES) as it:
for xop, a, b in it:
pyop, op = xop
c = pyop(a, b)
if not (INT64_MIN <= c <= INT64_MAX):
assert_raises(OverflowError, mt.extint_safe_binop, a, b, op)
else:
d = mt.extint_safe_binop(a, b, op)
if c != d:
# assert_equal is slow
assert_equal(d, c)
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def eval_expr(expr):
import ast
import operator as op
op = {
ast.Add: op.add,
ast.Sub: op.sub,
ast.Mult: op.mul,
ast.Div: op.truediv,
ast.Pow: op.pow,
ast.BitXor: op.xor,
ast.USub: op.neg,
}
def eval_(node):
if isinstance(node, ast.Num):
return fractions.Fraction(node.n)
elif isinstance(node, ast.BinOp):
return op[type(node.op)](eval_(node.left), eval_(node.right))
elif isinstance(node, ast.UnaryOp):
return op[type(node.op)](eval_(node.operand))
raise TypeError(node)
return eval_(ast.parse(str(expr), mode='eval').body)
def subtract(lhs, rhs):
""" Perform element-wise subtract
Parameters
----------
lhs : Array or float value
left hand side operand
rhs : Array of float value
right hand side operand
Returns
-------
out: Array
result array
"""
# pylint: disable= no-member, protected-access
return _ufunc_helper(
lhs,
rhs,
NDArray._minus,
operator.sub,
NDArray._minus_scalar,
NDArray._rminus_scalar)
# pylint: enable= no-member, protected-access
def parse_code(code):
lines = regex_flink.findall(code)
links = [[] for line in lines]
for index, line in enumerate(lines):
chains = links[index]
for word in regex_chain.findall(line):
chain = []
arity = str_arities.find(word[0])
for token in regex_token.findall(word):
if token in atoms:
chain.append(atoms[token])
elif token in quicks:
popped = []
while not quicks[token].condition(popped) and (chain or chains):
popped.insert(0, chain.pop() if chain else chains.pop())
chain += quicks[token].quicklink(popped, links, index)
elif token in hypers:
x = chain.pop() if chain else chains.pop()
chain.append(hypers[token](x, links))
else:
chain.append(create_literal(regex_liter.sub(parse_literal, token)))
chains.append(create_chain(chain, arity))
return links
def test_foldr(self):
"""
Apply a function of two arguments cumulatively to the items of
a sequence, from right to left, so as to reduce the sequence to
a single value.
"""
self.assertEqual(
irc._foldr(operator.sub, 0, [1, 2, 3, 4]),
-2)
def insertTop(l, x):
l.insert(0, x)
return l
self.assertEqual(
irc._foldr(insertTop, [], [[1], [2], [3], [4]]),
[[[[[], 4], 3], 2], 1])
def _gwas_interaction(data, *entities):
# Finding all the columns for all the targets
column_names = tuple(
tuple(name for name in data.columns if name.startswith(entity.id))
for entity in entities
)
# Finding the level column names if there are factors
factor_levels = tuple(
tuple(name[len(entity.id)+1:] for name in names)
for names, entity in zip(column_names, entities)
)
# Only creating the column name
out = {}
for cols, level_names in zip(itertools.product(*column_names),
itertools.product(*factor_levels)):
# Getting the key (for factors, if present)
key = re.sub(":{2,}", "", ":".join(level_names).strip(":"))
# Saving the columns to multiply with SNPs
out[key] = cols
return out
def test_safe_binop():
# Test checked arithmetic routines
ops = [
(operator.add, 1),
(operator.sub, 2),
(operator.mul, 3)
]
with exc_iter(ops, INT64_VALUES, INT64_VALUES) as it:
for xop, a, b in it:
pyop, op = xop
c = pyop(a, b)
if not (INT64_MIN <= c <= INT64_MAX):
assert_raises(OverflowError, mt.extint_safe_binop, a, b, op)
else:
d = mt.extint_safe_binop(a, b, op)
if c != d:
# assert_equal is slow
assert_equal(d, c)
def check_sub(Poly):
# This checks commutation, not numerical correctness
c1 = list(random((4,)) + .5)
c2 = list(random((3,)) + .5)
p1 = Poly(c1)
p2 = Poly(c2)
p3 = p1 - p2
assert_poly_almost_equal(p2 - p1, -p3)
assert_poly_almost_equal(p1 - c2, p3)
assert_poly_almost_equal(c2 - p1, -p3)
assert_poly_almost_equal(p1 - tuple(c2), p3)
assert_poly_almost_equal(tuple(c2) - p1, -p3)
assert_poly_almost_equal(p1 - np.array(c2), p3)
assert_poly_almost_equal(np.array(c2) - p1, -p3)
assert_raises(TypeError, op.sub, p1, Poly([0], domain=Poly.domain + 1))
assert_raises(TypeError, op.sub, p1, Poly([0], window=Poly.window + 1))
if Poly is Polynomial:
assert_raises(TypeError, op.sub, p1, Chebyshev([0]))
else:
assert_raises(TypeError, op.sub, p1, Polynomial([0]))
def get_numeric_subclasses(cls=numpy.number, ignore=None):
"""
Return subclasses of `cls` in the numpy scalar hierarchy.
We only return subclasses that correspond to unique data types.
The hierarchy can be seen here:
http://docs.scipy.org/doc/numpy/reference/arrays.scalars.html
"""
if ignore is None:
ignore = []
rval = []
dtype = numpy.dtype(cls)
dtype_num = dtype.num
if dtype_num not in ignore:
# Safety check: we should be able to represent 0 with this data type.
numpy.array(0, dtype=dtype)
rval.append(cls)
ignore.append(dtype_num)
for sub in cls.__subclasses__():
rval += [c for c in get_numeric_subclasses(sub, ignore=ignore)]
return rval
def op_to_opstr(op):
if op is operator.le:
return "<="
elif op is operator.ge:
return ">="
elif op is operator.eq:
return "="
elif op is operator.add:
return "+"
elif op is operator.sub:
return "-"
elif op is operator.mul:
return "*"
elif op is operator.div:
return "/"
else:
raise Exception("Operator '{}' not supported yet.", op)
def add_globals(self):
"Add some Scheme standard procedures."
import math, cmath, operator as op
from functools import reduce
self.update(vars(math))
self.update(vars(cmath))
self.update({
'+':op.add, '-':op.sub, '*':op.mul, '/':op.itruediv, 'níl':op.not_, 'agus':op.and_,
'>':op.gt, '<':op.lt, '>=':op.ge, '<=':op.le, '=':op.eq, 'mod':op.mod,
'frmh':cmath.sqrt, 'dearbhluach':abs, 'uas':max, 'íos':min,
'cothrom_le?':op.eq, 'ionann?':op.is_, 'fad':len, 'cons':cons,
'ceann':lambda x:x[0], 'tóin':lambda x:x[1:], 'iarcheangail':op.add,
'liosta':lambda *x:list(x), 'liosta?': lambda x:isa(x,list),
'folamh?':lambda x: x == [], 'adamh?':lambda x: not((isa(x, list)) or (x == None)),
'boole?':lambda x: isa(x, bool), 'scag':lambda f, x: list(filter(f, x)),
'cuir_le':lambda proc,l: proc(*l), 'mapáil':lambda p, x: list(map(p, x)),
'lódáil':lambda fn: load(fn), 'léigh':lambda f: f.read(),
'oscail_comhad_ionchuir':open,'dún_comhad_ionchuir':lambda p: p.file.close(),
'oscail_comhad_aschur':lambda f:open(f,'w'), 'dún_comhad_aschur':lambda p: p.close(),
'dac?':lambda x:x is eof_object, 'luacháil':lambda x: evaluate(x),
'scríobh':lambda x,port=sys.stdout:port.write(to_string(x) + '\n')})
return self
def sub(x, y, as_tuple=True):
""" Subtracts y from x. They can be iterables or scalars.
>>> Util.sub(1, 2)
(-1,)
>>> Util.sub((1,2,3), 1)
(0, 1, 2)
>>> Util.sub(1, [1,2,3])
(0, -1, -2)
>>> Util.sub([1, 2, 3], (4, 5, 5))
(-3, -3, -2)
"""
if Util.is_number(y): y = (y,)
x = Util.promote(x, len(y));
y = Util.promote(y, length=len(x))
assert len(x) == len(y), 'Assert that inputs are of the same length'
out = (i - j for i, j in zip(x, y))
return tuple(out) if as_tuple else out
def __init__(self, code, objects=None):
self._OPERATORS = [
('|', operator.or_),
('^', operator.xor),
('&', operator.and_),
('>>', operator.rshift),
('<<', operator.lshift),
('-', operator.sub),
('+', operator.add),
('%', operator.mod),
('/', operator.truediv),
('*', operator.mul),
]
self._ASSIGN_OPERATORS = [(op + '=', opfunc)
for op, opfunc in self._OPERATORS]
self._ASSIGN_OPERATORS.append(('=', lambda cur, right: right))
self._VARNAME_PATTERN = r'[a-zA-Z_$][a-zA-Z_$0-9]*'
if objects is None:
objects = {}
self.code = code
self._functions = {}
self._objects = objects
def calc(self):
self.nonStreamBucket = map(operator.sub, self.allBucket, self.streamBucket)
allCount = sum(self.allBucket)
streamCount = sum(self.streamBucket)
nonStreamCount = sum(self.nonStreamBucket)
return (allCount, sum(map(lambda x: fEntropy(x, allCount), self.allBucket)), streamCount, sum(map(lambda x: fEntropy(x, streamCount), self.streamBucket)), nonStreamCount, sum(map(lambda x: fEntropy(x, nonStreamCount), self.nonStreamBucket)))
def __sub__(self, trc):
return self.apply_op2(trc, operator.sub)
def __sub__(self, other):
return Vector(self._vectorOp(other, operator.sub), keep=True)