Python heapq 模块：使用堆和优先级队列

数据结构、堆和优先级队列

抽象数据结构指定操作及其之间的关系。例如，优先级队列抽象数据结构支持三种操作：

1. 是空的检查队列是否为空。
2. 添加元素将一个元素添加到队列中。
3. 弹出元素弹出具有最高优先级的元素。

优先级队列通常用于优化任务执行，其目标是处理具有最高优先级的任务。任务完成后，其优先级会降低，并返回到队列中。

有两种不同的约定来确定元素的优先级：

1. 这最大的元素具有最高优先级。
2. 这最小的元素具有最高优先级。

这两个约定是等效的，因为您始终可以颠倒有效顺序。例如，如果您的元素包含数字，那么使用负数将会颠倒约定。

蟒蛇heapq模块使用第二种约定，通常是两者中更常见的一种。根据该公约，最小的元素具有最高优先级。这听起来可能令人惊讶，但它通常非常有用。在稍后您将看到的现实示例中，此约定将简化您的代码。

具体数据结构实现抽象数据结构中定义的操作，并进一步指定性能保证。

优先级队列的堆实现保证了推送（添加）和弹出（删除）元素都是对数时间操作。这意味着推入和弹出所需的时间与以 2 为底的对数的元素数量。

对数慢慢成长。以 15 为底的对数约为 4，而以 2 为底的对数约为 40。这意味着，如果算法在 15 个元素上足够快，那么在 1 万亿个元素上它只会慢十倍，并且可能仍然足够快。

在任何关于性能的讨论中，最大的警告是，这些抽象的考虑没有实际测量一个具体的程序并了解该程序的实际意义来得重要。瓶颈是。一般性能保证对于对程序行为进行有用的预测仍然很重要，但这些预测应该得到确认。

堆的实现

堆将优先级队列实现为完全二叉树。在一个二叉树，每个节点最多有两个子节点。在一个完全的二叉树，除了可能最深的一层之外，所有级别始终都是满的。如果最深层不完整，则节点将尽可能位于左侧。

这完整性属性表示树的深度是元素数量以 2 为底的对数，向上舍入。这是一个完整二叉树的示例：

在此特定示例中，所有级别均已完成。除了最深的节点之外，每个节点都恰好有两个子节点。三级共七个节点。三是七的以 2 为底的对数，四舍五入。

底层的单个节点称为根节点。将树顶部的节点称为根可能看起来很奇怪，但这是编程和计算机科学中的常见约定。

堆中的性能保证取决于元素如何在树上上下渗透。实际结果是堆中的比较次数是树大小的以 2 为底的对数。

在堆树中，节点中的值始终小于其两个子节点的值。这被称为堆属性。这不同于二叉搜索树，其中只有左节点会小于其父节点的值。

压入和弹出的算法都依赖于暂时违反堆属性，然后通过在单个分支上或下进行比较和替换来修复堆属性。

例如，要将一个元素推送到堆上，Python 将新节点添加到下一个开放槽中。如果底层未满，则将该节点添加到底部的下一个空槽中。否则，将创建一个新级别，然后将元素添加到新的底层。

添加节点后，Python 会将其与其父节点进行比较。如果违反了堆属性，则交换该节点及其父节点，并再次从父节点开始检查。这一直持续到堆属性成立或到达根为止。

类似地，当弹出最小元素时，Python 知道，由于堆属性，该元素位于树的根部。它用最深层的最后一个元素替换该元素，然后检查分支下是否违反了堆属性。

优先级队列的用途

优先级队列以及作为优先级队列实现的堆对于涉及查找某种程度极端的元素的程序非常有用。例如，您可以将优先级队列用于以下任意任务：

• 从点击数据中获取三个最受欢迎的博客文章
• 寻找从一个点到另一个点的最快方法
• 根据到达频率预测哪辆巴士将最先到达车站

您可以使用优先级队列的另一个任务是安排电子邮件。想象一个有多种电子邮件的系统，每种电子邮件都需要以一定的频率发送。一种电子邮件需要每十五分钟发送一次，另一种电子邮件需要每四十分钟发送一次。

调度程序可以将两种类型的电子邮件添加到队列中时间戳指示下次需要发送电子邮件的时间。然后，调度程序可以查看具有最小时间戳的元素（表明它是下一个要发送的元素），并计算发送之前要休眠多长时间。

