新增升级程序相关功能

code/clibs.py

@@ -16,6 +16,7 @@ password = "luoshi2019"  # for real robot
 xService_port = 6666
 external_port = 8080
 modbus_port = 502
+upgrade_port = 4567
 interval = 0.5  # interval after actions being triggered, such as modbus/socket/external communication operations
 heartbeat_interval = 2
 RADIAN = 57.3  # 180 / 3.1415926

code/common.py

@@ -1,7 +1,13 @@
+import os.path
+import socket
 import time
 import openapi
 import json
 import clibs
+from socket import *
+from ctypes import *
+import hashlib
+import struct
 
 
 def initialization():
@@ -115,5 +121,190 @@ def initialization():
     md.r_clear_alarm()
 
 
+def fw_updater(local_file_path):
+    fw_size = os.path.getsize(local_file_path)
+    if fw_size > 10000000:
+        # get previous version of xCore
+        hr = openapi.HmiRequest()
+        version_previous = hr.get_robot_params["version"]
+        hr.close()
+
+    # var def
+    remote_file_path = './upgrade/lircos.zip'
+    fw_content = bytearray()
+    package_data = bytearray()
+    package_data_with_head = bytearray()
+
+    # socket connect
+    clibs.logger.info(f"正在连接 {clibs.ip_addr}:{clibs.upgrade_port}...")
+    try:
+        tcp_socket = socket(AF_INET, SOCK_STREAM)
+        tcp_socket.connect((clibs.ip_addr, clibs.upgrade_port))
+        tcp_socket.setblocking(True)
+    except Exception as Err:
+        clibs.logger.error(f"{Err} | 连接 {clibs.ip_addr}:{clibs.upgrade_port} 失败...")
+        exit(1)
+
+    # get firmware content of binary format
+    clibs.logger.info(f"正在读取 {local_file_path} 文件内容...")
+    with open(local_file_path, 'rb') as f_fw:
+        fw_content += f_fw.read()
+
+    # construct package data: http://confluence.i.rokae.com/pages/viewpage.action?pageId=15634148
+    clibs.logger.info("正在构造数据包，以及包头...")
+    # 1 protocol id
+    protocol_id = c_ushort(htons(0))  # 2 bytes
+    package_data += protocol_id
+
+    # 2 write type
+    write_type = c_ubyte(0)
+    package_data += bytes(write_type)
+
+    # 3 md5
+    md5_hash = hashlib.md5()
+    md5_hash.update(fw_content)
+    fw_md5 = md5_hash.hexdigest()
+    i = 0
+    while i < len(fw_md5):
+        _ = (fw_md5[i:i + 2])
+        package_data += bytes.fromhex(_)
+        i += 2
+
+    # 4 remote path len
+    remote_file_path_len = c_ushort(htons(len(remote_file_path)))
+    package_data += remote_file_path_len
+
+    # 5 remote path
+    package_data += remote_file_path.encode("ascii")
+
+    # 6 file
+    package_data += fw_content
+
+    # construct communication protocol: http://confluence.i.rokae.com/pages/viewpage.action?pageId=15634045
+    # 1 get package data with header
+    package_size = c_uint(htonl(len(package_data)))
+    package_type = c_ubyte(1)  # 0-无协议 1-文件 2-json
+    package_reserve = c_ubyte(0)  # 预留位
+
+    package_data_with_head += package_size
+    package_data_with_head += package_type
+    package_data_with_head += package_reserve
+    package_data_with_head += package_data
+
+    # 2 send data to server
+    clibs.logger.info("正在发送数据到 xCore，升级控制器无需重启，升级配置文件会自动软重启...")
+    start = 0
+    len_of_package = len(package_data_with_head)
+    while start < len_of_package:
+        end = 10240 + start
+        if end > len_of_package:
+            end = len_of_package
+        sent = tcp_socket.send((package_data_with_head[start:end]))
+        time.sleep(0.01)
+        if sent == 0:
+            raise RuntimeError("socket connection broken")
+        else:
+            start += sent
+
+    waited = 5 if fw_size > 10000000 else 25
+    time.sleep(waited)
+
+    if fw_size > 10000000:
+        # get current version of xCore
+        hr = openapi.HmiRequest()
+        version_current = hr.get_robot_params["version"]
+        hr.close()
+
+        clibs.logger.info(f"控制器升级成功：from {version_previous} to {version_current} :)")
+    else:
+        clibs.logger.info(f"配置文件升级成功 :)")
+
+    tcp_socket.close()
+
+
+class UpgradeJsonCmd(object):
+    def __init__(self):
+        self.__c = None
+        self.__sock_conn()
+
+    def __sock_conn(self):
+        # socket connect
+        clibs.logger.info(f"正在连接 {clibs.ip_addr}:{clibs.upgrade_port}...")
+        try:
+            self.__c = socket(AF_INET, SOCK_STREAM)
+            self.__c.connect((clibs.ip_addr, clibs.upgrade_port))
+            self.__c.setblocking(True)
+            self.__c.settimeout(3)
+        except Exception as Err:
+            clibs.logger.error(f"{Err} | 连接 {clibs.ip_addr}:{clibs.upgrade_port} 失败...")
+            exit(1)
+
+    @staticmethod
+    def __do_package(cmd):
+        package_size = struct.pack('!I', len(cmd))
+        package_type = struct.pack('B', 2)
+        reserved_byte = struct.pack('B', 0)
+        return package_size + package_type + reserved_byte + cmd
+
+    def __recv_result(self, cmd):
+        time.sleep(2)
+        try:
+            res = self.__c.recv(10240).decode()
+        except timeout:
+            res = "ResponseNone"
+        clibs.logger.info(f"请求命令 {cmd.decode()} 的返回信息：\n{res[8:]}")
+        self.__c.close()
+        time.sleep(2)
+
+    def __exec(self, command: dict):
+        try:
+            self.__c.recv(10240)
+        except timeout:
+            pass
+        cmd = json.dumps(command, separators=(",", ":")).encode("utf-8")
+        self.__c.sendall(self.__do_package(cmd))
+        self.__recv_result(cmd)
+
+    def erase_cfg(self):
+        # 一键抹除机器人配置(.rc_cfg)、交互数据配置(interactive_data)，但保留用户日志
+        # 场景：如果xCore版本升级跨度过大，配置文件可能不兼容导致无法启动，可以使用该功能抹除配置，重新生成配置。
+        # 机器人参数、零点等会丢失！
+        self.__exec({"cmd": "erase"})
+
+    def clear_rubbish(self):
+        self.__exec({"cmd": "clearRubbish"})
+
+    def soft_reboot(self):
+        self.__exec({"cmd": "soft_reboot"})
+
+    def version_query(self):
+        self.__exec({"query": "version"})
+
+    def robot_reboot(self):
+        self.__exec({"cmd": "reboot"})
+
+    def reset_passwd(self):
+        # 不生效，有问题
+        self.__exec({"cmd": "passwd"})
+
+    def backup_origin(self):
+        # xCore
+        # .rc_cfg /
+        # interactive_data /
+        # module /
+        # demo_project /
+        # robot_cfg /
+        # dev_eni.xml
+        # ecat_license
+        # libemllI8254x.so & libemllI8254x_v3.so
+        # set_network_parameters
+        self.__exec({"cmd": "backup_origin"})
+
+    def origin_recovery(self):
+        self.__exec({"cmd": "recover"})
+
+
 if __name__ == "__main__":
-    initialization()
+    # initialization()
+    # fw_updater()
+    pass