如何基于PyDot流程与DataFrame创建带时长热力图？

核心思路 #

通过DataFrame计算节点执行耗时和节点间流转耗时，将耗时映射到热力色阶（比如冷色对应短耗时、暖色对应长耗时），再把颜色赋值给PyDot的节点（填充色对应节点自身耗时）和边（颜色对应节点间流转耗时），替代单纯的边框颜色修改，实现真实热力效果。

1. 处理DataFrame，计算耗时 #

假设你的DataFrame结构如下（示例）：

先计算两个关键耗时：

• 节点自身执行耗时：task_duration = end_time - start_time（统一转成秒/分钟单位）
• 节点间流转耗时：transition_duration = next_task.start_time - current_task.end_time

代码示例：

import pandas as pd

# 示例DataFrame（替换成你实际的数据集）
df = pd.DataFrame({
    "task": ["task1", "task2", "task3"],
    "start_time": pd.to_datetime(["2024-05-01 10:00:00", "2024-05-01 10:03:00", "2024-05-01 10:05:45"]),
    "end_time": pd.to_datetime(["2024-05-01 10:02:30", "2024-05-01 10:05:15", "2024-05-01 10:08:00"])
})

# 计算节点执行耗时（转成秒）
df["task_duration"] = (df["end_time"] - df["start_time"]).dt.total_seconds()

# 计算节点间流转耗时（转成秒，最后一个节点无后续，填充0）
df["transition_duration"] = (df["start_time"].shift(-1) - df["end_time"]).dt.total_seconds().fillna(0)

2. 建立耗时到热力颜色的映射 #

用Matplotlib的色阶（推荐coolwarm：冷色短耗时、暖色长耗时），将耗时归一化后转成PyDot支持的十六进制颜色：

import matplotlib.pyplot as plt
import matplotlib.colors as mcolors

# 获取所有耗时值（节点+流转），用于统一归一化
all_durations = pd.concat([df["task_duration"], df["transition_duration"]]).values

# 创建归一化器，把耗时映射到0-1范围
norm = mcolors.Normalize(vmin=all_durations.min(), vmax=all_durations.max())
# 选择热力色阶
cmap = plt.cm.coolwarm

# 定义转换函数：输入耗时，返回十六进制颜色
def duration_to_color(duration):
    rgba = cmap(norm(duration))
    # 转成PyDot支持的#RRGGBB格式
    return mcolors.to_hex(rgba)

3. 修改PyDot流程图的颜色 #

假设你已经用PyDot创建了基础流程图，现在遍历节点和边，设置对应颜色：

处理节点（填充色对应自身耗时） #

import pydot

# 示例基础流程图（替换成你自己的创建代码）
graph = pydot.Dot(graph_type='digraph')
task1_node = pydot.Node("task1", label="Task 1")
task2_node = pydot.Node("task2", label="Task 2")
task3_node = pydot.Node("task3", label="Task 3")
graph.add_node(task1_node)
graph.add_node(task2_node)
graph.add_node(task3_node)
graph.add_edge(pydot.Edge(task1_node, task2_node))
graph.add_edge(pydot.Edge(task2_node, task3_node))

# 遍历DataFrame，给每个节点设置填充色
for idx, row in df.iterrows():
    task_name = row["task"]
    duration = row["task_duration"]
    color = duration_to_color(duration)
    # 找到对应节点并设置样式
    node = graph.get_node(task_name)[0]
    node.set_style("filled")  # 开启填充模式
    node.set_fillcolor(color)
    node.set_color("#333333")  # 边框设为深灰，突出填充色

处理边（颜色对应流转耗时） #

# 遍历流转关系，给边设置颜色
for idx, row in df.iterrows():
    if idx == len(df)-1:
        break  # 最后一个节点没有后续流转
    source_task = row["task"]
    target_task = df.iloc[idx+1]["task"]
    transition_duration = row["transition_duration"]
    color = duration_to_color(transition_duration)
    # 找到对应边并设置颜色
    edge = graph.get_edge(source_task, target_task)[0]
    edge.set_color(color)
    edge.set_penwidth("2")  # 加粗边，让颜色更醒目

4. 输出热力流程图 #

最后生成图片文件：

graph.write_png("process_heatmap.png")

内容的提问来源于stack exchange，提问作者Test