当调度程序醒来时，它将处理相关电子邮件，将电子邮件从优先级队列中取出，计算下一个时间戳，并将电子邮件放回到队列中的正确位置。

基本操作

蟒蛇heapq模块在列表上实现堆操作。与许多其他模块不同，它确实not定义一个自定义类。蟒蛇heapq模块具有直接作用于列表的函数。

通常，如上面的电子邮件示例所示，元素将从空堆开始一个接一个地插入到堆中。然而，如果已经有一个需要成为堆的元素列表，那么Pythonheapq模块包括heapify()用于将列表转换为有效的堆。

下面的代码使用heapify()转动a变成一个堆:

>>> import heapq
>>> a = [3, 5, 1, 2, 6, 8, 7]
>>> heapq.heapify(a)
>>> a
[1, 2, 3, 5, 6, 8, 7]

你可以检查一下，即使7紧接着8， 列表a仍然遵守堆性质。例如，a[2]，即3， 小于a[2*2 + 2]，即7.

如你看到的，heapify()就地修改列表但不对其进行排序。不必对堆进行排序即可满足堆属性。然而，由于每个排序列表做满足堆属性，运行heapify()在排序列表上不会改变列表中元素的顺序。

Python中的其他基本操作heapq模块假设列表已经是一个堆。值得注意的是，空列表或长度为 1 的列表始终是堆。

由于树的根是第一个元素，因此不需要专用函数来无损读取最小元素。第一个元素，a[0]，将永远是最小的元素。

为了在保留堆属性的同时弹出最小元素，Pythonheapq模块定义heappop().

使用方法如下heappop()弹出一个元素：

>>> import heapq
>>> a = [1, 2, 3, 5, 6, 8, 7]
>>> heapq.heappop(a)
1
>>> a
[2, 5, 3, 7, 6, 8]

该函数返回第一个元素，1，并保留堆属性a。例如，a[1]是5和a[1*2 + 2]是6.

蟒蛇heapq模块还包括heappush()用于将元素推入堆，同时保留堆属性。

以下示例显示将值推送到堆：

>>> import heapq
>>> a = [2, 5, 3, 7, 6, 8]
>>> heapq.heappush(a, 4)
>>> a
[2, 5, 3, 7, 6, 8, 4]
>>> heapq.heappop(a)
2
>>> heapq.heappop(a)
3
>>> heapq.heappop(a)
4

推后4到堆中，您从中弹出三个元素。自从2和3已经在堆中并且小于4，它们首先被弹出。

蟒蛇heapq模块还定义了另外两个操作：

1. heapreplace()相当于heappop()其次是heappush().
2. heappushpop()相当于heappush()其次是heappop().

这些在某些算法中很有用，因为它们比单独执行这两个操作更有效。

高级别操作

由于优先级队列经常用于合并排序序列，Pythonheapq模块有现成的功能，merge()，用于使用堆来合并多个可迭代对象。merge()假设其输入迭代已经排序并返回迭代器，不是一个列表。

作为使用的示例merge()，这是前面描述的电子邮件调度程序的实现：

import datetime
import heapq

def email(frequency, details):
    current = datetime.datetime.now()
    while True:
        current += frequency
        yield current, details

fast_email = email(datetime.timedelta(minutes=15), "fast email")
slow_email = email(datetime.timedelta(minutes=40), "slow email")

unified = heapq.merge(fast_email, slow_email)

输入到merge()在这个例子中是无限发电机。返回值分配给多变的 unified也是一个无限迭代器。该迭代器将生成要按照未来时间戳的顺序发送的电子邮件。

要调试并确认代码合并正确，您可以打印要发送的前十封电子邮件：

>>> for _ in range(10):
...    print(next(element))
(datetime.datetime(2020, 4, 12, 21, 27, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 21, 42, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 21, 52, 20, 305360), 'slow email')
(datetime.datetime(2020, 4, 12, 21, 57, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 22, 12, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 22, 27, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 22, 32, 20, 305360), 'slow email')
(datetime.datetime(2020, 4, 12, 22, 42, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 22, 57, 20, 305358), 'fast email')
(datetime.datetime(2020, 4, 12, 23, 12, 20, 305358), 'fast email')

