子头(2字节) + 网络号(1) + PLC编号(1) + IO编号(2) + CPU编号(1) + 命令(2) + 子命令(2) + 数据长度(2) + 数据块

import socket
import struct
2. 构造3E帧：批量读取字软元件 (命令0x0401)
def make_read_word_request(start_addr, count):
"""
读D寄存器，二进制3E帧
:param start_addr: int, 起始地址 (如100)
:param count: int, 字数
:return: bytes, 待发送的帧
"""
# 帧头固定部分
header = b'\x50\x00'  # 子头 (二进制3E帧)
header += b'\x00'     # 网络号
header += b'\xFF'     # PLC编号 (默认255)
header += b'\xFF\x03' # IO编号 (默认0xFF03)
header += b'\x00'     # CPU编号 (默认0)
# 命令: 0401 (读字软元件, 大端)
header += struct.pack('>H', 0x0401)
# 子命令: 0000 (批量读取)
header += struct.pack('>H', 0x0000)
# 数据块部分
data_block = b''
# 软元件代码: D = 0x00A8 (大端)
data_block += struct.pack('>H', 0x00A8)
# 起始地址: 4字节大端
data_block += struct.pack('>I', start_addr)
# 读取点数: 2字节大端
data_block += struct.pack('>H', count)
# 数据长度 = 数据块长度 (不含帧头)
data_len = len(data_block)
header += struct.pack('>H', data_len)
return header + data_block
3. 解析响应帧
def parse_read_word_response(response):
"""
解析3E帧响应，返回字列表（小端）
"""
if len(response) < 17:
raise ValueError("响应帧太短")
# 检查结束代码 (第11-12字节，大端)
end_code = struct.unpack_from('>H', response, 10)[0]  # 偏移10? 注意: 结束代码在第11~12字节(下标10~11)
if end_code != 0:
raise Exception(f"PLC返回错误代码: 0x{end_code:04X}")
# 数据长度 (第15-16字节)
data_len = struct.unpack_from('>H', response, 14)[0]
# 数据区从17字节开始
data = response[17:17+data_len]
# 每个字2字节小端
words = []
for i in range(0, len(data), 2):
word = struct.unpack_from('<H', data, i)[0]  # 小端!
words.append(word)
return words
4. 主通信函数
def read_plc_registers(ip, start_addr, count, port=2000, timeout=3):
"""
ip: PLC IP地址
start_addr: 起始D地址 (如100)
count: 字数
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(timeout)
try:
s.connect((ip, port))
request = make_read_word_request(start_addr, count)
s.sendall(request)
# 先接收帧头 (17字节)
header = b''
while len(header) < 17:
chunk = s.recv(17 - len(header))
if not chunk:
raise ConnectionError("连接断开")
header += chunk
# 读取数据长度 (帧头第15-16字节)
data_len = struct.unpack_from('>H', header, 14)[0]
# 接收剩余数据
data = b''
while len(data) < data_len:
chunk = s.recv(data_len - len(data))
if not chunk:
raise ConnectionError("读取数据中断")
data += chunk
full_response = header + data
return parse_read_word_response(full_response)
finally:
s.close()
5. 使用示例
if __name__ == '__main__':
values = read_plc_registers('192.168.1.100', 100, 10)
print("D100~D109:", values)

import csv
import sqlite3
import datetime
SQLite示例: 创建表并批量插入
def create_db():
conn = sqlite3.connect('plc_data.db')
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS registers (
timestamp TEXT,
start_addr INTEGER,
values TEXT
)''')
conn.commit()
conn.close()
def insert_data(start_addr, values):
conn = sqlite3.connect('plc_data.db')
c = conn.cursor()
now = datetime.datetime.now().isoformat()
# values转为逗号分隔字符串
val_str = ','.join(str(v) for v in values)
c.execute("INSERT INTO registers VALUES (?,?,?)", (now, start_addr, val_str))
conn.commit()
conn.close()

📚 推荐阅读

KEYWORDS

PLC, plc, MELSEC, 三菱, 函数

💡 如果你觉得这篇文章有帮助,请点个在看,分享给更多需要的人!

📝 关注我,获取更多实用干货~

🤝 有问题欢迎评论区留言交流!