日志解析与多时间维度设备状态可视化实现技术咨询

核心优化思路 #

• 数据预处理优先：清理无效记录、去重并按时间排序，确保后续时段合并逻辑正确
• 用时段集合替代固定数组：避免硬编码时间维度（比如原代码的60分钟模板），直接维护运行时段的起始/结束时间，最小化内存占用
• 直接处理datetime对象：利用Pandas的datetime类型原生支持跨年、月、日的时间计算，无需拆分维度

步骤拆解与代码实现 #

1. 数据预处理 #

先清理日志中的异常数据（比如结束时间早于开始时间的错误记录），并对重复测试记录去重，最后按开始时间排序：

def preprocess_data(self):
    # 过滤结束时间晚于开始时间的有效记录
    self.df = self.df[self.df['End_time'] > self.df['Start_time']]
    # 去重：同一SerialNumber的相同测试时段只保留一条
    self.df = self.df.drop_duplicates(subset=['SerialNumber', 'Start_time', 'End_time'])
    # 按开始时间全局排序，确保相邻时段可以正确计算间隔
    self.df = self.df.sort_values('Start_time').reset_index(drop=True)

2. 合并符合条件的运行时段 #

遍历排序后的记录，合并单设备时段以及间隔≤80秒的相邻时段：

def merge_running_periods(self):
    if self.df.empty:
        return []
    
    running_periods = []
    # 初始化第一个时段
    current_start = self.df.iloc[0]['Start_time']
    current_end = self.df.iloc[0]['End_time']
    
    for idx in range(1, len(self.df)):
        next_start = self.df.iloc[idx]['Start_time']
        next_end = self.df.iloc[idx]['End_time']
        time_gap = (next_start - current_end).total_seconds()
        
        # 间隔≤80秒则合并时段
        if time_gap <= self.define_time_gap:
            current_end = max(current_end, next_end)
        else:
            # 保存当前时段，开始新的时段
            running_periods.append((current_start, current_end))
            current_start = next_start
            current_end = next_end
    # 保存最后一个时段
    running_periods.append((current_start, current_end))
    return running_periods

3. 绘制状态时间图 #

用Matplotlib嵌入Tkinter，绘制时间轴上的运行/空闲色块。运行状态为绿色，空闲为蓝色：

def show_graph(self):
    # 预处理数据
    self.preprocess_data()
    running_periods = self.merge_running_periods()
    
    if not running_periods:
        print("无有效运行数据")
        return
    
    # 准备绘图数据：计算总时间范围
    all_times = [t for period in running_periods for t in period]
    start_total = min(all_times)
    end_total = max(all_times)
    total_duration = (end_total - start_total).total_seconds() / 3600  # 转为小时
    
    # 计算空闲时段：总时间减去运行时段
    idle_periods = []
    prev_end = start_total
    for run_start, run_end in running_periods:
        if run_start > prev_end:
            idle_periods.append((prev_end, run_start))
        prev_end = run_end
    if prev_end < end_total:
        idle_periods.append((prev_end, end_total))
    
    # 嵌入Matplotlib到Tkinter
    from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
    import matplotlib.pyplot as plt
    
    fig, ax = plt.subplots(figsize=(7, 2))
    
    # 绘制运行时段（绿色）
    for run_start, run_end in running_periods:
        run_hours_start = (run_start - start_total).total_seconds() / 3600
        run_duration = (run_end - run_start).total_seconds() / 3600
        ax.barh(0, run_duration, left=run_hours_start, color='green', label='运行' if run_start == running_periods[0][0] else "")
    
    # 绘制空闲时段（蓝色）
    for idle_start, idle_end in idle_periods:
        idle_hours_start = (idle_start - start_total).total_seconds() / 3600
        idle_duration = (idle_end - idle_start).total_seconds() / 3600
        ax.barh(0, idle_duration, left=idle_hours_start, color='blue', label='空闲' if idle_start == idle_periods[0][0] else "")
    
    # 设置坐标轴
    ax.set_yticks([0])
    ax.set_yticklabels(['设备状态'])
    ax.set_xlabel(f'时间（小时）\n起始时间：{start_total.strftime("%Y-%m-%d %H:%M:%S")}')
    ax.set_xlim(0, total_duration)
    ax.legend()
    
    # 嵌入到Tkinter窗口
    canvas = FigureCanvasTkAgg(fig, master=self.function_frame)
    canvas.draw()
    canvas.get_tk_widget().pack()

