我们从Python开源项目中，提取了以下17个代码示例，用于说明如何使用collections.Generator()。
def __init__(self, indent_step=4, indent_char=' ', repr_strings=False, simple_cutoff=10, width=120, yield_from_generators=True): self._indent_step = indent_step self._c = indent_char self._repr_strings = repr_strings self._repr_generators = not yield_from_generators self._simple_cutoff = simple_cutoff self._width = width self._type_lookup = [ (dict, self._format_dict), (str, self._format_str), (bytes, self._format_bytes), (tuple, self._format_tuples), ((list, set, frozenset), self._format_list_like), (collections.Generator, self._format_generators), ]
def _format(self, value: Any, indent_current: int, indent_first: bool): if indent_first: self._stream.write(indent_current * self._c) value_repr = repr(value) if len(value_repr) <= self._simple_cutoff and not isinstance(value, collections.Generator): self._stream.write(value_repr) else: indent_new = indent_current + self._indent_step for t, func in self._type_lookup: if isinstance(value, t): func(value, value_repr, indent_current, indent_new) return self._format_raw(value, value_repr, indent_current, indent_new)
def _format_generators(self, value: Generator, value_repr: str, indent_current: int, indent_new: int): if self._repr_generators: self._stream.write(value_repr) else: self._stream.write('(\n') for v in value: self._format(v, indent_new, True) self._stream.write(',\n') self._stream.write(indent_current * self._c + ')')
def patch(patch_inspect=True): """ Main entry point for patching the ``collections.abc`` and ``inspect`` standard library modules. """ PATCHED['collections.abc.Generator'] = _collections_abc.Generator = Generator PATCHED['collections.abc.Coroutine'] = _collections_abc.Coroutine = Coroutine PATCHED['collections.abc.Awaitable'] = _collections_abc.Awaitable = Awaitable if patch_inspect: import inspect PATCHED['inspect.isawaitable'] = inspect.isawaitable = isawaitable
def test_agent_group(agent_type, attributes): group = AgentGroup(size=SIZE, agent_type=agent_type, attributes=attributes) assert isinstance(group.size, int) assert issubclass(group.agent_type, AgentType) assert isinstance(group.attributes, (Collection, Generator, Callable)) assert isinstance(group.members, list) assert len(group.members) == SIZE
def _observe_members(self, change): if self.size > 0 and self.attributes is not None and self.agent_type is not None: if isinstance(self.attributes, Collection): self.members = [self.agent_type(**a) for a in self.attributes] elif isinstance(self.attributes, Generator): self.members = [self.agent_type(**next(self.attributes)) for _ in range(self.size)] elif isinstance(self.attributes, Callable): self.members = [self.agent_type(**self.attributes()) for _ in range(self.size)] else: raise TraitError
def mk_gen(): from abc import abstractmethod required_methods = ( '__iter__', '__next__' if hasattr(iter(()), '__next__') else 'next', 'send', 'throw', 'close') class Generator(_collections_abc.Iterator): __slots__ = () if '__next__' in required_methods: def __next__(self): return self.send(None) else: def next(self): return self.send(None) @abstractmethod def send(self, value): raise StopIteration @abstractmethod def throw(self, typ, val=None, tb=None): if val is None: if tb is None: raise typ val = typ() if tb is not None: val = val.with_traceback(tb) raise val def close(self): try: self.throw(GeneratorExit) except (GeneratorExit, StopIteration): pass else: raise RuntimeError('generator ignored GeneratorExit') @classmethod def __subclasshook__(cls, C): if cls is Generator: mro = C.__mro__ for method in required_methods: for base in mro: if method in base.__dict__: break else: return NotImplemented return True return NotImplemented generator = type((lambda: (yield))()) Generator.register(generator) return Generator
def add_non_overlapping_group(self, group, position_gen, obstacles=None): """Add group of agents Args: group (AgentGroup): position_gen (Generator|Callable): obstacles (numpy.ndarray): """ if self.agent_type is not group.agent_type: raise CrowdDynamicsException # resize self.array to fit new agents array = np.zeros(group.size, dtype=group.agent_type.dtype()) self.array = np.concatenate((self.array, array)) index = 0 overlaps = 0 overlaps_max = 10 * group.size while index < group.size and overlaps < overlaps_max: new_agent = group.members[index] new_agent.position = position_gen() if callable(position_gen) \ else next(position_gen) # Overlapping check neighbours = self._neighbours.nearest(new_agent.position, radius=1) if new_agent.overlapping(self.array[neighbours]): # Agent is overlapping other agent. overlaps += 1 continue if obstacles is not None and new_agent.overlapping_obstacles(obstacles): # Agent is overlapping with an obstacle. overlaps += 1 continue # Agent can be successfully placed self.array[self.index] = np.array(new_agent) self._neighbours[new_agent.position] = self.index self.index += 1 index += 1 # TODO: remove agents that didn't fit from self.array if self.index + 1 < self.array.size: pass # Array should remain contiguous assert self.array.flags.c_contiguous
