44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
from snap7_client import Snap7Client
|
||
import logging
|
||
|
||
|
||
class PLCManager:
|
||
"""PLC连接管理器，管理多个PLC连接"""
|
||
|
||
def __init__(self, plcs_config, reader_threads):
|
||
"""
|
||
初始化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"],
|
||
reader_threads
|
||
)
|
||
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 |