199 lines
5.6 KiB
Python
199 lines
5.6 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
'''
|
||
# @Time : 2025/6/19 11:20
|
||
# @Author : reenrr
|
||
# @File : EMV-test1.py
|
||
# 功能描述 ：通过网络控制继电器模块，实现对高周波开启和停止按钮的操作
|
||
'''
|
||
import socket
|
||
import binascii
|
||
import time
|
||
|
||
# 网络继电器的 IP 和端口
|
||
# 高周波 19
|
||
HOST = '192.168.58.19'
|
||
PORT = 50000
|
||
|
||
# 开启按钮
|
||
START_BUTTON = 'start_button'
|
||
# 紧急停止按钮
|
||
STOP_BUTTON = 'stop_button'
|
||
|
||
'''
|
||
控件控制报文,存储各按钮的开关命令
|
||
'''
|
||
valve_commands = {
|
||
START_BUTTON: {
|
||
'open': '00000000000601050000FF00',
|
||
'close': '000000000006010500000000',
|
||
},
|
||
STOP_BUTTON: {
|
||
'open': '00000000000601050001FF00',
|
||
'close': '000000000006010500010000',
|
||
},
|
||
}
|
||
|
||
'''
|
||
读取状态命令,获取设备当前状态的指令
|
||
'''
|
||
read_status_command = {
|
||
'button':'000000000006010100000008',
|
||
}
|
||
|
||
# 控件对应 DO 位（从低到高）
|
||
button_bit_map = {
|
||
START_BUTTON: 0,
|
||
STOP_BUTTON: 1,
|
||
}
|
||
|
||
# 控件名称映射表
|
||
button_name_map = {
|
||
START_BUTTON: "开启按钮",
|
||
STOP_BUTTON: "紧急停止按钮",
|
||
}
|
||
|
||
def send_command(command):
|
||
'''
|
||
发送控制命令到网络继电器，
|
||
|
||
参数：
|
||
command: 十六进制字符串，控制指令
|
||
返回：
|
||
设备响应数据（字节类型），失败返回 False
|
||
'''
|
||
|
||
# 将十六进制字符串转换为字节数据并发送
|
||
byte_data = binascii.unhexlify(command)
|
||
|
||
# 创建套接字并连接到继电器
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
try:
|
||
sock.connect((HOST, PORT))
|
||
sock.send(byte_data)
|
||
# 接收响应
|
||
response = sock.recv(1024)
|
||
print(f"收到响应: {binascii.hexlify(response)}")
|
||
# 校验响应
|
||
return response
|
||
except Exception as e:
|
||
print(f"通信错误: {e}")
|
||
return False
|
||
|
||
def get_all_button_status(command_type='button'):
|
||
'''
|
||
获取所有按钮的当前状态
|
||
|
||
参数：
|
||
command_type: 读取类型，默认为 'button'
|
||
返回：
|
||
字典，键为按钮名称，值为 True/False，失败返回空字典
|
||
'''
|
||
|
||
# 获取对应的读取命令
|
||
command = read_status_command.get(command_type)
|
||
if not command:
|
||
print(f"未知的读取类型: {command_type}")
|
||
return {}
|
||
|
||
response = send_command(command)
|
||
status_dict = {}
|
||
|
||
# 校验响应是否有效（至少10字节）
|
||
if response and len(response) >= 10:
|
||
# 状态信息存储在响应的第10个字节（索引9）
|
||
status_byte = response[9] # 状态在第10字节
|
||
# 将字节转换为8位二进制字符串，并反转使低位在前
|
||
status_bin = f"{status_byte:08b}"[::-1]
|
||
|
||
if command_type == 'button':
|
||
bit_map = button_bit_map
|
||
name_map = button_name_map
|
||
else:
|
||
print("不支持的映射类型")
|
||
return{}
|
||
|
||
# 解析每个按钮的状态
|
||
for key, bit_index in bit_map.items():
|
||
# 检查对应位是否为1（1表示开启，0表示关闭）
|
||
state = status_bin[bit_index] == '1'
|
||
status_dict[key] = state
|
||
else:
|
||
print("读取状态失败或响应无效")
|
||
|
||
return status_dict
|
||
|
||
def get_button_status(button_name, command_type='button'):
|
||
'''
|
||
获取单个控件的当前状态
|
||
|
||
参数：
|
||
device_name: 控件名称，如 RESIN_MOLDING_BUTTON、FINSHING_AGENT_BUTTON、STOP_BUTTON
|
||
command_type: 读取类型，默认为 'button'
|
||
返回：
|
||
True: 开启, False: 关闭, None: 无法读取
|
||
'''
|
||
status = get_all_button_status(command_type)
|
||
return status.get(button_name, None)
|
||
|
||
def open(start_button=False, stop_button=False):
|
||
'''
|
||
打开指定的按钮（仅在按钮当前处于关闭状态时才执行）
|
||
|
||
参数：
|
||
resin_molding_button: 素面操作按钮，默认为 False
|
||
finishing_agent_button: 烫金操作按钮，默认为 False
|
||
stop_button: 紧急停止按钮，默认为 False
|
||
'''
|
||
|
||
status = get_all_button_status()
|
||
|
||
if start_button and not status.get(START_BUTTON, False):
|
||
print("打开开启按钮")
|
||
send_command(valve_commands[START_BUTTON]['open'])
|
||
time.sleep(0.05)
|
||
|
||
if stop_button and not status.get(STOP_BUTTON, False):
|
||
print("打开紧急停止按钮")
|
||
send_command(valve_commands[STOP_BUTTON]['open'])
|
||
time.sleep(0.05)
|
||
|
||
def close(start_button=False, stop_button=False):
|
||
'''
|
||
关闭指定的按钮（仅在按钮当前处于开启状态时才执行）
|
||
|
||
参数：
|
||
resin_molding_button: 素面操作按钮，默认为 False
|
||
finishing_agent_button: 烫金操作按钮，默认为 False
|
||
stop_button: 紧急停止按钮，默认为 False
|
||
'''
|
||
status = get_all_button_status()
|
||
|
||
if start_button and status.get(START_BUTTON, True):
|
||
print("关闭开启按钮")
|
||
send_command(valve_commands[START_BUTTON]['close'])
|
||
time.sleep(0.05)
|
||
|
||
if stop_button and status.get(STOP_BUTTON, True):
|
||
print("关闭紧急停止按钮")
|
||
send_command(valve_commands[STOP_BUTTON]['close'])
|
||
time.sleep(0.05)
|
||
|
||
def start_high(h_time = 80):
|
||
# 操作按钮需要先打开后立即关闭
|
||
open(start_button=True)
|
||
time.sleep(0.5)
|
||
close(start_button=True)
|
||
print(f"等待高周波完成，等待时间为{h_time}秒......")
|
||
time.sleep(h_time)
|
||
print("高周波已经完成...")
|
||
|
||
if __name__ == '__main__':
|
||
# 操作按钮需要先打开后立即关闭
|
||
# open(start_button=True)
|
||
# time.sleep(0.5)
|
||
# close(start_button=True)
|
||
start_high()
|
||
|
||
|