zhangwei
/
VMiddleware
generated from zhangwei/Train_Identify
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

新增大量日志，明细对PLC的所有操作。

This commit is contained in:
Mr.V 2024-04-13 18:16:54 +08:00
parent 706abf1cfa
commit 5033872573
1 changed files with 62 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
utils/ConnectPLC.py
View File

@ -1,18 +1,26 @@
# import logging
import traceback import traceback
import re import re
from pylogix import PLC from pylogix import PLC
from .LogT import logging, Log
logger = Log(stdout_log_level=logging.FATAL,
fil_log_level=logging.FATAL).getLogger()
class PLC_Tool: class PLC_Tool:
def __init__(self, plcIp: str, plcSlot: int = 0, timeOut: float = 5): def __init__(self, plcIp: str, plcSlot: int = 0, timeOut: float = 1):
self.plcIp = plcIp self.plcIp = plcIp
self.plcSlot = plcSlot self.plcSlot = plcSlot
self.timeOut = timeOut self.timeOut = timeOut
self.comm = None self.comm = None
logger.info("准备连接PLC，IP：%s, slot: %d, timeout: %d", self.plcIp, self.plcSlot, self.timeOut)
self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut) self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut)
logger.info("连接PLC成功")
def __del__(self): def __del__(self):
logger.info("准备断开PLC连接")
self.comm.Close() self.comm.Close()
logger.info("断开PLC连接完成")
def __isIP(self, ip: str): def __isIP(self, ip: str):
""" """
@ -27,10 +35,14 @@ class PLC_Tool:
return False return False
def connect(self): def connect(self):
logger.warning("准备重新连接PLC")
self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut) self.comm = PLC(self.plcIp, self.plcSlot, self.timeOut)
logger.info("连接PLC成功")
def close(self): def close(self):
logger.info("准备断开PLC连接")
self.comm.Close() self.comm.Close()
logger.info("断开PLC连接完成")
def batch_write_tag(self, tags: list): def batch_write_tag(self, tags: list):
""" """
@ -54,7 +66,8 @@ class PLC_Tool:
if not isinstance(tag, tuple): if not isinstance(tag, tuple):
return 0, "输入的参数格式不对，请参照：[(\"AI.speed\", 10.32),(\"AI.start\", 1)]", None return 0, "输入的参数格式不对，请参照：[(\"AI.speed\", 10.32),(\"AI.start\", 1)]", None
print("需要写入的数据：", tags) # print("准备写入PLC的数据：%s" % str(tags))
logger.info("准备写入PLC的数据：%s", str(tags))
rets = self.comm.Write(tags) rets = self.comm.Write(tags)
hasNone = False hasNone = False
@ -68,24 +81,28 @@ class PLC_Tool:
if ret.Status != "Success": if ret.Status != "Success":
hasNone = True hasNone = True
if hasNone: if hasNone:
logger.warning("写入PLC的数据中，存在部分失败：%s", str(results))
return 2, "存在写入失败的标签", results return 2, "存在写入失败的标签", results
else: else:
logger.info("写入PLC成功,写入的内容：%s", str(results))
return 1, "", results return 1, "", results
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("写入PLC异常，计划写入的内容：%s 。 异常反馈：%s", str(tags), traceback.format_exc())
return 0, str(io_error), None return 0, str(io_error), None
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.batch_write_tag(tags) return self.batch_write_tag(tags)
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("写入PLC异常，计划写入的内容：%s 。 异常反馈：%s", str(tags), traceback.format_exc())
return 0, str(e), None return 0, str(e), None
def write_tag(self, tag: str, value: [str, int, float]): def write_tag(self, tag: str, value: [str, int, float]):
""" """
Write tags to PLC . 批量写入标签 Write tags to PLC . 批量写入标签
:param tag: [("AI.speed", 10.32),("AI.start", 1)] :param tag: "AI.speed"...
:param value: type is [str|int|float] :param value: type is [str|int|float]
:return: [bool, str:异常信息, list: 修改的标签tuple] :return: [bool, str:异常信息, list: 修改的标签tuple]
""" """
@ -100,22 +117,27 @@ class PLC_Tool:
# if len(value) == 0: # if len(value) == 0:
# return False, "请输入标签'%s'的新值value:str" % tag # return False, "请输入标签'%s'的新值value:str" % tag
print("需要写入的数据 %s:%s" % (tag, str(value))) # print("需要写入的数据 %s:%s" % (tag, str(value)))
logger.info("准备写入PLC的数据：%s:%s", tag, str(value))
ret = self.comm.Write(tag, value) ret = self.comm.Write(tag, value)
if ret.Status != "Success": if ret.Status != "Success":
logger.error("存在写入失败的标签, 计划写入的信息：%s:%s", tag, str(value))
return False, "存在写入失败的标签" return False, "存在写入失败的标签"
else: else:
logger.info("写入PLC成功，写入的数据：%s:%s", tag, str(value))
return True, "" return True, ""
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("写入PLC异常，计划写入的内容：%s:%s 。 异常反馈：%s", tag, str(value), traceback.format_exc())
return False, str(io_error) return False, str(io_error)
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.write_tag(tag, value) return self.write_tag(tag, value)
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("写入PLC异常，计划写入的内容：%s:%s 。 异常反馈：%s", tag, str(value), traceback.format_exc())
return False, str(e) return False, str(e)
def write_array(self, tag: str, array: list): def write_array(self, tag: str, array: list):
@ -143,18 +165,22 @@ class PLC_Tool:
# "status": ret.Status # "status": ret.Status
# }] # }]
if ret.Status != "Success": if ret.Status != "Success":
logger.error("数组信息写入失败，计划写入内容：%s:%s", tag, str(array))
return False, "数组信息写入失败" return False, "数组信息写入失败"
logger.info("数组信息写入成功，写入内容：%s:%s", tag, str(array))
return True, "" return True, ""
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("数组信息写入失败，计划写入内容：%s:%s， 反馈的异常信息：%s", tag, str(array), traceback.format_exc())
return False, str(io_error) return False, str(io_error)
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.write_array(tag, array) return self.write_array(tag, array)
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("数组信息写入失败，计划写入内容：%s:%s， 反馈的异常信息：%s", tag, str(array),
traceback.format_exc())
return False, str(e) return False, str(e)
def get_tags(self, tags: [str, list]): def get_tags(self, tags: [str, list]):
@ -173,6 +199,7 @@ class PLC_Tool:
if len(tags) == 0: if len(tags) == 0:
return False, "请输入标签tags:list", None return False, "请输入标签tags:list", None
logger.info("读取PLC的点位信息，所读点位：%s", str(tags))
rets = self.comm.Read(tags) rets = self.comm.Read(tags)
for ret in rets: for ret in rets:
results.append( results.append(
@ -180,17 +207,23 @@ class PLC_Tool:
"value": ret.Value, "value": ret.Value,
"status": ret.Status "status": ret.Status
}) })
if ret.Status != "Success":
logger.error("读取PLC的点位失败，反馈结果：%s", str(results))
return False, "读取PLC点位失败", results
logger.info("已读取到PLC的点位信息，读取结果：%s", str(results))
return True, "", results return True, "", results
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("读取PLC的点位失败，所读点位：%s， 反馈错误信息：%s", str(tags), traceback.format_exc())
return False, str(io_error), None return False, str(io_error), None
connect_count += 1 connect_count += 1
self.connect()
self.get_tags(tags)
print(traceback.format_exc()) print(traceback.format_exc())
self.connect()
return self.get_tags(tags)
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("读取PLC的点位失败，所读点位：%s， 反馈错误信息：%s", str(tags), traceback.format_exc())
return False, str(e), None return False, str(e), None
def get_array(self, tag: str, count: int = 1): def get_array(self, tag: str, count: int = 1):
@ -207,25 +240,32 @@ class PLC_Tool:
if not self.__isIP(self.plcIp): if not self.__isIP(self.plcIp):
return False, "PLC的IP地址格式异常", None return False, "PLC的IP地址格式异常", None
if len(tag) == 0: if len(tag) == 0:
return False, "请输入标签tags:list", None return False, "请输入标签tag:str", None
if count <= 0: if count <= 0:
return False, "取数个数应大于0", None return False, "取数个数应大于0", None
logger.info("读取PLC的数组信息，所读点位：%s[%d]", str(tag), count)
ret = self.comm.Read(tag, count) ret = self.comm.Read(tag, count)
results = {"tagName": ret.TagName, results = {"tagName": ret.TagName,
"value": ret.Value, "value": ret.Value,
"status": ret.Status "status": ret.Status
} }
if ret.Status != "Success":
logger.error("读取PLC的数组失败，所读点位：%s[%d]， 反馈结果：%s", tag, count, str(results))
return False, "读取PLC的数组失败", results
logger.info("已读取PLC的数组信息，读取结果：%s", str(results))
return True, "", results return True, "", results
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("读取PLC的数组失败，所读点位：%s[%d]， 反馈错误信息：%s", tag, count, traceback.format_exc())
return False, str(io_error), None return False, str(io_error), None
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.get_array(tag, count) return self.get_array(tag, count)
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("读取PLC的数组失败，所读点位：%s[%d]， 反馈错误信息：%s", tag, count, traceback.format_exc())
return False, str(e), None return False, str(e), None
def get_tag_list(self): def get_tag_list(self):
@ -236,19 +276,23 @@ class PLC_Tool:
connect_count: int = 0 connect_count: int = 0
results = [] results = []
try: try:
logger.info("获取PLC所有标签")
tags = self.comm.GetTagList() tags = self.comm.GetTagList()
for t in tags.Value: for t in tags.Value:
results.append({t.TagName: t.DataType}) results.append({t.TagName: t.DataType})
logger.info("已获取到PLC所有标签, 标签列表：%s", str(results))
return results return results
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("获取PLC所有标签失败，反馈错误信息：%s", traceback.format_exc())
return None return None
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.get_tag_list() return self.get_tag_list()
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("获取PLC所有标签失败，反馈错误信息：%s", traceback.format_exc())
return None return None
def get_plc_time(self, raw=False): def get_plc_time(self, raw=False):
@ -259,17 +303,21 @@ class PLC_Tool:
""" """
connect_count: int = 0 connect_count: int = 0
try: try:
logger.info("开始获取PLC时间")
time = self.comm.GetPLCTime(raw) time = self.comm.GetPLCTime(raw)
logger.info("获取PLC时间成功：%s", str(time))
return time return time
except IOError as io_error: except IOError as io_error:
if connect_count > 0: if connect_count > 0:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("获取PLC时间失败，反馈错误信息：%s", traceback.format_exc())
return None return None
connect_count += 1 connect_count += 1
self.connect() self.connect()
self.get_plc_time() return self.get_plc_time()
except Exception as e: except Exception as e:
print(traceback.format_exc()) print(traceback.format_exc())
logger.error("获取PLC时间失败，反馈错误信息：%s", traceback.format_exc())
return None return None
# if __name__ == "__main__": # if __name__ == "__main__":