heapq --- الگوریتم صف هیپ

کد منبع: Lib/heapq.py

این ماژول پیاده‌سازی‌ای از الگوریتم صف هیپ (heap queue) ارائه می‌دهد، که به‌عنوان الگوریتم صف اولویت نیز شناخته می‌شود.

هیپ‌های کمینه (min-heaps) درخت‌های دودویی هستند که در آن‌ها هر گره والد دارای مقداری کوچک‌تر یا مساوی هر یک از فرزندان خود است. این شرط را ناوردای هیپ (heap invariant) می‌نامیم.

برای هیپ‌های کمینه (min-heaps)، این پیاده‌سازی از فهرست‌هایی استفاده می‌کند که در آن‌ها heap[k] <= heap[2*k+1] و heap[k] <= heap[2*k+2] برای تمام kهایی که عناصر مقایسه‌شده موجود باشند، برقرار است. عناصر از صفر شمارش می‌شوند. ویژگی جالب هیپ کمینه (min-heap) این است که کوچک‌ترین عنصر آن همیشه ریشه، heap[0] است.

هیپ‌های بیشینه (max-heaps) ناوردا معکوس را برآورده می‌کنند: هر گره والد مقداری بزرگ‌تر از هر یک از فرزندان خود دارد. این‌ها به‌صورت فهرست‌هایی پیاده‌سازی شده‌اند که در آن‌ها maxheap[2*k+1] <= maxheap[k] و maxheap[2*k+2] <= maxheap[k] برای تمام kهایی که عناصر مقایسه‌شده موجود باشند، برقرار است. ریشه، maxheap[0]، شامل بزرگ‌ترین عنصر است؛ heap.sort(reverse=True) ناوردا هیپ بیشینه را حفظ می‌کند.

API ماژول heapq از دو جهت با الگوریتم‌های هیپدر کتاب‌های درسی تفاوت دارد: (الف) ما از اندیس‌گذاری مبتنی بر صفر استفاده می‌کنیم. این کار رابطه بین اندیس یک گره و اندیس‌های فرزندان آن را کمی کمتر بدیهی می‌کند، اما از آن‌جا که پایتون از اندیس‌گذاری مبتنی بر صفر استفاده می‌کند، مناسب‌تر است. (ب) کتاب‌های درسی اغلب به دلیل مناسب بودن هیپ‌های بیشینه برای مرتب‌سازی درجا، بر آن‌ها تمرکز دارند. پیاده‌سازی ما هیپ‌های کمینه را ترجیح می‌دهد، زیرا آن‌ها با lists پایتون تطابق بهتری دارند.

این دو جنبه به شما امکان می‌دهد هیپ را به‌عنوان یک فهرست معمولی پایتون بدون غافلگیری در نظر بگیرید: heap[0] کوچک‌ترین آیتم است، و heap.sort() ناوردایی هیپ را حفظ می‌کند!

مانند list.sort()، این پیاده‌سازی برای مقایسه‌ها تنها از عملگر < استفاده می‌کند، هم برای هیپ‌های کمینه (min-heaps) و هم برای پشته‌های بیشینه (max-heaps).

در API زیر و در این مستندات، اصطلاح بدون قید هیپ معمولاً به یک هیپ کمینه اشاره دارد. API برای هیپ‌های بیشینه با استفاده از پسوند _max نام‌گذاری شده است.

برای ایجاد یک هیپ، از یک فهرست مقداردهی‌شده به‌صورت [] استفاده کنید، یا یک فهرست موجود را با استفاده از توابع heapify() و heapify_max()، به‌ترتیب به یک هیپ کمینه یا هیپ بیشینه تبدیل کنید.

توابع زیر برای هیپ‌های کمینه فراهم شده‌اند:

heapq.heapify(x)

فهرست x را به‌صورت درجا و در زمان خطی به یک هیپ کمینه (min-heap) تبدیل می‌کند.

heapq.heappush(heap, item)

مقدار item را روی هیپ قرار دهید، به‌طوری که ناوردا هیپ کمینه حفظ شود.

heapq.heappop(heap)

کوچک‌ترین آیتم را از هیپ بردارید و برگردانید، به‌طوری که ناوردا (invariant) هیپ کمینه حفظ شود. اگر هیپ خالی باشد، IndexError پرتاب می‌شود. برای دسترسی به کوچک‌ترین آیتم بدون برداشتن آن، از heap[0] استفاده کنید.

heapq.heappushpop(heap, item)

