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-11-12 07:11:09 +08:00
parent cbda29e270
commit ed6cc098df
15 changed files with 365 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
EMV/EMV.py
View File

@ -55,6 +55,7 @@ class RelayController(QObject):
self.SENSOR2 = 'sensor2'
self.BELT = 'belt'
self.ALARM = 'alarm'
self.BLOW_SENSOR2 = 'blow_sensor2'
self.valve_commands = {
#包装机皮带
@ -71,6 +72,8 @@ class RelayController(QObject):
self.BELT: {'open': '00000000000601050006FF00', 'close': '000000000006010500060000'},
#D01 声控报警
self.ALARM: {'open': '00000000000601050000FF00', 'close': '000000000006010500000000'},
#DO6 吹传感器2
self.BLOW_SENSOR2: {'open': '00000000000601050005FF00', 'close': '000000000006010500050000'},
#滚筒反转
self.CONVEYOR2_REVERSE: {'open': '000100000006020620000022', 'close': '000100000006020620000001'}
}
@ -91,6 +94,7 @@ class RelayController(QObject):
self.CONVEYOR2_REVERSE: 5,
self.BELT: 6,
self.ALARM: 7,
#self.BLOW_SENSOR2: 8
}
self.sensor_bit_map = {
@ -99,14 +103,15 @@ class RelayController(QObject):
}
self.device_name_map = {
self.CONVEYOR1: "传送带1",
self.CONVEYOR1: "包装机皮带",
self.PUSHER: "推板开",
self.CONVEYOR2: "传送带2",
self.CONVEYOR2: "滚筒",
self.CLAMP: "机械臂夹爪",
self.PUSHER1: "推板关",
self.CONVEYOR2_REVERSE: "传送带2反转",
self.CONVEYOR2_REVERSE: "滚筒反转",
self.BELT: "皮带",
self.ALARM: "声控报警",
self.BLOW_SENSOR2: "吹传感器2"
}
self.sensor_name_map = {
@ -142,7 +147,7 @@ class RelayController(QObject):
sock.connect((self.host, self.port))
sock.send(byte_data)
response = sock.recv(1024)
hex_response = binascii.hexlify(response).decode('utf-8')
# hex_response = binascii.hexlify(response).decode('utf-8')
#if source == 'sensor':
#print(f"[传感器响应] {hex_response}")
#elif source == 'device':
@ -163,8 +168,8 @@ class RelayController(QObject):
# ===================== 状态读取方法 =====================
def get_all_device_status(self, command_type='devices'):
if Constant.DebugPosition:
return {self.SENSOR2:True}
# if Constant.DebugPosition:
# return {self.SENSOR2:True}
command = self.read_status_command.get(command_type)
if not command:
print(f"未知的网络继电器读取类型: {command_type}")
@ -178,7 +183,7 @@ class RelayController(QObject):
status_byte = response[9]
status_bin = f"{status_byte:08b}"[::-1]
bit_map = self.device_bit_map if command_type == 'devices' else self.sensor_bit_map
name_map = self.device_name_map if command_type == 'devices' else self.sensor_name_map
# name_map = self.device_name_map if command_type == 'devices' else self.sensor_name_map
for key, bit_index in bit_map.items():
status_dict[key] = status_bin[bit_index] == '1'
@ -234,7 +239,6 @@ class RelayController(QObject):
return loc_is_pressed
def parse_status_code(self, response):
"""
从 Modbus 响应字符串中提取状态码（后两位）
@ -364,36 +368,39 @@ class RelayController(QObject):
time.sleep(check_interval)
return False
# ===================== 动作控制方法 =====================
def open(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False, pusher1=False, conveyor2_reverse=False,belt=False,alarm=False):
if Constant.DebugPosition:
return
status = self.get_all_device_status()
if conveyor1 and not status.get(self.CONVEYOR1, False):
def open(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False, pusher1=False, conveyor2_reverse=False,belt=False,alarm=False,blow_sensor2=False):
# if Constant.DebugPosition:
# return
# status = self.get_all_device_status()
if conveyor1:
self.send_command(self.valve_commands[self.CONVEYOR1]['open'])
time.sleep(self.delay_conveyor)
if pusher and not status.get(self.PUSHER, False):
if pusher:
self.send_command(self.valve_commands[self.PUSHER]['open'])
time.sleep(self.delay_pusher)
if conveyor2 and not status.get(self.CONVEYOR2, False):
if conveyor2:
self.send_command(self.valve_commands[self.CONVEYOR2]['open'])
time.sleep(self.delay_conveyor)
if clamp and not status.get(self.CLAMP, False):
if clamp:
self.send_command(self.valve_commands[self.CLAMP]['open'])
time.sleep(self.delay_clamp)
if pusher1 and not status.get(self.PUSHER1, False):
if pusher1:
self.send_command(self.valve_commands[self.PUSHER1]['open'])
time.sleep(self.delay_pusher)
if conveyor2_reverse:
self.send_command(self.valve_commands[self.CONVEYOR2_REVERSE]['open'])
time.sleep(self.delay_conveyor)
if belt and not status.get(self.BELT, False):
if belt:
self.send_command(self.valve_commands[self.BELT]['open'])
# time.sleep(self.delay_belt)
if alarm and not status.get(self.ALARM, False):
if alarm:
self.send_command(self.valve_commands[self.ALARM]['open'])
# time.sleep(self.delay_alarm)
if blow_sensor2:
self.send_command(self.valve_commands[self.BLOW_SENSOR2]['open'])
# time.sleep(self.delay_blow_sensor2)
def close(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False, pusher1=False, conveyor2_reverse=False,belt=False,alarm=False):
def close(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False, pusher1=False, conveyor2_reverse=False,belt=False,alarm=False,blow_sensor2=False):
if conveyor1:
self.send_command(self.valve_commands[self.CONVEYOR1]['close'])
time.sleep(self.delay_conveyor)
@ -418,6 +425,9 @@ class RelayController(QObject):
if alarm:
self.send_command(self.valve_commands[self.ALARM]['close'])
# time.sleep(self.delay_alarm)
if blow_sensor2:
self.send_command(self.valve_commands[self.BLOW_SENSOR2]['close'])
# time.sleep(self.delay_blow_sensor2)
# ===================== 传感器处理线程 =====================
def handle_sensor1(self):
@ -497,11 +507,16 @@ class RelayController(QObject):
# 发送信号通知机器人取走物品
self.take_robot_signal.emit()
_is_signal=False
self.sensor2_ready=False #打开滚洞标识
self.sensor2_ready=False #打开滚洞标识，FPhoto控制打开
else:
if self.sensor2_ready:
#只有在FPhoto处才有效
_is_signal=True
_is_signal=True
if self.motor_stopped_by_sensor2:
print('开滚筒')
self.open(conveyor2=True)
self.motor_stopped_by_sensor2 = False
# time.sleep(0.1)
continue
elif self.sensor2_ready:
@ -523,6 +538,13 @@ class RelayController(QObject):
is_pause:True是，False否
"""
self._ispause = is_pause
if is_pause:
"""暂停皮带"""
self.close(belt=True)
else:
"""开启皮带"""
self.open(belt=True)
def stop_sensor(self,sensor1_thread,sensor2_thread):
@ -539,15 +561,21 @@ class RelayController(QObject):
def handle_emergency_pressed(self):
"处理急停按钮信号状态线程"
print('检查急停按钮状态1')
while self._running:
try:
print('检查急停按钮状态')
loc_is_pressed =self.get_emergency_is_pressed()
if loc_is_pressed:
# 处理急停按钮信号状态
self.log_signal.emit(logging.INFO,f"急停按钮被按下")
# 处理急停按钮信号状态
if not self.emergency_is_pressed:
print('急停按钮被按下')
self.log_signal.emit(logging.INFO,f"急停按钮被按下")
self.emergency_is_pressed=True
self.emergency_signal.emit(True)
else:
print('急停按钮未被按下')
self.emergency_is_pressed=False
self.emergency_signal.emit(False)
time.sleep(0.5)

27
EMV/all_sensors_device_test.py
View File

@ -1,4 +1,7 @@
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from EMV import RelayController
import time
import threading
@ -35,6 +38,18 @@ def test_device(device_name, action):
'clamp': {
'open': lambda: relay_controller.open(clamp=True),
'close': lambda: relay_controller.close(clamp=True)
},
'alarm': {
'open': lambda: relay_controller.open(alarm=True),
'close': lambda: relay_controller.close(alarm=True)
},
'belt': {
'open': lambda: relay_controller.open(belt=True),
'close': lambda: relay_controller.close(belt=True)
},
'blow_sensor2': {
'open': lambda: relay_controller.open(blow_sensor2=True),
'close': lambda: relay_controller.close(blow_sensor2=True)
}
}
@ -83,16 +98,18 @@ if __name__ == "__main__":
# print(sensors)
# time.sleep(3)
# test_device('conveyor2', 'open')
test_device('conveyor2_reverse', 'open')
# test_device('belt', 'close')
# time.sleep(3)
# test_device('conveyor2', 'open')
test_device('conveyor2', 'close')
# test_device('belt', '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')
test_device('blow_sensor2', 'close')
# test_device('blow_sensor2', 'close')
# while True:
# if relay_controller.is_valid_sensor_status_1('sensor2'):
# test_device('conveyor2', 'close')
@ -112,8 +129,10 @@ if __name__ == "__main__":
# time.sleep(0.1)
# relay_controller._running=True
# relay_controller.handle_emergency_pressed()
time.sleep(100)
print('aaaaa')