itertools
--- 効率的なループ実行のためのイテレータ生成関数¶
このモジュールは イテレータ を構築する部品を実装しています。プログラム言語 APL, Haskell, SML からアイデアを得ていますが、 Python に適した形に修正されています。
このモジュールは、高速でメモリ効率に優れ、単独でも組合せても使用することのできるツールを標準化したものです。同時に、このツール群は "イテレータの代数" を構成していて、pure Python で簡潔かつ効率的なツールを作れるようにしています。
例えば、SML の作表ツール tabulate(f)
は f(0), f(1), ...
のシーケンスを作成します。同じことを Python では map()
と count()
を組合せて map(f, count())
という形で実現できます。
これらのツールと組み込み関数は operator
モジュール内の高速な関数とともに使うことで見事に動作します。例えば、乗算演算子を2つのベクトルにわたってマップすることで効率的な内積計算を実現できます: sum(map(operator.mul, vector1, vector2))
。
無限イテレータ:
イテレータ |
引数 |
結果 |
使用例 |
---|---|---|---|
start, [step] |
start, start+step, start+2*step, ... |
|
|
p |
p0, p1, ... plast, p0, p1, ... |
|
|
elem [,n] |
elem, elem, elem, ... 無限もしくは n 回 |
|
一番短い入力シーケンスで止まるイテレータ:
イテレータ |
引数 |
結果 |
使用例 |
---|---|---|---|
p [,func] |
p0, p0+p1, p0+p1+p2, ... |
|
|
p, q, ... |
p0, p1, ... plast, q0, q1, ... |
|
|
iterable |
p0, p1, ... plast, q0, q1, ... |
|
|
data, selectors |
(d[0] if s[0]), (d[1] if s[1]), ... |
|
|
pred, seq |
seq[n], seq[n+1], pred が偽の場所から始まる |
|
|
pred, seq |
pred(elem) が偽になるseqの要素 |
|
|
iterable[, key] |
key(v) の値でグループ化したサブイテレータ |
||
seq, [start,] stop [, step] |
seq[start:stop:step] |
|
|
iterable |
(p[0], p[1]), (p[1], p[2]) |
|
|
func, seq |
func(*seq[0]), func(*seq[1]), ... |
|
|
pred, seq |
seq[0], seq[1], pred が偽になるまで |
|
|
it, n |
it1, it2 , ... itn 一つのイテレータを n 個に分ける |
||
p, q, ... |
(p[0], q[0]), (p[1], q[1]), ... |
|
組合せイテレータ:
イテレータ |
引数 |
結果 |
---|---|---|
p, q, ... [repeat=1] |
デカルト積、ネストしたforループと等価 |
|
p[, r] |
長さrのタプル列、重複なしのあらゆる並び |
|
p, r |
長さrのタプル列、ソートされた順で重複なし |
|
p, r |
長さrのタプル列、ソートされた順で重複あり |
使用例 |
結果 |
---|---|
|
|
|
|
|
|
|
|
Itertool関数¶
以下の関数は全て、イテレータを作成して返します。無限長のストリームのイテレータを返す関数もあり、この場合にはストリームを中断するような関数かループ処理から使用しなければなりません。
-
itertools.
accumulate
(iterable[, func, *, initial=None])¶ 累計や加算ではない (func オプション引数で指定される) 2変数関数による累積結果を返すイテレータを作成します。
func が与えられた場合、2つの引数を取る関数でなければなりません。 入力 iterable の要素は、 func が引数として取れる型を持ちます。 (例えば、デフォルトの演算の加算では、要素は
Decimal
やFraction
を含む、加算ができる型を持ちます。)通常、出力される要素数は入力 iterable の要素数と一致します。 ただし、キーワード引数 initial が指定されていれば、 iterable の要素は最初に initial 値を持つため、出力は入力されたイテラブルよりも要素を1つ多く持ちます。
およそ次と等価です:
def accumulate(iterable, func=operator.add, *, initial=None): 'Return running totals' # accumulate([1,2,3,4,5]) --> 1 3 6 10 15 # accumulate([1,2,3,4,5], initial=100) --> 100 101 103 106 110 115 # accumulate([1,2,3,4,5], operator.mul) --> 1 2 6 24 120 it = iter(iterable) total = initial if initial is None: try: total = next(it) except StopIteration: return yield total for element in it: total = func(total, element) yield total
func 引数の利用法はたくさんあります。最小値にするために
min()
を、最大値にするためにmax()
を、積にするためにoperator.mul()
を使うことができます。金利を累積し支払いを適用して償還表を作成することもできます。初期値をイテラブルに与えて func 引数で累積和を利用するだけで一階の 漸化式 をモデル化できます:>>> data = [3, 4, 6, 2, 1, 9, 0, 7, 5, 8] >>> list(accumulate(data, operator.mul)) # running product [3, 12, 72, 144, 144, 1296, 0, 0, 0, 0] >>> list(accumulate(data, max)) # running maximum [3, 4, 6, 6, 6, 9, 9, 9, 9, 9] # Amortize a 5% loan of 1000 with 4 annual payments of 90 >>> cashflows = [1000, -90, -90, -90, -90] >>> list(accumulate(cashflows, lambda bal, pmt: bal*1.05 + pmt)) [1000, 960.0, 918.0, 873.9000000000001, 827.5950000000001] # Chaotic recurrence relation https://en.wikipedia.org/wiki/Logistic_map >>> logistic_map = lambda x, _: r * x * (1 - x) >>> r = 3.8 >>> x0 = 0.4 >>> inputs = repeat(x0, 36) # only the initial value is used >>> [format(x, '.2f') for x in accumulate(inputs, logistic_map)] ['0.40', '0.91', '0.30', '0.81', '0.60', '0.92', '0.29', '0.79', '0.63', '0.88', '0.39', '0.90', '0.33', '0.84', '0.52', '0.95', '0.18', '0.57', '0.93', '0.25', '0.71', '0.79', '0.63', '0.88', '0.39', '0.91', '0.32', '0.83', '0.54', '0.95', '0.20', '0.60', '0.91', '0.30', '0.80', '0.60']
最終的な累積値だけを返す類似の関数については
functools.reduce()
を見てください。バージョン 3.2 で追加.
バージョン 3.3 で変更: オプションの func 引数が追加されました。
バージョン 3.8 で変更: オプションの initial パラメータが追加されました。
-
itertools.
chain
(*iterables)¶ 先頭の iterable の全要素を返し、次に2番目の iterable の全要素を返し、と全 iterable の要素を返すイテレータを作成します。連続したシーケンスを一つのシーケンスとして扱う場合に使用します。およそ次と等価です:
def chain(*iterables): # chain('ABC', 'DEF') --> A B C D E F for it in iterables: for element in it: yield element
-
classmethod
chain.
from_iterable
(iterable)¶ chain()
のためのもう一つのコンストラクタです。遅延評価される iterable 引数一つから連鎖した入力を受け取ります。この関数は、以下のコードとほぼ等価です:def from_iterable(iterables): # chain.from_iterable(['ABC', 'DEF']) --> A B C D E F for it in iterables: for element in it: yield element
-
itertools.
combinations
(iterable, r)¶ 入力 iterable の要素からなる長さ r の部分列を返します。
組合せ(combination)は入力 iterable に応じた辞書式順序で出力されます。したがって、入力 iterable がソートされていれば、出力される組合わせ(combination)タプルもソートされた順番で生成されます。
各要素は場所に基づいて一意に取り扱われ、値にはよりません。入力された要素がバラバラなら各組合せの中に重複した値は現れません。
およそ次と等価です:
def combinations(iterable, r): # combinations('ABCD', 2) --> AB AC AD BC BD CD # combinations(range(4), 3) --> 012 013 023 123 pool = tuple(iterable) n = len(pool) if r > n: return indices = list(range(r)) yield tuple(pool[i] for i in indices) while True: for i in reversed(range(r)): if indices[i] != i + n - r: break else: return indices[i] += 1 for j in range(i+1, r): indices[j] = indices[j-1] + 1 yield tuple(pool[i] for i in indices)
combinations()
のコードはpermutations()
のシーケンスから (入力プールでの位置に応じた順序で) 要素がソートされていないものをフィルターしたようにも表現できます:def combinations(iterable, r): pool = tuple(iterable) n = len(pool) for indices in permutations(range(n), r): if sorted(indices) == list(indices): yield tuple(pool[i] for i in indices)
返される要素の数は、
0 <= r <= n
の場合は、n! / r! / (n-r)!
で、r > n
の場合は 0 です。
-
itertools.
combinations_with_replacement
(iterable, r)¶ 入力 iterable から、それぞれの要素が複数回現れることを許して、長さ r の要素の部分列を返します。
組合せ(combination)は入力 iterable に応じた辞書式順序で出力されます。したがって、入力 iterable がソートされていれば、出力される組合わせ(combination)タプルもソートされた順番で生成されます。
要素は、値ではなく位置に基づいて一意に扱われます。ですから、入力の要素が一意であれば、生成された組合せも一意になります。
およそ次と等価です:
def combinations_with_replacement(iterable, r): # combinations_with_replacement('ABC', 2) --> AA AB AC BB BC CC pool = tuple(iterable) n = len(pool) if not n and r: return indices = [0] * r yield tuple(pool[i] for i in indices) while True: for i in reversed(range(r)): if indices[i] != n - 1: break else: return indices[i:] = [indices[i] + 1] * (r - i) yield tuple(pool[i] for i in indices)
combinations_with_replacement()
のコードは、product()
の部分列から、要素が (入力プールの位置に従って) ソートされた順になっていない項目をフィルタリングしたものとしても表せます:def combinations_with_replacement(iterable, r): pool = tuple(iterable) n = len(pool) for indices in product(range(n), repeat=r): if sorted(indices) == list(indices): yield tuple(pool[i] for i in indices)
返される要素の数は、
n > 0
のとき(n+r-1)! / r! / (n-1)!
です。バージョン 3.1 で追加.
-
itertools.
compress
(data, selectors)¶ data の要素から selectors の対応する要素が
True
と評価されるものだけをフィルタしたイテレータを作ります。data と selectors のいずれかが尽きたときに止まります。およそ次と等価です:def compress(data, selectors): # compress('ABCDEF', [1,0,1,0,1,1]) --> A C E F return (d for d, s in zip(data, selectors) if s)
バージョン 3.1 で追加.
-
itertools.
count
(start=0, step=1)¶ 数 start で始まる等間隔の値を返すイテレータを作成します。
map()
に渡して連続したデータを生成するのによく使われます。 また、zip()
に連続した番号を追加するのにも使われます。 およそ次と等価です:def count(start=0, step=1): # count(10) --> 10 11 12 13 14 ... # count(2.5, 0.5) -> 2.5 3.0 3.5 ... n = start while True: yield n n += step
浮動小数点数でカウントするときは
(start + step * i for i in count())
のように掛け算を使ったコードに置き換えたほうが正確にできることがあります。バージョン 3.1 で変更: step 引数が追加され、非整数の引数が許されるようになりました。
-
itertools.
cycle
(iterable)¶ iterable から要素を取得し、そのコピーを保存するイテレータを作成します。iterable の全要素を返すと、セーブされたコピーから要素を返します。これを無限に繰り返します。およそ次と等価です:
def cycle(iterable): # cycle('ABCD') --> A B C D A B C D A B C D ... saved = [] for element in iterable: yield element saved.append(element) while saved: for element in saved: yield element
cycle()
は大きなメモリ領域を使用します。使用するメモリ量は iterable の大きさに依存します。
-
itertools.
dropwhile
(predicate, iterable)¶ predicate (述語) が真である間は要素を飛ばし、その後は全ての要素を返すイテレータを作成します。このイテレータは、predicate が最初に偽になるまで 全く 要素を返さないため、要素を返し始めるまでに長い時間がかかる場合があります。およそ次と等価です:
def dropwhile(predicate, iterable): # dropwhile(lambda x: x<5, [1,4,6,4,1]) --> 6 4 1 iterable = iter(iterable) for x in iterable: if not predicate(x): yield x break for x in iterable: yield x
-
itertools.
filterfalse
(predicate, iterable)¶ iterable から predicate が
False
となる要素だけを返すイテレータを作成します。predicate がNone
の場合、偽の要素だけを返します。およそ次と等価です:def filterfalse(predicate, iterable): # filterfalse(lambda x: x%2, range(10)) --> 0 2 4 6 8 if predicate is None: predicate = bool for x in iterable: if not predicate(x): yield x
-
itertools.
groupby
(iterable, key=None)¶ 同じキーをもつような要素からなる iterable 中のグループに対して、キーとグループを返すようなイテレータを作成します。key は各要素に対するキー値を計算する関数です。キーを指定しない場合や
None
にした場合、key 関数のデフォルトは恒等関数になり要素をそのまま返します。通常、iterable は同じキー関数でソート済みである必要があります。groupby()
の操作は Unix のuniq
フィルターと似ています。 key 関数の値が変わるたびに休止または新しいグループを生成します (このために通常同じ key 関数でソートしておく必要があるのです)。この動作は SQL の入力順に関係なく共通の要素を集約する GROUP BY とは違います。返されるグループはそれ自体がイテレータで、
groupby()
と iterable を共有しています。もととなる iterable を共有しているため、groupby()
オブジェクトの要素取り出しを先に進めると、それ以前の要素であるグループは見えなくなってしまいます。従って、データが後で必要な場合にはリストの形で保存しておく必要があります:groups = [] uniquekeys = [] data = sorted(data, key=keyfunc) for k, g in groupby(data, keyfunc): groups.append(list(g)) # Store group iterator as a list uniquekeys.append(k)
groupby()
はおよそ次と等価です:class groupby: # [k for k, g in groupby('AAAABBBCCDAABBB')] --> A B C D A B # [list(g) for k, g in groupby('AAAABBBCCD')] --> AAAA BBB CC D def __init__(self, iterable, key=None): if key is None: key = lambda x: x self.keyfunc = key self.it = iter(iterable) self.tgtkey = self.currkey = self.currvalue = object() def __iter__(self): return self def __next__(self): self.id = object() while self.currkey == self.tgtkey: self.currvalue = next(self.it) # Exit on StopIteration self.currkey = self.keyfunc(self.currvalue) self.tgtkey = self.currkey return (self.currkey, self._grouper(self.tgtkey, self.id)) def _grouper(self, tgtkey, id): while self.id is id and self.currkey == tgtkey: yield self.currvalue try: self.currvalue = next(self.it) except StopIteration: return self.currkey = self.keyfunc(self.currvalue)
-
itertools.
islice
(iterable, stop)¶ -
itertools.
islice
(iterable, start, stop[, step]) iterable から要素を選択して返すイテレータを作成します。 start が0でない場合、iterable の要素は start に達するまでスキップされます。その後、 要素が順に返されます。
def islice(iterable, *args): # islice('ABCDEFG', 2) --> A B # islice('ABCDEFG', 2, 4) --> C D # islice('ABCDEFG', 2, None) --> C D E F G # islice('ABCDEFG', 0, None, 2) --> A C E G s = slice(*args) start, stop, step = s.start or 0, s.stop or sys.maxsize, s.step or 1 it = iter(range(start, stop, step)) try: nexti = next(it) except StopIteration: # Consume *iterable* up to the *start* position. for i, element in zip(range(start), iterable): pass return try: for i, element in enumerate(iterable): if i == nexti: yield element nexti = next(it) except StopIteration: # Consume to *stop*. for i, element in zip(range(i + 1, stop), iterable): pass
start が
None
の場合、イテレーションは0から始まります。step がNone
の場合、ステップはデフォルトの1になります。
-
itertools.
pairwise
(iterable)¶ Return successive overlapping pairs taken from the input iterable.
The number of 2-tuples in the output iterator will be one fewer than the number of inputs. It will be empty if the input iterable has fewer than two values.
およそ次と等価です:
def pairwise(iterable): # pairwise('ABCDEFG') --> AB BC CD DE EF FG a, b = tee(iterable) next(b, None) return zip(a, b)
バージョン 3.10 で追加.
-
itertools.
permutations
(iterable, r=None)¶ iterable の要素からなる長さ r の順列 (permutation) を連続的に返します。
r が指定されない場合や
None
の場合、r はデフォルトで iterable の長さとなり、可能な最長の順列の全てが生成されます。順列(permutation)は入力 iterable に応じた辞書式順序で出力されます。したがって、入力 iterable がソートされていれば、出力される組合わせタプルもソートされた順番で生成されます。
要素は値ではなく位置に基づいて一意的に扱われます。したがって入力された要素が全て異なっている場合、それぞれの順列に重複した要素が現れないことになります。
およそ次と等価です:
def permutations(iterable, r=None): # permutations('ABCD', 2) --> AB AC AD BA BC BD CA CB CD DA DB DC # permutations(range(3)) --> 012 021 102 120 201 210 pool = tuple(iterable) n = len(pool) r = n if r is None else r if r > n: return indices = list(range(n)) cycles = list(range(n, n-r, -1)) yield tuple(pool[i] for i in indices[:r]) while n: for i in reversed(range(r)): cycles[i] -= 1 if cycles[i] == 0: indices[i:] = indices[i+1:] + indices[i:i+1] cycles[i] = n - i else: j = cycles[i] indices[i], indices[-j] = indices[-j], indices[i] yield tuple(pool[i] for i in indices[:r]) break else: return
permutations()
のコードはproduct()
の列から重複 (それらは入力プールの同じ位置から取られたものです) を除くようフィルタしたものとしても表現できます:def permutations(iterable, r=None): pool = tuple(iterable) n = len(pool) r = n if r is None else r for indices in product(range(n), repeat=r): if len(set(indices)) == r: yield tuple(pool[i] for i in indices)
返される要素の数は、
0 <= r <= n
の場合n! / (n-r)!
で、r > n
の場合は 0 です。
-
itertools.
product
(*iterables, repeat=1)¶ 入力イテラブルのデカルト積です。
ジェネレータ式の入れ子になった for ループとおよそ等価です。たとえば
product(A, B)
は((x,y) for x in A for y in B)
と同じものを返します。入れ子ループは走行距離計と同じように右端の要素がイテレーションごとに更新されていきます。このパターンは辞書式順序を作り出し、入力のイテレート可能オブジェクトたちがソートされていれば、直積タプルもソートされた順に出てきます。
イテラブル自身との直積を計算するためには、オプションの repeat キーワード引数に繰り返し回数を指定します。たとえば
product(A, repeat=4)
はproduct(A, A, A, A)
と同じ意味です。この関数は以下のコードとおよそ等価ですが、実際の実装ではメモリ中に中間結果を作りません:
def product(*args, repeat=1): # product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy # product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111 pools = [tuple(pool) for pool in args] * repeat result = [[]] for pool in pools: result = [x+[y] for x in result for y in pool] for prod in result: yield tuple(prod)
product()
は動作する前に、入力のイテラブルを完全に読み取り、直積を生成するためにメモリ内に値を蓄えます。したがって、入力が有限の場合に限り有用です。
-
itertools.
repeat
(object[, times])¶ 繰り返し object を返すイテレータを作成します。 times 引数を指定しなければ、無限に値を返し続けます。
map()
の引数にして、呼び出された関数に同じ引数を渡すのに使います。またzip()
と使って、タプルの変わらない部分を作ります。およそ次と等価です:
def repeat(object, times=None): # repeat(10, 3) --> 10 10 10 if times is None: while True: yield object else: for i in range(times): yield object
repeat は map や zip に定数のストリームを与えるためによく利用されます:
>>> list(map(pow, range(10), repeat(2))) [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
-
itertools.
starmap
(function, iterable)¶ iterable の要素を引数として funtion を計算するイテレータを作成します。 function の引数が一つの iterable からタプルに既にグループ化されている (データが "zip済み") 場合、
map()
の代わりに使用します。map()
とstarmap()
の違いはfunction(a,b)
とfunction(*c)
の差に似ています。およそ次と等価です:def starmap(function, iterable): # starmap(pow, [(2,5), (3,2), (10,3)]) --> 32 9 1000 for args in iterable: yield function(*args)
-
itertools.
takewhile
(predicate, iterable)¶ predicate が真である限り iterable から要素を返すイテレータを作成します。およそ次と等価です:
def takewhile(predicate, iterable): # takewhile(lambda x: x<5, [1,4,6,4,1]) --> 1 4 for x in iterable: if predicate(x): yield x else: break
-
itertools.
tee
(iterable, n=2)¶ 一つの iterable から n 個の独立したイテレータを返します。
以下の Python コードは tee がすることについての理解を助けるでしょう (ただし実際の実装はより複雑で、下層の FIFO キューを一つしか使いませんが)。
およそ次と等価です:
def tee(iterable, n=2): it = iter(iterable) deques = [collections.deque() for i in range(n)] def gen(mydeque): while True: if not mydeque: # when the local deque is empty try: newval = next(it) # fetch a new value and except StopIteration: return for d in deques: # load it to all the deques d.append(newval) yield mydeque.popleft() return tuple(gen(d) for d in deques)
一度
tee()
でイテレータを分割すると、もとの iterable を他で使ってはいけません。さもなければ、tee()
オブジェクトの知らない間に iterable が先の要素に進んでしまうことになります。tee()
イテレーターはスレッドセーフではありません。同じtee()
から生じた複数のイテレータを同時に使用すると、元のイテラブルがスレッドセーフであっても、RuntimeError
が発生することがあります。tee()
はかなり大きなメモリ領域を使用するかもしれません (使用するメモリ量はiterableの大きさに依存します)。一般には、一つのイテレータが他のイテレータよりも先にほとんどまたは全ての要素を消費するような場合には、tee()
よりもlist()
を使った方が高速です。
-
itertools.
zip_longest
(*iterables, fillvalue=None)¶ 各 iterable の要素をまとめるイテレータを作成します。iterable の長さが違う場合、足りない値は fillvalue で埋められます。最も長い itarable が尽きるまでイテレーションします。およそ次と等価です:
def zip_longest(*args, fillvalue=None): # zip_longest('ABCD', 'xy', fillvalue='-') --> Ax By C- D- iterators = [iter(it) for it in args] num_active = len(iterators) if not num_active: return while True: values = [] for i, it in enumerate(iterators): try: value = next(it) except StopIteration: num_active -= 1 if not num_active: return iterators[i] = repeat(fillvalue) value = fillvalue values.append(value) yield tuple(values)
iterables の1つが無限になりうる場合
zip_longest()
は呼び出し回数を制限するような何かでラップしなければいけません(例えばislice()
ortakewhile()
)。 fillvalue は指定しない場合のデフォルトはNone
です。
Itertools レシピ¶
この節では、既存の itertools を素材としてツールセットを拡張するためのレシピを示します。
下記のレシピを包含した広範なレシピ集は Python Package Index 上の more-itertools project からインストールできます。
pip install more-itertools
iterable 全体を一度にメモリ上に置くよりも、要素を一つづつ処理する方がメモリ効率上の有利さを保てます。関数形式のままツールをリンクしてゆくと、コードのサイズを減らし、一時変数を減らす助けになります。インタプリタのオーバヘッドをもたらす for ループや ジェネレータ を使わずに、 "ベクトル化された" ビルディングブロックを使うと、高速な処理を実現できます。
def take(n, iterable):
"Return first n items of the iterable as a list"
return list(islice(iterable, n))
def prepend(value, iterator):
"Prepend a single value in front of an iterator"
# prepend(1, [2, 3, 4]) -> 1 2 3 4
return chain([value], iterator)
def tabulate(function, start=0):
"Return function(0), function(1), ..."
return map(function, count(start))
def tail(n, iterable):
"Return an iterator over the last n items"
# tail(3, 'ABCDEFG') --> E F G
return iter(collections.deque(iterable, maxlen=n))
def consume(iterator, n=None):
"Advance the iterator n-steps ahead. If n is None, consume entirely."
# Use functions that consume iterators at C speed.
if n is None:
# feed the entire iterator into a zero-length deque
collections.deque(iterator, maxlen=0)
else:
# advance to the empty slice starting at position n
next(islice(iterator, n, n), None)
def nth(iterable, n, default=None):
"Returns the nth item or a default value"
return next(islice(iterable, n, None), default)
def all_equal(iterable):
"Returns True if all the elements are equal to each other"
g = groupby(iterable)
return next(g, True) and not next(g, False)
def quantify(iterable, pred=bool):
"Count how many times the predicate is true"
return sum(map(pred, iterable))
def pad_none(iterable):
"""Returns the sequence elements and then returns None indefinitely.
Useful for emulating the behavior of the built-in map() function.
"""
return chain(iterable, repeat(None))
def ncycles(iterable, n):
"Returns the sequence elements n times"
return chain.from_iterable(repeat(tuple(iterable), n))
def dotproduct(vec1, vec2):
return sum(map(operator.mul, vec1, vec2))
def convolve(signal, kernel):
# See: https://betterexplained.com/articles/intuitive-convolution/
# convolve(data, [0.25, 0.25, 0.25, 0.25]) --> Moving average (blur)
# convolve(data, [1, -1]) --> 1st finite difference (1st derivative)
# convolve(data, [1, -2, 1]) --> 2nd finite difference (2nd derivative)
kernel = tuple(kernel)[::-1]
n = len(kernel)
window = collections.deque([0], maxlen=n) * n
for x in chain(signal, repeat(0, n-1)):
window.append(x)
yield sum(map(operator.mul, kernel, window))
def flatten(list_of_lists):
"Flatten one level of nesting"
return chain.from_iterable(list_of_lists)
def repeatfunc(func, times=None, *args):
"""Repeat calls to func with specified arguments.
Example: repeatfunc(random.random)
"""
if times is None:
return starmap(func, repeat(args))
return starmap(func, repeat(args, times))
def grouper(iterable, n, *, incomplete='fill', fillvalue=None):
"Collect data into non-overlapping fixed-length chunks or blocks"
# grouper('ABCDEFG', 3, fillvalue='x') --> ABC DEF Gxx
# grouper('ABCDEFG', 3, incomplete='strict') --> ABC DEF ValueError
# grouper('ABCDEFG', 3, incomplete='ignore') --> ABC DEF
args = [iter(iterable)] * n
if incomplete == 'fill':
return zip_longest(*args, fillvalue=fillvalue)
if incomplete == 'strict':
return zip(*args, strict=True)
if incomplete == 'ignore':
return zip(*args)
else:
raise ValueError('Expected fill, strict, or ignore')
def triplewise(iterable):
"Return overlapping triplets from an iterable"
# triplewise('ABCDEFG') -> ABC BCD CDE DEF EFG
for (a, _), (b, c) in pairwise(pairwise(iterable)):
yield a, b, c
def sliding_window(iterable, n):
# sliding_window('ABCDEFG', 4) -> ABCD BCDE CDEF DEFG
it = iter(iterable)
window = collections.deque(islice(it, n), maxlen=n)
if len(window) == n:
yield tuple(window)
for x in it:
window.append(x)
yield tuple(window)
def roundrobin(*iterables):
"roundrobin('ABC', 'D', 'EF') --> A D E B F C"
# Recipe credited to George Sakkis
num_active = len(iterables)
nexts = cycle(iter(it).__next__ for it in iterables)
while num_active:
try:
for next in nexts:
yield next()
except StopIteration:
# Remove the iterator we just exhausted from the cycle.
num_active -= 1
nexts = cycle(islice(nexts, num_active))
def partition(pred, iterable):
"Use a predicate to partition entries into false entries and true entries"
# partition(is_odd, range(10)) --> 0 2 4 6 8 and 1 3 5 7 9
t1, t2 = tee(iterable)
return filterfalse(pred, t1), filter(pred, t2)
def before_and_after(predicate, it):
""" Variant of takewhile() that allows complete
access to the remainder of the iterator.
>>> it = iter('ABCdEfGhI')
>>> all_upper, remainder = before_and_after(str.isupper, it)
>>> ''.join(all_upper)
'ABC'
>>> ''.join(remainder) # takewhile() would lose the 'd'
'dEfGhI'
Note that the first iterator must be fully
consumed before the second iterator can
generate valid results.
"""
it = iter(it)
transition = []
def true_iterator():
for elem in it:
if predicate(elem):
yield elem
else:
transition.append(elem)
return
def remainder_iterator():
yield from transition
yield from it
return true_iterator(), remainder_iterator()
def subslices(seq):
"Return all contiguous non-empty subslices of a sequence"
# subslices('ABCD') --> A AB ABC ABCD B BC BCD C CD D
slices = starmap(slice, combinations(range(len(seq) + 1), 2))
return map(operator.getitem, repeat(seq), slices)
def powerset(iterable):
"powerset([1,2,3]) --> () (1,) (2,) (3,) (1,2) (1,3) (2,3) (1,2,3)"
s = list(iterable)
return chain.from_iterable(combinations(s, r) for r in range(len(s)+1))
def unique_everseen(iterable, key=None):
"List unique elements, preserving order. Remember all elements ever seen."
# unique_everseen('AAAABBBCCDAABBB') --> A B C D
# unique_everseen('ABBCcAD', str.lower) --> A B C D
seen = set()
seen_add = seen.add
if key is None:
for element in filterfalse(seen.__contains__, iterable):
seen_add(element)
yield element
else:
for element in iterable:
k = key(element)
if k not in seen:
seen_add(k)
yield element
def unique_justseen(iterable, key=None):
"List unique elements, preserving order. Remember only the element just seen."
# unique_justseen('AAAABBBCCDAABBB') --> A B C D A B
# unique_justseen('ABBCcAD', str.lower) --> A B C A D
return map(next, map(operator.itemgetter(1), groupby(iterable, key)))
def iter_except(func, exception, first=None):
""" Call a function repeatedly until an exception is raised.
Converts a call-until-exception interface to an iterator interface.
Like builtins.iter(func, sentinel) but uses an exception instead
of a sentinel to end the loop.
Examples:
iter_except(functools.partial(heappop, h), IndexError) # priority queue iterator
iter_except(d.popitem, KeyError) # non-blocking dict iterator
iter_except(d.popleft, IndexError) # non-blocking deque iterator
iter_except(q.get_nowait, Queue.Empty) # loop over a producer Queue
iter_except(s.pop, KeyError) # non-blocking set iterator
"""
try:
if first is not None:
yield first() # For database APIs needing an initial cast to db.first()
while True:
yield func()
except exception:
pass
def first_true(iterable, default=False, pred=None):
"""Returns the first true value in the iterable.
If no true value is found, returns *default*
If *pred* is not None, returns the first item
for which pred(item) is true.
"""
# first_true([a,b,c], x) --> a or b or c or x
# first_true([a,b], x, f) --> a if f(a) else b if f(b) else x
return next(filter(pred, iterable), default)
def random_product(*args, repeat=1):
"Random selection from itertools.product(*args, **kwds)"
pools = [tuple(pool) for pool in args] * repeat
return tuple(map(random.choice, pools))
def random_permutation(iterable, r=None):
"Random selection from itertools.permutations(iterable, r)"
pool = tuple(iterable)
r = len(pool) if r is None else r
return tuple(random.sample(pool, r))
def random_combination(iterable, r):
"Random selection from itertools.combinations(iterable, r)"
pool = tuple(iterable)
n = len(pool)
indices = sorted(random.sample(range(n), r))
return tuple(pool[i] for i in indices)
def random_combination_with_replacement(iterable, r):
"Random selection from itertools.combinations_with_replacement(iterable, r)"
pool = tuple(iterable)
n = len(pool)
indices = sorted(random.choices(range(n), k=r))
return tuple(pool[i] for i in indices)
def nth_combination(iterable, r, index):
"Equivalent to list(combinations(iterable, r))[index]"
pool = tuple(iterable)
n = len(pool)
if r < 0 or r > n:
raise ValueError
c = 1
k = min(r, n-r)
for i in range(1, k+1):
c = c * (n - k + i) // i
if index < 0:
index += c
if index < 0 or index >= c:
raise IndexError
result = []
while r:
c, n, r = c*r//n, n-1, r-1
while index >= c:
index -= c
c, n = c*(n-r)//n, n-1
result.append(pool[-1-n])
return tuple(result)