🥒 前言
目前演示的是一个工具，但如此，未来完成有潜力可以演变为一整套系统。
👑现场人员自检失败表计点位教程V2.0
NOTE: 如果没有“logfiles-meter-tool“目录的请联系我们进行提供！
👇
进入<dist>目录
👇
【关键步骤一】、将我们需要分析的日志文件放到该目录中
👇
【关键步骤二】、配置<日志名称>和<任务ID>
config.ini配置文件内容和详细解析如下图：
@pararm:[logfile_path]是存放日志的路径，但由于与<应用程序>处于同目录下，所以相当于日志名，该日志包含您刚跑完测试的日志内容。
@pararm:[work_id] 是您任务的序号，如下图,Ftp图片路径下包含”task“/"Task"的字符串，也就是灰色框框住的那一串正式您此次任务的序号。
👇
【关键步骤三】、运行<应用程序>
👇
【异常】如果出现下面的红框信息，是因为任务ID输入错了，没有匹配结果，根据提示操作。【如果有匹配结果，列出来的任务id后面会打√的。】
【正常】正常的运行终端结果
👇
自动生成自检报表meterLog_checking-<任务ID>.txt，位于<分析报告生成处>目录下
👇
里面部分关键内容如下：
👇
接下来大家请对照这张表，找到【需要现场人员自检】的【错误】进行搜索排查，有多个，可以从上往下慢慢来。
👇
以【通用类】<序号7>"该点位没有录入"作为例子，打开自检文本meterlog_checking.txt，搜索指定错误。
👇
①NOTE：如果出现无需现场人员自检的错误，需要提供一下日志文件，可能后续还需提供图片我们这边进行优化。
②NOTE：如果点位出现多次，只会取最后一次也就是最新一次的结果。
🍉一些使用样例图:
👑Code
# -*- coding: utf-8 -*-
'''
参考diamagnetic：
# 兰江
python3 meterPoint_Self-Checking_sys.py -p meterlog -t 30M00000036658634_task1703485183168_20231225141946
# 金鼎
python3 meterPoint_Self-Checking_sys.py -p meterlog -t 30M00000036658634_task1703485183168_20231225141947
'''
import os
import re
import json
import configparser
def get_settings():
config = configparser.ConfigParser()
config.read('./config.ini')
log_file = config.get('settings', 'logfile_path')
work_id = config.get('settings', 'work_id')
return log_file, work_id
def extract_debug_segments(log_file):
debug_segments = []
with open(log_file, 'r', encoding='utf-8') as file:
lines = file.readlines()
start_line = None
end_line = None
segment = []
for i, line in enumerate(lines):
if 'Debug' in line or '收到请求' in line or '数据库信息' in line:
if start_line is None:
start_line = i
segment.append(line.strip())
elif '结果放入队列待发送' in line:
if start_line is not None:
end_line = i
segment.append(line)
debug_segments.append([segment, start_line, end_line])
segment = []
start_line = None
end_line = None
return debug_segments
def process_request(request_str):
target_index = request_str.index("{")
# 按照":"分割字符串
split_str = request_str[target_index:]
# 获取分割后数组中最后一个索引所保存的信息
json_str = split_str.strip().replace("—", "-").replace("'", "\"")
objectList_request_str = json.loads(json_str)['objectList'][0]
# for k in objectList_request_str:
# print(k)
return objectList_request_str
def get_pointList_length(json_str):
pattern = r"'Position': '(\[.*?\])'"
matches = re.search(pattern, json_str)
if matches is None:
return 0
position_list = json.loads(matches.group(1))
# print("position_list:", position_list)
return len(position_list)
def process_sql(json_str):
json_str = json_str[json_str.index("MinValue"):]
json_str = "{'" + json_str
json_str = json_str.replace("'", "\"")
sql_dict = json.loads(json_str)
return sql_dict
def process_result(json_str):
json_str = json_str[json_str.index("code"):-5]
json_str = "{'" + json_str
# print(json_str)
json_str = json_str.replace("'", "\"")
json_str = json_str.replace("None", "null")
sql_dict = json.loads(json_str)
return sql_dict
def contains_digit(string):
pattern = r'\d' # 正则表达式模式，匹配任意数字
if re.search(pattern, string):
return True
else:
return False
def get_path_separator(path):
if '/' in path:
return '/'
elif '\\' in path:
return '\\'
elif '\\' * 2 in path:
return '\\\\'
else:
return None
def extract_work_path_tool(goal_str):
split_str = get_path_separator(goal_str)
pathIdx = -1
splitPaths = goal_str.split(split_str)
for idx, ss in enumerate(splitPaths):
if ss == 'CCD':
pathIdx = idx
if pathIdx == -1:
raise Exception("您的任务路径中没有CCD路径")
work_path = splitPaths[pathIdx-1]
return work_path
if __name__ == "__main__":
print("---------------------------------------------------------------------")
# 摄像机偏移严重+模糊
Error_withoutDetctor = []
# 未识别出指针
Error_withoutPointer = []
# 读取ftp图失败
Error_loadftp = []
# minIO无图
Error_withoutMinioImage = []
# minIO错图
Error_minioErrorImage = []
# 点位未录入
Error_withoutId = []
# 表计类型录入错误
Error_clsType = []
# 最大最小值设置错误
Error_minMaxSet = []
# 最大最小值未设置
Error_withoutMinMax = []
# 未打刻度点位
Error_withoutPointList = []
# 刻度打点错误
Error_PointList = []
# 未识别到任何油面表！
Error_ymb = []
# 画框与推理出来的油面表无匹配
Error_withoutYmbMatch = []
# OCR没有检测出数字
Error_ocrRec = []
# OCR没有检测出表盘
Error_ocrDet = []
# ===========================核
# 获取命令行参数
log_file, work_id = get_settings()
debug_segments = extract_debug_segments(log_file)
error_num = 0
# not_reading_num = 0
# type_num = 0
ymb_num, sxb_num, bj_num = 0, 0, 0
ymb_errorNum, sxb_errorNum, bj_errorNum = 0, 0, 0
# 过滤一遍只剩下最新的
filter_schem = {}
piNums_schem = {}
not_del_ids = []
# 任务计算
workNUms_schem = {}
for idx, segment in enumerate(debug_segments):
strat_line = segment[1]
end_line = segment[2]
for line in segment[0]:
if "收到请求" in line:
# print('【请求信息】: ',end='')
objectList_request_str = process_request(line)
# 任务ID
work_path = extract_work_path_tool(objectList_request_str['imageUrlList'][0])
if not work_path in workNUms_schem:
workNUms_schem[work_path] = 1
else:
workNUms_schem[work_path] += 1
if work_path != work_id:
break
#点位ID
extract_objectId = objectList_request_str['objectId']
if not extract_objectId in filter_schem.keys():
# 新增
filter_schem[extract_objectId] = idx
piNums_schem[extract_objectId] = 1
else:
# 更新
filter_schem[extract_objectId] = idx
piNums_schem[extract_objectId] += 1
not_del_ids.append(idx)
break
print('|任务id |数量')
print("---------------------------------------------------------------------")
for wnn in workNUms_schem:
if work_id == wnn:
print(wnn, ' |',workNUms_schem[wnn],end=' ✔\n')
else:
print(wnn, ' |',workNUms_schem[wnn])
print('*********************************************************************')
if not work_id in workNUms_schem:
print("[告警]任务ID有误，本日志中无匹配任务。上方已列出所有任务ID以及他们的数量！请根据上面列出的任务ID，输入正确的任务ID。")
print('*********************************************************************')
work_id = input('[Input]:')
print("[提示]此次任务ID已经修改为:<{}>".format(work_id))
# 重置
filter_schem = {}
piNums_schem = {}
not_del_ids = []
for idx, segment in enumerate(debug_segments):
strat_line = segment[1]
end_line = segment[2]
for line in segment[0]:
if "收到请求" in line:
objectList_request_str = process_request(line)
# 任务ID
work_path = extract_work_path_tool(objectList_request_str['imageUrlList'][0])
if work_path != work_id:
break
# 点位ID
extract_objectId = objectList_request_str['objectId']
if not extract_objectId in filter_schem.keys():
# 新增
filter_schem[extract_objectId] = idx
piNums_schem[extract_objectId] = 1
else:
# 更新
filter_schem[extract_objectId] = idx
piNums_schem[extract_objectId] += 1
not_del_ids.append(idx)
break
print('*********************************************************************')
# print(piNums_schem)
# 找到第一次出现重复点位的位置
print("此次任务ID:<{}>中".format(work_id))
idsNums_result1 = len({key: value for key, value in piNums_schem.items() if value == 1})
print("点位 [=1] 的数量：",idsNums_result1)
idsNums_result2 = len({key: value for key, value in piNums_schem.items() if value > 1})
print("点位 [>1] 的数量：",idsNums_result2)
print('*********************************************************************')
# print(filter_schem, len(filter_schem))
# ------------------过滤结束
sumWorkNum, filter_workId_num, filter_objectId_num = 0, 0, 0
for idx, segment in enumerate(debug_segments):
# print(segment[0],'\n',len(segment[0]))
error_flag = False
ftpLoad_flag = False
# print('Start Line:', segment[1])
# print('End Line:', segment[2])
for line in segment[0]:
if "收到请求" in line:
# print('【请求信息】: ',end='')
objectList_request_str = process_request(line)
extract_objectId = objectList_request_str['objectId']
# print(extract_objectId)
# print(objectList_request_str['imageUrlList'][0], work_id)
# 过滤掉【不同任务】
if not work_id == extract_work_path_tool(objectList_request_str['imageUrlList'][0]):
filter_workId_num += 1
break
# 过滤掉【同任务相同点位取最新】
if ( piNums_schem[extract_objectId] > 1 ) and ( idx != filter_schem[extract_objectId] ):
# print(idx, filter_schem[extract_objectId])
filter_objectId_num += 1
break
# 这里才是没被break的真正点位数量
sumWorkNum += 1
elif '数据库信息' in line:
# print(line)
if line.split("【数据库信息】")[-1] == '{}':
# 数据库信息为空
# print('*pointList_length:0')
# print('{}')
Error_withoutId.append(extract_objectId)
error_num += 1
break
else:
# 数据库有信息
pointList_length = get_pointList_length(line)
sql_schem = process_sql(line)
MinValue = sql_schem['MinValue']
MaxValue = sql_schem['MaxValue']
meter_type = sql_schem['AlgorithmType']
ImagePath = sql_schem['ImagePath']
if meter_type == 'meter_v5':
bj_num += 1
if meter_type == 'meter_ywj':
ymb_num += 1
if meter_type == 'paddleocr':
sxb_num += 1
if meter_type == 'meter_v5':
if len(MinValue)== 0 or len(MaxValue) == 0:
Error_withoutMinMax.append(extract_objectId)
MinValue = float(0)
MaxValue = float(100)
error_flag = True
else:
MinValue = float(MinValue)
MaxValue = float(MaxValue)
# 表计类型录入错误(如果打点了，但表计类型不是meter_v5)
if meter_type != 'meter_v5' and pointList_length != 0:
Error_clsType.append(extract_objectId)
error_flag = True
# 未打刻度点位
if meter_type == 'meter_v5' and pointList_length == 0:
Error_withoutPointList.append(extract_objectId)
error_flag = True
# print(sql_schem, end=',')
# print("*pointList_length:", pointList_length)
elif '结果放入队列待发送' in line:
result_schem = process_result(line)
# print('【结果队列信息】：',end='')
# print(result_schem)
if result_schem['code'] == '2001':
Error_loadftp.append(extract_objectId)
ftpLoad_flag = True
error_flag = True
break
if result_schem['desc'] == '未识别到任何油面表！':
error_flag = True
Error_ymb.append(extract_objectId)
else:
splitContent = line.split("【Debug】")[-1]
if "成功检测到表盘！表盘信息是" in splitContent:
det_clsType = splitContent.split("：")[-1].strip().strip("").strip("[]").strip()
if splitContent.split("：")[-1].strip().strip("") == "[]":
Error_withoutDetctor.append(extract_objectId)
error_flag = True
if not 'sxb' in det_clsType and meter_type == 'paddleocr':
Error_ocrDet.append(extract_objectId)
error_flag = True
if 'ywb' in det_clsType:
ywb_minMax = [
[-20, 140],
[0, 160]
]
iter_minMax = [MinValue, MaxValue]
if not iter_minMax in ywb_minMax:
Error_minMaxSet.append(extract_objectId)
error_flag = True
elif 'xldlb' in det_clsType:
xldlb_minMax = [
[0, 3.0],
[0, 10],
[0, 9],
[0, 1]
]
iter_minMax = [MinValue, MaxValue]
if not iter_minMax in xldlb_minMax:
Error_minMaxSet.append(extract_objectId)
error_flag = True
# if '动作次数' in splitContent:
# print(splitContent)
# if '泄漏电流值' in splitContent:
# print(splitContent)
if 'OCR没有检测出数字' in splitContent:
Error_ocrRec.append(extract_objectId)
error_flag = True
if "没识别出指针" in splitContent:
Error_withoutPointer.append(extract_objectId)
error_flag = True
# 画框与推理出来的油面表无匹配
if '画框与推理出来的油面表无匹配' in splitContent:
Error_withoutYmbMatch.append(extract_objectId)
error_flag = True
if len(ImagePath) == 0 or "MinIo中缺失该点位基准图" in splitContent:
Error_withoutMinioImage.append(extract_objectId)
error_flag = True
# 用于验证
if '读数结果' in splitContent and not contains_digit(splitContent):
# not_reading_num +=1
# 验证后 无读数个数和错误个数基本一致->代表验证成功
# print(not_reading_num)
continue
if error_flag and not ftpLoad_flag:
if meter_type == 'meter_v5':
bj_errorNum += 1
if meter_type == 'meter_ywj':
ymb_errorNum += 1
if meter_type == 'paddleocr':
sxb_errorNum += 1
error_num += 1
elif error_flag and ftpLoad_flag:
error_num += 1
meter_type = ''
print("错误总数比：【{}/{}】-> 即正确率：{}%".format(error_num,sumWorkNum,round((1-error_num/sumWorkNum)*100, 2)))
# ===========================核
# 写入
# with open('meterLog_checking.txt', 'w') as output_file:
saveLogFile_path = './分析报告生成处'
if not os.path.exists(saveLogFile_path):
os.makedirs(saveLogFile_path)
with open(os.path.join(saveLogFile_path,'meterLog_checking-{}.txt'.format(work_id)), 'w', encoding='utf-8') as output_file:
output_file.write('您这次序号为[{}]的任务:\n---------------------------------\n一共测试表计数量：[{}]个, 错误点位为：[{}]个, 未打点个数为：[{}]。\n<在此之中>\n，指针类表计成功占[{}/{}]个\n，油面表成功占[{}/{}]个\n，数显表成功占[{}/{}]个。'.format(work_id,sumWorkNum,error_num,len(Error_withoutId),bj_num - bj_errorNum, bj_num,ymb_num - ymb_errorNum, ymb_num, sxb_num - sxb_errorNum, sxb_num))
# output_file.write("-> 即正确率：{}%".format(error_num,sumWorkNum,round((1-error_num/sumWorkNum)*100, 2)))
output_file.write('\n')
output_file.write('---------------------------------\n')
output_file.write('NOTE：接下来，请您根据所需要查询的错误名称，使用<ctrl+F>的方式进行查询。\n')
output_file.write('---------------------------------\n')
output_file.write("【错误】可能存在摄像机偏移严重/模糊<数量：{}>：".format(str(len(set(Error_withoutDetctor)))) + "\n")
output_file.write("\n".join(set(Error_withoutDetctor)))
output_file.write('\n')
output_file.write("【错误】未识别出指针<数量：{}>：".format(str(len(set(Error_withoutPointer)))) + "\n")
output_file.write("\n".join(set(Error_withoutPointer)))
output_file.write('\n')
output_file.write("【错误】读取ftp图失败<数量：{}>：".format(str(len(set(Error_loadftp)))) + "\n")
output_file.write("\n".join(set(Error_loadftp)))
output_file.write('\n')
output_file.write("【错误】minIO无图<数量：{}>：".format(str(len(set(Error_withoutMinioImage)))) + "\n")
output_file.write("\n".join(set(Error_withoutMinioImage)))
output_file.write('\n')
output_file.write("【错误】该点位没有录入<数量：{}>：".format(str(len(set(Error_withoutId)))) + "\n")
output_file.write("\n".join(set(Error_withoutId)))
output_file.write('\n')
output_file.write("【错误】表计类型录入错误<数量：{}>：".format(str(len(set(Error_clsType)))) + "\n")
output_file.write("\n".join(set(Error_clsType)))
output_file.write('\n')
output_file.write("【错误】最大最小值未设置<数量：{}>：".format(str(len(set(Error_withoutMinMax)))) + "\n")
output_file.write("\n".join(set(Error_withoutMinMax)))
output_file.write('\n')
output_file.write("【错误】未打刻度点位<数量：{}>：".format(str(len(set(Error_withoutPointList)))) + "\n")
output_file.write("\n".join(set(Error_withoutPointList)))
output_file.write('\n')
output_file.write("【错误】最大最小值设置错误<数量：{}>：".format(str(len(set(Error_minMaxSet)))) + "\n")
output_file.write("\n".join(set(Error_minMaxSet)))
output_file.write('\n')
output_file.write("【错误】存在刻度打点错误(暂未启用)<数量：{}>：".format(str(len(set(Error_PointList)))) + "\n")
output_file.write("\n".join(set(Error_PointList)))
output_file.write('\n')
for ey in Error_ymb:
if ey in Error_withoutYmbMatch:
Error_ymb.remove(ey)
output_file.write("【错误】未识别到任何油面<数量：{}>：".format(str(len(set(Error_ymb)))) + "\n")
output_file.write("\n".join(set(Error_ymb)))
output_file.write('\n')
output_file.write("【错误】画框与推理结果无匹配<数量：{}>：".format(str(len(set(Error_withoutYmbMatch)))) + "\n")
output_file.write("\n".join(set(Error_withoutYmbMatch)))
output_file.write('\n')
output_file.write("【错误】OCR没有检测出数字<数量：{}>：".format(str(len(set(Error_ocrRec)))) + "\n")
output_file.write("\n".join(set(Error_ocrRec)))
output_file.write('\n')
output_file.write("【错误】OCR没有检测出表盘<数量：{}>：".format(str(len(set(Error_ocrDet)))) + "\n")
output_file.write("\n".join(set(Error_ocrDet)))
output_file.write('\n')
print('<*总共统计数量:{}>\n<*过滤掉的非此次任务ID数量:{}>\n<*过滤掉的重复的点位ID数量:{}>'.format(len(debug_segments),filter_workId_num, filter_objectId_num))
print('*********************************************************************')
input("Press any key to exit...")
