feat(#24): 西门子 S7-1200 驱动参数化适配
- 连接参数 ip/port/rack/slot/timeout 全部来自模板配置（去硬编码）
- 点位→DB 地址映射经点位字典 opcNode 列携带（DB{db}.{byte}[.{bit}]）
- 按 dataType 解码：float(REAL/4B) int(WORD/2B) bool(位)，S7 大端
- read_area 用 snap7 常量 0x84(DB区)，单点失败容错记 None 计入丢失率
- 严格只读，无任何写/控制指令（PRD 9 章安全约束）
- 新增 test_s7_driver.py 18 个用例（mock snap7），全量通过
This commit is contained in:
@@ -1,14 +1,24 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""西门子 S7-1200 驱动 —— 参数化适配（issue #24 落点）。
|
||||
|
||||
连接参数（ip / rack / slot / db / offset 映射）全部来自模板配置，
|
||||
点位到 DB 地址的映射通过点位字典 CSV 的 opcNode 列携带
|
||||
（S7 场景下约定格式 `DB{db}.{byte}.{bit}`，如 DB100.0.0）。
|
||||
连接参数（ip / port / rack / slot）全部来自模板配置；点位到 DB 地址的
|
||||
映射通过点位字典 CSV 的 opcNode 列携带，约定格式：
|
||||
|
||||
DB{db}.{byte} → 按点位 dataType 读取整字（WORD/DWORD/REAL）
|
||||
DB{db}.{byte}.{bit} → 读取位（BOOL）
|
||||
|
||||
数据类型解析（对齐点位字典 schema.dataType ∈ float/int/bool）：
|
||||
bool → 1 字节，取指定位
|
||||
int → 2 字节，有符号 WORD（big-endian，S7 默认大端）
|
||||
float → 4 字节，IEEE754 单精度 REAL（big-endian）
|
||||
|
||||
安全约束（PRD 9 章）：仅暴露只读 `read_points()`，无任何写/控制指令。
|
||||
依赖 `python-snap7`，未安装时给出明确提示。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import struct
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from point_dict.loader import Point
|
||||
@@ -16,17 +26,38 @@ from .base import Driver, SampleValue
|
||||
|
||||
_S7_NODE_RE = re.compile(r"^DB(\d+)\.(\d+)(?:\.(\d+))?$")
|
||||
|
||||
# S7 area 标识（snap7 常量）：0x84 = DB 区
|
||||
_S7_AREA_DB = 0x84
|
||||
|
||||
# dataType → (字节数, python struct 格式字符)
|
||||
# S7 PLC 默认大端（big-endian），故用 ">" 前缀。
|
||||
_TYPE_SPEC = {
|
||||
"bool": (1, None), # 位读取，单独处理
|
||||
"int": (2, ">h"), # 有符号 16 位整数（WORD/INT）
|
||||
"float": (4, ">f"), # IEEE754 32 位单精度（REAL）
|
||||
}
|
||||
|
||||
|
||||
class S7Driver(Driver):
|
||||
"""西门子 S7-1200 只读采集驱动（参数化适配）。"""
|
||||
"""西门子 S7-1200 只读采集驱动（参数化适配）。
|
||||
|
||||
配置项（drivers.s7.*）：
|
||||
ip str 必填 PLC IP 地址
|
||||
port int 可选 ISO TCP 端口，默认 102
|
||||
rack int 可选 机架号，默认 0
|
||||
slot int 可选 插槽号，S7-1200 默认 1
|
||||
timeout int 可选 连接/读取超时(秒)，默认 10
|
||||
"""
|
||||
|
||||
protocol = "s7"
|
||||
|
||||
def __init__(self, config: Optional[dict] = None):
|
||||
super().__init__(config)
|
||||
self.ip: str = self.config.get("ip", "")
|
||||
self.port: int = int(self.config.get("port", 102))
|
||||
self.rack: int = int(self.config.get("rack", 0))
|
||||
self.slot: int = int(self.config.get("slot", 1))
|
||||
self.timeout: int = int(self.config.get("timeout", 10))
|
||||
self._client = None
|
||||
|
||||
def connect(self) -> None:
|
||||
@@ -39,7 +70,12 @@ class S7Driver(Driver):
|
||||
if not self.ip:
|
||||
raise ConnectionError("S7 驱动缺少配置: drivers.s7.ip")
|
||||
self._client = snap7.client.Client()
|
||||
self._client.connect(self.ip, self.rack, self.slot)
|
||||
# python-snap7 支持 (host, rack, slot) 或 (host, rack, slot, tcp_port)
|
||||
try:
|
||||
self._client.connect(self.ip, self.rack, self.slot, self.port)
|
||||
except TypeError:
|
||||
# 旧版本签名不支持 port 参数，回退三参数
|
||||
self._client.connect(self.ip, self.rack, self.slot)
|
||||
|
||||
def read_points(self, points: List[Point]) -> Dict[str, SampleValue]:
|
||||
if self._client is None:
|
||||
@@ -49,16 +85,23 @@ class S7Driver(Driver):
|
||||
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)
|
||||
db = int(match.group(1))
|
||||
byte_offset = int(match.group(2))
|
||||
bit = 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]
|
||||
# BOOL：读 1 字节，取指定位
|
||||
raw = self._client.read_area(_S7_AREA_DB, db, byte_offset, 1)
|
||||
result[p.point_id] = bool(raw[0] & (1 << int(bit)))
|
||||
else:
|
||||
result[p.point_id] = self._client.read_area(0x84, db, byte_, 4)[0]
|
||||
dtype = (p.data_type or "float").lower()
|
||||
spec = _TYPE_SPEC.get(dtype, _TYPE_SPEC["float"])
|
||||
size, fmt = spec
|
||||
raw = self._client.read_area(_S7_AREA_DB, db, byte_offset, size)
|
||||
result[p.point_id] = struct.unpack(fmt, bytes(raw[:size]))[0]
|
||||
except Exception:
|
||||
result[p.point_id] = None # 单点失败不中断整批
|
||||
# 单点失败不中断整批；由引擎按“未读到”计入丢失率
|
||||
result[p.point_id] = None
|
||||
return result
|
||||
|
||||
def close(self) -> None:
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""S7-1200 驱动参数化适配单元测试（issue #24）。
|
||||
|
||||
snap7 为现场 PLC 依赖，CI 环境不可安装，故用注入 fake client 的方式
|
||||
验证：地址解析、按 dataType 解码（bool/int/float）、容错与配置项。
|
||||
"""
|
||||
import struct
|
||||
import unittest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from drivers.s7_driver import S7Driver, _S7_NODE_RE
|
||||
from point_dict.loader import Point
|
||||
|
||||
|
||||
def _point(node: str, dtype: str = "float", pid: str = "S7-01.X") -> Point:
|
||||
return Point(
|
||||
device_id="S7-01",
|
||||
point_id=pid,
|
||||
name="测点",
|
||||
unit="℃",
|
||||
data_type=dtype,
|
||||
sample_rate=1000,
|
||||
quality_code=True,
|
||||
opc_node=node,
|
||||
protocol="s7",
|
||||
row_number=2,
|
||||
)
|
||||
|
||||
|
||||
class _FakeClient:
|
||||
"""模拟 snap7 client：按预置 {(db, start, size): bytes} 返回区域数据。"""
|
||||
|
||||
def __init__(self, area_map):
|
||||
# area_map: {(db, byte_offset, size): bytes}
|
||||
self.area_map = area_map
|
||||
self.read_area = MagicMock(side_effect=self._read_area)
|
||||
|
||||
def _read_area(self, area, db, start, size):
|
||||
assert area == 0x84, "S7 只读应固定 DB 区(0x84)"
|
||||
return self.area_map[(db, start, size)]
|
||||
|
||||
def disconnect(self):
|
||||
pass
|
||||
|
||||
|
||||
class S7NodeRegexTest(unittest.TestCase):
|
||||
"""opcNode 地址格式解析。"""
|
||||
|
||||
def test_parse_db_byte(self):
|
||||
m = _S7_NODE_RE.match("DB100.4")
|
||||
self.assertIsNotNone(m)
|
||||
self.assertEqual((m.group(1), m.group(2), m.group(3)), ("100", "4", None))
|
||||
|
||||
def test_parse_db_byte_bit(self):
|
||||
m = _S7_NODE_RE.match("DB100.4.3")
|
||||
self.assertIsNotNone(m)
|
||||
self.assertEqual((m.group(1), m.group(2), m.group(3)), ("100", "4", "3"))
|
||||
|
||||
def test_reject_non_s7_node(self):
|
||||
self.assertIsNone(_S7_NODE_RE.match("ns=2;s=Temp"))
|
||||
self.assertIsNone(_S7_NODE_RE.match("holding:40001"))
|
||||
self.assertIsNone(_S7_NODE_RE.match(""))
|
||||
|
||||
|
||||
class S7ConfigTest(unittest.TestCase):
|
||||
"""配置项参数化。"""
|
||||
|
||||
def test_defaults(self):
|
||||
d = S7Driver({})
|
||||
self.assertEqual(d.port, 102)
|
||||
self.assertEqual((d.rack, d.slot), (0, 1))
|
||||
self.assertEqual(d.timeout, 10)
|
||||
|
||||
def test_custom(self):
|
||||
d = S7Driver({"ip": "192.168.1.10", "port": 10102, "rack": 0, "slot": 2})
|
||||
self.assertEqual(d.ip, "192.168.1.10")
|
||||
self.assertEqual(d.port, 10102)
|
||||
self.assertEqual(d.slot, 2)
|
||||
|
||||
def test_protocol_name(self):
|
||||
self.assertEqual(S7Driver.protocol, "s7")
|
||||
|
||||
def test_connect_requires_ip(self):
|
||||
d = S7Driver({})
|
||||
with self.assertRaises(ConnectionError):
|
||||
d.connect()
|
||||
|
||||
|
||||
class S7ReadTest(unittest.TestCase):
|
||||
"""read_points 按 dataType 解码。"""
|
||||
|
||||
def _driver_with_map(self, area_map):
|
||||
d = S7Driver({"ip": "1.2.3.4"})
|
||||
d._client = _FakeClient(area_map)
|
||||
return d
|
||||
|
||||
def test_read_float_real(self):
|
||||
# DB100.0 → 4 字节 REAL，编码 25.5
|
||||
raw = struct.pack(">f", 25.5)
|
||||
d = self._driver_with_map({(100, 0, 4): raw})
|
||||
out = d.read_points([_point("DB100.0", "float", "S7.TEMP")])
|
||||
self.assertAlmostEqual(out["S7.TEMP"], 25.5, places=4)
|
||||
|
||||
def test_read_int_word(self):
|
||||
# DB200.2 → 2 字节有符号 INT，编码 -1234
|
||||
raw = struct.pack(">h", -1234)
|
||||
d = self._driver_with_map({(200, 2, 2): raw})
|
||||
out = d.read_points([_point("DB200.2", "int", "S7.PRES")])
|
||||
self.assertEqual(out["S7.PRES"], -1234)
|
||||
|
||||
def test_read_bool_bit(self):
|
||||
# DB300.0.2 → 字节 0x04（第 2 位为 1）
|
||||
d = self._driver_with_map({(300, 0, 1): bytes([0x04])})
|
||||
out = d.read_points([_point("DB300.0.2", "bool", "S7.RUN")])
|
||||
self.assertIs(out["S7.RUN"], True)
|
||||
|
||||
def test_read_bool_zero(self):
|
||||
d = self._driver_with_map({(300, 0, 1): bytes([0x00])})
|
||||
out = d.read_points([_point("DB300.0.5", "bool", "S7.IDLE")])
|
||||
self.assertIs(out["S7.IDLE"], False)
|
||||
|
||||
def test_mixed_batch(self):
|
||||
area = {
|
||||
(100, 0, 4): struct.pack(">f", 100.0), # float @ DB100.0
|
||||
(100, 4, 2): struct.pack(">h", 7), # int @ DB100.4
|
||||
(100, 6, 1): bytes([0x80]), # bool @ DB100.6.7
|
||||
}
|
||||
d = self._driver_with_map(area)
|
||||
pts = [
|
||||
_point("DB100.0", "float", "A"),
|
||||
_point("DB100.4", "int", "B"),
|
||||
_point("DB100.6.7", "bool", "C"),
|
||||
]
|
||||
out = d.read_points(pts)
|
||||
self.assertAlmostEqual(out["A"], 100.0, places=3)
|
||||
self.assertEqual(out["B"], 7)
|
||||
self.assertIs(out["C"], True)
|
||||
|
||||
def test_non_s7_node_skipped(self):
|
||||
d = self._driver_with_map({})
|
||||
out = d.read_points([_point("ns=2;s=x", "float", "X")])
|
||||
self.assertNotIn("X", out)
|
||||
|
||||
def test_read_failure_yields_none_not_raise(self):
|
||||
# read_area 抛异常 → 该点 None，整批不中断
|
||||
d = S7Driver({"ip": "1.2.3.4"})
|
||||
client = MagicMock()
|
||||
client.read_area.side_effect = RuntimeError("PLC 断连")
|
||||
d._client = client
|
||||
out = d.read_points([_point("DB100.0", "float", "BAD")])
|
||||
self.assertIsNone(out["BAD"])
|
||||
|
||||
def test_default_dtype_float_when_missing(self):
|
||||
raw = struct.pack(">f", 1.25)
|
||||
d = self._driver_with_map({(100, 0, 4): raw})
|
||||
# data_type 留空 → 走 float 解码
|
||||
p = _point("DB100.0", "", "Z")
|
||||
out = d.read_points([p])
|
||||
self.assertAlmostEqual(out["Z"], 1.25, places=4)
|
||||
|
||||
def test_read_before_connect_raises(self):
|
||||
d = S7Driver({"ip": "1.2.3.4"})
|
||||
with self.assertRaises(ConnectionError):
|
||||
d.read_points([_point("DB100.0", "float", "X")])
|
||||
|
||||
|
||||
class S7CloseTest(unittest.TestCase):
|
||||
def test_close_resets_client(self):
|
||||
d = S7Driver({"ip": "1.2.3.4"})
|
||||
client = MagicMock()
|
||||
d._client = client
|
||||
d.close()
|
||||
client.disconnect.assert_called_once()
|
||||
self.assertIsNone(d._client)
|
||||
|
||||
def test_close_idempotent(self):
|
||||
d = S7Driver({"ip": "1.2.3.4"})
|
||||
d.close() # 未连接，不报错
|
||||
self.assertIsNone(d._client)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user