آیتم را روی هیپ قرار دهید، سپس کوچک‌ترین آیتم را از هیپ بردارید و برگردانید. این عملیات ترکیبی، کارآمدتر از heappush() و به دنبال آن فراخوانی جداگانه heappop() اجرا می‌شود.

heapq.heapreplace(heap, item)

کوچک‌ترین آیتم را از هیپ برمی‌دارد و برمی‌گرداند، و همچنین آیتم جدید را نیز درج (push) می‌کند. اندازه‌ی هیپ تغییر نمی‌کند. اگر هیپ خالی باشد، IndexError پرتاب می‌شود.

این عملیات تک‌مرحله‌ای کارآمدتر از فراخوانی heappop() و سپس heappush() است و می‌تواند هنگام استفاده از هپبا اندازه‌ی ثابت مناسب‌تر باشد. ترکیب pop/push همیشه عنصری را از هپبرمی‌گرداند و آن را با item جایگزین می‌کند.

مقدار بازگشتی ممکن است بزرگ‌تر از آیتم اضافه‌شده باشد. اگر این مطلوب نیست، در عوض استفاده از heappushpop() را در نظر بگیرید. ترکیب درج/حذف (push/pop) آن کوچک‌تر بین دو مقدار را برمی‌گرداند و مقدار بزرگ‌تر را روی هیپ باقی می‌گذارد.

برای هیپ‌های بیشینه (max-heaps)، توابع زیر ارائه شده‌اند:

heapq.heapify_max(x)

فهرست x را به‌صورت درجا و در زمان خطی به یک هیپ بیشینه (max-heap) تبدیل می‌کند.

اضافه شده در نسخه‌ی 3.14.

heapq.heappush_max(heap, item)

مقدار item را بر روی هیپ بیشینه heap قرار دهید، به‌طوری که شرط هیپ بیشینه حفظ شود.

اضافه شده در نسخه‌ی 3.14.

heapq.heappop_max(heap)

بزرگ‌ترین آیتم را از هیپ بیشینه (max-heap) heap بردارید و برگردانید، با حفظ ناوردایی هیپ بیشینه. اگر هیپ بیشینه خالی باشد، IndexError پرتاب می‌شود. برای دسترسی به بزرگ‌ترین آیتم بدون برداشتن آن، از maxheap[0] استفاده کنید.

اضافه شده در نسخه‌ی 3.14.

heapq.heappushpop_max(heap, item)

آیتم item را روی هیپ بیشینه‌ی heap قرار دهید، سپس بزرگ‌ترین آیتم را از heap بردارید و برگردانید. این عملیات ترکیبی، کارآمدتر از فراخوانی heappush_max() و سپس یک فراخوانی جداگانه‌ی heappop_max() اجرا می‌شود.

اضافه شده در نسخه‌ی 3.14.

heapq.heapreplace_max(heap, item)

بزرگ‌ترین آیتم را از هیپ بیشینه (max-heap) heap خارج می‌کند و برمی‌گرداند و همچنین آیتم جدید item را وارد می‌کند. اندازه‌ی هیپ بیشینه تغییر نمی‌کند. اگر هیپ بیشینه خالی باشد، IndexError پرتاب می‌شود.

ممکن است مقدار برگردانده‌شده کوچک‌تر از آیتم افزوده‌شده باشد. برای نکات تفصیلی کاربرد، به تابع مشابه heapreplace() مراجعه کنید.

اضافه شده در نسخه‌ی 3.14.

این ماژول همچنین ۳ تابع همه‌منظوره مبتنی بر هیپ‌ها ارائه می‌دهد.

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

چندین ورودی مرتب‌شده را در یک خروجی مرتب‌شده واحد ادغام می‌کند (برای مثال، ورودی‌های دارای برچسب زمانی را از چندین پرونده گزارش ادغام می‌کند). یک iterator روی مقادیر مرتب‌شده برمی‌گرداند.

مشابه sorted(itertools.chain(*iterables)) است، اما یک پیمایش‌پذیر برمی‌گرداند، داده‌ها را به‌صورت یک‌جا در حافظه بارگذاری نمی‌کند و فرض می‌کند که هر یک از جریان‌های ورودی از قبل مرتب شده‌اند (از کوچک‌ترین به بزرگ‌ترین).

دارای دو آرگومان اختیاری است که باید به‌صورت آرگومان‌های کلیدواژه‌ای مشخص شوند.

key یک key function با یک آرگومان را مشخص می‌کند که برای استخراج کلید مقایسه از هر المان ورودی به کار می‌رود. مقدار پیش‌فرض None است (المان‌ها مستقیماً مقایسه می‌شوند).

