iteratori zip che affermano per uguale lunghezza in python

Question

Sto cercando un modo semplice per zip diversi iterabili che generano un'eccezione se le lunghezze dei iterabili non sono uguali.

Nel caso in cui i iterabili sono elenchi o avere un metodo len questa soluzione è pulito e facile:

def zip_equal(it1, it2): 
    if len(it1) != len(it2): 
     raise ValueError("Lengths of iterables are different") 
    return zip(it1, it2) 

Tuttavia, se it1 e it2 sono generatori, la funzione precedente non riesce perché la lunghezza non è definita TypeError: object of type 'generator' has no len().

Immagino che il modulo itertools offra un modo semplice per implementarlo, ma finora non sono stato in grado di trovarlo. Sono venuto su con questa soluzione fatta in casa:

def zip_equal(it1, it2): 
    exhausted = False 
    while True: 
     try: 
      el1 = next(it1) 
      if exhausted: # in a previous iteration it2 was exhausted but it1 still has elements 
       raise ValueError("it1 and it2 have different lengths") 
     except StopIteration: 
      exhausted = True 
      # it2 must be exhausted too. 
     try: 
      el2 = next(it2) 
      # here it2 is not exhausted. 
      if exhausted: # it1 was exhausted => raise 
       raise ValueError("it1 and it2 have different lengths") 
     except StopIteration: 
      # here it2 is exhausted 
      if not exhausted: 
       # but it1 was not exhausted => raise 
       raise ValueError("it1 and it2 have different lengths") 
      exhausted = True 
     if not exhausted: 
      yield (el1, el2) 
     else: 
      return 

La soluzione può essere testato con il seguente codice:

it1 = (x for x in ['a', 'b', 'c']) # it1 has length 3 
it2 = (x for x in [0, 1, 2, 3])  # it2 has length 4 
list(zip_equal(it1, it2))   # len(it1) < len(it2) => raise 
it1 = (x for x in ['a', 'b', 'c']) # it1 has length 3 
it2 = (x for x in [0, 1, 2, 3])  # it2 has length 4 
list(zip_equal(it2, it1))   # len(it2) > len(it1) => raise 
it1 = (x for x in ['a', 'b', 'c', 'd']) # it1 has length 4 
it2 = (x for x in [0, 1, 2, 3])   # it2 has length 4 
list(zip_equal(it1, it2))    # like zip (or izip in python2) 

Perchè sono affacciate qualsiasi soluzione alternativa? Esiste un'implementazione più semplice della mia funzione zip_equal?

PS: Ho scritto la domanda pensando in Python 3, ma una soluzione Python 2 è anche benvenuta.

Answer 1

posso una soluzione più semplice, utilizzare itertools.zip_longest() e sollevare un'eccezione se il valore sentinella utilizzato per riempire le iterables più brevi è presente nella tupla prodotto:

from itertools import zip_longest 

def zip_equal(*iterables): 
    sentinel = object() 
    for combo in zip_longest(*iterables, fillvalue=sentinel): 
     if sentinel in combo: 
      raise ValueError('Iterables have different lengths') 
     yield combo 

Purtroppo, non possiamo usare zip() con yield from per evitare un loop di codice Python con un test ogni iterazione; una volta esaurito l'iteratore più breve, zip() farà avanzare tutti gli iteratori precedenti e in tal modo inghiottirà la prova se c'è un solo oggetto in più in quelli.

Answer 2

Ecco un approccio che non richiede alcun controllo aggiuntivo con ogni ciclo dell'iterazione. Questo potrebbe essere auspicabile soprattutto per lunghi iterables.

L'idea è di riempire ogni iterabile con un "valore" alla fine che solleva un'eccezione quando viene raggiunto, quindi eseguire la verifica necessaria solo alla fine. L'approccio utilizza zip() e itertools.chain().

Il codice seguente è stato scritto per Python 3.5.

import itertools 

class ExhaustedError(Exception): 
    def __init__(self, index): 
     """The index is the 0-based index of the exhausted iterable.""" 
     self.index = index 

def raising_iter(i): 
    """Return an iterator that raises an ExhaustedError.""" 
    raise ExhaustedError(i) 
    yield 

def terminate_iter(i, iterable): 
    """Return an iterator that raises an ExhaustedError at the end.""" 
    return itertools.chain(iterable, raising_iter(i)) 

def zip_equal(*iterables): 
    iterators = [terminate_iter(*args) for args in enumerate(iterables)] 
    try: 
     yield from zip(*iterators) 
    except ExhaustedError as exc: 
     index = exc.index 
     if index > 0: 
      raise RuntimeError('iterable {} exhausted first'.format(index)) from None 
     # Check that all other iterators are also exhausted. 
     for i, iterator in enumerate(iterators[1:], start=1): 
      try: 
       next(iterator) 
      except ExhaustedError: 
       pass 
      else: 
       raise RuntimeError('iterable {} is longer'.format(i)) from None 

Di seguito è ciò che sembra essere utilizzato.

>>> list(zip_equal([1, 2], [3, 4], [5, 6])) 
[(1, 3, 5), (2, 4, 6)] 

>>> list(zip_equal([1, 2], [3], [4])) 
RuntimeError: iterable 1 exhausted first 

>>> list(zip_equal([1], [2, 3], [4])) 
RuntimeError: iterable 1 is longer 

>>> list(zip_equal([1], [2], [3, 4])) 
RuntimeError: iterable 2 is longer 

Answer 3

mi si avvicinò con una soluzione utilizzando Sentinel iterabile FYI:

class _SentinelException(Exception): 
    def __iter__(self): 
     raise _SentinelException 


def zip_equal(iterable1, iterable2): 
    i1 = iter(itertools.chain(iterable1, _SentinelException())) 
    i2 = iter(iterable2) 
    try: 
     while True: 
      yield (next(i1), next(i2)) 
    except _SentinelException: # i1 reaches end 
     try: 
      next(i2) # check whether i2 reaches end 
     except StopIteration: 
      pass 
     else: 
      raise ValueError('the second iterable is longer than the first one') 
    except StopIteration: # i2 reaches end, as next(i1) has already been called, i1's length is bigger than i2 
     raise ValueError('the first iterable is longger the second one.')