chuyiwen/AutoControlSystem-G
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

料袋末尾检测

Browse Source
This commit is contained in:
fujinliang
2025-09-28 09:48:12 +08:00
parent 61e6d39dbf
commit 1916fb0bbc
62 changed files with 29808 additions and 21103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
EMV/EMV.py
View File

@ -7,22 +7,22 @@ import threading
import logging
from PySide6.QtCore import Signal, QObject
import numpy as np
from pandas.core.arrays import boolean
import Constant
class RelayController:
need_origin_signal = Signal(str)
take_no_photo_sigal = Signal()
update_detect_image = Signal(np.ndarray)
class RelayController(QObject):
log_signal = Signal(int, str)
take_robot_signal = Signal()
def __init__(self, host='192.168.0.18', port=50000):
super().__init__()
# ===================== 全局线程延时参数 =====================
self.sensor1_loop_delay = 0.2 # SENSOR1 线程轮询间隔（秒）
self.sensor1_error_delay = 1.0 # SENSOR1 出错时延时（秒）
self.sensor1_loop_delay = 0.1 # SENSOR1 线程轮询间隔（秒）
self.sensor1_error_delay = 1.0 # SENSOR1 出错或暂停时延时（秒）
self.sensor1_post_action_delay = 0.2 # SENSOR1 每次循环后延时（秒）
self.sensor2_loop_delay = 0.1 # SENSOR2 线程轮询间隔（秒）
self.sensor2_loop_delay = 0.2 # SENSOR2 线程轮询间隔（秒）
self.sensor2_loop_lost=0.1 # SENSOR2 线程轮询间隔（秒）
# self.sensor2_loop_delay = 0.5 # SENSOR2 线程轮询间隔（秒）
self.sensor2_error_delay = 0.5 # SENSOR2 出错时延时（秒）
self.sensor2_post_action_delay = 0.2 # SENSOR2 每次循环后延时（秒）
@ -30,7 +30,7 @@ class RelayController:
self.delay_conveyor = 0.5 # 传送带开/关动作延时（一半时间，我在控制程序和线程都加了一样的延时）
self.delay_pusher = 0.05 # 推板开/关动作延时
self.delay_clamp = 0.5 # 夹爪动作延时
self.delay_after_pusher = 3.0 # 推板推出后到重启传动带时间
self.delay_after_pusher = 5.0 # 推板推出后到重启传动带时间
# ===================== 传感器稳定检测参数 =====================
self.sensor_stable_duration = 1.0 # 传感器状态稳定检测时间（秒）
@ -51,15 +51,16 @@ class RelayController:
self.SENSOR2 = 'sensor2'
self.valve_commands = {
self.CONVEYOR1: {'open': '00000000000601050000FF00', 'close': '000000000006010500000000'},
self.CONVEYOR1: {'open': '000000000006010500070000', 'close': '00000000000601050007FF00'},
# self.CONVEYOR11: {'open': '00000000000601050000FF00', 'close': '000000000006010500000000'},
self.PUSHER: {'open': '00000000000601050001FF00', 'close': '000000000006010500010000'},
#滚筒，2000 0012正转，2000 0022 2001变频器频率调整 2000正反转。
self.CONVEYOR2: {'open': '000100000006020620000012', 'close': '000100000006020620000001'},
#DO4
self.CLAMP: {'open': '00000000000601050003FF00', 'close': '000000000006010500030000'},
#DO5
#DO5 回 DO2推
self.PUSHER1: {'open': '00000000000601050004FF00', 'close': '000000000006010500040000'},
#滚筒反转
self.CONVEYOR2_REVERSE: {'open': '000100000006020620000022', 'close': '000100000006020620000001'}
}
@ -99,7 +100,8 @@ class RelayController:
}
# ===================== 状态控制变量 =====================
self._running = False
self._running = False #线程运行标识
self._ispause = False #线程暂停标识
self._sensor1_thread = None
self._sensor2_thread = None
@ -109,13 +111,14 @@ class RelayController:
self.sensor1_triggered = False
self.sensor1_last_time = 0
self.sensor2_ready = True
self.motor_stopped_by_sensor2 = False
self.sensor2_ready = False #默认不打开
self.motor_stopped_by_sensor2 = True
# ===================== 基础通信方法 =====================
def send_command(self, command_hex, retry_count=2, source='unknown'):
if Constant.DebugPosition:
return True
# print(f"[发送命令] {command_hex} ({source})")
return None
byte_data = binascii.unhexlify(command_hex)
for attempt in range(retry_count):
try:
@ -183,7 +186,7 @@ class RelayController:
if response and len(response) >= 10:
hex_response = binascii.hexlify(response).decode('utf-8')
print(f"[原始响应][{command_type}] {hex_response}")
# print(f"[原始响应][{command_type}] {hex_response}")
# 假设传感器数据从第 9 字节开始，长度为 2 字节
for name, bit_index in self.sensor_bit_map.items():
@ -244,6 +247,27 @@ class RelayController:
time.sleep(self.sensor2_loop_delay)
return False
def is_valid_sensor2_status_lost(self, sensor_name):
stable_count = 0
_try_nums=5 # 尝试次数
for _ in range(_try_nums):
responses = self.get_all_sensor_responses('sensors')
response = responses.get(sensor_name)
if not response:
print(f"[警告] 无法获取 {sensor_name} 的响应，尝试重试...")
stable_count = 0
else:
status_code = self.parse_status_code(response)
if status_code not in self.required_codes_1:
stable_count += 1
if stable_count >= self.sensor_max_attempts:
return True
else:
stable_count = 0
time.sleep(self.sensor2_loop_lost)
return False
# ===================== 动作控制方法 =====================
def open(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False, pusher1=False, conveyor2_reverse=False):
if Constant.DebugPosition:
@ -291,6 +315,11 @@ class RelayController:
# ===================== 传感器处理线程 =====================
def handle_sensor1(self):
while self._running:
if self._ispause and not self.sensor1_triggered:
#暂停线程， 保证线程1执行完传感器操作
time.sleep(self.sensor1_error_delay)
continue
try:
if self.is_valid_sensor_status(self.SENSOR1):
current_time = time.time()
@ -305,15 +334,16 @@ class RelayController:
self.open(pusher=True)
time.sleep(self.delay_pusher)
self.close(pusher=True)
# 推板推出后重新启动电机时间传送带1延时
# 推板推出后重新启动电机时间传送带1延时5秒
time.sleep(self.delay_after_pusher)
# 3.重新开启包装机皮带电机: 开启conveyor1
self.open(conveyor1=True)
time.sleep(self.delay_conveyor)
# 4.推板关闭：推板收回来动作
# 4.推板关闭：推板收回来动作
self.open(pusher1=True)
time.sleep(self.delay_pusher)
self.close(pusher1=True)
# 3.重新开启包装机皮带电机: 开启conveyor1
time.sleep(self.delay_conveyor)
self.open(conveyor1=True)
time.sleep(1)
# 5. 状态检查（可选）
status = self.get_all_device_status()
@ -330,47 +360,59 @@ class RelayController:
time.sleep(self.sensor1_error_delay)
def handle_sensor2(self):
_is_pause_close=True
#是否料袋尾部（有信号--》无信号）
_is_signal=False
while self._running:
if self._ispause:
#暂停
if _is_pause_close:
self.close(conveyor2=True)
self.motor_stopped_by_sensor2 = True
# self.sensor2_ready = True #初始值
_is_pause_close=False
time.sleep(self.sensor2_error_delay)
continue
#开启线程
_is_pause_close=True
try:
if self.is_valid_sensor_status_1(self.SENSOR2):
if not self.sensor2_ready:
if _is_signal or self.is_valid_sensor_status_1(self.SENSOR2):
#检测到信号，如果之前是没有信号，关闭滚筒
print('检查到sensor2信号')
if _is_signal and self.is_valid_sensor2_status_lost(self.SENSOR2):
print('检查到sensor2信号消失')
self.close(conveyor2=True)
#滚筒关闭标志
self.motor_stopped_by_sensor2 = True
self.sensor2_ready = True
else:
if self.sensor2_ready: #and self.motor_stopped_by_sensor2:
# 发送信号通知机器人取走物品
self.take_robot_signal.emit()
_is_signal=False
self.sensor2_ready=False #打开滚洞标识
else:
_is_signal=True
# time.sleep(0.1)
continue
elif self.sensor2_ready:
#sensor2_ready:通过Feeding:FPhoto处控制是否启动
if self.motor_stopped_by_sensor2:
print('开滚筒')
self.open(conveyor2=True)
self.motor_stopped_by_sensor2 = False
self.sensor2_ready = False
time.sleep(self.sensor2_loop_delay)
time.sleep(2)
except Exception as e:
print(f"SENSOR2 处理错误: {e}")
time.sleep(self.sensor2_error_delay)
# ===================== 线程控制方法 =====================
def start(self):
if self._running:
print("线程已经在运行")
return
print("启动传感器线程")
self._running = True
self._sensor1_thread = threading.Thread(target=self.handle_sensor1, daemon=True)
self._sensor2_thread = threading.Thread(target=self.handle_sensor2, daemon=True)
self._sensor1_thread.start()
self._sensor2_thread.start()
def stop(self):
if not self._running:
print("线程未在运行")
return
print("停止传感器线程")
self._running = False
if self._sensor1_thread:
self._sensor1_thread.join()
if self._sensor2_thread:
self._sensor2_thread.join()
print("传感器线程已终止。")
def pause_start_sensor(self,is_pause):
"""
暂停或开启传感器线程
is_pause:True是，False否
"""
self._ispause = is_pause
def stop_sensor(self,sensor1_thread,sensor2_thread):
if not self._running:
print("线程未在运行")
@ -382,22 +424,3 @@ class RelayController:
if sensor2_thread and sensor2_thread.is_alive():
sensor2_thread.join()
print("传感器线程已终止。")
def start_sensor1_only(self):
if self._running:
print("传感器线程已经在运行")
return
print("启动传感器1监听线程...")
self._running = True
self._sensor1_thread = threading.Thread(target=self.handle_sensor1, daemon=True)
self._sensor1_thread.start()
if __name__ == '__main__':
controller = RelayController()
controller.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
controller.stop()

