gitea/APIs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

给socket执行函数增加了一个装饰器，每次发送的请求都要验证返回值，修正了查找id的逻辑bug

Browse Source
This commit is contained in:
gitea 2024-09-26 17:09:23 +08:00
parent 4b832b3ed1
commit 7f1e794f48
1 changed files with 22 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
code/openapi.py
View File

@ -8,6 +8,8 @@ from time import time, sleep, strftime, localtime
from pymodbus.client.tcp import ModbusTcpClient from pymodbus.client.tcp import ModbusTcpClient
# from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder # from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
# from pymodbus.constants import Endian # from pymodbus.constants import Endian
from functools import wraps
from inspect import getcallargs
from paramiko import SSHClient, AutoAddPolicy from paramiko import SSHClient, AutoAddPolicy
import clibs import clibs
@ -338,7 +340,8 @@ class HmiRequest(object):
# 尝试连续三次发送心跳包，确认建联成功 # 尝试连续三次发送心跳包，确认建联成功
self.__error_code = 0 self.__error_code = 0
for i in range(3): for i in range(3):
self.execution("controller.heart") self.execution.__wrapped__(self, "controller.heart")
# self.execution("controller.heart")
sleep(clibs.interval/5) sleep(clibs.interval/5)
if not self.__error_code: if not self.__error_code:
self.__is_connected = True self.__is_connected = True
@ -412,12 +415,12 @@ class HmiRequest(object):
for _ in range(3): for _ in range(3):
res = find_response(clibs.log_data_debug) res = find_response(clibs.log_data_debug)
if res is not None: if res is not None:
return res["data"] return res.get("data", f"{msg_id} has no data item")
sleep(clibs.interval*2) sleep(clibs.interval*2)
else: # 尝试在上一次分割的日志中查找，只做一次 else: # 尝试在上一次分割的日志中查找，只做一次
res = find_response("".join([clibs.log_path, listdir(clibs.log_path)[0]])) res = find_response("".join([clibs.log_path, sorted(listdir(clibs.log_path))[-3]]))
if res is not None: if res is not None:
return res["data"] return res.get("data", f"{msg_id} has no data item")
elif flag == 1: elif flag == 1:
for _ in range(3): for _ in range(3):
res = find_response_xs(clibs.log_data_debug) res = find_response_xs(clibs.log_data_debug)
@ -425,13 +428,14 @@ class HmiRequest(object):
return res return res
sleep(clibs.interval*2) sleep(clibs.interval*2)
else: else:
res = find_response_xs("".join([clibs.log_path, listdir(clibs.log_path)[0]])) res = find_response_xs("".join([clibs.log_path, sorted(listdir(clibs.log_path))[-3]]))
if res is not None: if res is not None:
return res return res
self.__sth_wrong(11, f"无法找到请求 {msg_id} 的返回结果") self.__sth_wrong(11, f"无法找到请求 {msg_id} 的返回结果")
def __msg_storage(self, response, flag=0): def __msg_storage(self, response, flag=0):
# response是解码后的字符串 # response是解码后的字符串
if "move.monitor" not in response:
clibs.logger.debug(f"{loads(response)}") clibs.logger.debug(f"{loads(response)}")
messages = self.c_msg if flag == 0 else self.c_msg_xs messages = self.c_msg if flag == 0 else self.c_msg_xs
if "move.monitor" in response: if "move.monitor" in response:
@ -644,6 +648,17 @@ class HmiRequest(object):
except Exception as Err: except Exception as Err:
self.__sth_wrong(8, f"错误信息：{Err}") self.__sth_wrong(8, f"错误信息：{Err}")
@staticmethod
def validate_req(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
all_args = getcallargs(func, self, *args, **kwargs)
_id = func(self, *args, **kwargs)
self.get_from_id(_id, flag=all_args["protocol_flag"])
return _id
return wrapper
@validate_req
def execution(self, command, protocol_flag=0, **kwargs): def execution(self, command, protocol_flag=0, **kwargs):
def log_reqs(request): def log_reqs(request):
with open(clibs.log_data_reqs, mode="a", encoding="utf-8") as f_log: with open(clibs.log_data_reqs, mode="a", encoding="utf-8") as f_log:
@ -1052,7 +1067,7 @@ class HmiRequest(object):
def set_diagnosis_params(self, display_pdo_params: list, frequency: int = 50, version: str = "1.4.1"): # OK def set_diagnosis_params(self, display_pdo_params: list, frequency: int = 50, version: str = "1.4.1"): # OK
""" """
设置诊断功能显示参数 设置诊断功能显示参数 [{"name": "hw_joint_vel_feedback", "channel": 0}, ]
:param display_pdo_params: 指定要采集的曲线名称，具体可通过 get_diagnosis_params 函数功能获取所有目前已支持的曲线 :param display_pdo_params: 指定要采集的曲线名称，具体可通过 get_diagnosis_params 函数功能获取所有目前已支持的曲线
:param frequency: 采样频率，默认 50ms :param frequency: 采样频率，默认 50ms
:param version: xDiagnose的版本号 :param version: xDiagnose的版本号