INIT

code/clibs.py

@@ -0,0 +1,46 @@
+from sys import stdout
+from os import mkdir, remove, listdir
+from os.path import exists
+from socket import setdefaulttimeout
+from loguru import logger
+
+
+ip_addr = "192.168.2.160"
+# ip_addr = "192.168.168.128"
+# ip_addr = "192.168.40.130"
+ssh_port = 22
+socket_port = 5050
+username = "luoshi"
+password = "luoshi2019"  # for real robot
+# password = "forpqart"  # for robot vm
+xService_port = 6666
+external_port = 8080
+modbus_port = 502
+interval = 0.5  # interval after actions being triggered, such as modbus/socket/external communication operations
+heartbeat_interval = 2
+RADIAN = 57.3  # 180 / 3.1415926
+MAX_FRAME_SIZE = 1024
+TIMEOUT = 3
+setdefaulttimeout(TIMEOUT)
+PREFIX = "../assets"
+log_path = f"{PREFIX}/logs/"
+log_data_debug = f"{PREFIX}/logs/debug.log"
+
+if not exists(log_path):
+    mkdir(log_path)
+else:
+    for _ in listdir(log_path):
+        remove("".join([log_path, _]))
+logger.remove()
+logger.add(stdout, level="INFO")
+logger.add(
+    sink=log_data_debug,
+    level="DEBUG",
+    format="{time: YYYY-MM-DD HH:mm:ss} | {level} | {message}",
+    rotation="10 KB",
+    encoding="utf-8",
+    enqueue=True,
+    diagnose=True,
+    colorize=True,
+    # filter=lambda x: "DEBUG" in str(x["level"]).upper()
+)

code/common.py

@@ -0,0 +1,100 @@
+import time
+import openapi
+import json
+
+
+def initialization():
+    hr = openapi.HmiRequest()
+    pd = openapi.PreDos()
+    # 推送配置文件
+    robot_params = hr.get_robot_params()
+    robot_type = robot_params["robot_type"]
+    security_type = robot_params["security_type"]
+    controller_type = robot_params["controller_type"]
+    io_device_file = "_".join(["io_device", controller_type, security_type, "1"])
+
+    user_settings = "/home/luoshi/bin/controller/user_settings"
+    interactive_data = f"/home/luoshi/bin/controller/interactive_data/{robot_type}"
+
+    config_files = [
+        "..\\assets\\configs\\fieldbus_device.json",
+        "..\\assets\\configs\\registers.json",
+        "..\\assets\\configs\\registers.xml"
+    ]
+    for config_file in config_files:
+        filename = config_file.split("\\")[-1]
+        pd.push_file_to_server(config_file, f"{user_settings}/{filename}")
+        pd.push_file_to_server(config_file, f"{interactive_data}/{filename}")
+
+    io_device_autotest = {'ai_num': 0, 'ao_num': 0, 'di_num': 16, 'do_num': 16, 'extend_attr': {'mode': 'slaver', 'name': 'autotest', 'type': 'MODBUS'}, 'id': 5, 'name': 'autotest', 'type': 6}
+    io_device_file_local = f"..\\assets\\configs\\{io_device_file}"
+    io_device_file_local_tmp = f"..\\assets\\configs\\{io_device_file}_tmp"
+    io_device_file_remote = f"{user_settings}/{io_device_file}"
+    pd.pull_file_from_server(io_device_file_remote, io_device_file_local)
+    with open(io_device_file_local, mode="r", encoding="utf-8") as f:
+        data = json.load(f)
+        for _ in data["device_list"]:
+            if _["extend_attr"].get("name", None) == "autotest":
+                break
+        else:
+            data["device_list"].append(io_device_autotest)
+            with open(io_device_file_local_tmp, mode="w", encoding="utf-8") as f_tmp:
+                json.dump(data, f_tmp, indent=4)
+            pd.push_file_to_server(io_device_file_local_tmp, f"{user_settings}/{io_device_file}")
+            pd.push_file_to_server(io_device_file_local_tmp, f"{interactive_data}/{io_device_file}")
+
+    hr.reload_io()
+    hr.reload_registers()
+    hr.reload_fieldbus()
+    hr.set_fieldbus_device_params("autotest", True)
+
+    md = openapi.ModbusRequest()
+    # 触发急停并恢复
+    md.r_soft_estop(0)
+    md.r_soft_estop(1)
+
+    # 断开示教器连接
+    hr.switch_tp_mode("without")
+
+    # 清空 system IO 配置
+    hr.update_system_io_configuration([], [], [], [], [])
+
+    # 关闭缩减模式
+    md.r_reduced_mode(0)
+
+    # 关闭安全区域
+    hr.set_safety_area_overall(False)
+    hr.set_safety_area_signal(False)
+    for i in range(10):
+        hr.set_safety_area_enable(i, False)
+
+    # 打开外部通信
+    hr.set_socket_params(True, "", "name", "8080", "\r", 1, True, True, 0, 10)
+
+    # 关闭拖动
+    hr.set_drag_params(False, 1, 2)
+
+    # 关闭碰撞检测
+    hr.set_collision_params(False, 0, 1, 100)
+
+    # 清除所有错误码
+    hr.set_filtered_error_code("clear", [])
+
+    # 回拖动位姿
+    hr.switch_operation_mode("manual")
+    hr.switch_motor_state("on")
+    hr.set_quickturn_pos(enable_drag=True)
+    hr.move2quickturn("drag")
+    while True:
+        if md.w_robot_moving():
+            time.sleep(1)
+        else:
+            break
+    hr.stop_move(1)
+    hr.switch_motor_state("off")
+
+    # 清除所有告警
+    md.r_clear_alarm()
+
+
+initialization()