reverse یک مقدار بولی است. اگر روی True تنظیم شود، آنگاه عناصر ورودی به‌گونه‌ای ادغام می‌شوند که گویی هر مقایسه معکوس شده باشد. برای دستیابی به رفتاری مشابه sorted(itertools.chain(*iterables), reverse=True)، تمام پیمایش‌پذیرها باید از بزرگ‌ترین به کوچک‌ترین مرتب شده باشند.

تغییر یافته در نسخه‌ی 3.5: پارامترهای اختیاری key و reverse اضافه شدند.

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

فهرستی شامل n بزرگ‌ترین عنصر از مجموعه داده‌ی تعریف‌شده با iterable را برمی‌گرداند. key، در صورت ارائه، تابعی با یک آرگومان را مشخص می‌کند که برای استخراج کلید مقایسه از هر عنصر در iterable استفاده می‌شود (برای مثال، key=str.lower). معادل است با: sorted(iterable, key=key, reverse=True)[:n].

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

فهرستی شامل n عنصر از کوچک‌ترین عناصر مجموعه‌داده تعریف‌شده توسط iterable برمی‌گرداند. key، در صورت ارائه، تابعی با یک آرگومان را مشخص می‌کند که برای استخراج کلید مقایسه از هر عنصر در iterable استفاده می‌شود (برای مثال، key=str.lower). معادل است با: sorted(iterable, key=key)[:n].

دو تابع اخیر برای مقادیر کوچک‌تر n بهترین عملکرد را دارند. برای مقادیر بزرگ‌تر، استفاده از تابع sorted() کارآمدتر است. همچنین، وقتی n==1 باشد، استفاده از توابع توکار min() و max() کارآمدتر است. اگر استفاده‌ی مکرر از این توابع لازم است، تبدیل پیمایش‌پذیر به یک هیپ واقعی را در نظر بگیرید.

مثال‌های پایه

می‌توان یک مرتب‌سازی هیپ‌ای را با قرار دادن همه مقادیر در یک هیپ و سپس برداشتن کوچک‌ترین مقادیر به‌صورت یکی‌یکی پیاده‌سازی کرد:

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

این مشابه sorted(iterable) است، اما برخلاف sorted()، این پیاده‌سازی پایدار نیست.

عناصر هیپ می‌توانند تاپل باشند. این امر برای اختصاص مقادیر مقایسه‌ای (مانند اولویت‌های وظیفه) در کنار رکورد اصلی در حال پیگیری مفید است:

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

سایر برنامه‌های کاربردی

میانه‌ها معیاری از گرایش مرکزی برای مجموعه‌ای از اعداد هستند. در توزیع‌هایی که به‌دلیل داده‌های پرت اریب‌شده‌اند، میانه برآورد پایدارتری نسبت به میانگین (میانگین حسابی) ارائه می‌دهد. میانه جاری یک الگوریتم برخط است که با رسیدن داده‌های جدید به‌طور مداوم به‌روزرسانی می‌شود.

می‌توان میانه‌ی جاری را به‌صورت کارآمد با متعادل‌سازی دو هیپ پیاده‌سازی کرد، یک هیپ بیشینه برای مقادیر برابر با نقطه‌ی میانی یا کمتر از آن و یک هیپ کمینه برای مقادیر بیشتر از نقطه‌ی میانی. هنگامی که دو هیپ اندازه‌ی یکسانی داشته باشند، میانه‌ی جدید میانگین رأس‌های دو هیپ است؛ در غیر این صورت، میانه در رأس هیپ بزرگ‌تر قرار دارد:

def running_median(iterable):
    "Yields the cumulative median of values seen so far."

    lo = []  # max-heap
    hi = []  # min-heap (same size as or one smaller than lo)

    for x in iterable:
        if len(lo) == len(hi):
            heappush_max(lo, heappushpop(hi, x))
            yield lo[0]
        else:
            heappush(hi, heappushpop_max(lo, x))
            yield (lo[0] + hi[0]) / 2

برای مثال:

>>> list(running_median([5.0, 9.0, 4.0, 12.0, 8.0, 9.0]))
[5.0, 7.0, 5.0, 7.0, 8.0, 8.5]

یادداشت‌های پیاده‌سازی صف اولویت

