def test_Iterator(self):
non_samples = [None, 42, 3.14, 1j, "".encode('ascii'), "", (), [],
{}, set()]
for x in non_samples:
self.assertNotIsInstance(x, Iterator)
self.assertFalse(issubclass(type(x), Iterator), repr(type(x)))
samples = [iter(str()),
iter(tuple()), iter(list()), iter(dict()),
iter(set()), iter(frozenset()),
iter(dict().keys()), iter(dict().items()),
iter(dict().values()),
(lambda: (yield))(),
(x for x in []),
]
for x in samples:
self.assertIsInstance(x, Iterator)
self.assertTrue(issubclass(type(x), Iterator), repr(type(x)))
self.validate_abstract_methods(Iterator, 'next', '__iter__')
# Issue 10565
class NextOnly:
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnly(), Iterator)
class NextOnlyNew(object):
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnlyNew(), Iterator)
python类Iterator()的实例源码
def test_direct_subclassing(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C(B):
pass
self.assertTrue(issubclass(C, B))
self.assertFalse(issubclass(int, C))
def test_registration(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C:
__metaclass__ = type
__hash__ = None # Make sure it isn't hashable by default
self.assertFalse(issubclass(C, B), B.__name__)
B.register(C)
self.assertTrue(issubclass(C, B))
def test_range(self):
self.assertTrue(isinstance(range(0), Sequence))
self.assertTrue(isinstance(reversed(range(0)), Iterator))
def set_header_devpi_serial(response, tx):
if isinstance(response._app_iter, collections.Iterator):
return
if tx.commit_serial is not None:
serial = tx.commit_serial
else:
serial = tx.at_serial
response.headers[str("X-DEVPI-SERIAL")] = str(serial)
def to_params(expr, **kwargs):
if isinstance(expr, MesosDelayed):
return expr._params
if isinstance(expr, (Iterator, list, tuple, set)):
params = [to_params(e) for e in expr]
return flat_unique(params)
if isinstance(expr, dict):
params = [to_params(e) for e in expr.values()]
return flat_unique(params)
return []
def summary(x, axis=None, shorten=False):
if isinstance(x, Iterator):
x = list(x)
if isinstance(x, (tuple, list)):
x = np.array(x)
mean, std = np.mean(x, axis=axis), np.std(x, axis=axis)
median = np.median(x, axis=axis)
qu1, qu3 = np.percentile(x, [25, 75], axis=axis)
min_, max_ = np.min(x, axis=axis), np.max(x, axis=axis)
samples = ', '.join([str(i)
for i in np.random.choice(x.ravel(), size=8, replace=False).tolist()])
s = ""
if not shorten:
s += "***** Summary *****\n"
s += " Min : %s\n" % str(min_)
s += "1st Qu. : %s\n" % str(qu1)
s += " Median : %s\n" % str(median)
s += " Mean : %.8f\n" % mean
s += "3rd Qu. : %s\n" % str(qu3)
s += " Max : %s\n" % str(max_)
s += "-------------------\n"
s += " Std : %.8f\n" % std
s += "#Samples : %d\n" % len(x)
s += "Samples : %s\n" % samples
else:
s += "{#:%d|min:%s|qu1:%s|med:%s|mea:%.8f|qu3:%s|max:%s|std:%.8f}" %\
(len(x), str(min_), str(qu1), str(median), mean, str(qu3), str(max_), std)
return s
def iter_chunk(it, n):
""" Chunking an iterator into small chunk of size `n`
Note: this can be used to slice data into mini batches
"""
if not isinstance(it, Iterator):
it = iter(it)
obj = list(islice(it, n))
while obj:
yield obj
obj = list(islice(it, n))
def flatten_list(x, level=None):
"""
Parameters
----------
level: int, or None
how deep the function go into element of x to search for list and
flatten it. If None is given, flatten all list found.
Example
-------
>>> l = [1, 2, 3, [4], [[5], [6]], [[7], [[8], [9]]]]
>>> print(flatten_list(l, level=1))
>>> # [1, 2, 3, 4, [5], [6], [7], [[8], [9]]]
>>> print(flatten_list(l, level=2))
>>> # [1, 2, 3, 4, 5, 6, 7, [8], [9]]
>>> print(flatten_list(l, level=None))
>>> # [1, 2, 3, 4, 5, 6, 7, 8, 9]
"""
if isinstance(x, Iterator):
x = list(x)
if level is None:
level = 10e8
if not isinstance(x, (tuple, list)):
return [x]
if any(isinstance(i, (tuple, list)) for i in x):
_ = []
for i in x:
if isinstance(i, (tuple, list)) and level > 0:
_ += flatten_list(i, level - 1)
else:
_.append(i)
return _
return x
# ===========================================================================
# Python
# ===========================================================================
def _validate_texts(self, texts):
""" Valiate the input to `fit` and `transform` """
if not isinstance(texts, Iterable) and \
not isinstance(texts, Iterator) and \
not is_string(texts):
raise ValueError('texts must be an iterator, generator or a string.')
if is_string(texts):
texts = (texts,)
# convert to unicode
texts = (t.decode('utf-8') for t in texts)
return texts
# ==================== properties ==================== #
def test_iterators(self):
self.assertIsInstance(django_load.iterload('menus'), Iterator)
toolbars = get_cms_setting('TOOLBARS')
self.assertIsInstance(django_load.iterload_objects(toolbars), Iterator)
def test_Iterator(self):
non_samples = [None, 42, 3.14, 1j, "".encode('ascii'), "", (), [],
{}, set()]
for x in non_samples:
self.assertNotIsInstance(x, Iterator)
self.assertFalse(issubclass(type(x), Iterator), repr(type(x)))
samples = [iter(str()),
iter(tuple()), iter(list()), iter(dict()),
iter(set()), iter(frozenset()),
iter(dict().keys()), iter(dict().items()),
iter(dict().values()),
(lambda: (yield))(),
(x for x in []),
]
for x in samples:
self.assertIsInstance(x, Iterator)
self.assertTrue(issubclass(type(x), Iterator), repr(type(x)))
self.validate_abstract_methods(Iterator, 'next', '__iter__')
# Issue 10565
class NextOnly:
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnly(), Iterator)
class NextOnlyNew(object):
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnlyNew(), Iterator)
def test_direct_subclassing(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C(B):
pass
self.assertTrue(issubclass(C, B))
self.assertFalse(issubclass(int, C))
def test_registration(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C:
__metaclass__ = type
__hash__ = None # Make sure it isn't hashable by default
self.assertFalse(issubclass(C, B), B.__name__)
B.register(C)
self.assertTrue(issubclass(C, B))
def test_Iterator(self):
non_samples = [None, 42, 3.14, 1j, "".encode('ascii'), "", (), [],
{}, set()]
for x in non_samples:
self.assertNotIsInstance(x, Iterator)
self.assertFalse(issubclass(type(x), Iterator), repr(type(x)))
samples = [iter(str()),
iter(tuple()), iter(list()), iter(dict()),
iter(set()), iter(frozenset()),
iter(dict().keys()), iter(dict().items()),
iter(dict().values()),
(lambda: (yield))(),
(x for x in []),
]
for x in samples:
self.assertIsInstance(x, Iterator)
self.assertTrue(issubclass(type(x), Iterator), repr(type(x)))
self.validate_abstract_methods(Iterator, 'next', '__iter__')
# Issue 10565
class NextOnly:
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnly(), Iterator)
class NextOnlyNew(object):
def __next__(self):
yield 1
raise StopIteration
self.assertNotIsInstance(NextOnlyNew(), Iterator)
def test_direct_subclassing(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C(B):
pass
self.assertTrue(issubclass(C, B))
self.assertFalse(issubclass(int, C))
def test_registration(self):
for B in Hashable, Iterable, Iterator, Sized, Container, Callable:
class C:
__metaclass__ = type
__hash__ = None # Make sure it isn't hashable by default
self.assertFalse(issubclass(C, B), B.__name__)
B.register(C)
self.assertTrue(issubclass(C, B))
def _streamResponse(self, data, client):
if isinstance(data, collections.Iterator) or inspect.isgenerator(data):
if config.ITER_STREAMING:
if type(data) in self.__lazy_dict_iterator_types:
raise errors.PyroError("won't serialize or stream lazy dict iterators, convert to list yourself")
stream_id = str(uuid.uuid4())
self.streaming_responses[stream_id] = (client, time.time(), 0, data)
return True, stream_id
return True, None
return False, data
def pysub(arg, line, num):
"""substitutes the return value of a python statement for an arg"""
namespace.update(l=line)
value = eval(arg, namespace)
# return multiple args if the return value is a list, tuple or iterator
if isinstance(value, (list, tuple, collections.Iterator)):
return value
return [str(value)]
def test_propclass_propertyNames():
p = Properties({"key": "value", "apple": "zebra", "foo": "bar"})
names = p.propertyNames()
assert isinstance(names, collections.Iterator)
assert sorted(names) == ["apple", "foo", "key"]
