heapq — Algoritmo de colas montículos (heap)

Código fuente: Lib/heapq.py

Este módulo proporciona una implementación del algoritmo de montículos, también conocido como algoritmo de cola con prioridad.

Los montículos son árboles binarios para los cuales cada nodo padre tiene un valor menor o igual que cualquiera de sus hijos. Esta implementación utiliza matrices para las cuales heap[k] <= heap[2*k+1] y heap[k] <= heap[2*k+2] para todo k, contando los elementos desde cero. Para poder comparar, los elementos inexistentes se consideran infinitos. La propiedad interesante de un montículo es que su elemento más pequeño es siempre la raíz, heap[0].

El API que se presenta a continuación difiere de los algoritmos de los libros de texto en dos aspectos: (a) Utilizamos la indexación basada en cero. Esto hace que la relación entre el índice de un nodo y los índices de sus hijos sea un poco menos evidente, pero es más adecuado ya que Python utiliza la indexación basada en cero. (b) Nuestro método «pop» retorna el elemento más pequeño, no el más grande (llamado «min heap» o montículo por mínimo en los libros de texto; un «max heap» o montículo por máximos es más común en los textos debido a su idoneidad para la clasificación in situ).

Estos dos permiten ver el montículo como una lista Python normal sin sorpresas: heap[0] es el ítem más pequeño, y heap.sort() mantiene el montículo invariable!

Para crear un montículo, usa una lista inicializada como [], o puedes transformar una lista poblada en un montículo a través de la función heapify().

Las siguientes funciones están provistas:

heapq.heappush(heap, item)

Empujar el valor item en el heap, manteniendo el montículo invariable.

heapq.heappop(heap)

Desapila o pop y retorna el elemento más pequeño del heap, manteniendo el montículo invariable. Si el montículo está vacío, IndexError se lanza. Para acceder al elemento más pequeño sin necesidad de desapilar, usa heap[0].

heapq.heappushpop(heap, item)

Apila el elemento o iem en el montículo, y luego desapila y retorna el elemento más pequeño del montículo. La acción combinada se ejecuta más eficientemente que heappush() seguido de una llamada separada a heappop().

heapq.heapify(x)

Transformar la lista x en un montículo, en el lugar, en tiempo lineal.

heapq.heapreplace(heap, item)

Desapila y retorna el elemento más pequeño del heap, y también apile el nuevo item. El tamaño del montículo no cambia. Si el montículo está vacío, IndexError se lanza.

Esta operación de un solo paso es más eficiente que un heappop() seguido por heappush() y puede ser más apropiada cuando se utiliza un montículo de tamaño fijo. La combinación pop/push siempre retorna un elemento del montículo y lo reemplaza con item.

El valor retornado puede ser mayor que el item añadido. Si no se desea eso, considere usar heappushpop() en su lugar. Su combinación push/pop retorna el menor de los dos valores, dejando el mayor valor en el montículo.

El módulo también ofrece tres funciones de propósito general basadas en los montículos.

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

Fusionar varias entradas ordenadas en una sola salida ordenada (por ejemplo, fusionar entradas con marca de tiempo de varios archivos de registro). Retorna un iterator sobre los valores ordenados.

Similar a sorted(itertools.chain(*iterables)) pero retorna un iterable, no hala los datos a la memoria de una sola vez, y asume que cada uno de los flujos de entrada ya están ordenado (de menor a mayor).

Tiene dos argumentos opcionales que deben ser especificados como argumentos de palabras clave.

key especifica una key function de un argumento que se utiliza para extraer una clave de comparación de cada elemento de entrada. El valor por defecto es None (compara los elementos directamente).

reverse es un valor booleano. Si se establece en True, entonces los elementos de entrada se fusionan como si cada comparación se invirtiera. Para lograr un comportamiento similar a sorted(itertools.chain(*iterables), reverse=True), todos los iterables deben ser ordenados de mayor a menor.

Distinto en la versión 3.5: Añadió los parámetros opcionales de key y reverse.

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

Retorna una lista con los n elementos más grandes del conjunto de datos definidos por iterable. key, si se proporciona, especifica una función de un argumento que se utiliza para extraer una clave de comparación de cada elemento en iterable (por ejemplo, key=str.lower). Equivalente a: sorted(iterable, key=clave, reverse=True)[:n].

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

Retorna una lista con los n elementos más pequeños del conjunto de datos definidos por iterable. key, si se proporciona, especifica una función de un argumento que se utiliza para extraer una clave de comparación de cada elemento en iterable (por ejemplo, key=str.lower). Equivalente a: sorted(iterable, key=clave)[:n].

