使用Matplotlib、Pandas及Python绘制类broken_barh的事件状态水平图
基于Matplotlib Broken Barh实现带时间窗口的信号状态时间图
数据预处理流程
- 时间字段格式转换:将
timestamp、prev_timestamp统一通过pd.to_datetime()转成pandas原生datetime类型,适配示例数据的纳秒精度时间解析需求。 - 展示信号筛选:提取所有历史上出现过
error/warning状态的信号列表,过滤无异常记录的信号,避免无关数据冗余:# 提取符合展示要求的信号集合 qualified_signals = df.loc[df['value'].isin(['error', 'warning']), 'signal'].unique() # 仅保留这些信号的全量事件进入绘图数据集 df_plot = df[df['signal'].isin(qualified_signals)].copy() - 条带参数计算:
broken_barh的核心入参为每个条带的(起始位置, 宽度),需要注意原逻辑的差值方向错误:prev_timestamp是当前事件之前的事件时间,直接用prev_timestamp - timestamp会得到负的条带宽度,导致绘制位置偏移。正确的持续时间计算逻辑为:- 将绘图数据集按
signal分组、组内按timestamp正序排列 - 取同信号下当前事件的下一个事件时间作为当前状态的结束时间,无后续事件时用数据集全局最大时间兜底
- 计算条带宽度为结束时间减去当前事件发生时间,仅保留需要绘制的非ok状态行
df_plot = df_plot.sort_values(['signal', 'timestamp']).reset_index(drop=True) df_plot['state_end'] = df_plot.groupby('signal')['timestamp'].shift(-1) df_plot['state_end'] = df_plot['state_end'].fillna(df_plot['timestamp'].max()) # 过滤掉不需要展示的ok状态 df_plot = df_plot[df_plot['value'].isin(['error', 'warning'])] - 将绘图数据集按
绘图参数配置
- Y轴位置映射:为每个合格信号分配独立的Y轴行位置,行高固定为0.8,避免相邻信号的条带视觉重叠:
signal_y_map = {sig: idx for idx, sig in enumerate(qualified_signals)} - 条带颜色匹配:按给定规则绑定状态值和对应颜色:
colors = {'error': 'red', 'warning': 'yellow'} df_plot['bar_color'] = df_plot['value'].map(colors)
核心绘图逻辑
- 时间格式适配:Matplotlib的broken_barh不直接支持datetime类型作为X轴数值,需要通过
matplotlib.dates.date2num()将时间转成可识别的浮点数格式:import matplotlib.pyplot as plt import matplotlib.dates as mdates fig, ax = plt.subplots(figsize=(12, 6)) # 逐行绘制每个状态对应的水平条带 for _, row in df_plot.iterrows(): y_pos = signal_y_map[row['signal']] bar_start = mdates.date2num(row['timestamp']) bar_width = mdates.date2num(row['state_end'] - row['timestamp']) ax.broken_barh( xranges=[(bar_start, bar_width)], yrange=(y_pos - 0.4, 0.8), color=row['bar_color'], edgecolor='white' ) - 坐标轴格式优化:
# Y轴绑定信号名称 ax.set_yticks(list(signal_y_map.values())) ax.set_yticklabels(list(signal_y_map.keys())) # X轴时间格式适配,自动调整标签角度避免重叠 ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d %H:%M:%S')) fig.autofmt_xdate() ax.set_ylim(-0.5, len(qualified_signals)-0.5) ax.grid(axis='x', linestyle='--', alpha=0.3)
时间滑动窗口实现
- 基于Matplotlib内置的
RangeSlider组件实现时间范围滑动选择,不需要额外开发交互逻辑:from matplotlib.widgets import RangeSlider # 为滑块组件预留绘图区域 fig.subplots_adjust(bottom=0.2) slider_ax = fig.add_axes([0.2, 0.05, 0.6, 0.03]) # 滑块范围匹配数据集的时间跨度 time_start = mdates.date2num(df_plot['timestamp'].min()) time_end = mdates.date2num(df_plot['timestamp'].max()) time_slider = RangeSlider( ax=slider_ax, label="时间窗口", valmin=time_start, valmax=time_end, valinit=(time_start, time_end) ) # 滑块联动更新X轴显示范围 def update_display_window(val): ax.set_xlim(val[0], val[1]) fig.canvas.draw_idle() time_slider.on_changed(update_display_window)
注意事项:
- 示例数据中第三行的warning值和prev_timestamp之间缺少逗号,加载CSV时需要先做基础数据清洗
- 当数据集规模达到数千条时,逐行循环绘制的性能可以满足需求,不需要额外优化
内容的提问来源于stack exchange,提问作者Dennis Delin
相关产品推荐
相关产品推荐

