[openapi.py] 初步搭建起框架，完成了新老协议的封包/解包/异步采集日志的操作（未充分测试）

aio/README.md

@@ -247,4 +247,5 @@ v0.1.6.3(2024/06/18)
 > ！！WARNING：目前版本的电机电流程序还支持DriverMaster采集的数据处理，等明确后，将不再支持，也即所有的电机电流数据（工业+协作），都是用诊断曲线来采集
 
 v0.1.7.0(2024/06/29)
-1. [current.py] 适配电机电流中速度使用hw_joint_vel_feedback的数据，取消对device_servo_vel_feedback的支持，后续所有涉及到速度相关的数据均已前者为准，现已完成对单轴和场景的适配
+1. [openapi.py] 初步搭建起框架，完成了新老协议的封包/解包/异步采集日志的操作（未充分测试），其中解包操作未实现分帧的情况，因为暂无适应场景，也无测试实例，而且目前来看，不可能有消息超过文档中返回数据大小定义的上限（65535），故可以暂不做实现
+2. 

aio/code/automatic_test/openapi.py

@@ -1,67 +1,254 @@
 import json
-from socket import *
+import socket
 import threading
+import selectors
 import time
 import binascii
+import sys
+import inspect
 
 
 class HmiRequest(object):
 
     def __init__(self):
-        self.c = socket(AF_INET, SOCK_STREAM)
+        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_xs = socket(AF_INET, SOCK_STREAM)
-        self.c_xs.connect(('192.168.84.129', 6666))
         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.t = threading.Thread(target=self.__heartbeat_detection)
+        self.c_msg_xs = []
-        self.t.daemon = True
+        self.t_unpackage = threading.Thread(target=self.__unpackage, args=(self.c, ))
-        self.t.start()
+        self.t_unpackage.daemon = True
+        self.t_unpackage.start()
+        self.t_unpackage_xs = threading.Thread(target=self.__unpackage_xs, args=(self.c_xs, ))
+        self.t_unpackage_xs.daemon = True
+        self.t_unpackage_xs.start()
+        # self.t = threading.Thread(target=self.__heartbeat)
+        # self.t.daemon = True
+        # self.t.start()
+        self.flag = 0
+        self.response = ''
+        self.leftover = 0
+        self.flag_xs = 0
+        self.response_xs = ''
+        self.leftover_xs = 0
 
