2.38. 優先度付きキュー

優先度付きキューは、全体をソートすることなく、要素のうち最小（または最大）のものを常に把握しているコンテナです。heapq は、ストレージとして素の list を使用し、最小の要素をインデックス 0 に保つために 最小ヒープ（min-heap） と呼ばれる木構造ベースのアルゴリズムを用いて、MicroPython にこれを実装しています。

2.38.1. 考え方のモデル

最小ヒープとは、すべてのインデックスにある要素がインデックス 2*i + 1 と 2*i + 2 にある要素以下になるように並べられたリストです。この配置により、3 つの操作が安価になります。

  • 最小の要素を見つける -- それは heap[0] にあります。単一のインデックス参照で済み、走査は不要です。

  • 要素を追加する -- 要素は木構造を上に向かって浮上し、各レベルで 1 つの親と比較されます。1000 個の要素なら約 10 回の比較、100 万個でも約 20 回です。

  • 最小の要素を削除する -- 同じ経路を逆向きにたどり、同じわずかな回数の比較で済みます。

リストはソートされて いません。リストを反復処理しても、要素は任意の順序で得られます。最小であることが保証されているのは heap[0] だけです。

2.38.2. MicroPython における 3 つの関数

MicroPython の heapq は、ちょうど 3 つの関数を公開しています。

  • heapq.heappush() -- ヒープ不変条件を保ったまま要素を挿入します。

  • heapq.heappop() -- 最小の要素を削除して返します。

  • heapq.heapify() -- 既存のリストをその場でヒープに並べ替えます。

これがすべての機能です。CPython の heappushpop、heapreplace、nlargest、nsmallest、merge は存在 しません。以下のパターンは、よく使われるものを手作業で構築する方法を示しています。

ゼロから始める:

>>> import heapq
>>> heap = []
>>> heapq.heappush(heap, 5)
>>> heapq.heappush(heap, 1)
>>> heapq.heappush(heap, 3)
>>> heap
[1, 5, 3]
>>> heapq.heappop(heap)
1
>>> heapq.heappop(heap)
3
>>> heapq.heappop(heap)
5

既存のデータから始める:

>>> samples = [5, 1, 3, 2, 4]
>>> heapq.heapify(samples)
>>> samples
[1, 2, 3, 5, 4]
>>> heapq.heappop(samples)
1

ヒープ化（heapify）は各要素を 1 回ずつ処理し、同じ要素を 1 つずつプッシュするよりも明らかに高速です。

2.38.3. ストリームに対する上位 N 件

優先度付きキューの自然な使い方として、一度しか読めないストリームの中で N 個の最大要素を追跡する例があります。

コツは、サイズ N の最小ヒープを保つことです。新しい値はすべてプッシュされ、ヒープが N を超えて大きくなったら最小のものを取り除きます。最小値は常に heap[0] にあるので、入ってくる各要素を保持するか破棄するかは 1 回の比較で決まります:

def top_n_brightest(readings, n):
    heap = []
    for r in readings:
        if len(heap) < n:
            heapq.heappush(heap, r)
        elif r > heap[0]:
            heapq.heappop(heap)
            heapq.heappush(heap, r)
    return sorted(heap, reverse=True)

ヒープは N を超えて大きくなることがないため、ストリームがどれだけ長くても、入ってくる各値のコストは同じわずかな回数の比較で済みます。最後の sorted だけが、そのコストが N に直接依存する唯一のステップです。

2.38.4. スケジュールされたイベント

(deadline, payload) タプルの優先度付きキューは、イベントスケジューラの標準的な形です。タプルは辞書順で比較されるため、最小の締め切りを持つタプルが自動的に先頭に浮上します:

import heapq
import time

queue = []
heapq.heappush(queue, (time.time() + 5, 'check sensors'))
heapq.heappush(queue, (time.time() + 1, 'blink LED'))
heapq.heappush(queue, (time.time() + 3, 'send heartbeat'))

while queue:
    deadline, task = queue[0]
    now = time.time()
    if now < deadline:
        time.sleep(deadline - now)
    heapq.heappop(queue)
    run(task)

2 つのイベントが同じ締め切りを共有する場合、タプルの比較は 2 番目の要素へと進みます。本来比較できないペイロードについては、安定した順序を強制するために、タプルの 2 番目の要素としてカウンタを追加します:

counter = 0
def schedule(deadline, task):
    global counter
    counter += 1
    heapq.heappush(queue, (deadline, counter, task))

これで、締め切りの早いもの優先という動作はそのまま保たれ、同点の場合はカウンタで決着がつくため先にスケジュールされたタスクが先に実行され、task は比較可能である必要がなくなります。

2.38.5. heapq が提供しないもの

うまくいきそうに見えて、実際には機能しないパターンがいくつかあります。

  • for x in heap で反復しても、ソートされた出力は得られ ません -- リストの内部的な順序はヒープの木構造のレイアウトだからです。ヒープを順番に消費するには、空になるまで heappop してください。

  • すでにヒープ内にある要素の優先度を 変更する 方法はありません。標準的な回避策は、新しい優先度を新規エントリとしてプッシュし、古いものをキャンセル済みとしてマークする（それがポップされたときにスキップする）ことです。

  • ヒープは要素のコピーではなく参照を格納します。プッシュした後に要素を変更すると、ヒープ不変条件が壊れる可能性があります。比較キーには不変な値（数値、タプル）を使用してください。

素のリスト上の最小ヒープは、小型の組み込みデバイスにおける「最小は何か」や「最も緊急なものは何か」という問いに対する適切なツールです。3 つの関数と 1 つのリスト、あとはヒープが残りを引き受けてくれます。