请注意如何fast email安排在每15分钟，slow email安排在每40，并且电子邮件正确交错，以便它们按照时间戳的顺序排列。

merge()不会读取所有输入，而是动态工作。尽管两个输入都是无限迭代器，但打印前十项很快就会完成。

以类似的方式，当用于合并排序序列（如按时间戳排列的日志文件行）时，即使日志很大，这也会占用合理的内存量。

顶层代码

现在您已经了解了该算法，是时候编写代码来实现它了。在实现算法本身之前，编写一些支持代码很有用。

首先，您需要进口蟒蛇heapq模块：

import heapq

您将使用 Python 中的函数heapq模块来维护一个堆，这将帮助您在每次迭代时找到具有最短已知路径的位置。

下一步是将地图定义为代码中的变量：

map = """\
.......X..
.......X..
....XXXX..
..........
..........
"""

地图是一个三引号字符串显示机器人可以移动的区域以及任何障碍物。

虽然更现实的场景是让您从文件中读取地图，但出于教学目的，使用这个简单的地图在代码中定义变量会更容易。该代码适用于任何地图，但在简单地图上更容易理解和调试。

该映射经过优化，易于代码读者理解。点（.）足够轻，看起来很空，但它的优点是可以显示允许区域的尺寸。这X位置标记机器人无法通过的障碍。

支持代码

第一个函数会将地图转换为更易于在代码中解析的内容。parse_map()获取地图并对其进行分析：

def parse_map(map):
    lines = map.splitlines()
    origin = 0, 0
    destination = len(lines[-1]) - 1, len(lines) - 1
    return lines, origin, destination

该函数接受一个映射并返回一个包含三个元素的元组：

1. 的列表lines
2. 这origin
3. 这destination

这使得代码的其余部分可以在为计算机设计的数据结构上运行，而不是为人类视觉扫描的能力而设计。

名单lines可以通过索引(x, y)坐标。表达方式lines[y][x]将位置值返回为两个字符之一：

1. 一个点（".")表示该位置是一个空白区域。
2. 这封信"X"表示该位置是障碍物。

当您想要查找机器人可以占据哪些位置时，这将很有用。

功能is_valid()计算是否给定(x, y)位置有效：

def is_valid(lines, position):
    x, y = position
    if not (0 <= y < len(lines) and 0 <= x < len(lines[y])):
        return False
    if lines[y][x] == "X":
        return False
    return True

该函数有两个参数：

1. lines是作为线条列表的地图。
2. position是作为整数二元组进行检查的位置，指示(x, y)坐标。

为了有效，位置必须位于地图边界内并且没有障碍物。

该函数检查y通过检查长度是否有效lines列表。该函数接下来检查x确保它在内部是有效的lines[y]。最后，现在您知道两个坐标都在地图内，代码通过查看该位置的角色并将该角色与"X".

另一个有用的帮手是get_neighbors()，它找到一个位置的所有邻居：

def get_neighbors(lines, current):
    x, y = current
    for dx in [-1, 0, 1]:
        for dy in [-1, 0, 1]:
            if dx == 0 and dy == 0:
                continue
            position = x + dx, y + dy
            if is_valid(lines, position):
                yield position

该函数返回当前位置周围的所有有效位置。

get_neighbors()小心避免将某个位置识别为自己的邻居，但它确实允许对角邻居。这就是为什么至少有一个dx和dy一定不能为零，但两者都非零也可以。

最终的辅助函数是get_shorter_paths()，它找到较短的路径：

def get_shorter_paths(tentative, positions, through):
    path = tentative[through] + [through]
    for position in positions:
        if position in tentative and len(tentative[position]) <= len(path):
            continue
        yield position, path

get_shorter_paths()产生具有以下路径的位置through因为它的最后一步比当前已知的路径短。

get_shorter_paths()有三个参数：

1. tentative是将位置映射到已知最短路径的字典。
2. positions是您想要缩短路径的位置的迭代。
3. through是一个位置，通过该位置，也许有一条更短的路径可以到达positions可以被找寻到。

假设中的所有元素positions可以一步到达through.

功能get_shorter_paths()检查是否使用through因为最后一步将为每个位置提供更好的路径。如果没有已知的到达某个位置的路径，那么任何路径都会更短。如果存在已知路径，则仅在其长度较短的情况下生成新路径。为了使APIget_shorter_paths()更容易使用，一部分yield也是较短的路径。