-    def __handle_command(self, cmd):
+    def __header_check(self, index, data):
-        len_frame, len_pkg = len(cmd), len(cmd) + 6
+        try:
-        pkg_head = str(hex(len_pkg))[2:].rjust(4, '0')
+            _pkg_size = int(binascii.b2a_hex(data[index:index+2]), 16)
-        frame_head = str(hex(len_frame))[2:].rjust(4, '0')
+            _reserved = int(binascii.b2a_hex(data[index+2:index+4]), 16)
-        str0 = binascii.unhexlify(pkg_head)  # 报文
+            _frame_size = int(binascii.b2a_hex(data[index+4:index+6]), 16)
-        str1 = chr(0) + chr(0)  # 保留字段
+            _protocol = int(binascii.b2a_hex(data[index+6:index+8]), 16)
-        str2 = binascii.unhexlify(frame_head)  # 帧
+            if _reserved == 0 and _protocol == 512 and _pkg_size - _frame_size >= 6:
-        str3 = chr(2) + chr(0)  # 协议类型
+                return index+8, _frame_size, _pkg_size
-        return str0 + str1.encode() + str2 + str3.encode() + cmd.encode()
+            else:
+                print("数据有误，需要确认")
+                exit(9)
+        except:
+            print("无法读取数据，需要确认")
+            exit(10)
+
+    def __get_response(self, data):
+        _index = 0
+        while _index < len(data):
+            if self.flag == 0:
+                _index, _frame_size, _pkg_size = self.__header_check(_index, data)
+                if len(data) - _index >= _frame_size:
+                    # 说明剩余部分的数据正好就是完整的包数据
+                    self.response = data[_index:_index+_frame_size].decode()
+                    if len(self.c_msg) < 1000:
+                        self.c_msg.insert(0, self.response)
+                    else:
+                        self.c_msg.insert(0, self.response)
+                        while len(self.c_msg) > 1000:
+                            self.c_msg.pop()
+                    _index += _frame_size
+                    self.flag = 0
+                    self.response = ''
+                    self.leftover = 0
+                elif len(data) - _index < _frame_size:
+                    # 说有有分包的情况发生了，需要flag=1的处理
+                    self.flag = 1
+                    self.response = data[_index:].decode()
+                    self.leftover = _frame_size - (len(data) - _index)
+                    _index += _frame_size
+
+            elif self.flag == 1:
+                # 处理完之后，将flag重置为0
+                if self.leftover <= len(data):
+                    self.response += data[:self.leftover].decode()
+                    if len(self.c_msg) < 1000:
+                        self.c_msg.insert(0, self.response)
+                    else:
+                        self.c_msg.insert(0, self.response)
+                        while len(self.c_msg) > 1000:
+                            self.c_msg.pop()
+                    _index += self.leftover
+                    self.flag = 0
+                    self.response = ''
+                    self.leftover = 0
+                else:
+                    _index += self.leftover
+                    self.flag = 1
+                    self.response += data.decode()
+                    self.leftover -= len(data)
+
+    def __get_response_xs(self, data):
+        if self.flag_xs == 0:
+            if data[-1].decode() == '\r':
+                _responses = data.decode().split('\r')
+                for _response in _responses:
+                    if len(self.c_msg_xs) < 1000:
+                        self.c_msg_xs.insert(0, _response)
+                    else:
+                        self.c_msg_xs.insert(0, _response)
+                        while len(self.c_msg_xs) > 1000:
+                            self.c_msg_xs.pop()
+            else:
+                _responses = data.decode().split('\r')
+                for _response in _responses[:-1]:
+                    if not _response:
+                        break
+
+                    if len(self.c_msg_xs) < 1000:
+                        self.c_msg_xs.insert(0, _response)
+                    else:
+                        self.c_msg_xs.insert(0, _response)
+                        while len(self.c_msg_xs) > 1000:
+                            self.c_msg_xs.pop()
+                self.response_xs = _responses[-1]
+                self.flag_xs = 1
+        else:
+            if data[-1].decode() == '\r':
+                _responses = (self.response_xs.encode() + data).decode().split('\r')
+                for _response in _responses:
+                    if len(self.c_msg_xs) < 1000:
+                        self.c_msg_xs.insert(0, _response)
+                    else:
+                        self.c_msg_xs.insert(0, _response)
+                        while len(self.c_msg_xs) > 1000:
+                            self.c_msg_xs.pop()
+                self.response_xs = ''
+                self.flag_xs = 0
+            else:
+                _responses = (self.response_xs.encode() + data).decode().split('\r')
+                for _response in _responses[:-1]:
+                    if not _response:
+                        break
+
+                    if len(self.c_msg_xs) < 1000:
+                        self.c_msg_xs.insert(0, _response)
+                    else:
+                        self.c_msg_xs.insert(0, _response)
+                        while len(self.c_msg_xs) > 1000:
+                            self.c_msg_xs.pop()
+                self.response_xs = _responses[-1]
+                self.flag_xs = 1
+
+    def get_from_id(self, msg_id, flag=0):
+        messages = self.c_msg if flag == 0 else self.c_msg_xs
+        for i in range(3):
+            for msg in messages:
+                if msg_id in msg:
+                    return msg
+            time.sleep(1)
+        else:
+            print(f'无法查询到{msg_id}对应的响应')
+
+    def __package(self, cmd):
+        pkg_size = str(hex(len(cmd)+6))[2:].rjust(4, '0')
+        package = binascii.a2b_hex(pkg_size)      # 包的长度
+        reserved = chr(0) + chr(0)                # 保留字段
+        frame_size = str(hex(len(cmd)))[2:].rjust(4, '0')
+        frame = binascii.a2b_hex(frame_size)      # 帧的长度
+        protocol = chr(2) + chr(0)                # 协议类型
+        return package + reserved.encode() + frame + protocol.encode() + cmd.encode()
+
+    def __package_xs(self, cmd):
+        return f"{json.dumps(cmd, separators=(',', ':'))}\r".encode()
+
+    def __unpackage(self, sock):
+        def to_read(conn):
+            data = conn.recv(512)  # Should be ready
+            if data:
+                print(data)
+                self.__get_response(data)
+            else:
+                print('closing', sock)
+                sel.unregister(conn)
+                conn.close()
+
+        sel = selectors.DefaultSelector()
+        sel.register(sock, selectors.EVENT_READ, to_read)
 
