# -*- coding: utf-8 -*- """ run_diagnosis.py: PLC数据实时诊断脚本 功能描述: 1. 从API逐个查询最近40分钟的传感器历史数据（按秒采样） 2. 降采样到4秒间隔，合并成DataFrame 3. 调用WaterPlantDiagnoser进行诊断 4. 将诊断结果追加到output_format.txt 使用方式: python run_diagnosis.py # 从API查询数据 python run_diagnosis.py --local # 使用本地CSV测试 数据规格: - 查询时长: 40分钟 = 2400秒 - 采样间隔: 每4秒一个点 - 时间点数: 2400秒 ÷ 4秒 = 600 行 - PLC点位: 158 列 依赖: - test.py中的WaterPlantDiagnoser类 - 模型文件 models/ppo_tracing_model.pth - 阈值表 sensor_threshold.xlsx """ import os import json import requests import pandas as pd import numpy as np from datetime import datetime, timedelta import logging from config import config # 日志配置 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger(__name__) BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # ===================== API配置 ===================== # 历史数据查询API，与loop_main.py中的API保持一致 API_BASE_URL = "http://120.55.44.4:8900/api/v1/jinke-cloud/db/device/history-data" # 请求头配置（与loop_main.py保持一致） HEADERS = { "Content-Type": "application/json", "JWT-TOKEN": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJJRCI6NywiVXNlcm5hbWUiOiJhZG1pbiIsIkRlcCI6IjEzNSIsImV4cCI6MTc3NjExOTExNCwiaXNzIjoiZ2luLWJsb2cifQ.0HTtzHZjyd2mHo8VCy8icYROxmntRMuQhyoZsAYRL_M" } # 项目ID PROJECT_ID = "92" # 设备ID（默认为1，根据实际情况调整） DEVICE_ID = "1" # 采样间隔配置 SAMPLE_INTERVAL_SECONDS = 4 # 采样间隔：每4秒一个点 DURATION_MINUTES = 40 # 查询时长：40分钟 EXPECTED_ROWS = DURATION_MINUTES * 60 // SAMPLE_INTERVAL_SECONDS # 时间点数：2400秒 / 4秒 = 600行 def load_plc_points(input_file: str) -> list: """ 从input_format.txt加载PLC点位列表 Args: input_file: 输入文件路径 Returns: 点位名称列表（去掉第一行的'index'） """ points = [] with open(input_file, 'r', encoding='utf-8-sig') as f: # utf-8-sig自动去除BOM for line in f: # 去除空白字符和换行符 point = line.strip() # 跳过空行和index行（不区分大小写） if point and point.lower() != 'index': points.append(point) logger.info(f"加载了 {len(points)} 个PLC点位") return points def query_single_point_history(item_name: str, start_time: datetime, end_time: datetime) -> pd.Series: """ 通过API查询单个点位的历史数据（按秒采样） Args: item_name: 数据项名称（PLC点位） start_time: 开始时间 end_time: 结束时间 Returns: pd.Series: 以时间为索引的数据序列 """ # 转换时间为毫秒级时间戳 start_timestamp = int(start_time.timestamp() * 1000) end_timestamp = int(end_time.timestamp() * 1000) # API参数：按秒查询 params = { "deviceid": DEVICE_ID, "dataitemid": item_name, "project_id": PROJECT_ID, "stime": start_timestamp, "etime": end_timestamp, "size": "1", "interval": "s", # 按分钟采样（与loop_main.py一致） "aggregator": "new" # 取最新值 } try: response = requests.get(API_BASE_URL, params=params, headers=HEADERS, timeout=60) response.raise_for_status() data = response.json() # 检查API响应状态 if data.get("code") != 200: # 第一个点位时打印详细错误，帮助排查 if item_name == "C.M.UF1_PT_JS@out": logger.error(f"API返回错误: code={data.get('code')}, msg={data.get('msg', 'N/A')}") return pd.Series(dtype=float) if data.get("code") == 200 and data.get("data"): # 解析数据 time_list = [] val_list = [] for item in data["data"]: if item.get("val") is not None and item.get("htime_at") is not None: try: time_str = item.get("htime_at") val = float(item.get("val")) time_list.append(time_str) val_list.append(val) except (ValueError, TypeError): pass if time_list: # 创建Series series = pd.Series(val_list, index=pd.to_datetime(time_list, format='mixed', errors='coerce')) series = series[~series.index.isna()] # 去除无效时间 series = series.sort_index() return series return pd.Series(dtype=float) except requests.exceptions.RequestException as e: logger.warning(f"查询点位 {item_name} 网络失败: {e}") # 第一个点位时打印网络错误详情 if item_name == "C.M.UF1_PT_JS@out": logger.error(f"API网络请求失败: {e}") return pd.Series(dtype=float) except (json.JSONDecodeError, ValueError, KeyError) as e: logger.warning(f"解析点位 {item_name} 响应失败: {e}") return pd.Series(dtype=float) def resample_to_4s(series: pd.Series, start_time: datetime, end_time: datetime) -> pd.Series: """ 将原始数据降采样到4秒间隔 Args: series: 原始时间序列数据 start_time: 开始时间 end_time: 结束时间 Returns: 降采样后的Series """ # 创建4秒间隔的时间索引 time_index = pd.date_range(start=start_time, end=end_time, freq=f'{SAMPLE_INTERVAL_SECONDS}s') if series.empty: # 返回空序列 return pd.Series(np.nan, index=time_index) # 重采样：使用均值聚合 resampled = series.resample(f'{SAMPLE_INTERVAL_SECONDS}s').mean() # 重新索引到标准时间点 resampled = resampled.reindex(time_index, method='nearest', tolerance=pd.Timedelta(seconds=SAMPLE_INTERVAL_SECONDS)) # 前向填充缺失值 resampled = resampled.ffill().bfill() return resampled def query_all_points_history(plc_points: list, duration_minutes: int = DURATION_MINUTES) -> pd.DataFrame: """ 逐个查询所有PLC点位的历史数据，降采样后合并为DataFrame 异常处理: - 查询失败的点位填充NaN，后续通过ffill/bfill处理 Args: plc_points: PLC点位列表 duration_minutes: 查询时长（分钟），默认40分钟 Returns: DataFrame，第一列为时间索引，后续为PLC点位列 """ # 多查1分钟确保时间跨度 >= 40分钟 query_duration = duration_minutes + 1 # 41分钟 end_time = datetime.now() start_time = end_time - timedelta(minutes=query_duration) logger.info(f"查询时间范围: {start_time.strftime('%Y-%m-%d %H:%M:%S')} 至 {end_time.strftime('%Y-%m-%d %H:%M:%S')}") logger.info(f"查询时长: {query_duration}分钟 (确保>=40分钟)") logger.info(f"开始逐个查询 {len(plc_points)} 个点位...") # 创建标准时间索引（基于查询时长） time_index = pd.date_range(start=start_time, end=end_time, freq=f'{SAMPLE_INTERVAL_SECONDS}s') num_rows = len(time_index) # 使用字典收集所有列数据，避免DataFrame碎片化 data_dict = {'index': time_index} success_count = 0 fail_count = 0 for i, point in enumerate(plc_points): # 进度显示 if (i + 1) % 10 == 0 or i == 0: logger.info(f"查询进度: {i + 1}/{len(plc_points)} ({(i+1)/len(plc_points)*100:.1f}%)") # 查询单个点位 raw_series = query_single_point_history(point, start_time, end_time) if not raw_series.empty: # 查询成功：降采样到4秒间隔 resampled = resample_to_4s(raw_series, start_time, end_time) # 对齐到标准时间索引 if len(resampled) >= num_rows: point_data = resampled.values[:num_rows] else: point_data = np.full(num_rows, np.nan) point_data[:len(resampled)] = resampled.values data_dict[point] = point_data success_count += 1 else: # 查询失败：直接填充0 data_dict[point] = np.zeros(num_rows) fail_count += 1 logger.warning(f"点位 {point} 查询失败，填充0") logger.info(f"查询完成: 成功 {success_count}/{len(plc_points)}, 失败 {fail_count} 个") # 一次性创建DataFrame result_df = pd.DataFrame(data_dict) logger.info(f"最终DataFrame大小: {result_df.shape}") return result_df def run_diagnosis(df: pd.DataFrame) -> dict: """ 调用WaterPlantDiagnoser进行诊断 Args: df: 包含时间列的传感器数据DataFrame（600行×159列，含时间列） Returns: 诊断结果字典 """ # 延迟导入，避免循环依赖 from test import WaterPlantDiagnoser try: logger.info("初始化诊断器...") diagnoser = WaterPlantDiagnoser() logger.info(f"开始诊断，输入数据: {df.shape}") result = diagnoser.api_predict(df) return result except FileNotFoundError as e: logger.error(f"模型文件缺失: {e}") return {"status": "error", "message": str(e)} except Exception as e: logger.error(f"诊断过程出错: {e}") import traceback traceback.print_exc() return {"status": "error", "message": str(e)} def append_result_to_file(result: dict, output_file: str): """ 将诊断结果追加写入到输出文件 Args: result: 诊断结果字典 output_file: 输出文件路径 """ # 添加时间戳 result_with_time = { "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), **result } # 将结果转换为JSON字符串 result_str = json.dumps(result_with_time, ensure_ascii=False) # 追加写入文件 with open(output_file, 'a', encoding='utf-8') as f: f.write(result_str + '\n') logger.info(f"结果已追加到: {output_file}") def main(): """ 主函数：执行完整的诊断流程 """ logger.info("=" * 60) logger.info(" 水厂异常诊断系统 - 实时诊断") logger.info("=" * 60) # 文件路径 input_file = os.path.join(BASE_DIR, "input_format.txt") output_file = os.path.join(BASE_DIR, "output_format.txt") # 数据保存目录 data_save_dir = os.path.join(BASE_DIR, "data_history") os.makedirs(data_save_dir, exist_ok=True) # 步骤1: 加载PLC点位列表 if not os.path.exists(input_file): logger.error(f"点位配置文件不存在: {input_file}") return plc_points = load_plc_points(input_file) if not plc_points: logger.error("未能加载任何PLC点位") return # 步骤2: 逐个查询40分钟历史数据，降采样到4秒间隔 df = query_all_points_history(plc_points, duration_minutes=DURATION_MINUTES) if df.empty or len(df) < 100: result = {"status": "error", "message": "无法获取足够的历史数据"} append_result_to_file(result, output_file) return # 步骤3: 保存原始数据到文件（带时间戳，保留2位小数） timestamp_str = datetime.now().strftime('%Y%m%d_%H%M%S') data_filename = f"plc_data_{timestamp_str}.csv" data_filepath = os.path.join(data_save_dir, data_filename) # 数值列保留2位小数 df_to_save = df.copy() for col in df_to_save.columns: if col != 'index' and df_to_save[col].dtype in ['float64', 'float32']: df_to_save[col] = df_to_save[col].round(2) df_to_save.to_csv(data_filepath, index=False, encoding='utf-8') logger.info(f"原始数据已保存到: {data_filepath}") # 步骤4: 执行诊断 result = run_diagnosis(df) logger.info(f"诊断结果: {result}") # 步骤5: 追加结果到输出文件 append_result_to_file(result, output_file) logger.info("诊断完成") def run_loop(interval_minutes: int = 40): """ 循环运行模式：每隔指定分钟自动执行一次诊断 Args: interval_minutes: 运行间隔（分钟），默认40分钟 """ import time logger.info("=" * 60) logger.info(f" 水厂异常诊断系统 - 循环运行模式") logger.info(f" 运行间隔: 每 {interval_minutes} 分钟执行一次") logger.info("=" * 60) run_count = 0 while True: try: run_count += 1 logger.info(f"\n{'='*40}") logger.info(f"开始第 {run_count} 轮诊断...") logger.info(f"{'='*40}") # 执行一次诊断 main() # 计算下一次运行时间 next_run_time = datetime.now() + timedelta(minutes=interval_minutes) logger.info(f"本轮完成，下一轮将在 {next_run_time.strftime('%Y-%m-%d %H:%M:%S')} 开始") logger.info(f"等待 {interval_minutes} 分钟...") # 等待指定时间 time.sleep(interval_minutes * 60) except KeyboardInterrupt: logger.info("\n收到中断信号，停止循环运行") break except Exception as e: logger.error(f"诊断过程出错: {e}") import traceback traceback.print_exc() # 出错后继续等待下一轮 logger.info(f"等待 {interval_minutes} 分钟后重试...") time.sleep(interval_minutes * 60) # ===================== 本地模拟测试函数 ===================== def test_with_local_data(): """ 使用本地CSV文件进行测试（不依赖API） 用于在没有API连接时测试诊断逻辑 """ logger.info("=" * 60) logger.info(" 水厂异常诊断系统 - 本地数据测试") logger.info("=" * 60) output_file = os.path.join(BASE_DIR, "output_format.txt") # 尝试读取本地测试数据 test_file = os.path.join(config.DATASET_SENSOR_DIR, f"{config.SENSOR_FILE_PREFIX}1.csv") if not os.path.exists(test_file): logger.error(f"测试数据文件不存在: {test_file}") result = {"status": "error", "message": f"测试文件不存在: {test_file}"} append_result_to_file(result, output_file) return # 读取数据（40分钟 / 4秒采样间隔 = 600条，多读一些确保足够） logger.info(f"读取测试数据: {test_file}") df = pd.read_csv(test_file) # 取前900条数据确保超过40分钟 if len(df) > 900: df = df.iloc[:900] logger.info(f"测试数据大小: {df.shape} (预期至少600行)") # 执行诊断 result = run_diagnosis(df) logger.info(f"诊断结果: {result}") # 追加结果到输出文件 append_result_to_file(result, output_file) logger.info("本地测试完成") if __name__ == "__main__": import sys if len(sys.argv) > 1: if sys.argv[1] == "--local": # 使用本地数据测试 test_with_local_data() elif sys.argv[1] == "--once": # 单次运行 main() elif sys.argv[1] == "--help" or sys.argv[1] == "-h": print("用法:") print(" python run_diagnosis.py # 默认循环运行（每40分钟一次）") print(" python run_diagnosis.py 30 # 循环运行（每30分钟一次）") print(" python run_diagnosis.py --once # 单次运行") print(" python run_diagnosis.py --local # 本地数据测试") else: # 尝试解析为间隔分钟数 try: interval = int(sys.argv[1]) run_loop(interval_minutes=interval) except ValueError: print(f"未知参数: {sys.argv[1]}") print("使用 --help 查看用法") else: # 默认：循环运行模式（40分钟间隔） run_loop(interval_minutes=40)