From 50b51d4a275575b1ef01084eca3eb136f3168b6a Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Tue, 4 Aug 2026 23:34:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#25=20?= =?UTF-8?q?=E2=91=A0=20=E5=92=8C=E5=88=A9=E6=97=B6=20DCS=20OPC=20=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=8F=82=E6=95=B0=E5=8C=96=E9=80=82=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/edge-gateway/config/gateway.example.yaml | 17 ++- core/edge-gateway/drivers/__init__.py | 3 + core/edge-gateway/drivers/hollysys_driver.py | 113 ++++++++++++++ core/edge-gateway/drivers/opcua_driver.py | 14 ++ .../tests/test_hollysys_driver.py | 139 ++++++++++++++++++ 5 files changed, 283 insertions(+), 3 deletions(-) create mode 100644 core/edge-gateway/drivers/hollysys_driver.py create mode 100644 core/edge-gateway/tests/test_hollysys_driver.py diff --git a/core/edge-gateway/config/gateway.example.yaml b/core/edge-gateway/config/gateway.example.yaml index c6a7005..3130cf4 100644 --- a/core/edge-gateway/config/gateway.example.yaml +++ b/core/edge-gateway/config/gateway.example.yaml @@ -10,13 +10,24 @@ collector: cache_limit_bytes: 536870912 # 512MB # 协议可插拔驱动插槽（protocol 注册表见 drivers/__init__.py） drivers: - # 和利时 DCS —— OPC UA 只读 - - protocol: opcua + # 和利时 DCS —— OPC UA 只读（推荐；支持用户认证参数化，issue #25） + - protocol: hollysys device_prefixes: ["CLF"] # 设备编号前缀路由（CLF-01, CLF-02, ...） config: + mode: ua # ua（OPC UA）| da（OPC DA COM 接口） endpoint: "opc.tcp://10.20.1.10:4840" - security: "None" + security: "Basic256Sha256" # None | Basic256Sha256 等安全策略 + username: "iaop_read" # 和利时 UA 服务用户认证（只读账号） + password: "******" timeout: 10 + # 和利时 DCS —— OPC DA 传统接口（无 UA 服务时使用；opcNode 列写 OPC Item） + # - protocol: hollysys + # device_prefixes: ["CLF"] + # config: + # mode: da + # host: "10.20.1.10" # DCOM 主机（需 Windows + OpenOPC） + # prog_id: "Hollysys.OPCServer" # 或 cls_id 指定 CLSID + # timeout: 10 # 西门子 S7-1200 —— PLC 只读（点位 opcNode 列写 DB 地址，如 DB100.0.0） - protocol: s7 device_prefixes: ["S7"] diff --git a/core/edge-gateway/drivers/__init__.py b/core/edge-gateway/drivers/__init__.py index b044a55..8c4eb45 100644 --- a/core/edge-gateway/drivers/__init__.py +++ b/core/edge-gateway/drivers/__init__.py @@ -12,6 +12,7 @@ if TYPE_CHECKING: # pragma: no cover from .base import Driver from .energy_driver import EnergyDriver +from .hollysys_driver import HollysysDcsDriver from .modbus_driver import ModbusDriver from .opcua_driver import OpcUaDriver from .s7_driver import S7Driver @@ -20,6 +21,7 @@ from .weighing_driver import WeighingDriver _DRIVER_REGISTRY: Dict[str, Type["Driver"]] = { "opcua": OpcUaDriver, + "hollysys": HollysysDcsDriver, "s7": S7Driver, "modbus": ModbusDriver, "weighing": WeighingDriver, @@ -30,6 +32,7 @@ _DRIVER_REGISTRY: Dict[str, Type["Driver"]] = { __all__ = [ "Driver", "OpcUaDriver", + "HollysysDcsDriver", "S7Driver", "ModbusDriver", "WeighingDriver", diff --git a/core/edge-gateway/drivers/hollysys_driver.py b/core/edge-gateway/drivers/hollysys_driver.py new file mode 100644 index 0000000..5cc08bb --- /dev/null +++ b/core/edge-gateway/drivers/hollysys_driver.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +"""和利时 DCS OPC 接口驱动 —— 参数化适配（issue #25）。 + +和利时 DCS（HOLLiAS-M / MACS 系列）对外数据接口常见两种形态，全部参数化： +- `mode: ua` —— OPC UA 接口（推荐）：endpoint / security / username / password + 均来自模板配置（gateway.yaml 的 drivers.hollysys 段），委托 OpcUaDriver + （含用户认证参数化，见 opcua_driver.py）； +- `mode: da` —— OPC DA 传统 COM/DCOM 接口：host（DCOM 主机）/ prog_id + （如 "Hollysys.OPCServer"）/ cls_id 参数化，点位 opcNode 列写 OPC Item 路径 + （如 "CLF.Temp"）。 + +安全约束（PRD 9 章）：仅暴露只读 read_points()，无任何写/控制方法。 +依赖（未安装时给出明确提示，不静默失败）： +- ua 模式：`opcua`（同步）或 `asyncua`； +- da 模式：Windows 上 `OpenOPC`（依赖 pywin32）。 +""" +from __future__ import annotations + +from typing import Dict, List, Optional + +from point_dict.loader import Point +from .base import Driver, SampleValue +from .opcua_driver import OpcUaDriver + + +class _HollysysDaDriver(Driver): + """OPC DA（COM/DCOM）只读采集实现 —— 和利时传统接口。""" + + protocol = "hollysys-da" + + def __init__(self, config: Optional[dict] = None): + super().__init__(config) + self.host: str = self.config.get("host", "") + self.prog_id: str = self.config.get("prog_id", "Hollysys.OPCServer") + self.cls_id: str = self.config.get("cls_id", "") + self._client = None + + def connect(self) -> None: + try: + import OpenOPC # type: ignore + except ImportError as exc: # pragma: no cover - 依赖缺失路径 + raise ConnectionError( + "OPC DA 驱动依赖库未安装：请 `pip install OpenOPC`（Windows + pywin32）" + ) from exc + if not self.host: + raise ConnectionError("OPC DA 驱动缺少配置: drivers.hollysys.host") + self._client = OpenOPC.client() + self._client.connect(self.prog_id, self.host) + # 可选：按 CLSID 连接（部分和利时 OPC 服务器无 ProgID 注册时使用） + if self.cls_id: + self._client.connect(self.cls_id, self.host) + + def read_points(self, points: List[Point]) -> Dict[str, SampleValue]: + if self._client is None: + raise ConnectionError("OPC DA 客户端未连接，请先 connect()") + result: Dict[str, SampleValue] = {} + for p in points: + item = p.opc_node + if not item: + continue # 无 OPC Item 的点由其它驱动采集 + try: + value = self._client.read(item) + # OpenOPC.read 返回 (value, quality, timestamp) 或裸值 + if isinstance(value, (tuple, list)) and value: + value = value[0] + result[p.point_id] = value + except Exception: + # 单点读取失败不中断整批：记为未读到（计入丢失率） + result[p.point_id] = None + return result + + def close(self) -> None: + if self._client is not None: + try: + self._client.close() + finally: + self._client = None + + +class HollysysDcsDriver(Driver): + """和利时 DCS OPC 接口只读采集驱动（参数化适配，issue #25）。 + + 按 ``mode`` 参数路由到底层实现（与 energy_driver 的委托模式一致）： + - ua：OPC UA（委托 :class:`OpcUaDriver`，支持 endpoint/security/认证）； + - da：OPC DA COM 接口（委托 :class:`_HollysysDaDriver`）。 + """ + + protocol = "hollysys" + + def __init__(self, config: Optional[dict] = None): + super().__init__(config) + self.mode: str = self.config.get("mode", "ua") # ua | da + self._delegate: Optional[Driver] = None + + def connect(self) -> None: + if self.mode == "da": + self._delegate = _HollysysDaDriver(self.config) + else: + # ua 默认：和利时 OPC UA 服务，支持用户认证参数化 + self._delegate = OpcUaDriver(self.config) + self._delegate.connect() + + def read_points(self, points: List[Point]) -> Dict[str, SampleValue]: + if self._delegate is None: + raise ConnectionError("和利时驱动未连接，请先 connect()") + return self._delegate.read_points(points) + + def close(self) -> None: + if self._delegate is not None: + try: + self._delegate.close() + finally: + self._delegate = None diff --git a/core/edge-gateway/drivers/opcua_driver.py b/core/edge-gateway/drivers/opcua_driver.py index d63d541..e38c1cc 100644 --- a/core/edge-gateway/drivers/opcua_driver.py +++ b/core/edge-gateway/drivers/opcua_driver.py @@ -22,6 +22,8 @@ class OpcUaDriver(Driver): super().__init__(config) self.endpoint: str = self.config.get("endpoint", "") self.security: str = self.config.get("security", "None") + self.username: str = self.config.get("username", "") + self.password: str = self.config.get("password", "") self._client = None def connect(self) -> None: @@ -34,7 +36,19 @@ class OpcUaDriver(Driver): if not self.endpoint: raise ConnectionError("OPC UA 驱动缺少配置: drivers.opcua.endpoint") self._client = Client(self.endpoint, timeout=self.config.get("timeout", 10)) + # 安全策略参数化（如 Basic256Sha256；证书/密钥由现场安全通道提供） + if self.security and self.security != "None": + self._client.set_security_string(self.security) + # 用户认证参数化（和利时 DCS 的 OPC UA 服务常启用用户名/密码认证） + if self.username: + self._client.set_user(self.username) self._client.connect() + if self.username and self.password: + try: + self._client.activate_session(self.username, self.password) + except Exception: + # 部分服务端在 connect 时已完成认证；激活失败不阻断读取 + pass self._client.session_timeout = self.config.get("session_timeout", 60000) def read_points(self, points: List[Point]) -> Dict[str, SampleValue]: diff --git a/core/edge-gateway/tests/test_hollysys_driver.py b/core/edge-gateway/tests/test_hollysys_driver.py new file mode 100644 index 0000000..cfec957 --- /dev/null +++ b/core/edge-gateway/tests/test_hollysys_driver.py @@ -0,0 +1,139 @@ +# -*- coding: utf-8 -*- +"""和利时 DCS OPC 驱动参数化适配测试（issue #25）。 + +覆盖： +1. 驱动注册表包含 hollysys（from_template 可实例化）； +2. 参数化提取：ua 模式（endpoint/security/username/password）与 + da 模式（host/prog_id/cls_id）； +3. 只读约束：驱动无任何写/控制方法（PRD 9 章零控制指令下发）； +4. 依赖缺失路径：UA/DA 底层库未安装时 connect 抛出明确 ConnectionError； +5. 委托路由：connect 后 read_points/close 委托到底层实现。 +""" +import unittest +from unittest import mock + +from drivers import HollysysDcsDriver, from_template +from drivers.base import Driver +from drivers.hollysys_driver import _HollysysDaDriver +from drivers.opcua_driver import OpcUaDriver +from point_dict.loader import Point + +UA_CONFIG = { + "mode": "ua", + "endpoint": "opc.tcp://10.20.1.10:4840", + "security": "Basic256Sha256", + "username": "iaop_read", + "password": "secret", + "timeout": 10, +} +DA_CONFIG = { + "mode": "da", + "host": "10.20.1.10", + "prog_id": "Hollysys.OPCServer", + "timeout": 10, +} + + +def make_point(point_id="CLF-01.TEMP", opc_node="ns=2;s=CLF.Temp"): + return Point( + device_id="CLF-01", point_id=point_id, name="炉温", unit="℃", + data_type="float", sample_rate=1000, quality_code=True, + opc_node=opc_node, protocol="hollysys", + ) + + +class TestRegistry(unittest.TestCase): + """注册表与工厂。""" + + def test_registry_contains_hollysys(self): + driver = from_template("hollysys", UA_CONFIG) + self.assertIsInstance(driver, HollysysDcsDriver) + + def test_driver_is_read_only(self): + """只读约束：驱动只暴露 connect/read_points/close，无写方法。""" + driver = HollysysDcsDriver(UA_CONFIG) + methods = {m for m in dir(driver) if not m.startswith("_")} + self.assertIn("read_points", methods) + self.assertNotIn("write", methods) + self.assertNotIn("write_points", methods) + self.assertNotIn("set_value", methods) + + +class TestParameterization(unittest.TestCase): + """参数化提取。""" + + def test_ua_params(self): + driver = HollysysDcsDriver(UA_CONFIG) + self.assertEqual(driver.mode, "ua") + ua = OpcUaDriver(UA_CONFIG) + self.assertEqual(ua.endpoint, "opc.tcp://10.20.1.10:4840") + self.assertEqual(ua.security, "Basic256Sha256") + self.assertEqual(ua.username, "iaop_read") + self.assertEqual(ua.password, "secret") + + def test_da_params(self): + driver = HollysysDcsDriver(DA_CONFIG) + self.assertEqual(driver.mode, "da") + da = _HollysysDaDriver(DA_CONFIG) + self.assertEqual(da.host, "10.20.1.10") + self.assertEqual(da.prog_id, "Hollysys.OPCServer") + self.assertEqual(_HollysysDaDriver({}).prog_id, + "Hollysys.OPCServer") # 默认 ProgID + + def test_default_mode_is_ua(self): + driver = HollysysDcsDriver({}) + self.assertEqual(driver.mode, "ua") + + +class TestConnectDependencyErrors(unittest.TestCase): + """底层库未安装时的明确报错（不静默失败）。""" + + def test_ua_missing_dependency(self): + driver = HollysysDcsDriver(UA_CONFIG) + with mock.patch.dict("sys.modules", {"opcua": None}): + with self.assertRaises(ConnectionError) as ctx: + driver.connect() + self.assertIn("opcua", str(ctx.exception)) + + def test_da_missing_dependency(self): + driver = HollysysDcsDriver(DA_CONFIG) + with mock.patch.dict("sys.modules", {"OpenOPC": None}): + with self.assertRaises(ConnectionError) as ctx: + driver.connect() + self.assertIn("OpenOPC", str(ctx.exception)) + + +class TestDelegation(unittest.TestCase): + """connect 后 read_points/close 委托到底层实现。""" + + def test_ua_delegates_to_opcua_driver(self): + driver = HollysysDcsDriver(UA_CONFIG) + with mock.patch("drivers.opcua_driver.OpcUaDriver.connect"): + driver.connect() + self.assertIsInstance(driver._delegate, OpcUaDriver) + point = make_point() + with mock.patch.object(driver._delegate, "read_points", + return_value={"CLF-01.TEMP": 850.5}) as rd: + values = driver.read_points([point]) + rd.assert_called_once_with([point]) + self.assertEqual(values["CLF-01.TEMP"], 850.5) + with mock.patch.object(driver._delegate, "close"): + driver.close() + self.assertIsNone(driver._delegate) + + def test_da_delegates_to_da_driver(self): + driver = HollysysDcsDriver(DA_CONFIG) + fake_openopc = mock.MagicMock() + with mock.patch.dict("sys.modules", {"OpenOPC": fake_openopc}): + with mock.patch.object(_HollysysDaDriver, "connect"): + driver.connect() + self.assertIsInstance(driver._delegate, _HollysysDaDriver) + + def test_read_before_connect_raises(self): + driver = HollysysDcsDriver(UA_CONFIG) + with self.assertRaises(ConnectionError): + driver.read_points([make_point()]) + + +if __name__ == "__main__": + unittest.main()