louis
cfeedcf887
- 前馈冻结改用可配置的 3 秒流量绝对误差滑窗，修复采样抖动造成的重复解冻，并保留下游扰动后的自动更新 - 支持非单调阀特性反解、最小可测面积以下直接全关，以及闭阀端精细扫点与辨识开关 - 调整双压力判稳、最长等待时间、在线入口全开收尾和闭环四联图 - 更新配置、README、阀模型及离线测试，归档本轮实验数据与诊断产物
733 lines
27 KiB
Python
733 lines
27 KiB
Python
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("程序结束。")
|