import time import os import sys from pymodbus.client import ModbusTcpClient, ModbusSerialClient from pymodbus.exceptions import ModbusException from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder from pymodbus.constants import Endian def _motor_log(msg: str): """电机操作日志，直接写文件 + 刷盘""" try: log_dir = os.path.join(os.path.dirname(sys.executable), "logs") os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, "control_debug.log") with open(log_file, "a", encoding="utf-8") as f: f.write(f"[{time.strftime('%H:%M:%S.%f')[:-3]}] [MOTOR] {msg}\n") f.flush() os.fsync(f.fileno()) except Exception: pass volthege_min = 819 # 模拟量映射最小值（对应 0V/4mA）—— MT2 旧用 volthege_max = 4095 # 模拟量映射最大值（对应 10V/20mA）—— MT2 旧用 x_max = 1000 # 电机最大行程（PLC 沿用 0~1000 坐标，死区由 config 给定） raw_max = 32000 # PLC 模拟量映射最大值（0~32000，对应量程满量程） impulse_max = 163840 # 电机脉冲最大值（对应 x_max） def decode_signed_16(value): """将 Modbus 返回的 0~65535 解码为 16 位有符号整数。""" raw = int(value) if not 0 <= raw <= 0xFFFF: raise ValueError("16 位寄存器值必须在 0~65535 之间") return raw - 0x10000 if raw >= 0x8000 else raw def set_motor_limits(volthege_min_val=None, volthege_max_val=None, x_max_val=None): """更新电机限幅参数（由 GUI 高级设置页面调用） Args: volthege_min_val: 模拟量映射最小值，None 表示不更新 volthege_max_val: 模拟量映射最大值，None 表示不更新 x_max_val: 最大行程（对应 GUI 总限幅），None 表示不更新 """ global volthege_min, volthege_max, x_max if volthege_min_val is not None: volthege_min = volthege_min_val if volthege_max_val is not None: volthege_max = volthege_max_val if x_max_val is not None: x_max = x_max_val # ---------- PLC（Easy521）Modbus TCP 客户端 ---------- class Easy521ModbusClient: """ 信捷 Easy521 PLC 的 Modbus TCP 通讯类。 - D 寄存器为直接 Modbus 地址：读用功能码 03（保持寄存器），写用功能码 06。 - 模拟量映射：0~raw_max（32000）线性对应各量程满量程。 - 电机行程仍沿用 0~x_max（1000）坐标（死区由 config 给定），映射到 0~raw_max 输出。 对外接口沿用 MT2AM8Client 的 get_flow / get_pressure / set_motor_position， 只是这里的 ``channel`` 参数是寄存器地址。 """ def __init__( self, host="192.168.1.88", port=502, slave_id=1, flow_addr=700, flow_range=300.0, pressure_before_addr=720, pressure_before_range=300.0, pressure_after_addr=710, pressure_after_range=400.0, motor_output_addr=730, ): self.host = host self.port = port self.slave_id = slave_id self.flow_addr = int(flow_addr) self.flow_range = float(flow_range) self.pressure_before_addr = int(pressure_before_addr) self.pressure_before_range = float(pressure_before_range) self.pressure_after_addr = int(pressure_after_addr) self.pressure_after_range = float(pressure_after_range) self.motor_output_addr = int(motor_output_addr) self.client = ModbusTcpClient( host=host, port=port, timeout=3, retries=3, ) self.connected = False def connect(self): try: conn = self.client.connect() if conn: print(f"成功连接到 PLC {self.host}:{self.port}") self.connected = True else: print(f"无法连接到 PLC {self.host}:{self.port}") self.connected = False return conn except Exception as e: print(f"连接错误: {e}") self.connected = False return False def disconnect(self): self.client.close() self.connected = False print("连接已关闭") def _read_register(self, address): """读单个 16 位保持寄存器（功能码 03），返回原始整数，失败返回 None。""" try: result = self.client.read_holding_registers( address=int(address), count=1, slave=self.slave_id, ) if not result.isError(): return result.registers[0] print(f"读取寄存器 D{int(address)} 错误: {result}") return None except Exception as e: print(f"读取寄存器 D{int(address)} 异常: {e}") return None def _write_register(self, address, value): """写单个 16 位保持寄存器（功能码 06）。""" try: result = self.client.write_register( address=int(address), value=int(value), slave=self.slave_id, ) if result.isError(): print(f"写入寄存器 D{int(address)} 错误: {result}") return False return True except Exception as e: print(f"写入寄存器 D{int(address)} 异常: {e}") return False def get_flow(self, address): """读阀后流量，按 16 位有符号模拟量换算为 SLM。""" raw = self._read_register(address) if raw is None: return None raw = decode_signed_16(raw) return raw / raw_max * self.flow_range def get_pressure(self, address): """读压力，按 16 位有符号模拟量及通道量程换算为 kPa。""" address = int(address) if address == self.pressure_before_addr: rng = self.pressure_before_range elif address == self.pressure_after_addr: rng = self.pressure_after_range else: print(f"未知压力寄存器 D{address}，按阀后量程兜底") rng = self.pressure_after_range raw = self._read_register(address) if raw is None: return None raw = decode_signed_16(raw) return raw / raw_max * rng def set_motor_position(self, position, channel=None): """把 0~x_max 的行程映射到 0~raw_max 后写入电机输出寄存器。""" if channel is None: channel = self.motor_output_addr position = float(position) if not 0 <= position <= x_max: print(f"行程需在 0~{x_max:.0f} 之间") return False raw_value = int(position / x_max * raw_max) return self._write_register(channel, raw_value) # ---------- 新增：独立的电机 Modbus RTU 客户端类（含报文打印） ---------- class MotorModbusRTUClient: """电机 Modbus RTU 通讯客户端（增强调试报文打印）""" # 参数来源（GUI 页面1 Modbus RTU 区）： # port <- 端口号 (下拉) # baudrate <- 波特率 (默认 115200) # slave_id <- 站号 (默认 4) # bytesize <- 数据位 (默认 8) # stopbits <- 停止位 (默认 1) # parity <- 校验位 None/Odd/Even -> 'N'/'O'/'E' (默认 'N') def __init__(self, port='/dev/cu.usbserial-BG02B0IX', slave_id=4, baudrate=115200, bytesize=8, parity='N', stopbits=1): # def __init__(self, port='/dev/cu.usbserial-D30JITMY', slave_id=4, baudrate=115200): self.port = port self.slave_id = slave_id self.baudrate = baudrate self.bytesize = bytesize self.parity = parity self.stopbits = stopbits self.client = None def connect(self): """连接电机串口""" self.client = ModbusSerialClient( port=self.port, baudrate=self.baudrate, bytesize=self.bytesize, parity=self.parity, stopbits=self.stopbits, timeout=5 # 超时时间延长，便于观察 ) if self.client.connect(): print(f"电机串口 {self.port} 连接成功") return True else: print(f"电机串口 {self.port} 连接失败") return False def disconnect(self): """断开电机串口""" if self.client: self.client.close() self.client = None print("电机串口已关闭") @staticmethod def _compute_crc(data: bytes) -> bytes: """计算 Modbus CRC-16""" crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): if crc & 1: crc = (crc >> 1) ^ 0xA001 else: crc >>= 1 return crc.to_bytes(2, byteorder='little') def _print_sent_message(self, address, function_code, data_bytes): """构造完整报文并打印（含CRC）""" raw = bytes([self.slave_id, function_code]) + address.to_bytes(2, byteorder='big') + data_bytes crc = self._compute_crc(raw) full_msg = raw + crc hex_str = ' '.join(f'{b:02X}' for b in full_msg) # print(f"[发送] {hex_str}") def _write_single_register(self, address, value): """写单个寄存器（功能码 06）""" data_bytes = value.to_bytes(2, byteorder='big') self._print_sent_message(address, 0x06, data_bytes) try: result = self.client.write_register(address, value, slave=self.slave_id) if result.isError(): print(f"[接收] 错误响应: {result}") return False else: # print(f"[接收] 成功") return True except Exception as e: print(f"[接收] 异常: {e}") return False def _write_32bit(self, address, value): """ 写 32 位值到两个连续寄存器（功能码 10） 字节序：大端（高字节在前，高字在前） """ builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG) builder.add_32bit_uint(value) payload = builder.to_registers() # 构造数据部分：字节计数 + 各寄存器大端两字节 data_bytes = bytes([len(payload) * 2]) for reg in payload: data_bytes += reg.to_bytes(2, byteorder='big') self._print_sent_message(address, 0x10, data_bytes) try: result = self.client.write_registers(address, payload, slave=self.slave_id) if result.isError(): print(f"[接收] 错误响应: {result}") return False else: # print(f"[接收] 成功") return True except Exception as e: print(f"[接收] 异常: {e}") return False def init(self): """初始化电机参数（仅需调用一次）""" if not self.client or not self.client.connected: print("电机未连接，请先调用 connect()") return False print("开始初始化电机参数...") success = True # 1. 写模式 4 → 0x6007 print("--- 步骤1: 写模式 4 到 0x6007 ---") if not self._write_single_register(0x6007, 4): success = False print("模式写入失败") else: print("模式写入成功") # 2. 写速度 64000 → 0x6072 print("--- 步骤2: 写速度 64000 到 0x6072 ---") if not self._write_32bit(0x6072, 64000): success = False print("速度写入失败") else: print("速度写入成功") # 3. 写加速度 2400000 → 0x6067 print("--- 步骤3: 写加速度 96000 到 0x6067 ---") if not self._write_32bit(0x6067, 96000): success = False print("加速度写入失败") else: print("加速度写入成功") # 4. 写减速度 240000 → 0x6069 print("--- 步骤4: 写减速度 96000 到 0x6069 ---") if not self._write_32bit(0x6069, 96000): success = False print("减速度写入失败") else: print("减速度写入成功") if success: print("电机初始化完成。") else: print("电机初始化过程中出现错误。") return success def _read_single_register(self, address): """读取单个16位寄存器""" try: result = self.client.read_holding_registers(address, 1, slave=self.slave_id) if not result.isError(): return result.registers[0] else: print(f"读取寄存器 0x{address:X} 失败: {result}") return None except Exception as e: print(f"读取寄存器 0x{address:X} 异常: {e}") return None def set_position(self, position): """设置目标位置并立即启动（位置 0~impulse_max""" if not self.client or not self.client.connected: print("电机未连接，请先调用 connect()") return False position = int(position / x_max * impulse_max) if not (0 <= position <= impulse_max): print(f"位置值 {position} 超出范围 (0~{impulse_max})") return False self._write_32bit(0x6074, position) ret2 = self._write_single_register(0x6070, 112) if not ret2: print("第一次写入控制字失败，1ms后重试...") time.sleep(0.001) ret2 = self._write_single_register(0x6070, 112) if ret2: print("第二次重试成功") else: print("第二次重试仍然失败") return True def read_current_position(self): """ 读取电机当前位置（INT 型，32位有符号整数） 从寄存器 0x600E 开始，连续读取 2 个保持寄存器 字节序：大端（与写操作一致） :return: 当前位置值（int），读取失败返回 None """ if not self.client or not self.client.connected: print("电机未连接，无法读取位置") return None try: result = self.client.read_holding_registers( address=0x600E, count=2, slave=self.slave_id ) if result.isError(): print(f"读取位置寄存器失败: {result}") return None # 解码为 32 位有符号整数，使用与写操作相同的大端字节序 decoder = BinaryPayloadDecoder.fromRegisters( result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG ) position = decoder.decode_32bit_int() print(f"当前位置position: {position}") position_x = position / impulse_max * x_max return position_x except Exception as e: print(f"读取当前位置异常: {e}") return None # ---------- PC读取压力值 ---------- class PressureModbusRTUClient: """压力变送器 Modbus RTU 通讯客户端（读取16位压力值）""" def __init__(self, port='/dev/cu.usbserial-D30JITMY', slave_id=1, baudrate=9600, bytesize=8, parity='N', stopbits=1, timeout=3): """ 初始化压力客户端 :param port: 串口端口，如 COM3、/dev/ttyUSB0 :param slave_id: 从站地址（站号），默认 1 :param baudrate: 波特率，默认 9600 :param bytesize: 数据位，默认 8 :param parity: 校验位，默认 'N'（无校验） :param stopbits: 停止位，默认 1 :param timeout: 通讯超时时间（秒），默认 3 """ self.port = port self.slave_id = slave_id self.baudrate = baudrate self.bytesize = bytesize self.parity = parity self.stopbits = stopbits self.timeout = timeout self.client = None self.pressure_register_addr = 4 # 压力寄存器地址（04） def connect(self): """连接压力变送器串口""" self.client = ModbusSerialClient( port=self.port, baudrate=self.baudrate, bytesize=self.bytesize, parity=self.parity, stopbits=self.stopbits, timeout=self.timeout ) if self.client.connect(): print(f"压力串口 {self.port} 连接成功 (站号 {self.slave_id})") return True else: print(f"压力串口 {self.port} 连接失败") return False def disconnect(self): """断开压力串口""" if self.client: self.client.close() self.client = None print("压力串口已关闭") def get_current_p(self): """ 读取压力值 :return: 压力值（整数），若读取失败返回 None """ if not self.client or not self.client.connected: print("压力客户端未连接，请先调用 connect()") return None try: t1 = time.perf_counter() # 读取保持寄存器（功能码03），地址4，个数1 result = self.client.read_holding_registers( address=self.pressure_register_addr, count=1, slave=self.slave_id ) if result.isError(): print(f"压力读取错误: {result}") return None # 返回寄存器的第一个值（16位整数） pressure_raw = result.registers[0] # print(f"读压力用时：{time.perf_counter() - t1:.3f}s") return pressure_raw except ModbusException as e: print(f"压力读取 Modbus 异常: {e}") return None except Exception as e: print(f"压力读取未知异常: {e}") return None # ---------- 新增：MT2-AM8 模块 Modbus TCP 客户端 ---------- class MT2AM8Client: """ 艾莫迅 MT2-AM8 模块的 Modbus TCP 通讯类 - 默认 IP：192.168.1.12，端口 502，模块地址（站号）默认为 1 - 输入寄存器（AI）：地址 0x00~0x03（对应 PLC 地址 30001~30004） - 保持寄存器（AO）：地址 0x00~0x03（对应 PLC 地址 40001~40004） - 模拟量值范围：0～4095（对应 0～10V 或 0～20mA） """ def __init__(self, host="192.168.1.12", port=502, slave_id=1, pressure_range=400, flow_range=300): self.host = host self.port = port self.slave_id = slave_id self.pressure_range = pressure_range # 压力表量程上限 self.flow_range = flow_range # 流量计量程上限 self.client = ModbusTcpClient( host=host, port=port, timeout=3, retries=3 ) self.connected = False def connect(self): """连接模块""" try: conn = self.client.connect() if conn: print(f"成功连接到 MT2-AM8 模块 {self.host}:{self.port}") self.connected = True else: print(f"无法连接到 {self.host}:{self.port}") self.connected = False return conn except Exception as e: print(f"连接错误: {e}") self.connected = False return False def disconnect(self): """断开连接""" self.client.close() self.connected = False print("连接已关闭") def read_analog_input(self, channel): """ 读取单路模拟量输入原始值（16位无符号整数） :param channel: 通道号 0~3（对应 AI1~AI4） :return: 0~4095 的整数值，失败返回 None """ try: result = self.client.read_input_registers( address=channel, count=1, slave=self.slave_id ) if not result.isError(): return result.registers[0] else: print(f"读取输入寄存器错误: {result}") return None except Exception as e: print(f"读取模拟量输入异常: {e}") return None # def read_all_analog_inputs(self): # """ # 一次性读取全部 4 路模拟量输入 # :return: 长度为4的列表（int），失败返回 None # """ # try: # result = self.client.read_input_registers( # address=0, # count=4, # slave=self.slave_id # ) # if not result.isError(): # return result.registers # else: # print(f"读取全部输入寄存器错误: {result}") # return None # except Exception as e: # print(f"读取全部模拟量输入异常: {e}") # return None def write_analog_output(self, channel, value): """ 写入单路模拟量输出（保持寄存器） :param channel: 通道号 0~3（对应 AO1~AO4） :param value: {volthege_min}~{volthege_max} 的整数值 :return: True 成功，False 失败 """ # if not 0 <= channel <= 3: # print("通道号必须为 0~3") # return False if not 0 <= value <= volthege_max: print(f"值 {value} 超出范围 ({volthege_min}~{volthege_max})") return False try: result = self.client.write_register( address=channel, value=value, slave=self.slave_id ) return not result.isError() except Exception as e: print(f"写入模拟量输出异常: {e}") return False # def write_all_analog_outputs(self, values): # """ # 一次性写入全部 4 路模拟量输出（用于批量设置） # :param values: 长度为4的列表或元组，每个元素为 0~4095 # :return: True 成功，False 失败 # """ # if len(values) != 4: # print("需提供 4 个输出值") # return False # for v in values: # if not 0 <= v <= 4095: # print(f"值 {v} 超出范围 (0~4095)") # return False # try: # result = self.client.write_registers( # address=0, # values=list(values), # slave=self.slave_id # ) # return not result.isError() # except Exception as e: # print(f"批量写入模拟量输出异常: {e}") # return False def get_pressure(self, channel): """ 读取压力值并转换为实际物理量 转换公式：raw / (volthege_max - volthege_min) * pressure_range :param channel: 压力传感器模拟量通道地址 :return: 实际压力值（kPa），失败返回 None """ raw = self.read_analog_input(channel) if raw is None: return None pressure = (raw - volthege_min) / (volthege_max - volthege_min) * self.pressure_range # print(f"读取压力通道 {channel} 原始值: {raw}, 转换后压力: {pressure:.2f} kPa") return pressure def get_flow(self, channel): """ 读取流量值并转换为实际物理量 转换公式：raw / (volthege_max - volthege_min) * flow_range :param channel: 流量计模拟量通道地址 :return: 实际流量值（L/min），失败返回 None """ raw = self.read_analog_input(channel) if raw is None: return None flow = (raw - volthege_min) / (volthege_max - volthege_min) * self.flow_range # print(f"读取流量通道 {channel} 原始值: {raw}, 转换后流量: {flow:.2f} L/min") return flow # def set_motor_speed(self, voltage_percent): # """ # 通过模拟量输出控制电机（例如 0~100% 对应 0~10V） # :param voltage_percent: 0~100 的浮点数，表示百分比 # """ # if not 0 <= voltage_percent <= 100: # print("百分比需在 0~100 之间") # return False # # 将百分比映射到 0~4095 # raw_value = int(voltage_percent / 100.0 * 4095) # return self.write_analog_output(0, raw_value) # 假设电机接在 AO1 def set_motor_position(self, voltage_distance, channel=0): """ 通过模拟量输出控制电机（例如 0~1000 对应 0~10V） :param voltage_distance: 0~1000 的浮点数，表示行程 :param channel: 模拟量输出通道号，默认 0（AO1） """ if not (0 <= voltage_distance <= x_max): print(f"行程需在 0~{x_max} 之间") return False # 将行程映射到 0~4095 # Map the full travel range to the configured analog-output range. # The lower endpoint must include volthege_min; otherwise position 0 # produces raw value 0 and is rejected by write_analog_output(). raw_value = int( voltage_distance / x_max * volthege_max ) # print(f"设置电机行程为 {voltage_distance}，模拟量输出值 {raw_value}") return self.write_analog_output(channel, raw_value) # ---------- 主函数：测试示例 ---------- # if __name__ == "__main__": # # （可选）启用 pymodbus 详细日志，可观察底层收发帧 # # logging.basicConfig() # # logging.getLogger('pymodbus').setLevel(logging.DEBUG) # # motor = MotorModbusRTUClient() # # # print("连接电机 (Modbus RTU)...") # if not motor.connect(): # print("电机连接失败，退出。") # exit(1) # # # 增加短暂延时，等待驱动器接口就绪 # time.sleep(1) # # print("初始化电机参数...") # if not motor.init(): # print("电机初始化失败，退出。") # motor.disconnect() # exit(1) # # print("\n========== 电机位置控制测试 ==========") # print("输入目标位置 (0~60000)，输入 'q' 退出。函数已修改，只需输入开度。\n") # # try: # while True: # user_input = input("目标位置: ").strip() # if user_input.lower() in ('q', 'quit', 'exit'): # break # if not user_input: # continue # try: # pos = int(user_input) # motor.set_position(pos) # except ValueError: # print("错误：请输入有效的整数。") # except KeyboardInterrupt: # print("\n用户中断测试。") # finally: # motor.disconnect() # print("程序结束。")