Pyqt5-自动化电池监测工具

开源第二篇，书接上回，上回的工具用起来着实不方便，功能也少，不能满足大部分需求，体现在：钉钉发送数据，数据处理，以及接收数据，定时任务等这部分。

随后对其进行了优化

数据接收

首先是数据接收，为什么这么说？数据接收，存日志，实时处理，繁琐。

数据接收其实问题不是很大，利用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 = Pathself.SelectText = SelectTextself.dingding = DingTalkSendMsg()self.OnelyIphone = OnelyIphoneself.Devices = Devicesself.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参数以漂亮的格式缩进数据@staticmethoddef 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_timereturn str(over_time)def ReadLog(self):with open(self.Path, 'r', encoding="utf-8") as r:datas = r.read()return datasdef 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"] += 1MaxNumber.append(CountNumber)dictValue["JumpValue"].append(Jump[num + 1])if CountNumber < 0:  # 充电掉电dictValue["JumpNum"] += 1MaxNumber.append(CountNumber)dictValue["JumpValue"].append(Jump[num + 1])if MaxNumber:dictValue["MaxJump"] = max(MaxNumber)# print("ChargeInfoBatteryJump",dictValue)return dictValuedef 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"] += 1MaxNumber.append(CountNumber)dictValue["JumpValue"].append(Jump[num + 1])if CountNumber < 0:  # 放电跳电dictValue["JumpNum"] += 1MaxNumber.append(CountNumber)dictValue["JumpValue"].append(Jump[num + 1])if MaxNumber:dictValue["MaxJump"] = max(MaxNumber)return dictValuedef 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"] += 1MaxNumber.append(CountNumber)BatPutValue["JumpValue"].append(values[num + 1])if CountNumber < 0:"""回"""BatPutValue["JumpNum"] += 1MaxNumber.append(CountNumber)BatPutValue["JumpValue"].append(values[num + 1])if MaxNumber:BatPutValue["MaxJump"] = max(MaxNumber)return BatPutValuedef 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"] += 1MaxNumber.append(CountNumber)BatChargeValue["JumpValue"].append(values[num + 1])if CountNumber < 0:"""掉"""BatChargeValue["JumpNum"] += 1MaxNumber.append(CountNumber)BatChargeValue["JumpValue"].append(values[num + 1])if MaxNumber:BatChargeValue["MaxJump"] = max(MaxNumber)return BatChargeValuedef 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