-    def __heartbeat_detection(self):
-        data = {
-            "id": "system.controller.heart_0",
-            "module": "system",
-            "command": "controller.heart",
-        }
-        _id = 1
         while True:
-            data["id"] = f"#system.controller.heart_{_id}"
+            events = sel.select()
-            cmd = json.dumps(data, separators=(',', ':'))
+            for key, mask in events:
-            self.c.send(self.__handle_command(cmd))
+                callback = key.data
-            time.sleep(10)
+                callback(key.fileobj)
-            _id += 1
 
-    def motor_on(self):
+    def __unpackage_xs(self, sock):
-        """HMI上电"""
+        def to_read(conn):
-        data = {
+            data = conn.recv(1024)  # Should be ready
-            "command": "state.switch_motor_on",
+            if data:
-            "id": str(1234),
+                print(data)
-            "module": "system"
+                self.__get_response_xs(data)
-        }
+            else:
-        cmd = json.dumps(data, separators=(',', ':'))
+                print('closing', sock)
-        self.c.send(self.__handle_command(cmd))
+                sel.unregister(conn)
+                conn.close()
+
+        sel = selectors.DefaultSelector()
+        sel.register(sock, selectors.EVENT_READ, to_read)
+
+        while True:
+            events = sel.select()
+            for key, mask in events:
+                callback = key.data
+                callback(key.fileobj)
+
+    def __gen_id(self, command):
+        _now = time.time()
+        _id = f"{command}-{_now}"
+        return _id
+
+    def excution(self, command, **kwargs):
+        req = None
+        try:
+            with open(f'./templates/{command}.json', encoding='utf-8', mode='r') as f_json:
+                req = json.load(f_json)
+        except:
+            print(f"暂不支持 {command} 功能，或确认该功能存在...")
+            exit(1)
+
+        match command:
+            case 0:
+                pass
+            case 1:
+                pass
+
+        req['id'] = self.__gen_id(command)
+        cmd = json.dumps(req, separators=(',', ':'))
+        self.c.send(self.__package(cmd))
         time.sleep(2)
-        response = self.c.recv(102400)
+        return req['id']
-        print(response)
-        print(type(response))
-        print(response.decode())
-
-        # response = json.loads(self.c.recv(102400).decode('utf-16-be'))
-        # print(response)
 
 
 hr = HmiRequest()
-hr.motor_on()
+id_test = hr.excution('device.get_params')
+time.sleep(2)
+print(hr.c_msg)
+print(hr.get_from_id(id_test))
+# hr.excution('state.switch_manual')
+# time.sleep(2)
+# hr.excution('state.switch_motor_on')
+# time.sleep(2)
+# hr.excution('state.switch_motor_off')
+
+
 
 

aio/code/automatic_test/templates/device.get_params.json

@@ -0,0 +1,5 @@
+{
+  "id": "xxxxxxxxxxx",
+  "module": "system",
+  "command": "device.get_params"
+}

aio/code/automatic_test/templates/state.switch_auto.json

@@ -0,0 +1,5 @@
+{
+  "id": "xxxxxxxxxxx",
+  "module": "system",
+  "command": "state.switch_auto"
+}

aio/code/automatic_test/templates/state.switch_manual.json

@@ -0,0 +1,5 @@
+{
+  "id": "xxxxxxxxxxx",
+  "module": "system",
+  "command": "state.switch_manual"
+}

aio/code/automatic_test/templates/state.switch_motor_off.json

@@ -0,0 +1,5 @@
+{
+  "id": "xxxxxxxxxxx",
+  "module": "system",
+  "command": "state.switch_motor_off"
+}

aio/code/automatic_test/templates/state.switch_motor_on.json

@@ -0,0 +1,5 @@
+{
+  "id": "xxxxxxxxxxx",
+  "module": "system",
+  "command": "state.switch_motor_on"
+}