开源第二篇，书接上回，上回的工具用起来着实不方便，功能也少，不能满足大部分需求，体现在：钉钉发送数据，数据处理，以及接收数据，定时任务等这部分。
随后对其进行了优化
数据接收
首先是数据接收，为什么这么说？数据接收，存日志，实时处理，繁琐。
数据接收其实问题不是很大，利用Pyqt5的自定义信号槽机制可以优美的解决掉问题。对于实时数据处理，需要考虑到各种异常情况，所以也就伴随着很多个if-else，有些时候不得不嵌套3个左右的if-else。
如何解决？
实时数据照样拿，但是不做多的处理，只关注指标即可，指标达到，那就直接结束掉，从日志中将数据取出，统一做处理。
这样做，可以有效的减少if-else的使用，可以减少很多的不必要的判断，有些判断只需要一次即可。
数据处理
其次是数据处理，从实时数据修改成了读取日志数据。为什么这么做？
原因：实时数据需要考虑很多因素，处理起来极其麻烦，对于何时结束数据获取存在比较大的问题，尤其是代码书写的美观上。
解决：对于关键数据，分开来处理，分多个函数进行处理，可以有效的增加代码的可阅读性，对于不可避免的if-else也是有效的措施之一。对于数据组合，看个人情况使用，本次优化大部分采用的字典以及列表的的数据结构进行存储值的。保证数据的规范性，最后使用字典统一管理。
钉钉发送数据
在这之前，钉钉发送信息提示一直困扰着我，因为实时数据需要区分很多情况处理。在这之后，采用了关键数据甄别后，直接采用检测触发的方式进行即可，也就是检测到任务完成，我就读取数据，发送数据。避免了多种情况的发生。
定时任务
这个主要是拿来结束指令数据，当然也能拿来持续发送指令，但是鉴于设备端有持续返回数据的功能，我们就不是很需要发送指令了。
用途：数据满足要求后，直接执行自定义指令，来结束指令发送，不需要多余数据进行存储。对后续数据处理减少了判断逻辑以及简化了难易程度。
代码优化
这是本篇不会讲到的东西，因为代码并没有怎么优化出来。尤其是日志处理这块。有很多的冗余代码。是一下章会讲到的东西。先来预览一下：
class ReadLogThread(QThread): def __init__(self, Path, SelectText=None, OnelyIphone=None, Devices=None): super().__init__() self.Path = Path self.SelectText = SelectText self.dingding = DingTalkSendMsg() self.OnelyIphone = OnelyIphone self.Devices = Devices self.currentTime = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S") def run(self) -> None: info = self.Info() if info: self.SendDing(info) with open(sys_ + "\\" + "测试数据.json", "w") as json_file: json.dump(info, json_file, indent=4) # 使用indent参数以漂亮的格式缩进数据 @staticmethod def dataTimes(value, stime, etime): if len(value) >= 12: start_time = datetime.datetime.strptime(stime, '%H:%M:%S.%f') end_time = datetime.datetime.strptime(etime, '%H:%M:%S.%f') else: start_time = datetime.datetime.strptime(stime, '%H:%M:%S') end_time = datetime.datetime.strptime(etime, '%H:%M:%S') over_time = end_time - start_time return str(over_time) def ReadLog(self): with open(self.Path, 'r', encoding="utf-8") as r: datas = r.read() return datas def Info(self): datas = self.ReadLog() Infomations = self.VolCur() capValue = re.findall(r'\[(.*)\].*cap\s*:\s*(\d*)\s*%', datas) statusValue = re.findall(r'\[(.*)\].*status\s*:\s*(\w*)', datas) for key, value in zip(statusValue, capValue): if self.SelectText == '充电': # 充电时长 ChargeTime = self.dataTimes(stime=statusValue[0][0], etime=statusValue[-1][0], value=statusValue[0][0]) # 充电跳电情况 ChargeInfo = self.ChargeInfoBatteryJump() ChargeBat = self.ChargeBatJump() return {"PutTime": ChargeTime, "Currentbattery":capValue[-1][1], "PutInfo": ChargeInfo, "PutBat": ChargeBat, "Infomations": Infomations} elif self.SelectText == "放电": # 放电 # 放电时长 PutTime = self.dataTimes(stime=capValue[0][0], etime=capValue[-1][0], value=capValue[0][0]) # 放电跳电情况 PutInfo = self.PutInfoBatteryJump() PutBat = self.PutBatJump() return {"PutTime": PutTime, "Currentbattery":capValue[-1][1], "PutInfo": PutInfo, "PutBat": PutBat, "Infomations": Infomations} def ChargeInfoBatteryJump(self): """info充电跳电""" datas = self.ReadLog() MaxNumber = [] dictValue = {"JumpNum": 0, "JumpValue": [], "MaxJump": 0} Jump = re.findall(r'\[(.*)\].*cap\s*:\s*(\d*)\s*%', datas) for num in range(len(Jump) - 1): CountNumber = int(Jump[num + 1][1]) - int(Jump[num][1]) if CountNumber > 1: # 充电跳电 dictValue["JumpNum"] += 1 MaxNumber.append(CountNumber) dictValue["JumpValue"].append(Jump[num + 1]) if CountNumber < 0: # 充电掉电 dictValue["JumpNum"] += 1 MaxNumber.append(CountNumber) dictValue["JumpValue"].append(Jump[num + 1]) if MaxNumber: dictValue["MaxJump"] = max(MaxNumber) # print("ChargeInfoBatteryJump",dictValue) return dictValue def PutInfoBatteryJump(self): """info放电跳电数据""" datas = self.ReadLog() MaxNumber = [] dictValue = {"JumpNum": 0, "JumpValue": [], "MaxJump": 0} Jump = re.findall(r'\[(.*)\].*cap\s*:\s*(\d*)\s*%', datas) for num in range(len(Jump) - 1): CountNumber = int(Jump[num][1]) - int(Jump[num + 1][1]) if CountNumber > 1: # 放电回电 dictValue["JumpNum"] += 1 MaxNumber.append(CountNumber) dictValue["JumpValue"].append(Jump[num + 1]) if CountNumber < 0: # 放电跳电 dictValue["JumpNum"] += 1 MaxNumber.append(CountNumber) dictValue["JumpValue"].append(Jump[num + 1]) if MaxNumber: dictValue["MaxJump"] = max(MaxNumber) return dictValue def VolCur(self): """单数是充电电流电压，双数是电池电流电压""" datas = self.ReadLog() ChargeDictValue = {"Start": {"vol": None, "cur": None}, "End": {"vol": None, "cur": None}} BatteryDictValue = {"Start": {"vol": None, "cur": None}, "End": {"vol": None, "cur": None}} volValue = re.findall(r"\[(.*)\].*vol\s*:\s*(\d*)", datas) curValue = re.findall(r"\[(.*)\].*cur\s*:\s*(\d*)", datas) ChargeDict = {"ChargeVol": [], "ChargeCur": []} BatteryDict = {"BatteryVol": [], "BatteryCur": []} if (volValue and curValue) is not False: for num in range(len(volValue)): if num % 2 == 0: # 充电电流电压 ChargeDict["ChargeVol"].append(volValue[num]) ChargeDict["ChargeCur"].append(curValue[num]) else: # 电池电压电流 BatteryDict["BatteryVol"].append(volValue[num]) BatteryDict["BatteryCur"].append(curValue[num]) """充电电压电流数据""" ChargeDictValue["Start"].update({"vol": ChargeDict["ChargeVol"][0][1], "cur": ChargeDict["ChargeCur"][0][1]}) ChargeDictValue["End"].update({"vol": ChargeDict["ChargeVol"][-1][1], "cur": ChargeDict["ChargeCur"][-1][1]}) BatteryDictValue["Start"].update({"vol": BatteryDict["BatteryVol"][0][1], "cur": BatteryDict["BatteryCur"][0][1]}) BatteryDictValue["End"].update({"vol": BatteryDict["BatteryVol"][-1][1], "cur": BatteryDict["BatteryCur"][-1][1]}) # print("VolCur", ChargeDictValue) return {"ChargeDictValue": ChargeDictValue, "BatteryDictValue": BatteryDictValue} def PutBatJump(self): """Bat放电""" BatPutValue = {"JumpNum": 0, "JumpValue": [], "MaxJump": 0} MaxNumber = [] datas = self.ReadLog() values = re.findall('\[(.*)\].*cap\s*:.*,(.*)', datas) if values: for num in range(len(values) - 1): CountNumber = int(values[num][1]) - int(values[num + 1][1]) if CountNumber > 1: """掉""" BatPutValue["JumpNum"] += 1 MaxNumber.append(CountNumber) BatPutValue["JumpValue"].append(values[num + 1]) if CountNumber < 0: """回""" BatPutValue["JumpNum"] += 1 MaxNumber.append(CountNumber) BatPutValue["JumpValue"].append(values[num + 1]) if MaxNumber: BatPutValue["MaxJump"] = max(MaxNumber) return BatPutValue def ChargeBatJump(self): """Bat充电""" BatChargeValue = {"JumpNum": 0, "JumpValue": [], "MaxJump": 0} MaxNumber = [] datas = self.ReadLog() values = re.findall('\[(.*)\].*cap\s*:.*,(.*)', datas) if values: for num in range(len(values) - 1): CountNumber = int(values[num + 1][1]) - int(values[num][1]) if CountNumber > 1: """跳""" BatChargeValue["JumpNum"] += 1 MaxNumber.append(CountNumber) BatChargeValue["JumpValue"].append(values[num + 1]) if CountNumber < 0: """掉""" BatChargeValue["JumpNum"] += 1 MaxNumber.append(CountNumber) BatChargeValue["JumpValue"].append(values[num + 1]) if MaxNumber: BatChargeValue["MaxJump"] = max(MaxNumber) return BatChargeValue def SendDing(self, kwargs): message = f'\n --✉️ {self.Devices} Tests complete-- \n' \ f'\n📌 测试人员：Aiper \n' \ f'\n💡 当前电量：{kwargs["Currentbattery"]} % \n' \ f'\n📆 测试日期：{self.currentTime} \n' \ f'\n⌛ 跑机时长：{kwargs["PutTime"]} \n' \ f'\n📝 跳电次数：{kwargs["PutInfo"]["JumpNum"]} 次 \n' \ f'\n🚀 最大跳电：{kwargs["PutInfo"]["MaxJump"]} \n' \ f'\n ⚡ 开始电流：{kwargs["Infomations"]["ChargeDictValue"]["Start"]["cur"]} ma \n' \ f'\n ⚡ 开始电压：{kwargs["Infomations"]["ChargeDictValue"]["Start"]["vol"]} mv \n' \ f'\n ⚡ 结束电流：{kwargs["Infomations"]["ChargeDictValue"]["End"]["cur"]} ma \n' \ f'\n ⚡ 结束电压：{kwargs["Infomations"]["ChargeDictValue"]["End"]["vol"]} ma \n'\ f'\n📒 详细请参考"测试数据.json"文件。' mobiles = [] if self.OnelyIphone: mobiles.append(self.OnelyIphone) self.dingding.send_ding_notification(message, mobiles) def JsonPath(self, data, path): """取值""" return jsonpath.jsonpath(data, path)
这部分代码优化，下一章会讲到。感谢阅读。gitee地址：
https://gitee.com/qinganan_admin/Pyqt5_Battery_MONITOR_SYSTEM.git
