diff --git a/core/edge-gateway/drivers/s7_driver.py b/core/edge-gateway/drivers/s7_driver.py index c584114..a9b526e 100644 --- a/core/edge-gateway/drivers/s7_driver.py +++ b/core/edge-gateway/drivers/s7_driver.py @@ -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: diff --git a/core/edge-gateway/tests/test_s7_driver.py b/core/edge-gateway/tests/test_s7_driver.py new file mode 100644 index 0000000..3e018c6 --- /dev/null +++ b/core/edge-gateway/tests/test_s7_driver.py @@ -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()