import socket 
import struct
import time
class MCRemoteClient:
"""三菱MC协议3E帧通信客户端"""
def __init__(self, plc_ip='192.168.1.100', plc_port=1025):
self.plc_ip = plc_ip
self.plc_port = plc_port
self.sock = None
self.connect()
def connect(self):
"""建立TCP连接，超时5秒"""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(5.0)
try:
self.sock.connect((self.plc_ip, self.plc_port))
print(f"[OK] 连接成功 -> {self.plc_ip}:{self.plc_port}")
except Exception as e:
print(f"[ERROR] 连接失败: {e}")
self.sock = None
def _build_3e_frame(self, subcommand, data):
"""构建3E帧格式
帧结构: 帧头(2B) + 网络编号(1B) + PC编号(1B) +
请求目标(1B) + 定时器(2B) + 命令(2B) +
子命令(2B) + 数据体(N)
"""
frame = b'\\x50\\x00'  # 帧头固定
frame += b'\\x00'      # 网络编号
frame += b'\\xff'      # PC编号（默认255）
frame += b'\\xff\\x03' # 请求目标
frame += b'\\x00\\x3c' # 定时器 60秒
frame += b'\\x04\\x01' # 命令：批量读取
frame += struct.pack('>H', subcommand)  # 子命令
frame += data
return frame
def send_command(self, command_bytes):
"""发送命令并接收响应"""
if not self.sock:
print("[ERROR] 未连接")
return None
try:
self.sock.send(command_bytes)
response = self.sock.recv(2048)
return response
except socket.timeout:
print("[ERROR] 接收超时")
return None

def parse_mc_response(self, response_bytes): 
"""解析MC协议响应数据"""
if len(response_bytes) < 11:
print("[ERROR] 响应数据太短")
return None
# 检查响应头
if response_bytes[0:4] != b'\\xd0\\x00\\x00\\xff':
print("[ERROR] 响应头异常")
return None
# 解析数据长度
data_length = struct.unpack('>H', response_bytes[7:9])[0]
data_start = 11  # 数据从第11字节开始
# 读取16位整数
if data_length == 2:
return struct.unpack('>H', response_bytes[data_start:data_start+2])[0]
# 读取32位浮点
elif data_length == 4:
return struct.unpack('>f', response_bytes[data_start:data_start+4])[0]
# 读取字符串
else:
return response_bytes[data_start:data_start+data_length].decode('ascii', errors='ignore')

def read_with_retry(self, address, retries=3): 
"""带重连机制的读取函数"""
for attempt in range(retries):
response = self.send_command(address)
if response:
return self.parse_mc_response(response)
else:
print(f"[WARN] 第{attempt+1}次读取失败，重连中...")
self.connect()
time.sleep(1)
print("[ERROR] 重试3次后仍失败")
return None

import asyncio 
import aiohttp
class RemotePLCManager:
"""远程PLC管理平台"""
def __init__(self, plc_list):
self.plc_list = plc_list  # [{'ip':'192.168.1.100','name':'1号PLC'}, ...]
self.connection_pool = {}  # 连接池
self.data_buffer = {}  # 数据缓存
async def connect_all(self):
"""异步连接所有PLC"""
tasks = [self.connect_plc(plc) for plc in self.plc_list]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"[ERROR] {self.plc_list[i]['name']}连接失败: {result}")
else:
print(f"[OK] {self.plc_list[i]['name']}连接成功")
async def connect_plc(self, plc_info):
"""单台PLC连接（异步）"""
client = MCRemoteClient(plc_info['ip'], 1025)
self.connection_pool[plc_info['name']] = client
return client

async def batch_read(self, plc_name, addresses): 
"""批量读取多个地址（异步）"""
client = self.connection_pool.get(plc_name)
if not client:
return None
# 构建批量读取命令
data = b''
for addr in addresses:
# 地址格式：D100 → 0x0064
data += struct.pack('>H', addr)
command = client._build_3e_frame(0x0001, data)  # 子命令0x0001=批量读取
response = await self.async_send(client, command)
return self.parse_batch_response(response, len(addresses))

def evaluate_data_quality(self, current_value, previous_value, threshold=5): 
"""评估数据质量"""
# 1. 检查数值范围
if current_value < 0 or current_value > 10000:
return "异常值"
# 2. 检查变化率（防止跳变）
if previous_value and abs(current_value - previous_value) > threshold:
return "跳变异常"
# 3. 检查连续相同值（传感器可能卡死）
if current_value == previous_value:
self.same_count += 1
if self.same_count > 5:
return "传感器可能卡死"
else:
self.same_count = 0
return "正常"

def safe_write_plc(self, plc_name, address, value, confirm=True): 
"""安全写入PLC（带确认机制）"""
if confirm:
# 第一步：读取当前值
current = self.read_plc(plc_name, address)
print(f"[INFO] 当前值：{current}")
# 第二部：发送写入命令
confirm_msg = f"确认写入 {plc_name} 的 {address} 值为 {value}? (y/n)"
if input(confirm_msg).lower() != 'y':
print("[CANCEL] 写入已取消")
return False
# 执行写入
write_command = self._build_write_command(plc_name, address, value)
response = self.send_command(write_command)
return response

# Dockerfile 
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "main.py"]

📚 推荐阅读

KEYWORDS

PLC, plc, iQ-R, 三菱

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

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

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