zip iteratory zapewniające równą długość w python

Question

Szukam fajny sposób na zip kilka iterables podnosząc wyjątek, jeśli długość iterables nie są równe.

W przypadku, gdy iterables są listy lub mieć metodę len rozwiązanie to jest czyste i proste:

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

Jednak jeśli it1 i it2 są generatory, poprzednia funkcja nie powiedzie się, ponieważ długość nie jest zdefiniowana TypeError: object of type 'generator' has no len().

Wyobrażam sobie, że moduł itertools oferuje prosty sposób wdrożenia tego, ale jak dotąd nie udało mi się go znaleźć. I mają pochodzić z tego domowej roboty rozwiązania:

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 

Rozwiązaniem może być badana za pomocą następującego kodu:

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) 

jestem widokiem żadnego alternatywnego rozwiązania? Czy istnieje prostsza implementacja funkcji zip_equal?

PS: Napisałem pytanie myśląc w Pythonie 3, ale rozwiązanie Python 2 jest również mile widziane.

Answer 1

mogę prostsze rozwiązanie, użyj itertools.zip_longest() i podnieść wyjątek jeśli wartość Sentinel wykorzystywane do pad się krótsze iterables jest obecny w krotce Produkcja:

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 

Niestety, nie możemy wykorzystać zip() z yield from aby uniknąć pętli kodu Pythona z testem każdej iteracji; gdy skończy się najkrótszy iterator, zip() przesunie wszystkie poprzednie iteratory, a tym samym połknie dowody, jeśli w nich jest tylko jeden dodatkowy element.

Answer 2

Oto podejście, które nie wymaga żadnych dodatkowych kontroli z każdą pętlą iteracji. Może to być pożądane szczególnie w przypadku długich iteracji.

Chodzi o to, aby na każdej iteracji umieścić "wartość" na końcu, która podnosi wyjątek po osiągnięciu, a następnie przeprowadzić wymaganą weryfikację tylko na samym końcu. Podejście to używa zip() i itertools.chain().

Poniższy kod został napisany dla języka 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 

Poniżej opisano sposób użycia.

>>> 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

wymyśliłem rozwiązanie używając Sentinel iterowalny 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.')