BIN
EMV/__pycache__/EMV.cpython-39.pyc
View File

Binary file not shown.

38
EMV/all_sensors_device_test.py
View File

@ -58,10 +58,6 @@ def test_device(device_name, action):
# ✅ 示例调用（你可以取消注释你想测试的部分）
if __name__ == "__main__":
#test_device('conveyor1', 'open')
#test_device('conveyor1', 'close')
#已完成测试
#test_device('conveyor2', 'open')
#test_device('conveyor2', 'close')
@ -87,23 +83,35 @@ if __name__ == "__main__":
# print(sensors)
# time.sleep(3)
# test_device('conveyor2', 'open')
# test_device('conveyor2_reverse', 'open')
test_device('conveyor2_reverse', 'open')
# time.sleep(3)
# test_device('conveyor2', 'open')
# test_device('conveyor2', 'close')
test_device('conveyor2', 'close')
# sensors = relay_controller.get_all_device_status('sensors')
# sensor2_value = sensors.get(relay_controller.SENSOR2, False)
# relay_controller._running=True
# relay_controller.handle_sensor2()
test_device('conveyor2', 'close')
while True:
if relay_controller.is_valid_sensor_status_1('sensor2'):
test_device('conveyor2', 'close')
time.sleep(4)
test_device('conveyor2_reverse', 'open')
time.sleep(3)
test_device('conveyor2', 'open')
# test_device('conveyor2', 'close')
# while True:
# if relay_controller.is_valid_sensor_status_1('sensor2'):
# test_device('conveyor2', 'close')
# time.sleep(4)
# test_device('conveyor2_reverse', 'open')
# time.sleep(3)
# test_device('conveyor2', 'open')
# test_device('conveyor1', 'open')
# test_device('conveyor1', 'close')
# relay_controller._running=True
# relay_controller.handle_sensor1()
# test_device('conveyor1', 'open')
# while True:
# if relay_controller.is_valid_sensor_status(relay_controller.SENSOR1):
# test_device('conveyor1', 'close')
# time.sleep(0.1)