如何用Python/Pandas生成设备完整24小时运行状态时间区间？

我来帮你搞定这个设备运行时间线生成的问题！咱们一步步来，用Pandas就能完美实现从故障明细补全天24小时的working/pause状态区间。

核心思路 #

咱们的目标是把每个「工作站-设备-日期」组合的故障停机区间，和它们之间（以及当天首尾）的正常运行区间拼接起来，形成覆盖全天的完整时间线。具体来说要做这几件事：

• 先把输入数据按工作站、设备、日期分组，每组内按故障开始时间排序
• 给每组补上「当天开始到第一个故障」「故障之间」「最后一个故障到当天结束」的working区间
• 确保时间和对应的start_no/end_no完全匹配示例格式

具体实现步骤 #

1. 数据预处理 #

首先读取并清洗数据，把时间列转成datetime类型，方便后续计算：

import pandas as pd

# 读取你的样本数据（假设已保存为csv文件）
df = pd.read_csv('your_sample_data.csv')

# 转换时间列为datetime格式，这是后续时间计算的基础
df['start_date'] = pd.to_datetime(df['start_date'])
df['end_date'] = pd.to_datetime(df['end_date'])

# 提取日期字段，用于按天分组
df['date'] = df['start_date'].dt.date

# 按工作站、设备、日期分组，每组内按故障开始时间排序
grouped = df.groupby(['Stat.', 'Mac.', 'date'], group_keys=False)

2. 处理重叠故障区间（可选但建议） #

如果你的数据里存在故障区间重叠的情况，先合并这些重叠区间，避免生成重复的时段：

def merge_overlapping_pauses(group):
    # 先按故障开始时间排序
    sorted_group = group.sort_values('start_date').reset_index(drop=True)
    merged_intervals = []
    
    for _, row in sorted_group.iterrows():
        if not merged_intervals:
            merged_intervals.append(row.to_dict())
        else:
            last_interval = merged_intervals[-1]
            # 如果当前故障和上一个重叠或相邻，就合并
            if row['start_date'] <= last_interval['end_date']:
                last_interval['end_date'] = max(last_interval['end_date'], row['end_date'])
                last_interval['end_no'] = max(last_interval['end_no'], row['end_no'])
            else:
                merged_intervals.append(row.to_dict())
    
    return pd.DataFrame(merged_intervals)

# 应用合并函数，得到去重后的故障区间
df_merged = grouped.apply(merge_overlapping_pauses).reset_index(drop=True)
# 重新分组
grouped_merged = df_merged.groupby(['Stat.', 'Mac.', 'date'], group_keys=False)

3. 生成完整时间线 #

定义一个函数，给每个分组补全所有working区间：

def build_full_daily_timeline(group):
    # 确定当天的起始和结束时间（到23:59）
    day_start = pd.Timestamp(group['date'].iloc[0])
    day_end = day_start + pd.Timedelta(days=1) - pd.Timedelta(minutes=1)
    
    timeline = []
    
    # 处理当天开始到第一个故障的working时段
    first_fault_start = group['start_date'].iloc[0]
    if first_fault_start > day_start:
        # 计算对应的start_no和end_no（每分钟一个编号，0对应00:00）
        start_no = 0
        end_no = int((first_fault_start - day_start).total_seconds() // 60)
        timeline.append({
            'Stat.': group['Stat.'].iloc[0],
            'Mac.': group['Mac.'].iloc[0],
            'start_date': day_start,
            'end_date': first_fault_start,
            'start_no': start_no,
            'end_no': end_no,
            'status': 'working'
        })
    
    # 遍历每个故障区间，插入前后的working时段
    for i in range(len(group)):
        # 添加当前故障区间
        timeline.append(group.iloc[i].to_dict())
        
        # 如果不是最后一个故障，添加当前故障结束到下一个故障开始的working时段
        if i < len(group) - 1:
            current_fault_end = group['end_date'].iloc[i]
            next_fault_start = group['start_date'].iloc[i+1]
            if current_fault_end < next_fault_start:
                start_no = group['end_no'].iloc[i]
                end_no = group['start_no'].iloc[i+1]
                timeline.append({
                    'Stat.': group['Stat.'].iloc[0],
                    'Mac.': group['Mac.'].iloc[0],
                    'start_date': current_fault_end,
                    'end_date': next_fault_start,
                    'start_no': start_no,
                    'end_no': end_no,
                    'status': 'working'
                })
    
    # 处理最后一个故障到当天结束的working时段
    last_fault_end = group['end_date'].iloc[-1]
    if last_fault_end < day_end:
        start_no = group['end_no'].iloc[-1]
        end_no = int((day_end - day_start).total_seconds() // 60)
        timeline.append({
            'Stat.': group['Stat.'].iloc[0],
            'Mac.': group['Mac.'].iloc[0],
            'start_date': last_fault_end,
            'end_date': day_end,
            'start_no': start_no,
            'end_no': end_no,
            'status': 'working'
        })
    
    return pd.DataFrame(timeline)

4. 应用函数并输出结果 #

把函数应用到每个分组，得到最终的完整时间线：

# 生成完整时间线
full_timeline = grouped_merged.apply(build_full_daily_timeline).reset_index(drop=True)

# 调整列顺序，和你的示例格式一致
full_timeline = full_timeline[['Stat.', 'Mac.', 'start_date', 'end_date', 'start_no', 'end_no', 'status']]

# 可以保存为csv查看结果
full_timeline.to_csv('full_device_timeline.csv', index=False)

关键细节说明 #

• start_no和end_no是按每分钟计算的，0对应00:00，1对应00:01，1439对应23:59，完全匹配你的示例规则。
• 如果某个「工作站-设备-日期」当天没有故障，函数会自动生成一个从00:00到23:59的working区间。
• 合并重叠区间的步骤是可选的，但如果你的数据存在故障时间重叠，一定要加上，否则会出现逻辑错误。

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