#!/usr/bin/env python # -*- coding: utf-8 -*- """ 查询td_d_log_result表和td_d_log_file表，筛选符合条件的device_id和channel_no组合 筛选条件（两种类型）： - task_id = 2 第一种类型（value>30）： - 对于每个device_id和channel_no的组合，取最新的10条记录（按generate_time） - 计算这10个value值的平均值 - 如果平均值大于30，则记录该(device_id, channel_no)组合 第二种类型（abs>2）： - 对于每个device_id和channel_no的组合，取最新的10条file_name记录（按generate_time） - 计算每条记录的不对称性（abs正、abs负） - 如果有3条或以上记录的abs正或abs负绝对值大于2，则记录该(device_id, channel_no)组合 输出字段： - 设备ID - 设备名称 - 通道名称 - 问题结果（value>30时显示平均值；abs>2时显示10个文件的abs正和abs负值） """ import pymysql import sys import os import zlib import json from collections import defaultdict from datetime import datetime # 设置控制台输出编码（Windows需要，Linux默认UTF-8） if sys.platform == 'win32': try: os.system('chcp 65001 >nul 2>&1') except: pass else: # Linux环境下设置UTF-8编码 import locale try: locale.setlocale(locale.LC_ALL, 'en_US.UTF-8') except: try: locale.setlocale(locale.LC_ALL, 'C.UTF-8') except: pass def parse_file_data(file_name, use_fdfs=False, fdfs_client=None): """ 解析振动数据文件 参数: file_name: 文件路径（FDFS路径或本地路径） use_fdfs: 是否使用FDFS客户端（如果为True，需要提供fdfs_client） fdfs_client: FDFS客户端对象（可选） 返回: 解析后的数据（字典或列表） """ try: # 1.1 从 FDFS 下载文件数据，先把文件名转成 UTF-8 编码的二进制串 if use_fdfs and fdfs_client: # 使用FDFS客户端下载 data = fdfs_client.download_bytes(bytes(file_name, encoding="utf8")) else: # 如果没有FDFS客户端，尝试从本地文件系统读取 # 将FDFS路径转换为本地路径 if file_name.startswith('group1/M00/'): local_path = file_name.replace('group1/M00/', '/home/soft/data/fdfs/storage/data/', 1) elif file_name.startswith('group1/M00'): local_path = file_name.replace('group1/M00', '/home/soft/data/fdfs/storage/data', 1) else: local_path = file_name # 从本地文件系统读取 with open(local_path, 'rb') as f: data = f.read() # 1.2 检查下载的二进制数据开头是否是 {（JSON 格式的标志） if data[0:1] == b'{': # 如果是 JSON 二进制串，直接解码成字符串 data = data.decode() else: # 如果不是 JSON 开头，说明数据是压缩的，先解压 ss = zlib.decompress(data).decode() # 修正格式：单引号改双引号（JSON 要求双引号）、nan 改成 0（避免 JSON 解析报错） ss = ss.replace('\'', '"').replace("nan", "0") # 把修正后的字符串解析成 JSON 格式的 Python 数据（字典/列表） data = json.loads(ss) return data except FileNotFoundError as e: # 文件未找到，静默返回None return None except zlib.error as e: print(f" 警告: 解压缩失败 {file_name}: {e}") return None except json.JSONDecodeError as e: print(f" 警告: JSON解析失败 {file_name}: {e}") return None except Exception as e: print(f" 警告: 解析文件失败 {file_name}: {e}") return None def calculate_asymmetry(data): """ 计算不对称性 从data中提取raw_y数组，然后计算不对称性 返回 (abs_正, abs_负, 正半轴点数量, 负半轴点数量) """ try: # 如果data是字典，提取raw_y字段 if isinstance(data, dict): if 'raw_y' not in data: return None, None, None, None values_list = data['raw_y'] elif isinstance(data, list): # 如果是列表，直接使用（兼容旧代码） values_list = data else: return None, None, None, None # 确保values_list是列表 if not isinstance(values_list, list): return None, None, None, None # 转换为数值列表 values = [] for v in values_list: try: val = float(v) values.append(val) except (ValueError, TypeError): continue if len(values) == 0: return None, None, None, None # 分离正半轴和负半轴的值 positive_values = [v for v in values if v > 0] negative_values = [abs(v) for v in values if v < 0] # 负半轴取绝对值 # 计算正半轴和负半轴的点数量 count_positive = len(positive_values) count_negative = len(negative_values) # 计算和 sum_positive = sum(positive_values) sum_negative = sum(negative_values) # 计算不对称性 if sum_negative > 0: abs_positive = abs(sum_positive / sum_negative) else: abs_positive = None if sum_positive > 0: abs_negative = abs(sum_negative / sum_positive) else: abs_negative = None return abs_positive, abs_negative, count_positive, count_negative except Exception as e: return None, None, None, None def query_filtered_devices(fdfs_client=None): """ 查询符合条件的device_id和channel_no组合 参数: fdfs_client: FDFS客户端对象（可选，用于下载文件） """ # 数据库连接配置 db_config = { 'host': '127.0.0.1', 'port': 3306, 'user': 'prod', 'password': 'hmdmxjIvfIjIoflL', 'database': 'iot', 'charset': 'utf8mb4', 'connect_timeout': 10 } connection = None try: print("正在连接到数据库...") connection = pymysql.connect(**db_config) print("[成功] 数据库连接成功！\n") with connection.cursor() as cursor: # 查询所有唯一的device_id和channel_no组合，且task_id=2 print("查询所有device_id和channel_no的组合...") query_unique = """ SELECT DISTINCT device_id, channel_no FROM td_d_log_result WHERE task_id = '2' """ cursor.execute(query_unique) device_channel_pairs = cursor.fetchall() print(f"找到 {len(device_channel_pairs)} 个device_id和channel_no的组合\n") # 存储符合条件的(device_id, channel_no)组合及其问题结果 # 格式: {(device_id, channel_no): {'value>30': 平均值, 'abs>2': [(生成时间, abs正, abs负, 正半轴点数量, 负半轴点数量), ...]}} filtered_channel_dict = {} # ===== 阶段一：筛选value>30的故障测点（基于平均值） ===== print("=" * 60) print("阶段一：筛选value平均值>30的故障测点") print("=" * 60) for idx, (device_id, channel_no) in enumerate(device_channel_pairs, 1): # 查询该组合的最新10条记录（按generate_time降序） query_latest = """ SELECT value, generate_time FROM td_d_log_result WHERE device_id = %s AND channel_no = %s AND task_id = '2' ORDER BY generate_time DESC LIMIT 10 """ cursor.execute(query_latest, (device_id, channel_no)) records = cursor.fetchall() if len(records) < 10: # 如果记录数少于10条，跳过 continue # 计算最近10个value的平均值 values = [] for value_str, generate_time in records: try: # value字段是varchar类型，需要转换为float value = float(value_str) if value_str else 0 values.append(value) except (ValueError, TypeError): # 如果转换失败，跳过该值 continue # 如果有效值数量不足，跳过 if len(values) == 0: continue # 计算平均值 avg_value = sum(values) / len(values) # 如果平均值大于30，记录该(device_id, channel_no)组合（类型1：value>30） if avg_value > 30: key = (device_id, channel_no) if key not in filtered_channel_dict: filtered_channel_dict[key] = {} filtered_channel_dict[key]['value>30'] = avg_value print(f"[{idx}] device_id: {device_id}, channel_no: {channel_no}, " f"平均值: {avg_value:.2f}") # ===== 阶段二：筛选abs>2的故障测点（基于所有组合） ===== print("\n" + "=" * 60) print("阶段二：筛选abs>2的故障测点") print("=" * 60) for idx, (device_id, channel_no) in enumerate(device_channel_pairs, 1): # 查询该(device_id, channel_no)组合最近10条记录的file_name（按generate_time降序，task_id=2） query_file_name = """ SELECT lf.file_name, lf.generate_time as file_generate_time FROM td_d_log_file lf WHERE lf.device_id = %s AND lf.channel_no = %s AND lf.task_id = '2' ORDER BY lf.generate_time DESC LIMIT 10 """ cursor.execute(query_file_name, (device_id, channel_no)) file_results = cursor.fetchall() if len(file_results) < 10: # 如果文件记录数少于10条，跳过 continue # 统计abs正或abs负绝对值大于2的个数，同时保存所有10个文件的abs值、点数量和generate_time count_abs_over_2 = 0 abs_values_list = [] # 保存所有10个文件的(生成时间, abs正, abs负, 正半轴点数量, 负半轴点数量)值 for file_name, file_generate_time in file_results: if file_name: # 解析文件并计算不对称性 data = parse_file_data(file_name, use_fdfs=(fdfs_client is not None), fdfs_client=fdfs_client) result = calculate_asymmetry(data) if data is not None else (None, None, None, None) abs_positive, abs_negative, count_positive, count_negative = result # 保存abs值、点数量和generate_time abs_values_list.append((file_generate_time, abs_positive, abs_negative, count_positive, count_negative)) # 检查abs正或abs负的绝对值是否大于2（只要其中一个大于2就算一条） if (abs_positive is not None and abs(abs_positive) > 2) or (abs_negative is not None and abs(abs_negative) > 2): count_abs_over_2 += 1 else: # 如果file_name为空，也添加None值以保持列表长度 abs_values_list.append((file_generate_time, None, None, None, None)) # 如果有3条或以上记录的abs>2，记录该(device_id, channel_no)组合（类型2：abs>2） if count_abs_over_2 >= 3: key = (device_id, channel_no) if key not in filtered_channel_dict: filtered_channel_dict[key] = {} filtered_channel_dict[key]['abs>2'] = abs_values_list print(f"[{idx}] device_id: {device_id}, channel_no: {channel_no}, " f"abs>2的记录数量: {count_abs_over_2}/10") # 转换为列表格式（用于后续处理） filtered_channel_list = list(filtered_channel_dict.keys()) # 查询device_name print("\n查询设备名称...") device_info_map = {} if filtered_channel_list: # 获取所有唯一的device_id unique_device_ids = list(set([device_id for device_id, channel_no in filtered_channel_list])) # 批量查询device_name placeholders = ','.join(['%s'] * len(unique_device_ids)) query_device_name = f""" SELECT device_id, device_name FROM td_d_device_info WHERE device_id IN ({placeholders}) """ cursor.execute(query_device_name, tuple(unique_device_ids)) device_info_results = cursor.fetchall() # 构建device_id到device_name的映射 for device_id, device_name in device_info_results: device_info_map[device_id] = device_name if device_name else '(未设置设备名称)' # 查询channel_name print("查询通道名称...") channel_info_map = {} if filtered_channel_list: # 批量查询channel_name（使用device_id和channel_no组合） query_channel_name = """ SELECT device_id, channel_no, channel_name FROM td_d_device_channel WHERE (device_id, channel_no) IN ( """ # 构建IN子句的占位符 placeholders = ','.join(['(%s, %s)'] * len(filtered_channel_list)) query_channel_name += placeholders + ")" # 准备参数：将(device_id, channel_no)元组展开 params = [] for device_id, channel_no in filtered_channel_list: params.extend([device_id, int(channel_no)]) # channel_no需要转换为int cursor.execute(query_channel_name, tuple(params)) channel_info_results = cursor.fetchall() # 构建(device_id, channel_no)到channel_name的映射 for device_id, channel_no, channel_name in channel_info_results: channel_info_map[(device_id, str(channel_no))] = channel_name if channel_name else '(未设置通道名称)' # 生成输出文件（包含时间戳） timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_filename = f"query_result_{timestamp}.txt" output_filepath = os.path.join(os.getcwd(), output_filename) # 打印所有符合条件的设备信息 print("\n" + "=" * 60) print("符合条件的故障测点列表：") print("=" * 60) # 准备输出内容（同时用于打印和写入文件） output_lines = [] output_lines.append("=" * 60) output_lines.append("符合条件的故障测点列表：") output_lines.append("=" * 60) output_lines.append("") output_lines.append("字段说明：") output_lines.append("- 设备ID: 设备标识") output_lines.append("- 设备名称: 设备名称") output_lines.append("- 通道名称: 通道名称") output_lines.append("- 问题结果: value>30时显示平均值；abs>2时显示10个文件的abs正和abs负值") output_lines.append("") if filtered_channel_list: # 按device_id和channel_no排序后打印 sorted_channels = sorted(filtered_channel_list, key=lambda x: (x[0], int(x[1]) if x[1].isdigit() else x[1])) displayed_count = 0 # 记录实际显示的记录数 for device_id, channel_no in sorted_channels: device_name = device_info_map.get(device_id, '(未找到设备名称)') channel_name = channel_info_map.get((device_id, channel_no), '(未找到通道名称)') # 过滤掉"未找到通道名称"的记录 if channel_name == '(未找到通道名称)': continue displayed_count += 1 # 增加显示计数 # 获取问题结果 result_data = filtered_channel_dict.get((device_id, channel_no), {}) # 格式化输出 device_info_str = f"设备ID: {device_id}" device_name_str = f"设备名称: {device_name}" channel_name_str = f"通道名称: {channel_name}" print(device_info_str) print(device_name_str) print(channel_name_str) output_lines.append(device_info_str) output_lines.append(device_name_str) output_lines.append(channel_name_str) # 输出问题结果 problem_result_lines = [] if 'value>30' in result_data: avg_value = result_data['value>30'] result_line = f"问题结果(value>30): 平均值 = {avg_value:.2f}" print(result_line) problem_result_lines.append(result_line) if 'abs>2' in result_data: abs_values_list = result_data['abs>2'] result_line = f"问题结果(abs>2):" print(result_line) problem_result_lines.append(result_line) for idx, (generate_time, abs_pos, abs_neg, count_pos, count_neg) in enumerate(abs_values_list, 1): time_str = str(generate_time) if generate_time else "N/A" abs_pos_str = f"{abs_pos:.6f}" if abs_pos is not None else "N/A" abs_neg_str = f"{abs_neg:.6f}" if abs_neg is not None else "N/A" count_pos_str = str(count_pos) if count_pos is not None else "N/A" count_neg_str = str(count_neg) if count_neg is not None else "N/A" detail_line = f" 文件{idx}: generate_time = {time_str}, abs正 = {abs_pos_str}, abs负 = {abs_neg_str}, 正半轴点数量 = {count_pos_str}, 负半轴点数量 = {count_neg_str}" print(detail_line) problem_result_lines.append(detail_line) output_lines.extend(problem_result_lines) print("-" * 60) output_lines.append("-" * 60) total_str = f"\n总计: {displayed_count} 个故障测点（已过滤掉未找到通道名称的记录）" print(total_str) output_lines.append(total_str) else: no_result_str = "未找到符合条件的故障测点" print(no_result_str) output_lines.append(no_result_str) # 将结果写入文件 try: with open(output_filepath, 'w', encoding='utf-8') as f: f.write('\n'.join(output_lines)) f.write('\n') print(f"\n[成功] 查询结果已保存到文件: {output_filepath}") except Exception as e: print(f"\n[警告] 保存结果到文件失败: {e}") return filtered_channel_list except pymysql.Error as e: error_code, error_msg = e.args print("[失败] 数据库操作失败！") print(f"[失败] 错误代码: {error_code}") print(f"[失败] 错误信息: {error_msg}") return None except Exception as e: print("[失败] 发生未知错误！") print(f"[失败] 错误信息: {e}") import traceback traceback.print_exc() return None finally: # 关闭数据库连接 if connection: connection.close() print("\n[成功] 数据库连接已关闭") if __name__ == "__main__": print("=" * 60) print("设备和通道筛选工具") print("=" * 60) print("筛选条件：") print("- task_id = 2") print("") print("第一种类型（value>30）：") print(" - 每个(device_id, channel_no)组合的最新10条记录（按generate_time）") print(" - 计算这10个value值的平均值，如果平均值大于30") print("") print("第二种类型（abs>2）：") print(" - 每个(device_id, channel_no)组合的最新10条file_name记录（按generate_time）") print(" - 计算每条记录的不对称性（abs正、abs负）") print(" - 如果有3条或以上记录的abs正或abs负绝对值大于2") print("=" * 60 + "\n") result = query_filtered_devices() if result is not None: sys.exit(0) else: sys.exit(1)