我有一个问题。我想创建一个具有热力图的流程，以查看每个步骤所需的时间。
我使用
我该如何为我的流程创建一个热力图？
计算还应包括从一个步骤到另一个步骤的时间。 因此，您可以计算边缘时间，例如
我的MVP只会更改边框的颜色。但我想创建一个真正的热力图。
PyDot创建了流程，并为每个步骤创建了一个dataframe。我该如何为我的流程创建一个热力图？
计算还应包括从一个步骤到另一个步骤的时间。 因此，您可以计算边缘时间，例如
task1_start - start / task2_start - task1_end，
以及节点时间，例如 task1_end - task1_start / task2_end - task2_start。我的MVP只会更改边框的颜色。但我想创建一个真正的热力图。
进程
import pydot
from IPython.display import SVG
graph = pydot.Dot(graph_type='digraph')
task_node1 = pydot.Node("Task1", shape="box",)
task_node2 = pydot.Node("Task2", shape="box",)
graph.add_node(task_node1)
graph.add_node(task_node2)
task1_to_task2_edge = pydot.Edge("Task1", "Task2",)
graph.add_edge(task1_to_task2_edge)
graph.write_svg("diagram.svg")
SVG('diagram.svg')
Dataframe
id step timestamp
0 1 task1_start 2023-01-01
1 1 task1_End 2023-01-05
2 1 task2_start 2023-01-10
3 1 task2_end 2023-01-12
4 2 task1_start 2023-01-01
5 2 task1_End 2023-01-05
6 2 task2_start 2023-01-10
7 2 task2_end 2023-01-12
MVP
import pandas as pd
d = {'id': [1, 1, 1, 1,
2, 2, 2, 2,],
'step': ['task1_start', 'task1_End', 'task2_start', 'task2_end',
'task1_start', 'task1_End', 'task2_start', 'task2_end',],
'timestamp': ['2023-01-01', '2023-01-05', '2023-01-10', '2023-01-12',
'2023-01-01', '2023-01-05', '2023-01-10', '2023-01-12',]}
df = pd.DataFrame(data=d,)
df['timestamp'] = pd.to_datetime(df['timestamp'])
g = df.groupby('id')
out = (df
.assign(duration=df['timestamp'].sub(g['timestamp'].shift()),
step=lambda d: (df['step']+'/'+g['step'].shift()).str.replace(
r'([^_]+)[^/]*/([^_]+)[^/]*',
lambda m: m.group(1) if m.group(1)==m.group(2) else f"{m.group(2)}_to_{m.group(1)}",
regex=True)
)
[['id', 'step', 'duration']].dropna(subset=['duration'])
)
df = out
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
colors = mcolors.LinearSegmentedColormap.from_list(
'LightBlueGreenYellowRed', ['#B0E0E6', '#87CEEB', '#00FF00', '#ADFF2F', '#FFFF00', '#FFD700', '#FFA500', '#FF4500', '#FF0000', '#FF6347', '#FF7F50', '#FFA07A', '#FFC0CB', '#FFB6C1', '#FF69B4', '#DB7093', '#FF1493', '#C71585', '#FF00FF']
)
def get_color(value, vmin, vmax):
norm = (value - vmin) / (vmax - vmin)
cmap = colors(norm)
return mcolors.to_hex(cmap)
vmin = df['duration'].min()
vmax = df['duration'].max()
df['color'] = df['duration'].apply(lambda x: get_color(x, vmin, vmax))
def get_color(id):
if (df['step'] == id).any():
color = df.loc[df['step'] == id, 'color'].values[0]
if pd.isnull(color):
return '#808080'
else:
return color
else:
return '#808080'
import pydot
from IPython.display import SVG
graph = pydot.Dot(graph_type='digraph')
task_node1 = pydot.Node("Task1", shape="box", color = get_color('task1'))
task_node2 = pydot.Node("Task2", shape="box", color = get_color('task2'))
graph.add_node(task_node1)
graph.add_node(task_node2)
task1_to_task2_edge = pydot.Edge("Task1", "Task2", color = get_color('task1_to_task2'))
graph.add_edge(task1_to_task2_edge)
graph.write_svg("diagram.svg")
SVG('diagram.svg')
