heapq --- Heap queue algorithm

Code source : Lib/heapq.py

Ce module expose une implémentation de l'algorithme de file de priorité, basée sur un tas.

Min-heaps are binary trees for which every parent node has a value less than or equal to any of its children. We refer to this condition as the heap invariant.

For min-heaps, this implementation uses lists for which heap[k] <= heap[2*k+1] and heap[k] <= heap[2*k+2] for all k for which the compared elements exist. Elements are counted from zero. The interesting property of a min-heap is that its smallest element is always the root, heap[0].

Max-heaps satisfy the reverse invariant: every parent node has a value greater than any of its children. These are implemented as lists for which maxheap[2*k+1] <= maxheap[k] and maxheap[2*k+2] <= maxheap[k] for all k for which the compared elements exist. The root, maxheap[0], contains the largest element; heap.sort(reverse=True) maintains the max-heap invariant.

The heapq API differs from textbook heap algorithms in two aspects: (a) We use zero-based indexing. This makes the relationship between the index for a node and the indexes for its children slightly less obvious, but is more suitable since Python uses zero-based indexing. (b) Textbooks often focus on max-heaps, due to their suitability for in-place sorting. Our implementation favors min-heaps as they better correspond to Python lists.

These two aspects make it possible to view the heap as a regular Python list without surprises: heap[0] is the smallest item, and heap.sort() maintains the heap invariant!

Like list.sort(), this implementation uses only the < operator for comparisons, for both min-heaps and max-heaps.

In the API below, and in this documentation, the unqualified term heap generally refers to a min-heap. The API for max-heaps is named using a _max suffix.

To create a heap, use a list initialized as [], or transform an existing list into a min-heap or max-heap using the heapify() or heapify_max() functions, respectively.

The following functions are provided for min-heaps:

heapq.heappush(heap, item)

Push the value item onto the heap, maintaining the min-heap invariant.

heapq.heappop(heap)

Pop and return the smallest item from the heap, maintaining the min-heap invariant. If the heap is empty, IndexError is raised. To access the smallest item without popping it, use heap[0].

heapq.heappushpop(heap, item)

Introduit l'élément item dans le tas, puis extraie le plus petit élément de heap. Cette action combinée est plus efficace que heappush() suivie par un appel séparé à heappop().

heapq.heapify(x)

Transform list x into a min-heap, in-place, in linear time.

heapq.heapreplace(heap, item)

Extraie le plus petit élément de heap et introduit le nouvel élément item. La taille du tas ne change pas. Si le tas est vide, une exception IndexError est levée.

Cette opération en une étape est plus efficace qu'un appel à heappop() suivi d'un appel à heappush() et est plus appropriée lorsque le tas est de taille fixe. La combinaison pop/push renvoie toujours un élément du tas et le remplace par item.

La valeur renvoyée peut être plus grande que l'élément item ajouté. Si cela n'est pas souhaitable, utilisez plutôt heappushpop() à la place. Sa combinaison push/pop renvoie le plus petit élément des deux valeurs et laisse la plus grande sur le tas.

For max-heaps, the following functions are provided:

heapq.heapify_max(x)

Transform list x into a max-heap, in-place, in linear time.

Ajouté dans la version 3.14.

heapq.heappush_max(heap, item)

Push the value item onto the max-heap heap, maintaining the max-heap invariant.

Ajouté dans la version 3.14.

heapq.heappop_max(heap)

Pop and return the largest item from the max-heap heap, maintaining the max-heap invariant. If the max-heap is empty, IndexError is raised. To access the largest item without popping it, use maxheap[0].

Ajouté dans la version 3.14.

heapq.heappushpop_max(heap, item)

Push item on the max-heap heap, then pop and return the largest item from heap. The combined action runs more efficiently than heappush_max() followed by a separate call to heappop_max().

Ajouté dans la version 3.14.

heapq.heapreplace_max(heap, item)

Pop and return the largest item from the max-heap heap and also push the new item. The max-heap size doesn't change. If the max-heap is empty, IndexError is raised.

The value returned may be smaller than the item added. Refer to the analogous function heapreplace() for detailed usage notes.

Ajouté dans la version 3.14.

Ce module contient également trois fonctions génériques utilisant les tas.

heapq.merge(*iterables, key=None, reverse=False)

Fusionne plusieurs entrées ordonnées en une unique sortie ordonnée (par exemple, fusionne des entrées datées provenant de multiples journaux applicatifs). Renvoie un iterator sur les valeurs ordonnées.

