70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""西门子 S7-1200 驱动 —— 参数化适配（issue #24 落点）。
|
|||
|
|
|
|||
|
|
连接参数（ip / rack / slot / db / offset 映射）全部来自模板配置，
|
|||
|
|
点位到 DB 地址的映射通过点位字典 CSV 的 opcNode 列携带
|
|||
|
|
（S7 场景下约定格式 `DB{db}.{byte}.{bit}`，如 DB100.0.0）。
|
|||
|
|
依赖 `python-snap7`，未安装时给出明确提示。
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import re
|
|||
|
|
from typing import Dict, List, Optional
|
|||
|
|
|
|||
|
|
from point_dict.loader import Point
|
|||
|
|
from .base import Driver, SampleValue
|
|||
|
|
|
|||
|
|
_S7_NODE_RE = re.compile(r"^DB(\d+)\.(\d+)(?:\.(\d+))?$")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class S7Driver(Driver):
|
|||
|
|
"""西门子 S7-1200 只读采集驱动（参数化适配）。"""
|
|||
|
|
|
|||
|
|
protocol = "s7"
|
|||
|
|
|
|||
|
|
def __init__(self, config: Optional[dict] = None):
|
|||
|
|
super().__init__(config)
|
|||
|
|
self.ip: str = self.config.get("ip", "")
|
|||
|
|
self.rack: int = int(self.config.get("rack", 0))
|
|||
|
|
self.slot: int = int(self.config.get("slot", 1))
|
|||
|
|
self._client = None
|
|||
|
|
|
|||
|
|
def connect(self) -> None:
|
|||
|
|
try:
|
|||
|
|
import snap7 # type: ignore
|
|||
|
|
except ImportError as exc: # pragma: no cover - 依赖缺失路径
|
|||
|
|
raise ConnectionError(
|
|||
|
|
"S7 驱动依赖库未安装：请 `pip install python-snap7`"
|
|||
|
|
) from exc
|
|||
|
|
if not self.ip:
|
|||
|
|
raise ConnectionError("S7 驱动缺少配置: drivers.s7.ip")
|
|||
|
|
self._client = snap7.client.Client()
|
|||
|
|
self._client.connect(self.ip, self.rack, self.slot)
|
|||
|
|
|
|||
|
|
def read_points(self, points: List[Point]) -> Dict[str, SampleValue]:
|
|||
|
|
if self._client is None:
|
|||
|
|
raise ConnectionError("S7 客户端未连接，请先 connect()")
|
|||
|
|
result: Dict[str, SampleValue] = {}
|
|||
|
|
for p in points:
|
|||
|
|
match = _S7_NODE_RE.match(p.opc_node or "")
|
|||
|
|
if not match:
|
|||
|
|
continue # 非 S7 点位由其它驱动采集
|
|||
|
|
db, byte_, bit = int(match.group(1)), int(match.group(2)), match.group(3)
|
|||
|
|
try:
|
|||
|
|
if bit is not None:
|
|||
|
|
result[p.point_id] = self._client.read_area(
|
|||
|
|
0x84, db, byte_ * 8 + int(bit), 1
|
|||
|
|
)[0]
|
|||
|
|
else:
|
|||
|
|
result[p.point_id] = self._client.read_area(0x84, db, byte_, 4)[0]
|
|||
|
|
except Exception:
|
|||
|
|
result[p.point_id] = None # 单点失败不中断整批
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def close(self) -> None:
|
|||
|
|
if self._client is not None:
|
|||
|
|
try:
|
|||
|
|
self._client.disconnect()
|
|||
|
|
finally:
|
|||
|
|
self._client = None
|