# coding=utf-8 import socket import time from loguru import logger def write_jxb(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) send_data_list = jxb_num_to_send_data(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(send_data_list) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) print(response,"response") except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def read_button(): # 读取 D301 -- D350 # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) read_list = ['50 00 00 FF FF 03 00 ', '0C 00 ', '10 00 01 04 00 00 ', '2D 01 00 A8 ', '64 00 '] relist = [] try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) # print("成功连接到PLC") data = ''.join(read_list) command = bytearray.fromhex(data) # print(command) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) # 处理响应数据，提取寄存器的值 # 需要根据协议格式解析响应，这里假设响应包含一个D100寄存器的值 for i in range(11, len(response) - 1, 2): relist.append(int.from_bytes(response[i: i + 2], byteorder='little')) # print(relist) # print(len(relist)) # 读取301 - 320的值 # print(relist) except socket.error as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() # print("连接已关闭") return relist def write_button(): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = ['50 00 00 FF FF 03 00 ', '30 00 ', '10 00 01 14 00 00 ', '5E 01 00 A8 ', '12 00 ', '01 00 01 00 ', '01 00 01 00 ', '01 00 01 00 ', '01 00 01 00 ', '01 00 01 00 ', '00 00 01 00 ', '01 00 01 00 ', '01 00 01 00 ', '00 00 01 00 ', ] # 350 - 367 所有 try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) print("成功连接到PLC") data = ''.join(bon_data) # data = '50 00 00 FF FF 03 00 0C 00 10 00 01 04 00 00 64 00 00 A8 14 00' # data = '50 00 00 FF FF 03 00 12 00 10 00 01 14 00 00 64 00 00 A8 03 00 0C 00 0D 00 0E 00' # command = bytes.fromhex(data) command = bytearray.fromhex(data) print(command) # command = b'\x50\x00\x00\xFF\xFF\x03\x00\x0C\x00\x10\x00\x01\x04\x01\x00\x00\xA8\x00\x00\x00' # 示例请求报文 # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) print(response[11:]) print(len(response)) # 处理响应数据，提取寄存器的值 # 需要根据协议格式解析响应，这里假设响应包含一个D100寄存器的值 d100_value = int.from_bytes(response[11:13], byteorder='little') d101_value = int.from_bytes(response[13:15], byteorder='little') d102_value = int.from_bytes(response[15:17], byteorder='little') d104_value = int.from_bytes(response[21:23], byteorder='little') print(f"D100的值: {d100_value}, {d101_value}, {d102_value}, {d104_value}") except socket.error as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_agv(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = agv_num_to_send_data(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def num_to_hex(value): # 将值限制在16位有符号整型范围内 if value < -32768: value = -32768 if value > 32767: value = 32767 # 将负数转换为补码形式 if value < 0: value = (1 << 16) + value # 转为16进制，去掉'0x'前缀，并且保证长度为4位，不足则补齐 hex_value = format(value, '04X') hex_str = hex_value[-4:] # 返回最后四位 return f"{hex_str[2:]} {hex_str[:2]} " def jxb_num_to_send_data(datas): head = ['50 00 00 FF FF 03 00 ', '4A 00 ', '10 00 01 14 00 00 ', 'C8 00 00 A8 ', '1F 00 '] send_data = head + [num_to_hex(data) for data in datas] return send_data def agv_num_to_send_data(datas): head = ['50 00 00 FF FF 03 00 ', '22 00 ', '10 00 01 14 00 00 ', '64 00 00 A8 ', '0B 00 '] send_data = head + [num_to_hex(data) for data in datas] return send_data def write_stop_agv(status): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_stop_agv_status(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: logger.error(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() # print("连接已关闭") def write_stop_jxb(status): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_stop_jxb_status(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: logger.error(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() def write_stop_robot(status): """传输机械臂异常""" # PLC的IP地址和端口号 print("连接成功") PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_stop_robot_status(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() def write_once_status(status): """传输指示灯异常""" print("连接成功") # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_once_status_send(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("") def write_once_status_send(status): if status: rss = '01 00 00 00' else: rss = '00 00 01 00' head = ['50 00 00 FF FF 03 00 ', '10 00 ', '10 00 01 14 00 00 ', '6d 01 00 A8 ', '02 00 ', rss] return head def write_stop_agv_status(status): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '64 00 00 A8 ', '01 00 ', num_to_hex(status)] return head def write_stop_jxb_status(status): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', 'E5 00 00 A8 ', '01 00 ', num_to_hex(status)] return head def write_stop_robot_status(status): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '54 01 00 A8 ', '01 00 ', num_to_hex(status)] return head def write_network_status(status): """ 向plc传递 网络异常警报""" # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_status_send(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() def write_scram_button_status(status): """ 向plc传递 网络异常警报""" # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_scram_button_status_send(status) # print(bon_data) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() def write_network_status_send(status): head = ['50 00 00 FF FF 03 00 ', '14 00 ', '10 00 01 14 00 00 ', 'F1 00 00 A8 ', '04 00 ', status] return head def write_scram_button_status_send(status): head = ['50 00 00 FF FF 03 00 ', '14 00 ', '10 00 01 14 00 00 ', 'E7 00 00 A8 ', '04 00 ', status] return head def write_network_status_run_time_total(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '08 00 00 A8 ', '01 00 ', ] send_data = head + [num_to_hex(data) for data in run_time] print(send_data) return send_data def write_run_time_zongshijian(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_status_run_time_total(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_status_run_time_line1(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '0C 00 00 A8 ', '01 00 ', ] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_run_time_line1(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_status_run_time_line1(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) logger.error("写入成功") except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_status_run_time_line2(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '0E 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_run_time_line2(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_status_run_time_line2(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_status_run_time_line3(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '10 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_run_time_line3(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_status_run_time_line3(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_total_count(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '14 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_total_count(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_total_count(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_ok_count(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '16 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_ok_count(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_ok_count(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_NG_count(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '18 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_NG_count(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_NG_count(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_photo_count(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', '1C 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_photo_count(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_photo_count(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") def write_network_engine_number_count(run_time): head = ['50 00 00 FF FF 03 00 ', '0E 00 ', '10 00 01 14 00 00 ', 'D5 00 00 A8 ', '01 00 '] send_data = head + [num_to_hex(data) for data in run_time] return send_data def write_engine_number_count(datas): # PLC的IP地址和端口号 PLC_IP = '192.168.2.1' # 替换为你的PLC的实际IP地址 PLC_PORT = 4096 # 通常是MC协议的默认端口，也可能是SLMP的默认端口 # 创建一个TCP/IP socket client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) bon_data = write_network_engine_number_count(datas) try: # 连接到PLC client_socket.connect((PLC_IP, PLC_PORT)) data = ''.join(bon_data) # command = bytes.fromhex(data) command = bytearray.fromhex(data) # 发送请求 client_socket.sendall(command) # 接收PLC的响应 response = client_socket.recv(1024) except Exception as e: print(f"通信时发生错误: {e}") finally: # 关闭连接 client_socket.close() print("连接已关闭") if __name__ == '__main__': write_scram_button_status('00 00 01 00 00 00 00 00') # write_run_time_zongshijian([1000]) # time.sleep(0.1) # write_run_time_line1([40]) # time.sleep(0.1) # write_run_time_line2([50]) # time.sleep(0.1) # write_run_time_line3([90]) # write_stop_jxb(1) # write_NG_count([50]) # write_engine_number_count([412151512]) # time.sleep(0.1) # write_ok_count([60]) # time.sleep(0.1) # write_photo_count([90]) # time.sleep(0.1) # write_total_count([100]) # write_stop_jxb(1) # write_agv([2] *11) # write_agv() # write_jxb() # print(num_to_hex(4369)) # while True: # data = read_button() # if len(data)==100: # print(data[13]) # # print(data[20]) # # 自动启动 绿色按钮 # if len(data) == 20: # start_button = data[14] # line_button = data[19] # print(start_button, ' ', line_button) # print(len(data)) # """ # agv 充电中 准备中 工作中 关闭 故障中 # jxb 准备中 启动中 关闭 故障 # """ # write_jxb() # read_bou() # while True: # read_bou() # time.sleep(0.5)