pengqi/wire_controlsystem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改传送带电机、舵机、达妙电机等相关接口

Browse Source
This commit is contained in:
pengqi
2026-02-26 15:42:22 +08:00
parent 5c9946dfc7
commit 8d2e91b8b3
10 changed files with 1642 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

392
conveyor_controller/conveyor_motor.py Normal file
View File

@ -0,0 +1,392 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
# @Time : 2026/2/10 09:42
# @Author : reenrr
# @File : conveyor_motor.py
# @Desc : 传送带电机相关接口 需要测试
"""
import time
from modbus import (write_single_register, open_serial_port, RTU_HANDLE_MAP, RTU_HANDLE_LOCK, read_holding_register)
from error_code import ModbusError
PORT = 'COM4'
BAUDRATE = 115200
DATABITS = 8
STOPBITS = 1
PARITY = 0
# --------寄存器地址---------
MODE_REG_ADDR = 0x6200
SPEED_REG_ADDR = 0x6203
ACCELERATION_REG_ADDR = 0x6204
DECELERATION_REG_ADDR = 0x6205
START_OR_STOP_REG_ADDR = 0x6002
# ---------相关写入值------------
START_CMD = 0x0010
STOP_CMD = 0x0040
SPEED_CMD = 0x0002
ABSOLUTE_POSITION_CMD = 0x0001
handle1 = open_serial_port(
port=PORT,
baudrate=BAUDRATE,
databits=DATABITS,
stopbits=STOPBITS,
parity=PARITY
)
def _check_handle_valid():
"""通用句柄校验函数"""
if handle1 is None:
err_msg = ModbusError.get_error_desc(ModbusError.MODBUS_ERR_SERIAL_HANDLE)
print(f"错误：{err_msg}")
raise RuntimeError(err_msg)
with RTU_HANDLE_LOCK:
handle_obj = RTU_HANDLE_MAP.get(handle1)
if not handle_obj or not handle_obj.is_open:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_SERIAL_HANDLE)}（句柄{handle1}）"
print(f"错误：{err_msg}")
raise RuntimeError(err_msg)
def set_motor_speed(station_addr: int, speed_value: int) -> bool:
"""
设置伺服电机速度
:param station_addr: 从机地址
:param speed_value: 速度值
:return: True--设置成功，False--设置失败
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
value = handle_obj.decimal_to_16bit(speed_value)
print(f"速度值转换：{speed_value} → {value}（0x{value:04X}）")
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_CONVERT)} - {str(e)}"
print(f"错误：{err_msg}")
raise ValueError(err_msg) from e # 转换失败抛ValueError，保留原始异常栈
try:
ret_code = write_single_register(handle1, station_addr, SPEED_REG_ADDR, value, 0, 1)
if ret_code == ModbusError.MODBUS_SUCCESS:
print(f"成功：电机速度设置完成（从站{station_addr}，地址0x{SPEED_REG_ADDR:04X}，值{value}）")
return True
else:
# 写寄存器返回错误码：抛OSError
err_desc = ModbusError.get_error_desc(
ret_code) if ret_code in ModbusError.__members__.values() else f"未知错误码{ret_code}"
err_msg = f"速度设置失败，{err_desc}"
print(f"失败：{err_msg}")
raise OSError(err_msg, ret_code) # 携带错误码，便于上层解析
except OSError:
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_WRITE)} - {str(e)}"
print(f"异常：{err_msg}")
raise Exception(err_msg) from e
def set_motor_acceleration(station_addr: int, value: int) -> bool:
"""
设置伺服电机加速度
:param station_addr: 从机地址
:param value: 加速度值
:return: True--设置成功，False--设置失败
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
conv_value = handle_obj.decimal_to_16bit(value)
print(f"加速度值转换：{value} → {conv_value}（0x{conv_value:04X}）")
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_CONVERT)} - {str(e)}"
print(f"错误：{err_msg}")
raise ValueError(err_msg) from e
try:
ret_code = write_single_register(handle1, station_addr, ACCELERATION_REG_ADDR, conv_value, 0, 1)
if ret_code == ModbusError.MODBUS_SUCCESS:
print(f"成功：电机加速度设置完成（从站{station_addr}，地址0x{ACCELERATION_REG_ADDR:04X}，值{conv_value}）")
return True
else:
# 写寄存器返回错误码：抛OSError（携带错误码）
err_desc = ModbusError.get_error_desc(
ret_code) if ret_code in ModbusError.__members__.values() else f"未知错误码{ret_code}"
err_msg = f"加速度设置失败，{err_desc}"
print(f"失败：{err_msg}")
raise OSError(err_msg, ret_code) # 错误码存入args，上层可解析
except OSError:
# 主动抛出的OSError直接上抛，不做额外处理
raise
except Exception as e:
# 其他未知异常：抛通用Exception，保留原始栈
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_WRITE)} - {str(e)}"
print(f"异常：{err_msg}")
raise Exception(err_msg) from e
def set_motor_deceleration(station_addr: int, value: int) -> bool:
"""
设置伺服电机减速度
:param station_addr: 从机地址
:param value: 减速度值
:return: True--设置成功，False--设置失败
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
conv_value = handle_obj.decimal_to_16bit(value)
print(f"减速度值转换：{value} → {conv_value}（0x{conv_value:04X}）")
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_CONVERT)} - {str(e)}"
print(f"错误：{err_msg}")
raise ValueError(err_msg) from e
try:
ret_code = write_single_register(handle1, station_addr, DECELERATION_REG_ADDR, conv_value, 0, 1)
if ret_code == ModbusError.MODBUS_SUCCESS:
print(f"成功：电机减速度设置完成（从站{station_addr}，地址0x{DECELERATION_REG_ADDR:04X}，值{conv_value}）")
return True
else:
err_desc = ModbusError.get_error_desc(
ret_code) if ret_code in ModbusError.__members__.values() else f"未知错误码{ret_code}"
err_msg = f"减速度设置失败，{err_desc}"
print(f"失败：{err_msg}")
raise OSError(err_msg, ret_code)
except OSError:
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_WRITE)} - {str(e)}"
print(f"异常：{err_msg}")
raise Exception(err_msg) from e
def set_motor_mode(station_addr: int, value: int) -> bool:
"""
设置电机模式
:param station_addr: 从机地址
:param value: 0--速度模式 1--绝对位置模式
:return: bool 设置是否成功
"""
_check_handle_valid()
try:
if value == 0:
write_cmd = SPEED_CMD
mode_desc = "速度模式"
elif value == 1:
write_cmd = ABSOLUTE_POSITION_CMD
mode_desc = "绝对位置模式"
ret_code = write_single_register(handle1, station_addr, MODE_REG_ADDR, write_cmd, 0, 1)
if ret_code == ModbusError.MODBUS_SUCCESS:
print(f"成功：电机模式设置完成（从站{station_addr}，地址0x{MODE_REG_ADDR:04X}，模式：{mode_desc}）")
return True
else:
err_desc = ModbusError.get_error_desc(
ret_code) if ret_code in ModbusError.__members__.values() else f"未知错误码{ret_code}"
err_msg = f"电机{mode_desc}设置失败，{err_desc}"
print(f"失败：{err_msg}")
raise OSError(err_msg, ret_code)
except OSError:
# 主动抛出的OSError直接上抛，不做额外处理
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_MOTOR_MODE)} - {str(e)}"
print(f"异常：{err_msg}")
raise Exception(err_msg) from e
def get_motor_speed(station_addr: int) -> int:
"""
获取电机速度
:param station_addr: 从站地址
:return: 返回电机速度值
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
reg_values = read_holding_register(handle1, station_addr, SPEED_REG_ADDR, 1, 0, [], 1)
if not reg_values or len(reg_values) != 1:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_FORMAT)}，数据:{reg_values}"
print(f"错误：{err_msg}")
raise OSError(err_msg)
unsigned_speed = reg_values[0]
signed_speed = handle_obj.uint16_to_int16(unsigned_speed)
print(f"成功：获取电机速度（从站{station_addr})-->无符号值:{unsigned_speed}-->有符号值:{signed_speed}")
return signed_speed
except OSError:
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_READ)} - {str(e)}"
print(f"错误：{err_msg}")
raise Exception(err_msg) from e
def get_motor_acceleration(station_addr: int) -> int:
"""
获取电机加速度
:param station_addr: 从站地址
:return: 电机加速度值
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
reg_values = read_holding_register(handle1, station_addr, ACCELERATION_REG_ADDR, 1, 0, [], 1)
if not reg_values or len(reg_values) != 1:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_FORMAT)}，数据:{reg_values}"
print(f"错误：{err_msg}")
raise OSError(err_msg)
unsigned_speed = reg_values[0]
signed_speed = handle_obj.uint16_to_int16(unsigned_speed)
print(f"成功：获取电机加速度（从站{station_addr})-->无符号值:{unsigned_speed}-->有符号值:{signed_speed}")
return signed_speed
except OSError:
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_READ)} - {str(e)}"
print(f"错误：{err_msg}")
raise Exception(err_msg) from e
def get_motor_deceleration(station_addr: int) -> int:
"""
获取电机减速度
:param station_addr: 从站地址
:return: i电机减速度值
"""
_check_handle_valid()
handle_obj = RTU_HANDLE_MAP.get(handle1)
try:
reg_values = read_holding_register(handle1, station_addr, DECELERATION_REG_ADDR, 1, 0, [], 1)
if not reg_values or len(reg_values) != 1:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_DATA_FORMAT)}，数据:{reg_values}"
print(f"错误：{err_msg}")
raise OSError(err_msg)
unsigned_speed = reg_values[0]
signed_speed = handle_obj.uint16_to_int16(unsigned_speed)
print(f"成功：获取电机减速度（从站{station_addr})-->无符号值:{unsigned_speed}-->有符号值:{signed_speed}")
return signed_speed
except OSError:
raise
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_REG_READ)} - {str(e)}"
print(f"错误：{err_msg}")
raise Exception(err_msg) from e
def move_motor(station_addr: int, value: bool) -> bool:
"""
启停电机
:param station_addr: 从站地址
:param value: True--启动电机，False--停止电机
:return: 是否设置成功
"""
_check_handle_valid()
try:
if value:
write_cmd = START_CMD
op = "启动"
else:
write_cmd = STOP_CMD
op = "停止"
ret_code = write_single_register(handle1, station_addr, START_OR_STOP_REG_ADDR, write_cmd, 0,1)
if ret_code == ModbusError.MODBUS_SUCCESS:
print(f"成功：电机{op}指令已发送（从站{station_addr}）")
return True
else:
err_desc = ModbusError.get_error_desc(
ret_code) if ret_code in ModbusError.__members__.values() else f"未知错误码{ret_code}"
err_msg = f"电机{op}指令发送失败，{err_desc}"
print(f"失败：{err_msg}")
raise OSError(err_msg, ret_code)
except OSError:
raise
except Exception as e:
op = "启动" if value else "停止"
err_msg = f"异常：电机{op}指令执行失败 - {str(e)}"
print(f"异常:{err_msg}")
raise Exception(err_msg)
def sync_motor_move(value: bool):
"""
同步传送带电机
:param value: 启停传送点电机 True--同步启动传送带电机 False--同步停止传送带电机
"""
op_desc = "启动" if value else "停止"
if not isinstance(value, bool):
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_PARAM_TYPE)}（同步{op_desc}）：参数必须是布尔值，当前值：{value}（类型：{type(value)}）"
print(f"错误：{err_msg}")
raise TypeError(err_msg)
try:
print(f"开始同步{op_desc}传送带电机（从站1、2）...")
move_motor(1, value)
move_motor(2, value)
print(f"成功：所有传送带电机已同步{op_desc}完成")
except Exception as e:
err_msg = f"{ModbusError.get_error_desc(ModbusError.MODBUS_ERR_MOTOR_SYNC)}（{op_desc}）：{str(e)}"
print(f"错误：{err_msg}")
# 抛RuntimeError，保留原始异常链，便于上层定位具体失败电机
raise RuntimeError(err_msg) from e
# ------------调试接口----------
if __name__ == '__main__':
# 配置传送带电机参数 只需配置一次
set_motor_mode(1, 0) # 配置成速度模式
set_motor_speed(1, -30)
set_motor_acceleration(1, 50)
set_motor_deceleration(1, 50)
set_motor_mode(2, 0) # 配置成速度模式
set_motor_speed(2, -30)
set_motor_acceleration(2, 50)
set_motor_deceleration(2, 50)
sync_motor_move(True)
time.sleep(1)
sync_motor_move(False)