یک صف اولویت کاربرد رایجی برای هیپ است، و چندین چالش پیاده‌سازی به همراه دارد:

  • پایداری مرتب‌سازی: چگونه می‌توانید دو وظیفه با اولویت‌های برابر را به همان ترتیبی که در ابتدا افزوده شده‌اند دریافت کنید؟

  • مقایسه‌ی تاپل برای جفت‌های (priority, task) در صورتی با شکست مواجه می‌شود که اولویت‌ها برابر باشند و کارها ترتیب مقایسه‌ی پیش‌فرض نداشته باشند.

  • اگر اولویت یک وظیفه تغییر کند، چگونه آن را به جایگاه جدیدی در هیپ جابه‌جا می‌کنید؟

  • یا اگر یک وظیفه‌ی در انتظار باید حذف شود، چگونه آن را پیدا می‌کنید و از صف حذف می‌کنید؟

راه‌حلی برای دو چالش نخست این است که ورودی‌ها به‌صورت فهرست‌های ۳‌عنصری شامل اولویت، شماره‌ی ورودی و وظیفه ذخیره شوند. شماره‌ی ورودی به‌عنوان عامل تساوی‌شکن عمل می‌کند تا دو وظیفه با اولویت یکسان به همان ترتیبی که اضافه شده‌اند برگردانده شوند. و از آن‌جا که هیچ دو شماره‌ی ورودی‌ای یکسان نیستند، مقایسه‌ی تاپل هرگز تلاش نمی‌کند دو وظیفه را مستقیماً مقایسه کند.

راه‌حل دیگر برای مشکل وظایف غیرقابل‌مقایسه، ایجاد یک کلاس پوششی است که آیتم وظیفه را نادیده می‌گیرد و فقط فیلد اولویت را مقایسه می‌کند:

from dataclasses import dataclass, field
from typing import Any

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

چالش‌های باقی‌مانده حول یافتن یک وظیفه‌ی در انتظار و ایجاد تغییرات در اولویت آن یا حذف کامل آن می‌چرخند. یافتن یک وظیفه را می‌توان با دیکشنری‌ای که به یک ورودی در صف اشاره می‌کند، انجام داد.

حذف ورودی یا تغییر اولویت آن دشوارتر است، زیرا باعث نقض ناورداهای ساختار هیپمی‌شود. بنابراین، یک راه‌حل ممکن این است که ورودی را به‌عنوان حذف‌شده علامت‌گذاری کنید و ورودی جدیدی با اولویت اصلاح‌شده اضافه کنید:

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

نظریه

هیپ‌ها آرایه‌هایی هستند که در آن‌ها برای همه‌ی k، a[k] <= a[2*k+1] و a[k] <= a[2*k+2] برقرار است، با شمارش عناصر از ۰. برای مقایسه، عناصر ناموجود بی‌نهایت در نظر گرفته می‌شوند. ویژگی جالب یک هیپ این است که a[0] همیشه کوچک‌ترین عنصر آن است.

ناوردای عجیب بالا به‌عنوان یک بازنمایی کارآمد در حافظه برای یک تورنمنت (tournament) در نظر گرفته شده است. اعداد زیر k هستند، نه 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

در درخت بالا، هر سلول k بر 2*k+1 و 2*k+2 قرار دارد. در یک تورنمنت دودویی معمول که در ورزش‌ها می‌بینیم، هر سلول برنده‌ی دو سلولی است که بر آن‌ها قرار دارد، و می‌توانیم برنده را در درخت به سمت پایین دنبال کنیم تا همه‌ی حریفانی را که داشته است ببینیم. با این حال، در بسیاری از کاربردهای رایانه‌ای چنین تورنمنت‌هایی، نیازی به پیگیری تاریخچه‌ی یک برنده نیست. برای کارآمدتر بودن از نظر حافظه، هنگامی که یک برنده ارتقا می‌یابد، سعی می‌کنیم آن را با چیز دیگری در سطحی پایین‌تر جایگزین کنیم، و قاعده این می‌شود که یک سلول و دو سلولی که بر آن‌ها قرار دارد شامل سه آیتم متفاوت هستند، اما سلول بالایی بر دو سلول زیرین خود «برنده می‌شود».

اگر این ناوردایی هیپ همواره حفظ شود، اندیس ۰ به‌وضوح برنده‌ی کلی است. ساده‌ترین روش الگوریتمی برای حذف آن و یافتن برنده‌ی «بعدی» این است که یکی از بازنده‌ها (مثلاً خانه‌ی ۳۰ در شکل بالا) را به موقعیت ۰ منتقل کنیم و سپس این ۰ جدید را با تبادل مقادیر، در درخت به پایین جابه‌جا کنیم تا ناوردایی دوباره برقرار شود. این کار به‌وضوح بر حسب تعداد کل آیتم‌های درخت لگاریتمی است. با پیمایش همه‌ی آیتم‌ها، یک مرتب‌سازی O(n log n) به دست می‌آورید.

