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

20240624

Browse Source 
11. [openapi.py] 建联部分做容错处理，并将读写文件做自适应处理
12. [aio.py] 将读写文件做自适应处理，引入openapi模块并生成实例，做心跳检测，将socket超时时间修改为3s
This commit is contained in:
gitea
2024-06-24 19:22:56 +08:00
parent a4009eb17c
commit 6604f0ba06
4 changed files with 42 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
aio/code/automatic_test/openapi.py
View File

@ -3,24 +3,36 @@ import socket
import threading
import selectors
import time
import os
MAX_FRAME_SIZE = 1024
socket.setdefaulttimeout(3)
class HmiRequest(object):
def __init__(self):
super().__init__()
self.c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# self.c.connect(('192.168.0.160', 5050))
self.c.connect(('192.168.84.129', 5050))
self.c.setblocking(False)
self.c_msg = []
self.c_xs = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# self.c_xs.connect(('192.168.0.160', 6666))
self.c_xs.connect(('192.168.84.129', 6666))
self.c_xs.setblocking(False)
self.c_msg_xs = []
while True:
try:
self.c = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# self.c.connect(('192.168.0.160', 5050))
self.c.connect(('192.168.84.129', 5050))
self.c.setblocking(False)
self.c_msg = []
self.c_xs = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# self.c_xs.connect(('192.168.0.160', 6666))
self.c_xs.connect(('192.168.84.129', 6666))
self.c_xs.setblocking(False)
self.c_msg_xs = []
break
except Exception as Err:
current_path = os.path.dirname(__file__)
with open(f"{current_path}/../../assets/templates/heartbeat", "w", encoding='utf-8') as f_h:
f_h.write('0')
print("Connection failed, will try again after 2 seconds...")
time.sleep(2)
self.t_heartbeat = threading.Thread(target=self.__heartbeat)
self.t_heartbeat.daemon = True
self.t_heartbeat.start()
@ -54,10 +66,11 @@ class HmiRequest(object):
exit(10)
def __heartbeat(self):
current_path = os.path.dirname(__file__)
while True:
_id = self.excution('controller.heart')
_flag = 1 if self.get_from_id(_id) else 0
with open("./templates/heartbeat", "w", encoding='utf-8') as f_h:
with open(f"{current_path}/../../assets/templates/heartbeat", "w", encoding='utf-8') as f_h:
f_h.write(str(_flag))
time.sleep(10)
@ -222,10 +235,12 @@ class HmiRequest(object):
return _id
def excution(self, command, flag=0, **kwargs):
current_path = os.path.dirname(__file__)
if flag == 0: # for old protocols
req = None
try:
with open(f'./templates/{command}.json', encoding='utf-8', mode='r') as f_json:
with open(f'{current_path}/../../assets/templates/{command}.json', encoding='utf-8', mode='r') as f_json:
req = json.load(f_json)
except:
print(f"暂不支持 {command} 功能，或确认该功能存在...")