274 lines
10 KiB
Python
274 lines
10 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
import socket
|
||
import binascii
|
||
import time
|
||
import threading
|
||
|
||
class RelayController:
|
||
def __init__(self, host='192.168.0.18', port=50000):
|
||
self.host = host
|
||
self.port = port
|
||
|
||
# 控件映射
|
||
self.CONVEYOR1 = 'conveyor1'
|
||
self.PUSHER = 'pusher'
|
||
self.CONVEYOR2 = 'conveyor2'
|
||
self.CLAMP = 'clamp'
|
||
self.SENSOR1 = 'sensor1'
|
||
self.SENSOR2 = 'sensor2'
|
||
|
||
self.valve_commands = {
|
||
self.CONVEYOR1: {'open': '00000000000601050000FF00', 'close': '000000000006010500000000'},
|
||
self.PUSHER: {'open': '00000000000601050001FF00', 'close': '000000000006010500010000'},
|
||
self.CONVEYOR2: {'open': '00000000000601050002FF00', 'close': '000000000006010500020000'},
|
||
self.CLAMP: {'open': '00000000000601050003FF00', 'close': '000000000006010500030000'}
|
||
}
|
||
|
||
self.read_status_command = {
|
||
'devices': '000000000006010100000008',
|
||
'sensors': '000000000006010200000008'
|
||
}
|
||
|
||
self.device_bit_map = {
|
||
self.CONVEYOR1: 0,
|
||
self.PUSHER: 1,
|
||
self.CONVEYOR2: 2,
|
||
self.CLAMP: 3,
|
||
}
|
||
|
||
self.sensor_bit_map = {
|
||
self.SENSOR1: 0,
|
||
self.SENSOR2: 1,
|
||
}
|
||
|
||
self.device_name_map = {
|
||
self.CONVEYOR1: "传送带1",
|
||
self.PUSHER: "推板",
|
||
self.CONVEYOR2: "传送带2",
|
||
self.CLAMP: "机械臂"
|
||
}
|
||
|
||
self.sensor_name_map = {
|
||
self.SENSOR1: '位置传感器1',
|
||
self.SENSOR2: '位置传感器2',
|
||
}
|
||
|
||
self._running = False
|
||
self._sensor1_thread = None
|
||
self._sensor2_thread = None
|
||
|
||
def send_command_old(self, command_hex):
|
||
byte_data = binascii.unhexlify(command_hex)
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
try:
|
||
sock.connect((self.host, self.port))
|
||
sock.send(byte_data)
|
||
response = sock.recv(1024)
|
||
print(f"收到响应: {binascii.hexlify(response)}")
|
||
return response
|
||
except Exception as e:
|
||
print(f"通信错误: {e}")
|
||
return None
|
||
|
||
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()
|
||
|
||
def send_command(self, command_hex, retry_count=5, source='unknown'):
|
||
byte_data = binascii.unhexlify(command_hex)
|
||
for attempt in range(retry_count):
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(10)
|
||
sock.connect((self.host, self.port))
|
||
sock.send(byte_data)
|
||
response = sock.recv(1024)
|
||
hex_response = binascii.hexlify(response).decode('utf-8')
|
||
|
||
# 根据 source 区分响应来源
|
||
if source == 'sensor':
|
||
print(f"[传感器响应] {hex_response}")
|
||
elif source == 'device':
|
||
print(f"[设备控制响应] {hex_response}")
|
||
else:
|
||
print(f"[通信响应] {hex_response}")
|
||
|
||
return response
|
||
except Exception as e:
|
||
print(f"通信错误 ({source}): {e}, 尝试重连... ({attempt + 1}/{retry_count})")
|
||
if attempt == retry_count - 1:
|
||
self.trigger_alarm()
|
||
time.sleep(5)
|
||
return None
|
||
|
||
def trigger_alarm(self):
|
||
"""当通信错误超过最大重试次数时触发报警"""
|
||
print("警告：连续多次通信失败，请检查设备连接！")
|
||
# 这里可以添加更多的报警措施，如发送邮件、短信或声音警报等
|
||
|
||
def get_all_device_status(self, command_type='devices'):
|
||
command = self.read_status_command.get(command_type)
|
||
if not command:
|
||
print(f"未知的读取类型: {command_type}")
|
||
return {}
|
||
|
||
# 设置来源标识
|
||
source = 'sensor' if command_type == 'sensors' else 'device'
|
||
|
||
response = self.send_command(command, source=source)
|
||
status_dict = {}
|
||
|
||
if response and len(response) >= 10:
|
||
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
|
||
|
||
for key, bit_index in bit_map.items():
|
||
status_dict[key] = status_bin[bit_index] == '1'
|
||
else:
|
||
print(f"[{command_type}] 读取状态失败或响应无效")
|
||
|
||
return status_dict
|
||
|
||
def get_device_status_old(self, name, command_type='devices'):
|
||
return self.get_all_device_status(command_type).get(name, None)
|
||
|
||
def get_device_status(self, name, command_type='devices', stable_duration=1.0, max_attempts=3):
|
||
"""
|
||
获取指定设备/传感器的状态，只有在连续稳定检测到信号后才返回 True。
|
||
|
||
参数:
|
||
name (str): 设备/传感器名称，如 SENSOR1
|
||
command_type (str): 类型，'devices' 或 'sensors'
|
||
stable_duration (float): 信号需要稳定的持续时间（秒）
|
||
max_attempts (int): 最大尝试次数（用于稳定性判断）
|
||
|
||
返回:
|
||
bool or None: 稳定检测到信号返回 True，否则返回 False 或 None（失败）
|
||
"""
|
||
stable_count = 0
|
||
interval = 0.2 # 每隔多久检查一次
|
||
|
||
for _ in range(int(stable_duration / interval)):
|
||
statuses = self.get_all_device_status(command_type)
|
||
status = statuses.get(name)
|
||
|
||
if status is True:
|
||
stable_count += 1
|
||
if stable_count >= max_attempts:
|
||
return True
|
||
elif status is False:
|
||
return False
|
||
else:
|
||
# None 表示读取失败
|
||
print(f"[警告] 读取 {name} 状态失败，尝试重试...")
|
||
stable_count = 0
|
||
|
||
time.sleep(interval)
|
||
|
||
return False # 默认返回 False，避免误触发
|
||
|
||
def open(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False):
|
||
status = self.get_all_device_status()
|
||
if conveyor1 and not status.get(self.CONVEYOR1, False):
|
||
print("打开传送带1")
|
||
self.send_command(self.valve_commands[self.CONVEYOR1]['open'], source='device')
|
||
time.sleep(1)
|
||
|
||
if pusher and not status.get(self.PUSHER, False):
|
||
print("打开推板")
|
||
self.send_command(self.valve_commands[self.PUSHER]['open'], source='device')
|
||
time.sleep(0.05)
|
||
|
||
if conveyor2 and not status.get(self.CONVEYOR2, False):
|
||
print("打开传送带2")
|
||
self.send_command(self.valve_commands[self.CONVEYOR2]['open'], source='device')
|
||
time.sleep(1)
|
||
|
||
if clamp and not status.get(self.CLAMP, False):
|
||
print("启动机械臂抓夹")
|
||
self.send_command(self.valve_commands[self.CLAMP]['open'], source='device')
|
||
time.sleep(0.5)
|
||
|
||
def close(self, conveyor1=False, pusher=False, conveyor2=False, clamp=False):
|
||
status = self.get_all_device_status()
|
||
if conveyor1 and status.get(self.CONVEYOR1, True):
|
||
print("关闭传送带1")
|
||
self.send_command(self.valve_commands[self.CONVEYOR1]['close'], source='device')
|
||
time.sleep(1)
|
||
|
||
if pusher and status.get(self.PUSHER, True):
|
||
print("关闭推板")
|
||
self.send_command(self.valve_commands[self.PUSHER]['close'], source='device')
|
||
time.sleep(0.05)
|
||
|
||
if conveyor2 and status.get(self.CONVEYOR2, True):
|
||
print("关闭传送带2")
|
||
self.send_command(self.valve_commands[self.CONVEYOR2]['close'], source='device')
|
||
time.sleep(1)
|
||
|
||
if clamp and status.get(self.CLAMP, True):
|
||
print("停止机械臂抓夹")
|
||
self.send_command(self.valve_commands[self.CLAMP]['close'], source='device')
|
||
time.sleep(0.5)
|
||
|
||
def handle_sensor1(self):
|
||
while self._running:
|
||
try:
|
||
if self.get_device_status(self.SENSOR1, 'sensors'):
|
||
print("SENSOR1 检测到信号，执行流程")
|
||
self.close(conveyor1=True)
|
||
time.sleep(2)
|
||
self.open(pusher=True)
|
||
time.sleep(5)
|
||
self.close(pusher=True)
|
||
time.sleep(2)
|
||
self.open(conveyor1=True)
|
||
time.sleep(0.5)
|
||
except Exception as e:
|
||
print(f"SENSOR1 处理错误: {e}")
|
||
|
||
def handle_sensor2(self):
|
||
while self._running:
|
||
try:
|
||
if self.get_device_status(self.SENSOR2, 'sensors'):
|
||
print("SENSOR2 检测到信号，执行流程")
|
||
self.close(conveyor2=True)
|
||
time.sleep(2)
|
||
self.open(clamp=True)
|
||
time.sleep(2)
|
||
self.open(conveyor2=True)
|
||
time.sleep(0.5)
|
||
except Exception as e:
|
||
print(f"SENSOR2 处理错误: {e}")
|
||
|
||
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 # 设置标志位为 False
|
||
if self._sensor1_thread is not None:
|
||
self._sensor1_thread.join() # 等待线程结束
|
||
if self._sensor2_thread is not None:
|
||
self._sensor2_thread.join()
|
||
print("传感器线程已终止。") |