Similaire à sorted(itertools.chain(*iterables)) mais renvoie un itérable, ne stocke pas toutes les données en mémoire en une fois et suppose que chaque flux d'entrée est déjà classé (en ordre croissant).

A deux arguments optionnels qui doivent être fournis par mot clef.

key spécifie une key function d'un argument utilisée pour extraire une clef de comparaison de chaque élément de la liste. La valeur par défaut est None (compare les éléments directement).

reverse est une valeur booléenne. Si elle est True, la liste d'éléments est fusionnée comme si toutes les comparaisons étaient inversées. Pour obtenir un comportement similaire à sorted(itertools.chain(*iterables), reverse=True), tous les itérables doivent être classés par ordre décroissant.

Modifié dans la version 3.5: Ajout des paramètres optionnels key et reverse.

heapq.nlargest(n, iterable, key=None)

Renvoie une liste contenant les n plus grands éléments du jeu de données défini par iterable. Si l'option key est fournie, celle-ci spécifie une fonction à un argument qui est utilisée pour extraire la clé de comparaison de chaque élément dans iterable (par exemple, key=str.lower). Équivalent à : sorted(iterable, key=key, reverse=True)[:n].

heapq.nsmallest(n, iterable, key=None)

Renvoie une liste contenant les n plus petits éléments du jeu de données défini par iterable. Si l'option key est fournie, celle-ci spécifie une fonction à un argument qui est utilisée pour extraire la clé de comparaison de chaque élément dans iterable (par exemple, key=str.lower). Équivalent à : sorted(iterable, key=key)[:n].

Les deux fonctions précédentes sont les plus efficaces pour des petites valeurs de n. Pour de grandes valeurs, il est préférable d'utiliser la fonction sorted(). En outre, lorsque n==1, il est plus efficace d'utiliser les fonctions natives min() et max(). Si vous devez utiliser ces fonctions de façon répétée, il est préférable de transformer l'itérable en tas.

Exemples simples

Un tri par tas peut être implémenté en introduisant toutes les valeurs dans un tas puis en effectuant l'extraction des éléments un par un :

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(len(h))]
...
>>> heapsort([1, 3, 5, 7, 9, 2, 4, 6, 8, 0])
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

Ceci est similaire à sorted(iterable) mais, contrairement à sorted(), cette implémentation n'est pas stable.

Les éléments d'un tas peuvent être des n-uplets. C'est pratique pour assigner des valeurs de comparaison (par exemple, des priorités de tâches) en plus de l'élément qui est suivi :

>>> h = []
>>> heappush(h, (5, 'write code'))
>>> heappush(h, (7, 'release product'))
>>> heappush(h, (1, 'write spec'))
>>> heappush(h, (3, 'create tests'))
>>> heappop(h)
(1, 'write spec')

Notes d'implémentation de la file de priorité

Une file de priorité est une application courante des tas et présente plusieurs défis d'implémentation :

  • Stabilité du classement : comment s'assurer que deux tâches avec la même priorité sont renvoyées dans l'ordre de leur ajout ?

  • La comparaison des couples (priorité, tâche) échoue si les priorités sont identiques et que les tâches n'ont pas de relation d'ordre par défaut.

  • Si la priorité d'une tâche change, comment la déplacer à sa nouvelle position dans le tas ?

  • Si une tâche en attente doit être supprimée, comment la trouver et la supprimer de la file ?

Une solution aux deux premiers problèmes consiste à stocker les entrées sous forme de liste à 3 éléments incluant la priorité, le numéro d'ajout et la tâche. Le numéro d'ajout sert à briser les égalités de telle sorte que deux tâches avec la même priorité sont renvoyées dans l'ordre de leur insertion. Puisque deux tâches ne peuvent jamais avoir le même numéro d'ajout, la comparaison des triplets ne va jamais chercher à comparer des tâches entre elles.

Une autre solution au fait que les tâches ne possèdent pas de relation d'ordre est de créer une classe d'encapsulation qui ignore l'élément tâche et ne compare que le champ priorité :

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

Le problème restant consiste à trouver une tâche en attente et modifier sa priorité ou la supprimer. Trouver une tâche peut être réalisé à l'aide d'un dictionnaire pointant vers une entrée dans la file.

Supprimer une entrée ou changer sa priorité est plus difficile puisque cela romprait l'invariant de la structure de tas. Une solution possible est de marquer l'entrée comme supprimée et d'ajouter une nouvelle entrée avec sa priorité modifiée :

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

