Source code for libutilitaspy.data_structures.priority_queues
"""
Implements priority queues using heaps.
"""
__all__ = ['PriorityQueue']
from collections import MutableSequence
from heaps import Heap
[docs]class PriorityQueue(object):
"""
A priority queue is a queue of items such that its elements are extracted in
order of their priority (see http://en.wikipedia.org/wiki/Priority_queue).
The priority of the items determines how they are compared, thus this class
assumes that if two objects a and b are to be put in a queue with ascending
order, a < b if and only if a's priority is higher than b's. Dually, if they
are put in a queue with descending order, a > b if and only if a's priority
is higher than b's.
"""
def __init__(self, data=None, ascending=False):
"""
The constructor creates a new queue from a given sequence.
:param MutableSequence data: an initial sequence of comparable objects.
:param bool ascending: True if the items are sorted in ascending order, False for descending order.
:TODO: map data items to PQE
"""
assert isinstance(data, MutableSequence)
if data is None:
data = []
self.heap = Heap(data,ascending)
def isempty(self):
return self.heap.isempty()
def __contains__(self, datum):
for item in self.heap:
if item.data == datum:
return True
return False
def index(self, datum):
i = 0
for item in self.heap:
if item.data == datum:
return i
i += 1
raise IndexError()
def add(self, datum, priority):
self.heap.add(PriorityQueueEntry(datum, priority))
def peek(self):
return self.heap.peek_first().data
def remove_first(self):
return self.heap.remove_first().data
def remove(self, datum):
self.heap.delete(self.index(datum))
def __repr__(self):
return "PriorityQueue(%s)" % self.heap
class PriorityQueueEntry:
def __init__(self, data, priority):
self.data = data
self.priority = priority
def __lt__(self, other):
return self.priority < other.priority
def __repr__(self):
return "PQE(%s,%s)" % (self.priority, self.data)