import os import pickle import glob def load_and_merge_pickle_chunks(folder_path, file_pattern="*.pkl"): """ 从指定文件夹中读取所有匹配的分片文件，解包并合并成一个总的数据列表。 Args: folder_path: 存放 .pkl 分片文件的文件夹路径 file_pattern: 文件匹配模式，默认匹配所有 .pkl 文件 """ all_episodes = [] # 获取所有匹配的 pkl 文件路径，并按名称排序（确保 part1, part2 顺序或逻辑清晰） search_path = os.path.join(folder_path, file_pattern) file_list = sorted(glob.glob(search_path)) if not file_list: print(f"❌ 未在路径 【{folder_path}】 下找到任何匹配 【{file_pattern}】 的文件！") return [] print(f"📂 找到 {len(file_list)} 个数据分片文件，开始加载...") for file_path in file_list: try: with open(file_path, 'rb') as f: # 每个分片解包出来都是一个 list [ep1, ep2, ...] chunk_data = pickle.load(f) if isinstance(chunk_data, list): all_episodes.extend(chunk_data) print(f" ✅ 成功加载: {os.path.basename(file_path)} (包含 {len(chunk_data)} 个 Episode)") else: print(f" ⚠️ 警告: {os.path.basename(file_path)} 解析出的数据格式不是列表，跳过。") except Exception as e: print(f" ❌ 读取文件 {os.path.basename(file_path)} 失败: {e}") print(f"整个序列加载完成，共合并了 {len(all_episodes)} 个 Episode。") return all_episodes def analyze_episodes_data(episode_data_raw): """ 分析 Episode 数据，统计超调情况。 """ total_episodes = len(episode_data_raw) if total_episodes == 0: print("没有数据可供分析。") return invalid_count = 0 # 最后一步误差绝对值 > 2 kPa 的无效 episode invalid_high_flow = 0 # 无效 episode 中流量 > 200 invalid_low_flow = 0 # 无效 episode 中流量 < 100 all_steady_abs_errors = [] # 所有有效 episode 的稳态误差（绝对值） no_overshoot_count = 0 no_overshoot_abs_errors = [] # 绝对值稳态误差 no_overshoot_raw_errors = [] # 带符号稳态误差（+ = 高于目标, - = 低于目标） overshoot_lt_1_count = 0 overshoot_1_to_2_count = 0 overshoot_2_to_3_count = 0 overshoot_3_to_4_count = 0 overshoot_4_to_5_count = 0 overshoot_5_to_10_count = 0 overshoot_gt_10_count = 0 overshoots_5_to_10 = [] overshoots_gt_10 = [] for idx, ep in enumerate(episode_data_raw): pressures = ep.get('pressures', []) target_p = ep.get('target_pressure', 0.0) if not pressures: continue # 最后一步误差绝对值 > 2 kPa → 无效 episode，跳过 errors = ep.get('errors', []) if errors and abs(errors[-1]) > 2: invalid_count += 1 q = ep.get('Q_in', 0) if q > 200: invalid_high_flow += 1 elif q < 100: invalid_low_flow += 1 continue initial_p = pressures[0] # 所有有效 episode 的稳态误差（最后 30 步绝对值均值） if errors: last_n = errors[-30:] if len(errors) >= 30 else errors all_steady_abs_errors.append(sum(abs(e) for e in last_n) / len(last_n)) is_step_up = target_p >= initial_p # 升压为 True，降压为 False overshoot = 0.0 if is_step_up: # 升压：最大值大于目标压力为超调 max_p = max(pressures) if max_p > target_p: overshoot = max_p - target_p else: # 降压：最小值小于目标压力为超调 min_p = min(pressures) if min_p < target_p: overshoot = target_p - min_p # 统计区间 if overshoot == 0: no_overshoot_count += 1 elif overshoot < 1.0: overshoot_lt_1_count += 1 # 最后 30 步的平均误差作为稳态误差（分别记录绝对值和带符号值） if len(errors) >= 30: last_30 = errors[-30:] elif errors: last_30 = errors else: last_30 = [] if last_30: no_overshoot_abs_errors.append(sum(abs(e) for e in last_30) / len(last_30)) no_overshoot_raw_errors.append(sum(last_30) / len(last_30)) elif 1.0 <= overshoot < 2.0: overshoot_1_to_2_count += 1 elif 2.0 <= overshoot < 3.0: overshoot_2_to_3_count += 1 elif 3.0 <= overshoot < 4.0: overshoot_3_to_4_count += 1 elif 4.0 <= overshoot <= 5.0: overshoot_4_to_5_count += 1 else: item = { "index": idx, "direction": "升压" if is_step_up else "降压", "initial_p": initial_p, "target_p": target_p, "overshoot_value": round(overshoot, 3), "Q_in": ep.get("Q_in", 0), } if overshoot <= 10.0: overshoot_5_to_10_count += 1 overshoots_5_to_10.append(item) else: overshoot_gt_10_count += 1 overshoots_gt_10.append(item) # 打印报告 def _pct(n): return f"{n / total_episodes * 100:.1f}%" print("\n" + "="*25 + " 离线数据分析 " + "="*25) valid_episodes = total_episodes - invalid_count print(f"合并后的总 Episode 数 : {total_episodes}") print(f" - 无效 Episode（末步误差>2）: {invalid_count} ({_pct(invalid_count)})") if invalid_count > 0: print(f" ├ 流量 > 200 L/min : {invalid_high_flow}") print(f" └ 流量 < 100 L/min : {invalid_low_flow}") print(f" - 有效 Episode 数 : {valid_episodes}") print(f" - 未超调的 Episode 数 : {no_overshoot_count} ({_pct(no_overshoot_count)})") print(f" - 超调 < 1 kPa : {overshoot_lt_1_count} ({_pct(overshoot_lt_1_count)})") print(f" - 超调在 1 ~ 2 kPa 之间 : {overshoot_1_to_2_count} ({_pct(overshoot_1_to_2_count)})") print(f" - 超调在 2 ~ 3 kPa 之间 : {overshoot_2_to_3_count} ({_pct(overshoot_2_to_3_count)})") print(f" - 超调在 3 ~ 4 kPa 之间 : {overshoot_3_to_4_count} ({_pct(overshoot_3_to_4_count)})") print(f" - 超调在 4 ~ 5 kPa 之间 : {overshoot_4_to_5_count} ({_pct(overshoot_4_to_5_count)})") print(f" - 超调在 5 ~ 10 kPa 之间 : {overshoot_5_to_10_count} ({_pct(overshoot_5_to_10_count)})") print(f" - 超调 > 10 kPa : {overshoot_gt_10_count} ({_pct(overshoot_gt_10_count)})") print("=" * 68) def _print_detail(title, items): if items: print(f"\n[⚠️ {title}]:") for item in items: print(f" * Episode [{item['index']}] ({item['direction']}): " f"初始 {item['initial_p']:.2f} -> 目标 {item['target_p']:.2f} | " f"超调量: {item['overshoot_value']:.2f} kPa | " f"流量: {item['Q_in']:.1f} L/min") _print_detail("超调在 5 ~ 10 kPa", overshoots_5_to_10) _print_detail("超调大于 10 kPa", overshoots_gt_10) if not overshoots_5_to_10 and not overshoots_gt_10: print("\n🎉 极好！没有发现超调大于 5 kPa 的数据。") # ---- 流量分布统计 ---- flow_bins = [ (0, 10), (10, 50), (50, 100), (100, 150), (150, 200), (200, 250), (250, 300), ] flow_counts = {f"{lo}~{hi}": 0 for lo, hi in flow_bins} flow_counts["300+"] = 0 for ep in episode_data_raw: q = ep.get('Q_in', 0) placed = False for lo, hi in flow_bins: if lo <= q < hi: flow_counts[f"{lo}~{hi}"] += 1 placed = True break if not placed: flow_counts["300+"] += 1 print(f"\n📊 流量分布统计 (共 {total_episodes} 个 Episode):") for lo, hi in flow_bins: label = f"{lo}~{hi}" print(f" {label:>10} L/min : {flow_counts[label]:>5} ({flow_counts[label]/total_episodes*100:5.1f}%)") print(f" {'300+':>10} L/min : {flow_counts['300+']:>5} ({flow_counts['300+']/total_episodes*100:5.1f}%)") if all_steady_abs_errors: avg_all = sum(all_steady_abs_errors) / len(all_steady_abs_errors) print(f"\n📊 所有有效 Episode 平均稳态误差（最后 30 步绝对值均值）: {avg_all:.3f} kPa" f" ({len(all_steady_abs_errors)} 个 Episode)") if no_overshoot_abs_errors: avg_abs = sum(no_overshoot_abs_errors) / len(no_overshoot_abs_errors) avg_raw = sum(no_overshoot_raw_errors) / len(no_overshoot_raw_errors) print(f"\n📊 超调0～1kpa Episode 平均稳态误差（最后 30 步）:") print(f" 绝对值均值 : {avg_abs:.3f} kPa") print(f" 带符号均值 : {avg_raw:.3f} kPa ({'偏高于目标' if avg_raw > 0 else '偏低' if avg_raw < 0 else '无偏'})" f" ({no_overshoot_count} 个 Episode)") def print_episode_detail(episode_data_raw, index): """打印指定 episode 的完整数据""" if index < 0 or index >= len(episode_data_raw): print(f"❌ Episode 索引 {index} 超出范围 (0~{len(episode_data_raw)-1})") return ep = episode_data_raw[index] print(f"\n{'='*60}") print(f" Episode [{index}] 完整数据") print(f"{'='*60}") for key in ['Q_in', 'volume', 'target_pressure', 'mode']: if key in ep: print(f" {key}: {ep[key]}") pressures = ep.get('pressures', []) errors = ep.get('errors', []) valve_openings = ep.get('valves', []) print(f"\n 步数: {len(pressures)}") if pressures: print(f" 初始压力: {pressures[0]:.2f} kPa") print(f" 最终压力: {pressures[-1]:.2f} kPa") print(f" 目标压力: {ep.get('target_pressure', 'N/A')} kPa") if errors: print(f" 最终误差: {errors[-1]:.3f} kPa") print(f"\n {'步':>4s} {'压力(kPa)':>10s} {'误差(kPa)':>10s} {'开度(%)':>8s}") print(f" {'-'*36}") n = len(pressures) for i in range(n): p = pressures[i] e = errors[i] if i < len(errors) else float('nan') vo = valve_openings[i] if i < len(valve_openings) else float('nan') print(f" {i:4d} {p:10.2f} {e:10.3f} {vo:8.2f}") print(f"{'='*60}\n") # --- 执行离线分析 --- if __name__ == "__main__": # 💡 数据存放文件夹路径 DATA_FOLDER = "/Users/menglingrui/Documents/DominatedConvergence/cloud_down_file/永久/data_8L" # 1. 读取并合并分片 merged_data = load_and_merge_pickle_chunks(DATA_FOLDER, file_pattern="*part*.pkl") # 2. 执行分析 if merged_data: analyze_episodes_data(merged_data) # 3. 找出无效 episode（末步误差绝对值 > 2 kPa），打印前 3 个的完整数据 # invalid_indices = [] # for idx, ep in enumerate(merged_data): # errors = ep.get('errors', []) # if errors and abs(errors[-1]) > 2: # invalid_indices.append(idx) # if len(invalid_indices) >= 3: # break # if invalid_indices: # print(f"\n找到 {len(invalid_indices)} 个无效 Episode，索引: {invalid_indices}") # for idx in invalid_indices: # print_episode_detail(merged_data, idx) # else: # print("\n未找到无效 Episode") print_episode_detail(merged_data, 2500)