Théorie

Les tas sont des tableaux pour lesquels a[k] <= a[2*k+1] et a[k] <= a[2*k+2] pour tout k en comptant les éléments à partir de 0. Pour simplifier la comparaison, les éléments inexistants sont considérés comme étant infinis. L'intérêt des tas est que a[0] est toujours leur plus petit élément.

L'invariant étrange ci-dessus est une représentation efficace en mémoire d'un tournoi. Les nombres ci-dessous sont k et non a[k] :

                               0

              1                                 2

      3               4                5               6

  7       8       9       10      11      12      13      14

15 16   17 18   19 20   21 22   23 24   25 26   27 28   29 30

Dans l'arbre ci-dessus, chaque nœud k a pour enfants 2*k+1 et 2*k+2. Dans les tournois binaires habituels dans les compétitions sportives, chaque nœud est le vainqueur des deux nœuds inférieurs et nous pouvons tracer le chemin du vainqueur le long de l'arbre afin de voir qui étaient ses adversaires. Cependant, dans de nombreuses applications informatiques de ces tournois, nous n'avons pas besoin de produire l'historique du vainqueur. Afin d'occuper moins de mémoire, on remplace le vainqueur lors de sa promotion par un autre élément à un plus bas niveau. La règle devient alors qu'un nœud et les deux nœuds qu'il chapeaute contiennent trois éléments différents, mais le nœud supérieur « gagne » contre les deux nœuds inférieurs.

If this heap invariant is protected at all time, index 0 is clearly the overall winner. The simplest algorithmic way to remove it and find the "next" winner is to move some loser (let's say cell 30 in the diagram above) into the 0 position, and then percolate this new 0 down the tree, exchanging values, until the invariant is re-established. This is clearly logarithmic on the total number of items in the tree. By iterating over all items, you get an O(n log n) sort.

Une propriété agréable de cet algorithme est qu'il est possible d'insérer efficacement de nouveaux éléments en cours de classement, du moment que les éléments insérés ne sont pas « meilleurs » que le dernier élément qui a été extrait. Ceci s'avère très utile dans des simulations où l'arbre contient la liste des événements arrivants et que la condition de « victoire » est le plus petit temps d'exécution planifié. Lorsqu'un événement programme l'exécution d'autres événements, ceux-ci sont planifiés pour le futur et peuvent donc rejoindre le tas. Ainsi, le tas est une bonne structure pour implémenter un ordonnanceur (et c'est ce que j'ai utilisé pour mon séquenceur MIDI ☺).

Plusieurs structures ont été étudiées en détail pour implémenter des ordonnanceurs et les tas sont bien adaptés : ils sont raisonnablement rapides, leur vitesse est presque constante et le pire cas ne diffère pas trop du cas moyen. S'il existe des représentations qui sont plus efficaces en général, les pires cas peuvent être terriblement mauvais.

Les tas sont également très utiles pour ordonner les données sur de gros disques. Vous savez probablement qu'un gros tri implique la production de séquences pré-classées (dont la taille est généralement liée à la quantité de mémoire CPU disponible), suivie par une passe de fusion qui est généralement organisée de façon très intelligente [1]. Il est très important que le classement initial produise des séquences les plus longues possibles. Les tournois sont une bonne façon d'arriver à ce résultat. Si, en utilisant toute la mémoire disponible pour stocker un tournoi, vous remplacez et faites percoler les éléments qui s'avèrent acceptables pour la séquence courante, vous produirez des séquences d'une taille égale au double de la mémoire pour une entrée aléatoire et bien mieux pour une entrée approximativement triée.

Qui plus est, si vous écrivez l'élément 0 sur le disque et que vous recevez en entrée un élément qui n'est pas adapté au tournoi actuel (parce que sa valeur « gagne » par rapport à la dernière valeur de sortie), alors il ne peut pas être stocké dans le tas donc la taille de ce dernier diminue. La mémoire libérée peut être réutilisée immédiatement pour progressivement construire un deuxième tas, qui croit à la même vitesse que le premier décroît. Lorsque le premier tas a complètement disparu, vous échangez les tas et démarrez une nouvelle séquence. Malin et plutôt efficace !

Pour résumer, les tas sont des structures de données qu'il est bon de connaître. Je les utilise dans quelques applications et je pense qu'il est bon de garder le module heap sous le coude. ☺

Notes