add new laiyu liquid driver, yaml and json files

unilabos/devices/laiyu_liquid_test/driver_enable_move_test.py

@@ -0,0 +1,138 @@
+
+import os
+import time
+import json
+import logging
+from xyz_stepper_driver import ModbusRTUTransport, ModbusClient, XYZStepperController, MotorStatus
+
+# ========== 日志配置 ==========
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("XYZ_Debug")
+
+
+def create_controller(port: str = "/dev/ttyUSB1", baudrate: int = 115200) -> XYZStepperController:
+    """
+    初始化通信层与三轴控制器
+    """
+    logger.info(f"🔧 初始化控制器: {port} @ {baudrate}bps")
+    transport = ModbusRTUTransport(port=port, baudrate=baudrate)
+    transport.open()
+    client = ModbusClient(transport)
+    return XYZStepperController(client=client, port=port, baudrate=baudrate)
+
+
+def load_existing_soft_zero(ctrl: XYZStepperController, path: str = "work_origin.json") -> bool:
+    """
+    如果已存在软零点文件则加载，否则返回 False
+    """
+    if not os.path.exists(path):
+        logger.warning("⚠ 未找到已有软零点文件，将等待人工定义新零点。")
+        return False
+
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        origin = data.get("work_origin_steps", {})
+        ctrl.work_origin_steps = origin
+        ctrl.is_homed = True
+        logger.info(f"✔ 已加载软零点文件：{path}")
+        logger.info(f"当前软零点步数: {origin}")
+        return True
+    except Exception as e:
+        logger.error(f"读取软零点文件失败: {e}")
+        return False
+
+
+def test_enable_axis(ctrl: XYZStepperController):
+    """
+    依次使能 X / Y / Z 三轴
+    """
+    logger.info("=== 测试各轴使能 ===")
+    for axis in ["X", "Y", "Z"]:
+        try:
+            result = ctrl.enable(axis, True)
+            if result:
+                vals = ctrl.get_status(axis)
+                st = MotorStatus(vals[3])
+                logger.info(f"{axis} 轴使能成功，当前状态: {st.name}")
+            else:
+                logger.error(f"{axis} 轴使能失败")
+        except Exception as e:
+            logger.error(f"{axis} 轴使能异常: {e}")
+        time.sleep(0.5)
+
+
+def test_status_read(ctrl: XYZStepperController):
+    """
+    读取各轴当前状态（调试）
+    """
+    logger.info("=== 当前各轴状态 ===")
+    for axis in ["X", "Y", "Z"]:
+        try:
+            vals = ctrl.get_status(axis)
+            st = MotorStatus(vals[3])
+            logger.info(
+                f"{axis}: steps={vals[0]}, speed={vals[1]}, "
+                f"current={vals[2]}, status={st.name}"
+            )
+        except Exception as e:
+            logger.error(f"获取 {axis} 状态失败: {e}")
+        time.sleep(0.2)
+
+
+def redefine_soft_zero(ctrl: XYZStepperController):
+    """
+    手动重新定义软零点
+    """
+    logger.info("=== ⚙️ 重新定义软零点 ===")
+    ctrl.define_current_as_zero("work_origin.json")
+    logger.info("✅ 新软零点已写入 work_origin.json")
+
+
+def test_soft_zero_move(ctrl: XYZStepperController):
+    """
+    以软零点为基准执行三轴运动测试
+    """
+    logger.info("=== 测试软零点相对运动 ===")
+    ctrl.move_xyz_work(x=100.0, y=100.0, z=40.0, speed=100, acc=800)
+
+    for axis in ["X", "Y", "Z"]:
+        ctrl.wait_complete(axis)
+
+    test_status_read(ctrl)
+    logger.info("✅ 软零点运动测试完成")
+
+
+def main():
+    ctrl = create_controller(port="/dev/ttyUSB1", baudrate=115200)
+
+    try:
+        test_enable_axis(ctrl)
+        test_status_read(ctrl)
+
+        # === 初始化或加载软零点 ===
+        loaded = load_existing_soft_zero(ctrl)
+        if not loaded:
+            logger.info("👣 首次运行，定义软零点并保存。")
+            ctrl.define_current_as_zero("work_origin.json")
+
+        # === 软零点回归动作 ===
+        ctrl.return_to_work_origin()
+
+        # === 可选软零点运动测试 ===
+        # test_soft_zero_move(ctrl)
+
+    except KeyboardInterrupt:
+        logger.info("🛑 手动中断退出")
+
+    except Exception as e:
+        logger.exception(f"❌ 调试出错: {e}")
+
+    finally:
+        if hasattr(ctrl.client, "transport"):
+            ctrl.client.transport.close()
+        logger.info("串口已安全关闭 ✅")
+
+
+if __name__ == "__main__":
+    main()

unilabos/devices/laiyu_liquid_test/driver_status_test.py

@@ -0,0 +1,58 @@
+
+import logging
+from xyz_stepper_driver import (
+    ModbusRTUTransport,
+    ModbusClient,
+    XYZStepperController,
+    MotorAxis,
+)
+
+logger = logging.getLogger("XYZStepperCommTest")
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+
+def test_xyz_stepper_comm():
+    """仅测试 Modbus 通信是否正常（并输出寄存器数据，不做电机运动）"""
+    port = "/dev/ttyUSB1"
+    baudrate = 115200
+    timeout = 1.2  # 略长避免响应被截断
+
+    logger.info(f"尝试连接 Modbus 设备 {port} ...")
+    transport = ModbusRTUTransport(port, baudrate=baudrate, timeout=timeout)
+    transport.open()
+
+    client = ModbusClient(transport)
+    ctrl = XYZStepperController(client)
+
+    try:
+        logger.info("✅ 串口已打开，开始读取三个轴状态（打印寄存器内容） ...")
+        for axis in [MotorAxis.X, MotorAxis.Y, MotorAxis.Z]:
+            addr = ctrl.axis_addr[axis]
+
+            try:
+                # # 在 get_status 前打印原始寄存器内容
+                # regs = client.read_registers(addr, ctrl.REG_STATUS, 6)
+                # hex_regs = [f"0x{val:04X}" for val in regs]
+                # logger.info(f"[{axis.name}] 原始寄存器 ({len(regs)} 个): {regs} -> {hex_regs}")
+
+                # 调用 get_status() 正常解析
+                status = ctrl.get_status(axis)
+                logger.info(
+                    f"[{axis.name}] ✅ 通信正常: steps={status.steps}, speed={status.speed}, "
+                    f"current={status.current}, status={status.status.name}"
+                )
+
+            except Exception as e_axis:
+                logger.error(f"[{axis.name}] ❌ 通信失败: {e_axis}")
+
+
+    except Exception as e:
+        logger.error(f"❌ 通讯测试失败: {e}")
+
+    finally:
+        transport.close()
+        logger.info("🔌 串口已关闭")
+
+
+if __name__ == "__main__":
+    test_xyz_stepper_comm()

unilabos/devices/laiyu_liquid_test/work_origin.json

@@ -0,0 +1,8 @@
+{
+    "work_origin_steps": {
+        "x": 11799,
+        "y": 11476,
+        "z": 3312
+    },
+    "timestamp": "2025-11-04T15:31:09.802155"
+}