42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
|
from snap7_client import Snap7Client
|
|||
|
|
import logging
|
|||
|
|
|
|||
|
|
class PLCManager:
|
|||
|
|
"""PLC连接管理器，管理多个PLC连接"""
|
|||
|
|
|
|||
|
|
def __init__(self, plcs_config):
|
|||
|
|
"""
|
|||
|
|
初始化PLC管理器
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
plcs_config: PLC配置列表
|
|||
|
|
"""
|
|||
|
|
self.plcs = {}
|
|||
|
|
for plc_config in plcs_config:
|
|||
|
|
name = plc_config["name"]
|
|||
|
|
self.plcs[name] = Snap7Client(
|
|||
|
|
plc_config["ip"],
|
|||
|
|
plc_config["rack"],
|
|||
|
|
plc_config["slot"]
|
|||
|
|
)
|
|||
|
|
self.logger = logging.getLogger("PLCManager")
|
|||
|
|
|
|||
|
|
def get_plc(self, name):
|
|||
|
|
"""获取指定名称的PLC客户端"""
|
|||
|
|
return self.plcs.get(name)
|
|||
|
|
|
|||
|
|
def connect_all(self):
|
|||
|
|
"""连接所有配置的PLC"""
|
|||
|
|
for name, client in self.plcs.items():
|
|||
|
|
client.connect()
|
|||
|
|
|
|||
|
|
def get_connection_status(self):
|
|||
|
|
"""获取所有PLC的连接状态"""
|
|||
|
|
status = {}
|
|||
|
|
for name, client in self.plcs.items():
|
|||
|
|
status[name] = {
|
|||
|
|
"ip": client.ip,
|
|||
|
|
"connected": client.connected,
|
|||
|
|
"retry_count": client.retry_count
|
|||
|
|
}
|
|||
|
|
return status
|