ویژگی خوب این مرتب‌سازی این است که می‌توانید در حین انجام مرتب‌سازی، آیتم‌های جدید را به‌طور کارآمدی درج کنید، مشروط بر اینکه آیتم‌های درج‌شده «بهتر» از آخرین عنصر صفرمی که استخراج کرده‌اید نباشند. این ویژگی به‌ویژه در زمینه‌های شبیه‌سازی مفید است، جایی که درخت همه‌ی رویدادهای ورودی را در خود نگه می‌دارد و شرط «برد» به معنای کم‌ترین زمان زمان‌بندی‌شده است. هنگامی که رویدادی، رویدادهای دیگری را برای اجرا زمان‌بندی می‌کند، آن‌ها برای آینده زمان‌بندی می‌شوند، بنابراین می‌توانند به‌راحتی وارد هیپ شوند. بنابراین، هیپ ساختار خوبی برای پیاده‌سازی زمان‌بندها است (این همان چیزی است که من برای سکانس‌کننده‌ی MIDI خود از آن استفاده کردم :-).

ساختارهای گوناگونی برای پیاده‌سازی زمان‌بندها به‌طور گسترده بررسی شده‌اند و هیپ‌ها برای این کار مناسب هستند، زیرا نسبتاً سریع هستند، سرعت آن‌ها تقریباً ثابت است و بدترین حالت تفاوت چندانی با حالت میانگین ندارد. با این حال، بازنمایی‌های دیگری وجود دارند که به‌طور کلی کارآمدتر هستند، اما بدترین حالت‌ها ممکن است بسیار بد باشند.

هیپ‌ها همچنین در مرتب‌سازی‌های بزرگ دیسکی بسیار مفید هستند. احتمالاً همگی می‌دانید که یک مرتب‌سازی بزرگ مستلزم تولید «دنباله‌های مرتب‌شده» (runs) است (که دنباله‌هایی از پیش مرتب‌شده هستند و اندازه‌ی آن‌ها معمولاً به مقدار حافظه‌ی CPU بستگی دارد) و پس از آن، گذرهای ادغام برای این دنباله‌ها انجام می‌شود؛ ادغامی که اغلب بسیار هوشمندانه سازمان‌دهی می‌شود [1]. بسیار مهم است که مرتب‌سازی اولیه طولانی‌ترین دنباله‌های مرتب‌شده‌ی ممکن را تولید کند. تورنمنت‌ها (tournaments) راه خوبی برای دستیابی به آن هستند. اگر با استفاده از تمام حافظه‌ی موجود برای نگه‌داری یک تورنمنت، آیتم‌هایی را که اتفاقاً با دنباله‌ی جاری سازگار هستند جایگزین کنید و نفوذ دهید (percolate)، دنباله‌هایی تولید خواهید کرد که برای ورودی تصادفی دو برابر اندازه‌ی حافظه هستند و برای ورودی‌ای که به‌صورت تقریبی مرتب است بسیار بهتر خواهند بود.

علاوه بر این، اگر آیتم صفرم را روی دیسک خروجی دهید و ورودی‌ای دریافت کنید که ممکن است در مسابقه‌ی جاری نگنجد (زیرا مقدار آن بر مقدار خروجی آخر «پیروز می‌شود»)، آن ورودی در هیپ نمی‌گنجد، بنابراین اندازه‌ی هیپ کاهش می‌یابد. حافظه‌ی آزادشده می‌تواند بلافاصله و به‌صورت هوشمندانه برای ساخت تدریجی یک هیپ دوم دوباره استفاده شود، هیپ‌ای که دقیقاً با همان آهنگی که هیپ اول در حال آب شدن است، رشد می‌کند. هنگامی که پشته‌ی اول کاملاً ناپدید شود، پشته‌ها را جابه‌جا می‌کنید و یک اجرای جدید را آغاز می‌کنید. هوشمندانه و کاملاً مؤثر!

به‌اختصار، هیپ‌ها ساختارهای حافظه‌ی مفیدی برای آشنایی هستند. من از آن‌ها در چند برنامه استفاده می‌کنم و فکر می‌کنم خوب است که یک ماژول 'heap' در دسترس باشد. :-)

پانویس‌ها