完整修改后的代码 #

from tkinter import *
import pandas as pd
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import matplotlib.pyplot as plt

class gui():
    def __init__(self, container):
        self.container = container
        self.container.title('Machine monitor')
        self.container.geometry('800x600')
        self.df = pd.read_csv("full_log.txt", parse_dates=['Start_time', 'End_time'])
        self.define_time_gap = 80

    def set_up(self):
        self.function_frame = Frame(self.container, borderwidth=2, height="300", width="600",
                                    highlightbackground="black",
                                    highlightcolor="red", highlightthickness=1)
        self.function_frame.pack()
        self.show_chart = Button(self.function_frame, text='show chart',
                                command=self.show_graph)
        self.show_chart.pack()

    def preprocess_data(self):
        # 过滤无效记录（结束时间早于开始时间）
        self.df = self.df[self.df['End_time'] > self.df['Start_time']]
        # 去重：同一设备的相同测试时段只保留一条
        self.df = self.df.drop_duplicates(subset=['SerialNumber', 'Start_time', 'End_time'])
        # 按开始时间全局排序
        self.df = self.df.sort_values('Start_time').reset_index(drop=True)

    def merge_running_periods(self):
        if self.df.empty:
            return []
        
        running_periods = []
        current_start = self.df.iloc[0]['Start_time']
        current_end = self.df.iloc[0]['End_time']
        
        for idx in range(1, len(self.df)):
            next_start = self.df.iloc[idx]['Start_time']
            next_end = self.df.iloc[idx]['End_time']
            time_gap = (next_start - current_end).total_seconds()
            
            if time_gap <= self.define_time_gap:
                current_end = max(current_end, next_end)
            else:
                running_periods.append((current_start, current_end))
                current_start = next_start
                current_end = next_end
        running_periods.append((current_start, current_end))
        return running_periods

    def show_graph(self):
        self.preprocess_data()
        running_periods = self.merge_running_periods()
        
        if not running_periods:
            print("无有效运行数据")
            return
        
        # 计算总时间范围
        all_times = [t for period in running_periods for t in period]
        start_total = min(all_times)
        end_total = max(all_times)
        total_duration = (end_total - start_total).total_seconds() / 3600
        
        # 生成空闲时段
        idle_periods = []
        prev_end = start_total
        for run_start, run_end in running_periods:
            if run_start > prev_end:
                idle_periods.append((prev_end, run_start))
            prev_end = run_end
        if prev_end < end_total:
            idle_periods.append((prev_end, end_total))
        
        # 绘图
        fig, ax = plt.subplots(figsize=(7, 2))
        
        # 绘制运行状态
        for idx, (run_start, run_end) in enumerate(running_periods):
            run_hours_start = (run_start - start_total).total_seconds() / 3600
            run_duration = (run_end - run_start).total_seconds() / 3600
            ax.barh(0, run_duration, left=run_hours_start, color='green', label='运行' if idx == 0 else "")
        
        # 绘制空闲状态
        for idx, (idle_start, idle_end) in enumerate(idle_periods):
            idle_hours_start = (idle_start - start_total).total_seconds() / 3600
            idle_duration = (idle_end - idle_start).total_seconds() / 3600
            ax.barh(0, idle_duration, left=idle_hours_start, color='blue', label='空闲' if idx == 0 else "")
        
        # 美化图表
        ax.set_yticks([0])
        ax.set_yticklabels(['设备状态'])
        ax.set_xlabel(f'时间（小时）\n起始时间：{start_total.strftime("%Y-%m-%d %H:%M:%S")}')
        ax.set_xlim(0, total_duration)
        ax.legend(loc='upper right')
        
        # 嵌入到Tkinter
        canvas = FigureCanvasTkAgg(fig, master=self.function_frame)
        canvas.draw()
        canvas.get_tk_widget().pack()

if __name__ == "__main__":
    mainview = Tk()
    interface = gui(mainview)
    interface.set_up()
    mainview.mainloop()

关键说明 #

• 不再依赖固定长度的数组，而是用时段元组列表存储运行状态，完全适配任意时间维度（年、月、日、小时都无需额外处理）
• 预处理步骤过滤了日志中的错误记录（比如2023-12-30开始但2023-08-30结束的无效数据），保证数据有效性
• 绘图时将绝对时间转换为相对小时数，避免时间轴显示混乱，同时保留起始时间标注

内容的提问来源于stack exchange，提问作者Hoang Minh Nguyen