python - libreria - Cómo no perderse el siguiente elemento después de itertools.takewhile()

itertools python español

Question

Digamos que deseamos procesar un iterador y queremos manejarlo por trozos.
La lógica por trozo depende de trozos calculados previamente, por lo que groupby() no ayuda.

Nuestro amigo en este caso es itertools.takewhile ():

while True:
    chunk = itertools.takewhile(getNewChunkLogic(), myIterator)
    process(chunk)

El problema es que takewhile() necesita ir más allá del último elemento que cumple con la nueva lógica de fragmentos, por lo que "come" el primer elemento para el siguiente fragmento.

Hay varias soluciones para eso, incluyendo el ungetc() o a la C''s ungetc() , etc.
Mi pregunta es: ¿hay una solución elegante ?

Answer 1

Dada la GetNewChunkLogic() informará True largo de la primera parte y False después.
El siguiente fragmento

1. resuelve el problema del "siguiente paso adicional" de takewhile .
2. es elegante porque no tiene que implementar la lógica de un paso atrás.

def partition(pred, iterable):
    ''Use a predicate to partition entries into true entries and false entries''
    # partition(is_odd, range(10)) -->  1 3 5 7 9 and 0 2 4 6 8
    t1, t2 = tee(iterable)
    return filter(pred, t1), filterfalse(pred, t2)

while True:
    head, tail = partition(GetNewChunkLogic(), myIterator)
    process(head)
    myIterator = tail

Sin embargo, la forma más elegante es modificar su GetNewChunkLogic en un generador y eliminar el bucle while.

Answer 2

takewhile() hecho necesita mirar el siguiente elemento para determinar cuándo cambiar el comportamiento.

Podría usar un envoltorio que rastree el último elemento visto, y que se pueda ''reiniciar'' para hacer una copia de seguridad de un elemento:

_sentinel = object()

class OneStepBuffered(object):
    def __init__(self, it):
        self._it = iter(it)
        self._last = _sentinel
        self._next = _sentinel
    def __iter__(self):
        return self
    def __next__(self):
        if self._next is not _sentinel:
            next_val, self._next = self._next, _sentinel
            return next_val
        try:
            self._last = next(self._it)
            return self._last
        except StopIteration:
            self._last = self._next = _sentinel
            raise
    next = __next__  # Python 2 compatibility
    def step_back(self):
        if self._last is _sentinel:
            raise ValueError("Can''t back up a step")
        self._next, self._last = self._last, _sentinel

Envuelve tu iterador en este antes de usarlo con takewhile() :

myIterator = OneStepBuffered(myIterator)
while True:
    chunk = itertools.takewhile(getNewChunkLogic(), myIterator)
    process(chunk)
    myIterator.step_back()

Manifestación:

>>> from itertools import takewhile
>>> test_list = range(10)
>>> iterator = OneStepBuffered(test_list)
>>> list(takewhile(lambda i: i < 5, iterator))
[0, 1, 2, 3, 4]
>>> iterator.step_back()
>>> list(iterator)
[5, 6, 7, 8, 9]