import requests import logging import pandas as pd from datetime import datetime, timedelta # --- 1. 基础配置 --- # 配置日志记录器，方便调试和追踪 logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # 禁止传播到root logger，避免重复输出 logger.propagate = False # 只有在logger没有handler时才添加（防止重复） if not logger.handlers: console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') console_handler.setFormatter(formatter) logger.addHandler(console_handler) # --- 2. 辅助函数：填充缺失的小时数据 --- def fill_missing_hourly_data(df, start_date, end_date): """ 确保DataFrame在给定的日期范围内拥有完整的小时索引。 缺失的小时数据点将通过前向填充（forward-fill）的方式补全。 Args: df (pd.DataFrame): 经过数据透视处理的DataFrame，必须有名为 'timestamp' 的时间列。 start_date (datetime): 期望时间范围的开始时间。 end_date (datetime): 期望时间范围的结束时间。 Returns: pd.DataFrame: 填充了缺失小时数据后的DataFrame。 """ if df.empty: logger.warning("原始数据为空，无法进行数据补充。") return df # 将 'index' 列设为索引，这是进行时间序列分析的标准操作 df.set_index('index', inplace=True) # 创建一个完整的小时级别的时间范围 # inclusive='left' 表示包含开始时间点，但不包含结束时间点，符合常规查询逻辑 complete_time_range = pd.date_range( start=start_date.replace(minute=0, second=0, microsecond=0), periods=180 * 24, # 指定需要 4320 个小时 freq='H' ) logger.info( f"期望生成的时间范围: 从 {start_date.strftime('%Y-%m-%d %H:00')} 到 {end_date.strftime('%Y-%m-%d %H:00')}") logger.info(f"期望的小时总数: {len(complete_time_range)} 小时") logger.info(f"数据填充前的行数: {len(df)} 行") # 使用 reindex 和 ffill 高效地填充缺失值 # 1. reindex: 将DataFrame的索引与完整时间范围对齐，缺失的时间点会产生NaN值 # 2. ffill: 使用前一个有效观测值向前填充NaN值 filled_df = df.reindex(complete_time_range).ffill() filled_df.reset_index(inplace=True) filled_df.rename(columns={'index': 'index'}, inplace=True) # 通常 reset_index 默认列名就是 'index'，但这行可以确保万无一失 logger.info(f"数据填充后的最终行数: {len(filled_df)} 行") if not filled_df.empty: logger.info(f"最终时间范围: 从 {filled_df.index.min()} 到 {filled_df.index.max()}") return filled_df # --- 3. 主函数：通过API获取并处理传感器数据 --- def get_sensor_data(end_date_str=None, API_BASE_URL=None, HEADERS=None): """ 通过循环调用API获取多个传感器的数据，合并、处理并填充成一个干净、完整的DataFrame。 Args: end_date_str (str, optional): 查询的结束日期，格式为 'YYYY-MM-DD HH:MM:SS'。 如果为None，则默认为当前时间。 Returns: pd.DataFrame: 一个处理完成的DataFrame，索引是时间，每列是一个传感器。 """ # 步骤 1: 计算时间范围和时间戳 if end_date_str is None: end_date = datetime.now() else: try: end_date = datetime.strptime(end_date_str, '%Y-%m-%d %H:%M:%S') except ValueError: logger.error(f"日期格式错误: '{end_date_str}'。请使用 'YYYY-MM-DD HH:MM:SS' 格式。") return pd.DataFrame() start_date = end_date - timedelta(days=180) start_timestamp = int(start_date.timestamp() * 1000) end_timestamp = int(end_date.timestamp() * 1000) logger.info( f"开始查询数据，时间范围: {start_date.strftime('%Y-%m-%d %H:%M:%S')} 到 {end_date.strftime('%Y-%m-%d %H:%M:%S')}") # 步骤 2: 定义要查询的14个传感器列表 item_names = [ 'C.M.RO1_FT_JS@out', 'C.M.RO2_FT_JS@out', 'C.M.RO3_FT_JS@out', 'C.M.RO4_FT_JS@out', 'C.M.RO_TT_ZJS@out', 'C.M.RO_Cond_ZJS@out', 'C.M.RO1_DB@DPT_1', 'C.M.RO1_DB@DPT_2', 'C.M.RO2_DB@DPT_1', 'C.M.RO2_DB@DPT_2', 'C.M.RO3_DB@DPT_1', 'C.M.RO3_DB@DPT_2', 'C.M.RO4_DB@DPT_1', 'C.M.RO4_DB@DPT_2' ] # 步骤 3: 循环调用API获取所有传感器的数据 all_records = [] for item_name in item_names: params = { "deviceid": "1", "dataitemid": item_name, "project_id": "92", "stime": start_timestamp, "etime": end_timestamp, "size": "1", "interval": "h", "aggregator": "new" } try: response = requests.get(API_BASE_URL, params=params, headers=HEADERS, timeout=60) response.raise_for_status() # 如果请求失败 (如 404, 500)，则会抛出异常 api_response = response.json() if api_response.get('code') == 200 and api_response.get('data'): records = api_response['data'] # 将 item_name 添加到每条记录中，为后续数据透视做准备 for record in records: record['item_name'] = item_name # 使用查询时的 itemid 作为列名 all_records.extend(records) logger.info(f"成功获取 '{item_name}' 的 {len(records)} 条数据。") else: logger.warning(f"'{item_name}' 未返回有效数据。API消息: {api_response.get('msg', '无')}") except requests.exceptions.RequestException as e: logger.error(f"查询 '{item_name}' 时发生网络错误: {e}") continue # 跳过当前失败的传感器，继续下一个 # 步骤 4: 将原始数据转换为DataFrame并进行清洗 if not all_records: logger.error("未能从API获取任何有效数据，处理终止。") return pd.DataFrame() logger.info(f"API数据获取完成，总共获取了 {len(all_records)} 条原始记录。") data_origin = pd.DataFrame(all_records) # 数据清洗：转换数据类型，并处理可能存在的错误 # 使用 'coerce' 会将无法转换的值变为 NaT (时间) 或 NaN (数值)，更稳健 data_origin['index'] = pd.to_datetime(data_origin['htime_at'], errors='coerce') data_origin['val'] = pd.to_numeric(data_origin['val'], errors='coerce') # 删除时间或数值转换失败的无效行 data_origin.dropna(subset=['index', 'val'], inplace=True) # 步骤 5: 数据透视，将长表转换为宽表 logger.info("正在进行数据透视，将数据整理为每行一个时间点，每列一个传感器...") pivot_df = data_origin.pivot_table(index='index', columns='item_name', values='val', aggfunc='first') pivot_df.reset_index(inplace=True) # 将索引 'index' 变回普通列，方便传入填充函数 # 步骤 6: 填充缺失的小时数据并设置最终的索引 logger.info("正在填充缺失的小时数据以确保时间序列的完整性...") # pivot_df.columns.name = None final_df = fill_missing_hourly_data(pivot_df, start_date, end_date) # 步骤 7: 对 final_df 进行插值处理，消除所有 NaN 值 if not final_df.empty: # 检查是否存在 NaN 值 nan_count_before = final_df.isna().sum().sum() if nan_count_before > 0: logger.info(f"检测到 {nan_count_before} 个 NaN 值，开始进行插值处理...") # 保存 'index' 列（时间列） time_column = final_df['index'].copy() if 'index' in final_df.columns else None # 对数值列进行插值处理 numeric_columns = final_df.select_dtypes(include=['float64', 'int64']).columns if len(numeric_columns) > 0: # 1. 线性插值（适合时间序列数据） final_df[numeric_columns] = final_df[numeric_columns].interpolate(method='linear', limit_direction='both') # 2. 前向填充（处理开头的 NaN） final_df[numeric_columns] = final_df[numeric_columns].ffill() # 3. 后向填充（处理末尾的 NaN） final_df[numeric_columns] = final_df[numeric_columns].bfill() # 4. 如果仍有 NaN（整列为空的情况），用0填充 final_df[numeric_columns] = final_df[numeric_columns].fillna(0) # 恢复 'index' 列（确保时间列不被修改） if time_column is not None: final_df['index'] = time_column nan_count_after = final_df.isna().sum().sum() logger.info(f"插值处理完成，剩余 NaN 值: {nan_count_after} 个") if nan_count_after > 0: logger.warning(f"警告: 仍有 {nan_count_after} 个 NaN 值未能填充") else: logger.info("数据中没有 NaN 值，无需插值处理") # final_df = final_df.sort_values('index').reset_index(drop=True) return final_df # --- 4. 脚本执行入口 --- if __name__ == "__main__": # 示例: 不传入日期，默认使用当前时间作为结束时间，查询过去180天的数据 print("--- 开始执行数据获取任务 ---") sensor_df = get_sensor_data() if not sensor_df.empty: print("\n[成功] 数据获取与处理完成！") print("\n[结果] DataFrame (前5行):") print(sensor_df.head()) print("\n[结果] DataFrame (后5行):") print(sensor_df.tail()) print("\n[结果] DataFrame 信息:") # .info() 会打印出维度、列名、非空值数量和数据类型等关键信息 sensor_df.info() else: print("\n[失败] 未能生成最终的DataFrame，请检查上面的日志输出获取详细错误信息。")