Las dos últimas funciones funcionan mejor para valores más pequeños de n. Para valores más grandes, es más eficiente usar la función sorted(). Además, cuando n==1, es más eficiente usar las funciones incorporadas min() y max`(). Si se requiere el uso repetido de estas funciones, considere convertir lo iterable en un verdadero montículo.

Ejemplos Básicos

Un heapsort» puede ser implementado empujando todos los valores en un montículo y luego desapilando los valores más pequeños uno a la vez:

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

Esto es similar a sorted(iterable), pero a diferencia de sorted(), esta implementación no es estable.

Los elementos del montículo pueden ser tuplas. Esto es útil para asignar valores de comparación (como las prioridades de las tareas) junto con el registro principal que se está rastreando:

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

Notas de Aplicación de la Cola de Prioridades

Una cola de prioridad es de uso común para un montículo, y presenta varios desafíos de implementación:

  • Estabilidad de la clasificación: ¿cómo se consigue que dos tareas con iguales prioridades sean retornadas en el orden en que fueron añadidas originalmente?

  • Interrupciones de comparación en tupla para pares (prioridad, tarea) si las prioridades son iguales y las tareas no tienen un orden de comparación por defecto.

  • ¿Si la prioridad de una tarea cambia, cómo la mueves a una nueva posición en el montículo?

  • ¿O si una tarea pendiente necesita ser borrada, cómo la encuentras y la eliminas de la cola?

Una solución a los dos primeros desafíos es almacenar las entradas como una lista de 3 elementos que incluya la prioridad, un recuento de entradas y la tarea. El recuento de entradas sirve como un desempate para que dos tareas con la misma prioridad sean retornadas en el orden en que fueron añadidas. Y como no hay dos recuentos de entradas iguales, la comparación tupla nunca intentará comparar directamente dos tareas.

Otra solución al problema de las tareas no comparables es crear una clase envolvente que ignore el elemento de la tarea y sólo compare el campo de prioridad:

from dataclasses import dataclass, field
from typing import Any

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

Los desafíos restantes giran en torno a encontrar una tarea pendiente y hacer cambios en su prioridad o eliminarla por completo. Encontrar una tarea se puede hacer con un diccionario que apunta a una entrada en la cola.

Eliminar la entrada o cambiar su prioridad es más difícil porque rompería las invariantes de la estructura del montículo. Por lo tanto, una posible solución es marcar la entrada como eliminada y añadir una nueva entrada con la prioridad revisada:

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

Teoría

Los montículos son conjuntos para los cuales a[k] <= a[2*k+1] y a[k] <= a[2*k+2] para todos los k, contando los elementos desde 0. Para comparar, los elementos no existentes se consideran infinitos. La interesante propiedad de un montículo es que a[0] es siempre su elemento más pequeño.

La extraña invariante de arriba intenta ser una representación eficiente de la memoria para un torneo. Los números de abajo son k, no 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

En el árbol de arriba, cada celda k está coronada por 2*k+1 y 2*k+2. En un torneo binario habitual que vemos en los deportes, cada celda es el ganador sobre las dos celdas que supera, y podemos rastrear al ganador hasta el árbol para ver todos los oponentes que tuvo. Sin embargo, en muchas aplicaciones informáticas de tales torneos, no necesitamos rastrear la historia de un ganador. Para ser más eficientes en la memoria, cuando un ganador es ascendido, tratamos de reemplazarlo por algo más en un nivel inferior, y la regla se convierte en que una celda y las dos celdas que supera contienen tres elementos diferentes, pero la celda superior «gana» sobre las dos celdas superiores.

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.

Una buena característica de este tipo es que puedes insertar nuevos elementos de manera eficiente mientras se realiza la clasificación, siempre y cuando los elementos insertados no sean «mejores» que el último 0’th elemento que has extraído. Esto es especialmente útil en contextos de simulación, donde el árbol contiene todos los eventos entrantes, y la condición de «ganar» significa el menor tiempo programado. Cuando un evento programa otros eventos para su ejecución, se programan en el futuro, para que puedan ir fácilmente al montículo. Por lo tanto, un montículo es una buena estructura para implementar planificadores o schedulers (esto es lo que usé para mi secuenciador MIDI :-).

Se han estudiado extensamente varias estructuras para implementar los planificadores, y los montículos son buenos para ello, ya que son razonablemente rápidos, la velocidad es casi constante, y el peor de los casos no es muy diferente del caso promedio. Sin embargo, hay otras representaciones que son más eficientes en general, aunque los peores casos podrían ser terribles.

Los montículos también son muy útiles en las grandes ordenaciones de elementos en discos de memoria. Lo más probable es que todos sepan que un tipo grande implica la producción de «ejecuciones» (que son secuencias preclasificadas, cuyo tamaño suele estar relacionado con la cantidad de memoria de la CPU), seguidas de una fusión de pases para estas ejecuciones, cuya fusión suele estar muy inteligentemente organizada [1]. Es muy importante que la clasificación inicial produzca las ejecuciones posibles más largas. Los torneos son una buena manera de lograrlo. Si, utilizando toda la memoria disponible para celebrar un torneo, sustituyes y filtras los elementos que encajan en la carrera actual, producirás carreras que tienen el doble del tamaño de la memoria para la entrada aleatoria, y mucho mejor para la entrada ordenada de forma difusa.

Además, si se da salida al 0’th item en el disco y se obtiene una entrada que no puede caber en el torneo actual (porque el valor «gana» sobre el último valor de salida), no puede caber en el montículo, por lo que el tamaño del montículo disminuye. La memoria liberada podría ser ingeniosamente reutilizada inmediatamente para construir progresivamente un segundo montículo, que crece exactamente al mismo ritmo que el primer montículo se está fundiendo. Cuando el primer montículo se desvanece completamente, se cambia de montículo y se inicia una nueva carrera. ¡Ingenioso y muy efectivo!

En una palabra, los montículos son estructuras de memoria útiles a conocer. Las uso en algunas aplicaciones, y creo que es bueno tener un módulo “heap” alrededor. :-)

Notas al pie de página