所有辅助函数都被编写为纯函数，这意味着它们不修改任何数据结构，只返回值。这使得更容易遵循核心算法，该算法完成所有数据结构更新。

核心算法代码

回顾一下，您正在寻找起点和目的地之间的最短路径。

您保留三部分数据：

1. certain是某些位置的集合。
2. candidates是候选人堆。
3. tentative是将节点映射到当前最短已知路径的字典。

一个位置是在certain如果您可以确定最短的已知路径是最短的可能路径。如果目的地是在certain设置，那么到达目的地的最短已知路径无疑是最短可能路径，并且您可以返回该路径。

堆的candidates按已知最短路径的长度组织，并借助 Python 中的函数进行管理heapq模块。

在每一步中，您都会查看具有最短已知路径的候选者。这是堆被弹出的地方heappop()。没有到达该候选者的更短路径——所有其他路径都经过candidates，而且所有这些都更长。因此，当前候选人可以被标记为certain.

然后，您查看所有尚未访问过的邻居，如果遍历当前节点是一种改进，那么您将它们添加到candidates堆使用heappush().

功能find_path()实现这个算法：

 1def find_path(map):
 2    lines, origin, destination = parse_map(map)
 3    tentative = {origin: []}
 4    candidates = [(0, origin)]
 5    certain = set()
 6    while destination not in certain and len(candidates) > 0:
 7        _ignored, current = heapq.heappop(candidates)
 8        if current in certain:
 9            continue
10        certain.add(current)
11        neighbors = set(get_neighbors(lines, current)) - certain
12        shorter = get_shorter_paths(tentative, neighbors, current)
13        for neighbor, path in shorter:
14            tentative[neighbor] = path
15            heapq.heappush(candidates, (len(path), neighbor))
16    if destination in tentative:
17        return tentative[destination] + [destination]
18    else:
19        raise ValueError("no path")

find_path()收到一个map作为字符串，并以位置列表的形式返回从起点到目的地的路径。

这个函数有点长而且复杂，所以让我们一次一点地看一下：

• 第 2 至 5 行设置循环将查看和更新​​的变量。您已经知道从原点到自身的路径，该路径是长度为 0 的空路径。

• 6号线定义循环的终止条件。如果没有candidates，则无法缩短路径。如果destination是在certain，那么路径destination不能再短了。

• 第 7 行至第 10 行让候选人使用heappop()，如果已经存在则跳过循环certain，否则将候选人添加到certain。这确保每个候选者最多被循环处理一次。

• 第 11 至 15 行使用get_neighbors()和get_shorter_paths()找到到相邻位置的更短路径并更新tentative字典和candidates堆。

• 第 16 至 19 行处理返回正确的结果。如果找到路径，该函数将返回它。虽然计算路径没有最终位置使算法的实现更简单，这是一个更好的 API 来返回它和目的地。如果未找到路径，则会引发异常。

将函数分成单独的部分可以让您一次理解它的一部分。

可视化代码

如果该算法实际上由机器人使用，那么机器人可能会通过它应该经过的位置列表表现得更好。然而，为了使结果对人类来说更好看，最好将它们可视化。

show_path()在地图上绘制一条路径：

def show_path(path, map):
    lines = map.splitlines()
    for x, y in path:
        lines[y] = lines[y][:x] + "@" + lines[y][x + 1 :]
    return "\n".join(lines) + "\n"

该函数采用path和map作为参数。它返回一个新的映射，其路径由 at 符号 ("@").

运行代码

最后，您需要调用函数。这可以从Python交互式解释器.

以下代码将运行该算法并显示漂亮的输出：

>>> path = find_path(map)
>>> print(show_path(path, map))
@@.....X..
..@....X..
...@XXXX..
....@@@@@.
.........@

首先你得到最短路径find_path()。然后你把它传递给show_path()渲染带有标记路径的地图。最后，你print()映射到标准输出。

路径向右移动一步，然后向右下角移动几步，然后向右移动几步，最后以向右下角的对角一步结束。

恭喜！您已经使用 Python 解决了一个问题heapq模块。

这些类型的寻路问题可以通过动态规划和优先级队列的组合来解决，在以下领域很常见：工作面试和编程挑战。例如，2019 年代码问世包含一个问题这可以通过此处描述的技术来解决。