def test_dot_override(self):
# 2016-01-29: NUMPY_UFUNC_DISABLED
return
class A(object):
def __numpy_ufunc__(self, ufunc, method, pos, inputs, **kwargs):
return "A"
class B(object):
def __numpy_ufunc__(self, ufunc, method, pos, inputs, **kwargs):
return NotImplemented
a = A()
b = B()
c = np.array([[1]])
assert_equal(np.dot(a, b), "A")
assert_equal(c.dot(a), "A")
assert_raises(TypeError, np.dot, b, c)
assert_raises(TypeError, c.dot, b)
python类object()的实例源码
def test_ufunc_override_normalize_signature(self):
# 2016-01-29: NUMPY_UFUNC_DISABLED
return
# gh-5674
class SomeClass(object):
def __numpy_ufunc__(self, ufunc, method, i, inputs, **kw):
return kw
a = SomeClass()
kw = np.add(a, [1])
assert_('sig' not in kw and 'signature' not in kw)
kw = np.add(a, [1], sig='ii->i')
assert_('sig' not in kw and 'signature' in kw)
assert_equal(kw['signature'], 'ii->i')
kw = np.add(a, [1], signature='ii->i')
assert_('sig' not in kw and 'signature' in kw)
assert_equal(kw['signature'], 'ii->i')
def write_csv(df, filename):
""" Write a pandas dataframe to CSV in a standard way """
# Verify that the data do not contain commas
for col in df.select_dtypes([np.object]):
if df[col].str.contains(',').any():
raise ValueError("Failed write to %s: Column %s contains commas" % (filename, col))
df.to_csv(filename, header=True, index=False, sep=',')
def append_data_column(ds, column):
# Extend the dataset to fit the new data
new_count = column.shape[0]
existing_count = ds.shape[0]
ds.resize((existing_count + new_count,))
levels = get_levels(ds)
if levels is not None:
# update levels if we have new unique values
if type(column.values) == p.Categorical:
added_levels = set(column.values.categories) - set(levels)
elif len(column) == 0:
# Workaround for bug in pandas - get a crash in .unique() for an empty series
added_levels = set([])
else:
added_levels = set(column.unique()) - set(levels)
new_levels = list(levels)
new_levels.extend(added_levels)
# Check if the new categorical column has more levels
# than the current bit width supports.
# If so, rewrite the existing column data w/ more bits
if len(new_levels) > np.iinfo(ds.dtype).max:
new_dtype = pick_cat_dtype(len(new_levels))
ds = widen_cat_column(ds, new_dtype)
new_levels = np.array(new_levels, dtype=np.object)
new_data = make_index_array(new_levels, column.values, ds.dtype)
clear_levels(ds)
create_levels(ds, new_levels)
else:
new_data = column
# Append new data
ds[existing_count:(existing_count + new_count)] = new_data
def analyze_pd_dataframe(dataframe, target_attributes):
"""Analyze pandas.Dataframe and convert it into internal representation.
Parameters
----------
dataframe : pd.Dataframe
input data, can contain float, int, object
target_attributes : int, str or list
Index the target attribute. If this is
* an int, use this as an index (only works with positive indices)
* a str, use this to compare with the column values
* a list (which must either consist of all ints or strs), of which
all elements that matched are assumed to be targets.
Returns
-------
np.ndarray
Data. All columns are converted to type float. Categorical data is
encoded by positive integers.
dict
Attribute types. Contains the following keys:
* `type`: `categorical` or 'numerical`
* `name`: column name of the dataframe
* `is_target`: whether this column was designated as a target column
"""
dataframe = _normalize_pd_column_names(dataframe)
attribute_types = _get_pd_attribute_types(dataframe, target_attributes)
dataframe = _replace_objects_by_integers(dataframe, attribute_types)
return dataframe.values, attribute_types
def _compute_optimal(self):
not_visited = {
(y, x)
for x in range(self.width) for y in range(self.height)
}
queue = collections.deque()
queue.append(tuple(j[0] for j in np.where(self.grid == G)))
policy = np.empty(self.grid.shape, dtype=np.object)
print("INITIAL POLICY")
print(policy)
while len(queue) > 0:
current = queue.pop()
if current in not_visited:
not_visited.remove(current)
possible_actions = self.possible_next_actions(
self._index(current), True
)
for action in possible_actions:
self._state = self._index(current)
next_state, _, _, _ = self.step(action)
next_state_pos = self._pos(next_state)
if next_state_pos not in not_visited:
continue
not_visited.remove(next_state_pos)
if not self.is_terminal(next_state) and \
self.grid[next_state_pos] != W:
policy[next_state_pos] = self.invert_action(action)
queue.appendleft(self._pos(next_state))
print("FINAL POLICY")
print(policy)
return policy
def test_run(self):
"""Only test hash runs at all."""
for t in [np.int, np.float, np.complex, np.int32, np.str, np.object,
np.unicode]:
dt = np.dtype(t)
hash(dt)
def test_shape_sequence(self):
# Any sequence of integers should work as shape, but the result
# should be a tuple (immutable) of base type integers.
a = np.array([1, 2, 3], dtype=np.int16)
l = [1, 2, 3]
# Array gets converted
dt = np.dtype([('a', 'f4', a)])
assert_(isinstance(dt['a'].shape, tuple))
assert_(isinstance(dt['a'].shape[0], int))
# List gets converted
dt = np.dtype([('a', 'f4', l)])
assert_(isinstance(dt['a'].shape, tuple))
#
class IntLike(object):
def __index__(self):
return 3
def __int__(self):
# (a PyNumber_Check fails without __int__)
return 3
dt = np.dtype([('a', 'f4', IntLike())])
assert_(isinstance(dt['a'].shape, tuple))
assert_(isinstance(dt['a'].shape[0], int))
dt = np.dtype([('a', 'f4', (IntLike(),))])
assert_(isinstance(dt['a'].shape, tuple))
assert_(isinstance(dt['a'].shape[0], int))
def test_empty_string_to_object(self):
# Pull request #4722
np.array(["", ""]).astype(object)
def test_object_nans(self):
# Multiple checks to give this a chance to
# fail if cmp is used instead of rich compare.
# Failure cannot be guaranteed.
for i in range(1):
x = np.array(float('nan'), np.object)
y = 1.0
z = np.array(float('nan'), np.object)
assert_(np.maximum(x, y) == 1.0)
assert_(np.maximum(z, y) == 1.0)
def test_object_array(self):
arg1 = np.arange(5, dtype=np.object)
arg2 = arg1 + 1
assert_equal(np.maximum(arg1, arg2), arg2)
def test_object_array(self):
arg1 = np.arange(5, dtype=np.object)
arg2 = arg1 + 1
assert_equal(np.minimum(arg1, arg2), arg1)
def test_sign_dtype_object(self):
# In reference to github issue #6229
foo = np.array([-.1, 0, .1])
a = np.sign(foo.astype(np.object))
b = np.sign(foo)
assert_array_equal(a, b)
def test_sign_dtype_nan_object(self):
# In reference to github issue #6229
def test_nan():
foo = np.array([np.nan])
a = np.sign(foo.astype(np.object))
assert_raises(TypeError, test_nan)
def test_old_wrap(self):
class with_wrap(object):
def __array__(self):
return np.zeros(1)
def __array_wrap__(self, arr):
r = with_wrap()
r.arr = arr
return r
a = with_wrap()
x = ncu.minimum(a, a)
assert_equal(x.arr, np.zeros(1))
def test_failing_wrap(self):
class A(object):
def __array__(self):
return np.zeros(1)
def __array_wrap__(self, arr, context):
raise RuntimeError
a = A()
self.assertRaises(RuntimeError, ncu.maximum, a, a)
def test_default_prepare(self):
class with_wrap(object):
__array_priority__ = 10
def __array__(self):
return np.zeros(1)
def __array_wrap__(self, arr, context):
return arr
a = with_wrap()
x = ncu.minimum(a, a)
assert_equal(x, np.zeros(1))
assert_equal(type(x), np.ndarray)
def test_failing_prepare(self):
class A(object):
def __array__(self):
return np.zeros(1)
def __array_prepare__(self, arr, context=None):
raise RuntimeError
a = A()
self.assertRaises(RuntimeError, ncu.maximum, a, a)
def test_array_with_context(self):
class A(object):
def __array__(self, dtype=None, context=None):
func, args, i = context
self.func = func
self.args = args
self.i = i
return np.zeros(1)
class B(object):
def __array__(self, dtype=None):
return np.zeros(1, dtype)
class C(object):
def __array__(self):
return np.zeros(1)
a = A()
ncu.maximum(np.zeros(1), a)
self.assertTrue(a.func is ncu.maximum)
assert_equal(a.args[0], 0)
self.assertTrue(a.args[1] is a)
self.assertTrue(a.i == 1)
assert_equal(ncu.maximum(a, B()), 0)
assert_equal(ncu.maximum(a, C()), 0)
def test_ufunc_override_disabled(self):
# 2016-01-29: NUMPY_UFUNC_DISABLED
# This test should be removed when __numpy_ufunc__ is re-enabled.
class MyArray(object):
def __numpy_ufunc__(self, *args, **kwargs):
self._numpy_ufunc_called = True
my_array = MyArray()
real_array = np.ones(10)
assert_raises(TypeError, lambda: real_array + my_array)
assert_raises(TypeError, np.add, real_array, my_array)
assert not hasattr(my_array, "_numpy_ufunc_called")
