python数据结构与算法-16

python数据结构与算法-16_优先级队列

优先级队列

你可能比较奇怪，队列不是早就讲了嘛。这里之所以放到这里讲优先级队列，是因为虽然名字有队列，
但其实是使用堆来实现的。上一章讲完了堆，这一章我们就趁热打铁来实现一个优先级队列。

实现优先级队列

优先级队列(Priority Queue) 顾名思义，就是入队的时候可以给一个优先级，通常是个数字或者时间戳等，
当出队的时候我们希望按照给定的优先级出队，我们按照 TDD(测试驱动开发) 的方式先来写测试代码：

def test_priority_queue():size = 5pq = PriorityQueue(size)pq.push(5, 'purple')    # priority, valuepq.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):self.maxsize = maxsizeself._maxheap = MaxHeap(maxsize)def push(self, priority, value):# 注意这里把这个 tuple push 进去，python 比较 tuple 从第一个开始比较# 这样就很巧妙地实现了按照优先级排序entry = (priority, value)    # 入队的时候会根据 priority 维持堆的特性self._maxheap.add(entry)def pop(self, with_priority=False):entry = self._maxheap.extract()if with_priority:return entryelse:return entry[1]def is_empty(self):return len(self._maxheap) == 0

源码

# -*- coding:utf-8 -*-# 第二章拷贝的 Array 代码class Array(object):def __init__(self, size=32):self._size = sizeself._items = [None] * sizedef __getitem__(self, index):return self._items[index]def __setitem__(self, index, value):self._items[index] = valuedef __len__(self):return self._sizedef clear(self, value=None):for i in range(len(self._items)):self._items[i] = valuedef __iter__(self):for item in self._items:yield item#####################################################
# heap 实现
#####################################################class MaxHeap(object):"""Heaps:完全二叉树，最大堆的非叶子节点的值都比孩子大，最小堆的非叶子结点的值都比孩子小Heap包含两个属性，order property 和 shape property(a complete binary tree)，在插入一个新节点的时候，始终要保持这两个属性插入操作：保持堆属性和完全二叉树属性, sift-up 操作维持堆属性extract操作：只获取根节点数据，并把树最底层最右节点copy到根节点后，sift-down操作维持堆属性用数组实现heap，从根节点开始，从上往下从左到右给每个节点编号，则根据完全二叉树的性质，给定一个节点i， 其父亲和孩子节点的编号分别是:parent = (i-1) // 2left = 2 * i + 1rgiht = 2 * i + 2使用数组实现堆一方面效率更高，节省树节点的内存占用，一方面还可以避免复杂的指针操作，减少调试难度。"""def __init__(self, maxsize=None):self.maxsize = maxsizeself._elements = Array(maxsize)self._count = 0def __len__(self):return self._countdef add(self, value):if self._count >= self.maxsize:raise Exception('full')self._elements[self._count] = valueself._count += 1self._siftup(self._count-1)  # 维持堆的特性def _siftup(self, ndx):if ndx > 0:parent = int((ndx-1)/2)if self._elements[ndx] > self._elements[parent]:    # 如果插入的值大于 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]    # 保存 root 值self._count -= 1self._elements[0] = self._elements[self._count]    # 最右下的节点放到root后siftDownself._siftdown(0)    # 维持堆特性return valuedef _siftdown(self, ndx):left = 2 * ndx + 1right = 2 * ndx + 2# determine which node contains the larger valuelargest = ndxif (left < self._count and     # 有左孩子self._elements[left] >= self._elements[largest] andself._elements[left] >= self._elements[right]):  # 原书这个地方没写实际上找的未必是largestlargest = leftelif right < self._count and self._elements[right] >= self._elements[largest]:largest = rightif largest != ndx:self._elements[ndx], self._elements[largest] = self._elements[largest], self._elements[ndx]self._siftdown(largest)class PriorityQueue(object):def __init__(self, maxsize):self.maxsize = maxsizeself._maxheap = MaxHeap(maxsize)def push(self, priority, value):entry = (priority, value)    # 注意这里把这个 tuple push进去，python 比较 tuple 从第一个开始比较self._maxheap.add(entry)def pop(self, with_priority=False):entry = self._maxheap.extract()if with_priority:return entryelse:return entry[1]def is_empty(self):return len(self._maxheap) == 0def test_priority_queue():size = 5pq = 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']def test_buildin_PriorityQueue():  # python3"""测试内置的 PriorityQueuehttps://pythonguides.com/priority-queue-in-python/"""from queue import PriorityQueueq = PriorityQueue()q.put((10, 'Red balls'))q.put((8, 'Pink balls'))q.put((5, 'White balls'))q.put((4, 'Green balls'))while not q.empty():item = q.get()print(item)def test_buildin_heapq_as_PriorityQueue():"""测试使用 heapq 实现优先级队列，保存一个 tuple 比较元素(tuple第一个元素是优先级)"""import heapqs_roll = []heapq.heappush(s_roll, (4, "Tom"))heapq.heappush(s_roll, (1, "Aruhi"))heapq.heappush(s_roll, (3, "Dyson"))heapq.heappush(s_roll, (2, "Bob"))while s_roll:deque_r = heapq.heappop(s_roll)print(deque_r)# python3 没有了 __cmp__ 魔法函数 https://stackoverflow.com/questions/8276983/why-cant-i-use-the-method-cmp-in-python-3-as-for-python-2
class Item:def __init__(self, key, weight):self.key, self.weight = key, weightdef __lt__(self, other): # 看其来 heapq 实现只用了 小于 比较，这里定义了就可以 push 一个 item 类return self.weight < other.weightdef __eq__(self, other):return self.weight == other.weightdef __str__(self):return '{}:{}'.format(self.key,self.weight)def test_heap_item():"""测试使用 Item 类实现优先级队列，因为 heapq 内置使用的是小于运算法，重写魔术 < 比较方法即可实现"""import heapqpq = []heapq.heappush(pq, Item('c', 3))heapq.heappush(pq, Item('a', 1))heapq.heappush(pq, Item('b', 2))while pq:print(heapq.heappop(pq))