1135 lines
43 KiB
Python
1135 lines
43 KiB
Python
import socket
|
||
import struct
|
||
import time
|
||
from typing import Optional, List, Tuple, Dict
|
||
|
||
PLC_TIMEOUT = 1
|
||
RECONNECT_DELAY = 3 # seconds
|
||
|
||
|
||
class PlcClient:
|
||
def __init__(self, ip: str, port: int):
|
||
self.ip = ip
|
||
self.port = port
|
||
self.timeout = PLC_TIMEOUT
|
||
self.socket = None
|
||
self.connected = False
|
||
self._connect_with_retry()
|
||
|
||
def _connect_with_retry(self) -> bool:
|
||
"""尝试连接PLC，失败后无限重试，每次间隔5秒"""
|
||
while True:
|
||
try:
|
||
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
self.socket.settimeout(self.timeout)
|
||
self.socket.connect((self.ip, self.port))
|
||
self.connected = True
|
||
print(f"成功连接到PLC {self.ip}:{self.port}")
|
||
return True
|
||
except Exception as ex:
|
||
print(f"连接PLC失败: {str(ex)}，{RECONNECT_DELAY}秒后重试...")
|
||
self.connected = False
|
||
time.sleep(RECONNECT_DELAY)
|
||
|
||
# 清理旧的socket对象
|
||
if self.socket:
|
||
try:
|
||
self.socket.close()
|
||
except:
|
||
pass
|
||
self.socket = None
|
||
|
||
def _ensure_connection(self) -> bool:
|
||
"""确保连接有效，如果断开则尝试重连"""
|
||
if not self.connected or self.socket is None:
|
||
return self._connect_with_retry()
|
||
return True
|
||
|
||
@staticmethod
|
||
def _hex_str_to_bytes(hex_str: str) -> bytes:
|
||
return bytes.fromhex(hex_str.replace(" ", ""))
|
||
|
||
def _send_and_receive(self, cmd_hex: str, expected_response_length: int = 1024) -> Tuple[
|
||
Optional[bytes], Optional[str]]:
|
||
"""发送命令并接收响应，处理连接问题"""
|
||
if not self._ensure_connection():
|
||
return None, "PLC连接不可用"
|
||
|
||
try:
|
||
self.socket.sendall(self._hex_str_to_bytes(cmd_hex))
|
||
buffer = self.socket.recv(expected_response_length)
|
||
return buffer, None
|
||
except socket.timeout:
|
||
return None, "PLC响应超时"
|
||
except socket.error as ex:
|
||
self.connected = False
|
||
return None, f"通信错误: {str(ex)}"
|
||
except Exception as ex:
|
||
return None, f"未知错误: {str(ex)}"
|
||
|
||
def read_m(self, address: int) -> Tuple[Optional[bool], Optional[str]]:
|
||
try:
|
||
addr_low = address & 0xFF
|
||
addr_high = (address >> 8) & 0xFF
|
||
|
||
# 读1个位
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
"0C 00 " # 长度 12
|
||
"10 00 01 04 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 90 "
|
||
"01 00"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return None, f"M{address}读取失败: {err}"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return None, f"M{address}读取失败，响应: {buffer.hex()}"
|
||
|
||
# 数据部分通常从第11字节开始
|
||
if len(buffer) < 12:
|
||
return None, f"M{address}响应长度不足"
|
||
|
||
m_value = buffer[11] & 0x01 # 取最低位表示 ON/OFF
|
||
return bool(m_value), None
|
||
|
||
except Exception as ex:
|
||
return None, f"M{address}通信失败: {str(ex)}"
|
||
|
||
def read_multiple_m(self, start_address: int, count: int) -> Tuple[Optional[List[bool]], Optional[str]]:
|
||
"""
|
||
一次性读取多个M寄存器
|
||
:param start_address: 起始地址
|
||
:param count: 要读取的寄存器数量
|
||
:return: 布尔值列表和错误信息
|
||
"""
|
||
try:
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
count_low = count & 0xFF
|
||
count_high = (count >> 8) & 0xFF
|
||
|
||
# 批量读取M寄存器命令
|
||
# 子头: 10 00 01 04 00 00
|
||
# 地址: addr_low addr_high 00 90 (M寄存器类型)
|
||
# 数量: count_low count_high
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
"0C 00 " # 长度 12
|
||
"10 00 01 04 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 90 "
|
||
f"{count_low:02X} {count_high:02X}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex, 1024)
|
||
if err:
|
||
return None, f"M{start_address}-M{start_address + count - 1}批量读取失败: {err}"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return None, f"M{start_address}-M{start_address + count - 1}批量读取失败，响应: {buffer.hex()}"
|
||
|
||
# 数据部分通常从第11字节开始，每个位对应一个M寄存器
|
||
required_len = 11 + (count + 7) // 8 # 计算需要的字节数
|
||
if len(buffer) < required_len:
|
||
return None, f"M{start_address}-M{start_address + count - 1}响应长度不足"
|
||
|
||
# 解析位数据
|
||
m_values = []
|
||
data_start = 11
|
||
for i in range(count):
|
||
byte_index = i // 8
|
||
bit_index = i % 8
|
||
|
||
if data_start + byte_index < len(buffer):
|
||
byte_value = buffer[data_start + byte_index]
|
||
bit_value = (byte_value >> bit_index) & 0x01
|
||
m_values.append(bool(bit_value))
|
||
else:
|
||
m_values.append(False) # 如果数据不足，默认为False
|
||
|
||
return m_values, None
|
||
|
||
except Exception as ex:
|
||
return None, f"M{start_address}-M{start_address + count - 1}批量读取失败: {str(ex)}"
|
||
|
||
def write_m(self, address: int, value: bool) -> Tuple[bool, Optional[str]]:
|
||
try:
|
||
# 小端地址
|
||
addr_low = address & 0xFF
|
||
addr_high = (address >> 8) & 0xFF
|
||
value_hex = "01 00" if value else "00 00"
|
||
|
||
# 报文长度 14 字节 = 子头6 + 地址2 + 软元件2 + 点数2 + 数据2
|
||
length_hex = "0E 00"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 90 "
|
||
"01 00 "
|
||
f"{value_hex}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"M{address}写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"M{address}无响应"
|
||
|
||
# EndCode通常在响应报文的第9、10字节
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"M{address}写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"M{address}写入失败: {str(ex)}"
|
||
|
||
def write_d(self, address: int, value: int, data_type: str = "16bit") -> Tuple[bool, Optional[str]]:
|
||
"""
|
||
写入D寄存器
|
||
|
||
Args:
|
||
address: 寄存器地址
|
||
value: 写入的值
|
||
data_type: 数据类型 "16bit" 或 "32bit"
|
||
|
||
Returns:
|
||
(成功标志, 错误信息)
|
||
"""
|
||
try:
|
||
# 小端地址
|
||
addr_low = address & 0xFF
|
||
addr_high = (address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置软元件代码和点数
|
||
if data_type == "16bit":
|
||
# 16位写入：1个点，软元件代码 0xA8
|
||
device_code = "A8"
|
||
points = "01 00"
|
||
# 16位值处理（小端）
|
||
value_low = value & 0xFF
|
||
value_high = (value >> 8) & 0xFF
|
||
value_hex = f"{value_low:02X} {value_high:02X} 00 00"
|
||
data_length = 2 # 数据长度2字节
|
||
|
||
elif data_type == "32bit":
|
||
# 32位写入：2个点，软元件代码 0xA8
|
||
device_code = "A8"
|
||
points = "02 00"
|
||
# 32位值处理（小端）
|
||
value_low = value & 0xFFFF
|
||
value_high = (value >> 16) & 0xFFFF
|
||
value_low_low = value_low & 0xFF
|
||
value_low_high = (value_low >> 8) & 0xFF
|
||
value_high_low = value_high & 0xFF
|
||
value_high_high = (value_high >> 8) & 0xFF
|
||
value_hex = f"{value_low_low:02X} {value_low_high:02X} {value_high_low:02X} {value_high_high:02X}"
|
||
data_length = 4 # 数据长度4字节
|
||
|
||
else:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据N
|
||
total_length = 6 + 2 + 2 + 2 + data_length
|
||
length_hex = f"{total_length:02X} 00" # 小端格式
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points} "
|
||
f"{value_hex}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{address}写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{address}无响应"
|
||
|
||
# EndCode通常在响应报文的第9、10字节
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{address}写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{address}写入失败: {str(ex)}"
|
||
|
||
def write_d_batch(self, start_address: int, values: List[int], data_type: str = "16bit") -> Tuple[
|
||
bool, Optional[str]]:
|
||
# """
|
||
# 批量写入D寄存器
|
||
#
|
||
# Args:
|
||
# start_address: 起始地址
|
||
# values: 要写入的值列表
|
||
# data_type: 数据类型 "16bit" 或 "32bit"
|
||
#
|
||
# Returns:
|
||
# (成功标志, 错误信息)
|
||
# """
|
||
try:
|
||
# 小端地址
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置软元件代码和每个值的数据长度
|
||
if data_type == "16bit":
|
||
device_code = "A8"
|
||
bytes_per_value = 2
|
||
points = len(values)
|
||
elif data_type == "32bit":
|
||
device_code = "A8"
|
||
bytes_per_value = 4
|
||
points = len(values) * 2 # 32bit占用2个点
|
||
else:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
# 构建数据部分
|
||
data_hex_parts = []
|
||
for value in values:
|
||
if data_type == "16bit":
|
||
value_low = value & 0xFF
|
||
value_high = (value >> 8) & 0xFF
|
||
data_hex_parts.append(f"{value_low:02X} {value_high:02X}")
|
||
else: # 32bit
|
||
value_low = value & 0xFFFF
|
||
value_high = (value >> 16) & 0xFFFF
|
||
value_low_low = value_low & 0xFF
|
||
value_low_high = (value_low >> 8) & 0xFF
|
||
value_high_low = value_high & 0xFF
|
||
value_high_high = (value_high >> 8) & 0xFF
|
||
data_hex_parts.append(
|
||
f"{value_low_low:02X} {value_low_high:02X} {value_high_low:02X} {value_high_high:02X}")
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据
|
||
total_length = 6 + 2 + 2 + 2 + (bytes_per_value * len(values))
|
||
length_hex = f"{total_length:02X} 00" # 小端格式
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{start_address}批量写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def read_d(self, address: int, data_type: str = "16bit") -> Tuple[Optional[int], Optional[str]]:
|
||
"""
|
||
读取D寄存器
|
||
|
||
Args:
|
||
address: 寄存器地址
|
||
data_type: 数据类型 "16bit" 或 "32bit"
|
||
|
||
Returns:
|
||
(读取的值, 错误信息)
|
||
"""
|
||
try:
|
||
# 小端地址
|
||
addr_low = address & 0xFF
|
||
addr_high = (address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置软元件代码和点数
|
||
if data_type == "16bit":
|
||
device_code = "A8"
|
||
points = "01 00" # 1个点
|
||
response_data_length = 2 # 期望返回数据长度
|
||
elif data_type == "32bit":
|
||
device_code = "A8"
|
||
points = "02 00" # 2个点
|
||
response_data_length = 4 # 期望返回数据长度
|
||
else:
|
||
return None, f"不支持的数据类型: {data_type}"
|
||
|
||
# 报文长度 12 字节 = 子头6 + 地址2 + 软元件2 + 点数2
|
||
length_hex = "0C 00"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 04 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return None, f"D{address}读取失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return None, f"D{address}无响应"
|
||
|
||
# EndCode通常在响应报文的第9、10字节
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return None, f"D{address}读取被拒绝，响应: {buffer.hex()}"
|
||
|
||
# 数据通常在响应报文的第11字节开始
|
||
# 响应结构: 头9字节 + EndCode 2字节 + 数据
|
||
data_start = 11
|
||
if len(buffer) < data_start + response_data_length:
|
||
return None, f"D{address}响应数据长度不足"
|
||
|
||
data_bytes = buffer[data_start:data_start + response_data_length]
|
||
|
||
# 解析数据（小端格式）
|
||
if data_type == "16bit":
|
||
# 2字节数据，小端转整数
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
else: # 32bit
|
||
# 4字节数据，小端转整数
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
|
||
return value, None
|
||
|
||
except Exception as ex:
|
||
return None, f"D{address}读取失败: {str(ex)}"
|
||
|
||
def read_r(self, address: int, data_type: str = "16bit") -> Tuple[Optional[int], Optional[str]]:
|
||
"""
|
||
读取R寄存器
|
||
|
||
Args:
|
||
address: 寄存器地址
|
||
data_type: 数据类型 "16bit" 或 "32bit"
|
||
|
||
Returns:
|
||
(读取的值, 错误信息)
|
||
"""
|
||
try:
|
||
# R寄存器地址处理（16位地址）
|
||
addr_low = address & 0xFF
|
||
addr_high = (address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置软元件代码和点数
|
||
if data_type == "16bit":
|
||
device_code = "AF" # R寄存器软元件代码
|
||
points = "01 00" # 1个点
|
||
response_data_length = 2 # 期望返回数据长度
|
||
elif data_type == "32bit":
|
||
device_code = "AF"
|
||
points = "02 00" # 2个点
|
||
response_data_length = 4 # 期望返回数据长度
|
||
else:
|
||
return None, f"不支持的数据类型: {data_type}"
|
||
|
||
# 报文长度 12 字节 = 子头6 + 地址2 + 软元件2 + 点数2
|
||
length_hex = "0C 00"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 04 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points}"
|
||
)
|
||
|
||
print(f"发送R读取命令: {cmd_hex}") # 调试信息
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return None, f"R{address}读取失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return None, f"R{address}无响应"
|
||
|
||
print(f"接收R读取响应: {buffer.hex()}") # 调试信息
|
||
|
||
# EndCode通常在响应报文的第9、10字节
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
error_msg = self._get_error_message(end_code)
|
||
return None, f"R{address}读取被拒绝，错误代码: 0x{end_code:04X} - {error_msg}"
|
||
|
||
# 数据通常在响应报文的第11字节开始
|
||
data_start = 11
|
||
if len(buffer) < data_start + response_data_length:
|
||
return None, f"R{address}响应数据长度不足"
|
||
|
||
data_bytes = buffer[data_start:data_start + response_data_length]
|
||
|
||
# 解析数据（小端格式）
|
||
if data_type == "16bit":
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
else: # 32bit
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
|
||
return value, None
|
||
|
||
except Exception as ex:
|
||
return None, f"R{address}读取失败: {str(ex)}"
|
||
|
||
def read_r_batch(self, start_address: int, count: int, data_type: str = "16bit") -> Tuple[
|
||
Optional[List[str]], Optional[str]]:
|
||
"""
|
||
批量读取R寄存器
|
||
|
||
Args:
|
||
start_address: 起始地址
|
||
count: 读取数量
|
||
data_type: 数据类型 "16bit" 或 "32bit"
|
||
|
||
Returns:
|
||
(读取的值列表, 错误信息)
|
||
"""
|
||
try:
|
||
if count <= 0:
|
||
return None, "读取数量必须大于0"
|
||
|
||
# R寄存器地址处理
|
||
addr_low = start_address & 0xFF
|
||
addr_mid = (start_address >> 8) & 0xFF
|
||
addr_high = (start_address >> 16) & 0xFF
|
||
|
||
# 根据数据类型设置
|
||
if data_type == "16bit":
|
||
# device_code = "AF"
|
||
device_code = "B0"
|
||
points = count
|
||
bytes_per_value = 2
|
||
elif data_type == "32bit":
|
||
device_code = "B0"
|
||
# device_code = "AF"
|
||
points = count * 2
|
||
bytes_per_value = 4
|
||
else:
|
||
return None, f"不支持的数据类型: {data_type}"
|
||
|
||
# 计算报文长度
|
||
total_length = 6 + 2 + 2 + 2 # 子头6 + 地址2 + 软元件2 + 点数2
|
||
length_hex = f"{total_length:02X} 00"
|
||
|
||
points_low = points & 0xFF
|
||
points_high = (points >> 8) & 0xFF
|
||
points_hex = f"{points_low:02X} {points_high:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 04 00 00 "
|
||
f"{addr_low:02X} {addr_mid:02X} {addr_high:02X} {device_code} "
|
||
f"{points_hex}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return None, f"R{start_address}批量读取失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return None, f"R{start_address}批量读取无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
error_msg = self._get_error_message(end_code)
|
||
return None, f"R{start_address}批量读取被拒绝，错误代码: 0x{end_code:04X} - {error_msg}"
|
||
|
||
# 解析返回数据
|
||
data_start = 11
|
||
total_data_length = count * bytes_per_value
|
||
|
||
if len(buffer) < data_start + total_data_length:
|
||
return None, f"R{start_address}响应数据长度不足，期望{total_data_length}字节"
|
||
|
||
values = []
|
||
for i in range(count):
|
||
data_offset = i * bytes_per_value
|
||
data_bytes = buffer[data_start + data_offset:data_start + data_offset + bytes_per_value]
|
||
|
||
if data_type == "16bit":
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
else: # 32bit
|
||
value = int.from_bytes(data_bytes, byteorder='little', signed=False)
|
||
|
||
values.append(self.hex_2_ascii(value))
|
||
|
||
return values, None
|
||
|
||
except Exception as ex:
|
||
return None, f"R{start_address}批量读取失败: {str(ex)}"
|
||
|
||
def hex_2_ascii(self, decimal_data):
|
||
hex_value = format(decimal_data, '04X') # 4位大写十六进制
|
||
|
||
# 分割为高位和低位
|
||
high_hex = hex_value[:2] # 前两位
|
||
low_hex = hex_value[2:] # 后两位
|
||
|
||
|
||
# 将16进制高位和低位转换为字符
|
||
high_char = chr(int(high_hex, 16))
|
||
low_char = chr(int(low_hex, 16))
|
||
|
||
result = low_char + high_char
|
||
|
||
return result
|
||
|
||
def write_r_batch(self, start_address: int, values: List[int], data_type: str = "16bit") -> Tuple[
|
||
bool, Optional[str]]:
|
||
"""
|
||
批量写入R寄存器
|
||
|
||
Args:
|
||
start_address: 起始地址
|
||
values: 要写入的值列表
|
||
data_type: 数据类型 "16bit" 或 "32bit"
|
||
|
||
Returns:
|
||
(成功标志, 错误信息)
|
||
"""
|
||
try:
|
||
# R寄存器地址处理
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置
|
||
if data_type == "16bit":
|
||
device_code = "AF"
|
||
bytes_per_value = 2
|
||
points = len(values)
|
||
elif data_type == "32bit":
|
||
device_code = "AF"
|
||
bytes_per_value = 4
|
||
points = len(values) * 2
|
||
else:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
# 构建数据部分
|
||
data_hex_parts = []
|
||
for value in values:
|
||
if data_type == "16bit":
|
||
value_low = value & 0xFF
|
||
value_high = (value >> 8) & 0xFF
|
||
data_hex_parts.append(f"{value_low:02X} {value_high:02X}")
|
||
else: # 32bit
|
||
value_low = value & 0xFFFF
|
||
value_high = (value >> 16) & 0xFFFF
|
||
value_low_low = value_low & 0xFF
|
||
value_low_high = (value_low >> 8) & 0xFF
|
||
value_high_low = value_high & 0xFF
|
||
value_high_high = (value_high >> 8) & 0xFF
|
||
data_hex_parts.append(
|
||
f"{value_low_low:02X} {value_low_high:02X} {value_high_low:02X} {value_high_high:02X}")
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算
|
||
total_length = 6 + 2 + 2 + 2 + (bytes_per_value * len(values))
|
||
length_hex = f"{total_length:02X} 00"
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"R{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"R{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
error_msg = self._get_error_message(end_code)
|
||
return False, f"R{start_address}批量写入被拒绝，错误代码: 0x{end_code:04X} - {error_msg}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"R{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def _get_error_message(self, error_code: int) -> str:
|
||
"""
|
||
解析三菱PLC错误代码
|
||
"""
|
||
error_messages = {
|
||
0xC061: "地址超出范围或软元件访问错误",
|
||
0xC056: "数据长度错误",
|
||
0xC059: "命令不支持",
|
||
0xC05A: "设备号错误",
|
||
0xC05B: "点数超出范围",
|
||
0xC05C: "头数据错误",
|
||
0xC05D: "帧长度错误",
|
||
0xC05F: "读写禁止",
|
||
}
|
||
return error_messages.get(error_code, f"未知错误: 0x{error_code:04X}")
|
||
|
||
def write_d_shen_7_batch(self, start_address: int, values: str, data_type: str = "16bit"):
|
||
try:
|
||
# 小端地址
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置参数
|
||
if data_type not in ["16bit", "32bit"]:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
device_code = "A8"
|
||
|
||
def get_16bit_data(chars):
|
||
"""获取16位数据：两个字符打包为一个16位值"""
|
||
low = ord(chars[0]) if len(chars) > 0 else 0
|
||
high = ord(chars[1]) if len(chars) > 1 else 0
|
||
return f"{low & 0xFF:02X} {high & 0xFF:02X}"
|
||
|
||
def get_32bit_data(char):
|
||
"""获取32位数据：一个字符扩展为32位值"""
|
||
code = ord(char)
|
||
# 小端序：低位字节在前
|
||
byte1 = code & 0xFF # 最低位字节
|
||
byte2 = (code >> 8) & 0xFF # 次低位字节
|
||
byte3 = 0x00 # 次高位字节（补0）
|
||
byte4 = 0x00 # 最高位字节（补0）
|
||
return f"{byte1:02X} {byte2:02X} {byte3:02X} {byte4:02X}"
|
||
|
||
# 构建数据部分
|
||
if data_type == "16bit":
|
||
# 16位：每2个字符为一组，每组2字节
|
||
data_hex_parts = [
|
||
get_16bit_data(values[i:i + 2])
|
||
for i in range(0, len(values), 2)
|
||
]
|
||
points = len(data_hex_parts) # 实际占用的寄存器点数
|
||
bytes_per_value = 2
|
||
else: # 32bit
|
||
# 32位：每个字符为一组，每组4字节
|
||
data_hex_parts = [get_32bit_data(char) for char in values]
|
||
points = len(values) * 2 # 每个32位值占用2个寄存器点
|
||
bytes_per_value = 4
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据
|
||
actual_data_bytes = len(data_hex_parts) * bytes_per_value
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据字节数
|
||
total_length = 6 + 2 + 2 + 2 + actual_data_bytes
|
||
length_hex = f"{total_length & 0xFF:02X} {(total_length >> 8) & 0xFF:02X}" # 小端格式
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
# "4E 01"
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
# cmd_hex = (
|
||
# "50 00 00 FF FF 03 00 "
|
||
# "10 00 "
|
||
# "10 00 01 14 00 00 "
|
||
# "78 00 00 A8 "
|
||
# "02 00 "
|
||
# "41 42 43 00"
|
||
# )
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{start_address}批量写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def write_d_130_batch(self, start_address: int, values: str, data_type: str = "16bit"):
|
||
try:
|
||
# 小端地址
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置参数
|
||
if data_type not in ["16bit", "32bit"]:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
device_code = "A8"
|
||
|
||
def get_16bit_data(value):
|
||
"""获取16位数据：数值转换为16位小端格式"""
|
||
value_int = int(value)
|
||
low = value_int & 0xFF
|
||
high = (value_int >> 8) & 0xFF
|
||
return f"{low:02X} {high:02X}"
|
||
|
||
def get_32bit_data(value):
|
||
"""获取32位数据：数值转换为32位小端格式"""
|
||
value_int = int(value)
|
||
# 小端序：低位字节在前
|
||
byte1 = value_int & 0xFF # 最低位字节
|
||
byte2 = (value_int >> 8) & 0xFF # 次低位字节
|
||
byte3 = (value_int >> 16) & 0xFF # 次高位字节
|
||
byte4 = (value_int >> 24) & 0xFF # 最高位字节
|
||
return f"{byte1:02X} {byte2:02X} {byte3:02X} {byte4:02X}"
|
||
|
||
# 构建数据部分
|
||
if data_type == "16bit":
|
||
# 16位：每个数值转换为2字节
|
||
data_hex_parts = [get_16bit_data(value) for value in values]
|
||
points = len(values) # 占用的寄存器点数
|
||
bytes_per_value = 2
|
||
else: # 32bit
|
||
# 32位：每个数值转换为4字节
|
||
data_hex_parts = [get_32bit_data(value) for value in values]
|
||
points = len(values) * 2 # 每个32位值占用2个寄存器点
|
||
bytes_per_value = 4
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据
|
||
actual_data_bytes = len(data_hex_parts) * bytes_per_value
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据字节数
|
||
total_length = 6 + 2 + 2 + 2 + actual_data_bytes
|
||
length_hex = f"{total_length & 0xFF:02X} {(total_length >> 8) & 0xFF:02X}" # 小端格式
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{start_address}批量写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def write_d_130_ascii_batch(self, start_address: int, values: str, data_type: str = "16bit"):
|
||
try:
|
||
# 小端地址
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置参数
|
||
if data_type not in ["16bit", "32bit"]:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
device_code = "A8"
|
||
|
||
def get_16bit_data(char1, char2):
|
||
"""将两个ASCII字符组合到一个16位寄存器中（小端格式）"""
|
||
# 第一个字符在低位，第二个字符在高位
|
||
value = (ord(char2) << 8) | ord(char1)
|
||
low = value & 0xFF
|
||
high = (value >> 8) & 0xFF
|
||
return f"{low:02X} {high:02X}"
|
||
|
||
def get_32bit_data(char1, char2, char3, char4):
|
||
"""将四个ASCII字符组合到一个32位寄存器中（小端格式）"""
|
||
# 小端序：char1在最低位，char4在最高位
|
||
value = (ord(char4) << 24) | (ord(char3) << 16) | (ord(char2) << 8) | ord(char1)
|
||
byte1 = value & 0xFF # 最低位字节
|
||
byte2 = (value >> 8) & 0xFF
|
||
byte3 = (value >> 16) & 0xFF
|
||
byte4 = (value >> 24) & 0xFF # 最高位字节
|
||
return f"{byte1:02X} {byte2:02X} {byte3:02X} {byte4:02X}"
|
||
|
||
# 构建数据部分
|
||
if data_type == "16bit":
|
||
# 16位：每两个字符组合到一个寄存器
|
||
data_hex_parts = []
|
||
# 将字符串按每两个字符分组，不足的用空格填充
|
||
for i in range(0, len(values), 2):
|
||
if i + 1 < len(values):
|
||
char1 = values[i]
|
||
char2 = values[i + 1]
|
||
else:
|
||
# 最后一个字符不足时，用空格填充
|
||
char1 = values[i]
|
||
char2 = ' '
|
||
data_hex_parts.append(get_16bit_data(char1, char2))
|
||
|
||
points = (len(values) + 1) // 2 # 向上取整计算需要的寄存器数量
|
||
bytes_per_value = 2
|
||
else: # 32bit
|
||
# 32位：每四个字符组合到一个寄存器
|
||
data_hex_parts = []
|
||
# 将字符串按每四个字符分组，不足的用空格填充
|
||
for i in range(0, len(values), 4):
|
||
chars = list(values[i:i + 4])
|
||
# 填充不足的字符
|
||
while len(chars) < 4:
|
||
chars.append(' ')
|
||
data_hex_parts.append(get_32bit_data(chars[0], chars[1], chars[2], chars[3]))
|
||
|
||
points = (len(values) + 3) // 4 * 2 # 每个32位值占用2个寄存器点
|
||
bytes_per_value = 4
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据
|
||
actual_data_bytes = len(data_hex_parts) * bytes_per_value
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据字节数
|
||
total_length = 6 + 2 + 2 + 2 + actual_data_bytes
|
||
length_hex = f"{total_length & 0xFF:02X} {(total_length >> 8) & 0xFF:02X}" # 小端格式
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{start_address}批量写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def write_d_130_ascii_batch(self, start_address: int, values: str, data_type: str = "16bit"):
|
||
try:
|
||
# 小端地址
|
||
addr_low = start_address & 0xFF
|
||
addr_high = (start_address >> 8) & 0xFF
|
||
|
||
# 根据数据类型设置参数
|
||
if data_type not in ["16bit", "32bit"]:
|
||
return False, f"不支持的数据类型: {data_type}"
|
||
|
||
device_code = "A8"
|
||
|
||
def get_16bit_data(char1, char2):
|
||
"""将两个ASCII字符组合到一个16位寄存器中（小端格式）"""
|
||
# 第一个字符在低位，第二个字符在高位
|
||
value = (ord(char2) << 8) | ord(char1)
|
||
low = value & 0xFF
|
||
high = (value >> 8) & 0xFF
|
||
return f"{low:02X} {high:02X}"
|
||
|
||
def get_32bit_data(char1, char2, char3, char4):
|
||
"""将四个ASCII字符组合到一个32位寄存器中（小端格式）"""
|
||
# 小端序：char1在最低位，char4在最高位
|
||
value = (ord(char4) << 24) | (ord(char3) << 16) | (ord(char2) << 8) | ord(char1)
|
||
byte1 = value & 0xFF # 最低位字节
|
||
byte2 = (value >> 8) & 0xFF
|
||
byte3 = (value >> 16) & 0xFF
|
||
byte4 = (value >> 24) & 0xFF # 最高位字节
|
||
return f"{byte1:02X} {byte2:02X} {byte3:02X} {byte4:02X}"
|
||
|
||
# 构建数据部分
|
||
if data_type == "16bit":
|
||
# 16位：每两个字符组合到一个寄存器
|
||
data_hex_parts = []
|
||
# 将字符串按每两个字符分组，不足的用空格填充
|
||
for i in range(0, len(values), 2):
|
||
if i + 1 < len(values):
|
||
char1 = values[i]
|
||
char2 = values[i + 1]
|
||
else:
|
||
# 最后一个字符不足时，用空格填充
|
||
char1 = values[i]
|
||
char2 = ' '
|
||
data_hex_parts.append(get_16bit_data(char1, char2))
|
||
|
||
points = (len(values) + 1) // 2 # 向上取整计算需要的寄存器数量
|
||
bytes_per_value = 2
|
||
else: # 32bit
|
||
# 32位：每四个字符组合到一个寄存器
|
||
data_hex_parts = []
|
||
# 将字符串按每四个字符分组，不足的用空格填充
|
||
for i in range(0, len(values), 4):
|
||
chars = list(values[i:i + 4])
|
||
# 填充不足的字符
|
||
while len(chars) < 4:
|
||
chars.append(' ')
|
||
data_hex_parts.append(get_32bit_data(chars[0], chars[1], chars[2], chars[3]))
|
||
|
||
points = (len(values) + 3) // 4 * 2 # 每个32位值占用2个寄存器点
|
||
bytes_per_value = 4
|
||
|
||
data_hex = " ".join(data_hex_parts)
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据
|
||
actual_data_bytes = len(data_hex_parts) * bytes_per_value
|
||
|
||
# 报文长度计算：子头6 + 地址2 + 软元件2 + 点数2 + 数据字节数
|
||
total_length = 6 + 2 + 2 + 2 + actual_data_bytes
|
||
length_hex = f"{total_length & 0xFF:02X} {(total_length >> 8) & 0xFF:02X}" # 小端格式
|
||
|
||
points_hex = f"{points & 0xFF:02X} {(points >> 8) & 0xFF:02X}"
|
||
|
||
cmd_hex = (
|
||
"50 00 00 FF FF 03 00 "
|
||
f"{length_hex} "
|
||
"10 00 01 14 00 00 "
|
||
f"{addr_low:02X} {addr_high:02X} 00 {device_code} "
|
||
f"{points_hex} "
|
||
f"{data_hex}"
|
||
)
|
||
buffer, err = self._send_and_receive(cmd_hex)
|
||
if err:
|
||
return False, f"D{start_address}批量写入失败: {err}"
|
||
|
||
if len(buffer) == 0:
|
||
return False, f"D{start_address}批量写入无响应"
|
||
|
||
# EndCode检查
|
||
end_code = int.from_bytes(buffer[9:11], byteorder='little')
|
||
if end_code != 0x0000:
|
||
return False, f"D{start_address}批量写入被拒绝，响应: {buffer.hex()}"
|
||
|
||
return True, None
|
||
|
||
except Exception as ex:
|
||
return False, f"D{start_address}批量写入失败: {str(ex)}"
|
||
|
||
def close(self):
|
||
"""关闭连接"""
|
||
if self.socket:
|
||
try:
|
||
self.socket.close()
|
||
except:
|
||
pass
|
||
self.connected = False
|
||
self.socket = None
|
||
|
||
def __del__(self):
|
||
"""析构函数，确保连接被关闭"""
|
||
self.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 使用示例
|
||
|
||
plc = PlcClient('192.168.2.1', 4995)
|
||
|
||
if plc.connected:
|
||
# data = [1,2,3]
|
||
# str_data = ''.join(data)
|
||
# print(len(str_data))
|
||
|
||
# print(plc.write_d_shen_7_batch(1200, str_data))
|
||
# print(plc.write_d_130_batch(1200, data))
|
||
# print(plc.read_d(1200))
|
||
# print(plc.read_d(1201))
|
||
# print(plc.read_d(1202))
|
||
|
||
print(plc.write_m(1009, True))
|
||
# print(plc.write_m(1007, True))
|
||
# # 读取130电池
|
||
# address = 200000
|
||
# count = 4100
|
||
# # 单次最多读 500
|
||
# zr_data = []
|
||
# for i in range(0, count, 500):
|
||
# if i < count - 500:
|
||
# temp_data, _ = plc.read_r_batch(address + i, 500)
|
||
# else:
|
||
# temp_data, _ = plc.read_r_batch(address + i, count - i)
|
||
# if temp_data:
|
||
# zr_data.extend(temp_data)
|
||
# else:
|
||
# print(f"ZR{address + i} 批量读取失败")
|
||
#
|
||
# print(f"ZR{address} 原始数据长度: {len(zr_data)}")
|
||
#
|
||
# # 读取566电池
|
||
# address = 210000
|
||
# count = 13018
|
||
# # 单次最多读 500
|
||
# zr_data_566 = []
|
||
# for i in range(0, count, 500):
|
||
# if i < count - 500:
|
||
# temp_data, _ = plc.read_r_batch(address + i, 500)
|
||
# else:
|
||
# temp_data, _ = plc.read_r_batch(address + i, count - i)
|
||
#
|
||
# if temp_data:
|
||
# zr_data_566.extend(temp_data)
|
||
# else:
|
||
# print(f"ZR{address + i} 批量读取失败")
|
||
#
|
||
# print(f"ZR{address} 原始数据长度: {len(zr_data_566)}")
|
||
#
|
||
# print(" 130 数据: \n ", zr_data)
|
||
# print(" 566 数据: \n ", zr_data_566)
|
||
|
||
plc.close()
|
||
else:
|
||
print("无法连接到PLC")
|