优先级队列
Queue : FIFO
TDD方式写代码,先写测试代码。
def test_priority_queue():
size = 5
pq = PriorityQueue(size)
pq.push(5, 'purple')
pq.push(0, 'white')
pq.push(3, 'orange')
pq.push(1, 'black')
res = []
while not pq.is_empty():
res.append(pq.pop())
assert res == ['purple', 'orange', 'black', 'white']
优先级队列推出数值比较大的值,可以使用最大堆来做,每次推出最大,但是需要推出两个值,所以使用元祖tuple,把这优先级和元素放进去。
比较tuple, (1,3)>(2,6)
还是粘贴数组代码,最大堆代码,然后实现优先级队列
class Array(object):
def __init__(self, size=32):
self._size = size
self._items = [None] * size
def __getitem__(self, index):
return self._items[index]
def __setitem__(self, index, value):
self._items[index] = value
def __len__(self):
return self._size
def clear(self, value=None):
for i in range(len(self._items)):
self._items[i] = value
def __iter__(self):
for item in self._items:
yield item
"""heap 实现"""
class Maxheap(object):
def __init__(self, maxsize=None):
self.maxsize = maxsize
self._elements = Array(maxsize)
self._count = 0
def __len__(self):
return self._count
def add(self, value):
if self._count >= self.maxsize:
raise Exception('full')
# 开始加入,先把值放在最后一位,最后一位就是_count
self._elements[self._count] = value
self._count += 1
self._siftup(self._count - 1) # 定义_siftup函数,传入的值是添加元素的位置
def _siftup(self, ndx): # 递归交换,直到满足最大堆的特性。
if ndx > 0:
parent = int((ndx - 1 / 2))
if self._elements[ndx] > self._elements[parent]: # 如果他的值大于父亲就交换
self._elements[ndx], self._elements[parent] = self._elements[parent], self._elements[ndx]
self._siftup(parent) # 递归
def extract(self):
if self._count <= 0:
raise Exception('empty')
value = self._elements[0]
self._count -= 1
self._elements[0] = self._elements[self._count]
self._siftdown(0)
return value
def _siftdown(self, ndx):
left = 2 * ndx + 1
right = 2 * ndx + 2
largest = ndx
if (left < self._count and self._elements[left] >= self._elements[largest] and self._elements[left] >=
self._elements[right]):
largest = left
elif right < self._count and self._elements[right] >= self._elements[largest]:
largest = right
if largest != ndx:
self._elements[ndx], self._elements[largest] = self._elements[largest], self._elements[ndx]
self._siftdown(largest)
def test_priority_queue():
size = 5
pq = PriorityQueue(size)
pq.push(5, 'purple')
pq.push(0, 'white')
pq.push(3, 'orange')
pq.push(1, 'black')
res = []
while not pq.is_empty():
res.append(pq.pop())
assert res == ['purple', 'orange', 'black', 'white']
class PriorityQueue(object):
def __init__(self, maxsize=None):
self.maxsize = maxsize
self._maxheap = Maxheap(maxsize)
def push(self, priority, value):
entry = (priority, value) # push a tuple
self._maxheap.add(entry)
def pop(self, with_priority=False):
entry = self._maxheap.extract()
if with_priority:
return entry
else:
return entry[1]
def is_empty(self):
return len(self._maxheap) == 0