114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
|
# -*- 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
|