跳到主要内容

构造函数

OpcUaSession 单一构造函数, 全部参数关键字传递, 默认值覆盖了大部分常用场景.

前置阅读

签名​

class OpcUaSession:
    def __init__(self,
                 endpoint_url: str,
                 security_mode: MessageSecurityMode = MessageSecurityMode.NONE,
                 username: Optional[str] = None,
                 password: Optional[str] = None,
                 session_timeout_ms: int = 600_000,
                 request_timeout_ms: int = 10_000,
                 connect_timeout_ms: int = 10_000,
                 keepalive_interval_ms: int = 10_000,
                 auto_reconnect: bool = True,
                 auto_publish_on_connect: bool = True,
                 client_cert_path: Optional[str] = None,
                 client_key_path: Optional[str] = None,
                 server_cert_path: Optional[str] = None,
                 security_policy_uri: Optional[str] = None):
        ...

参数​

类别属性类型访问说明
必填endpoint_urlstrinOPC UA endpoint, 形如 opc.tcp://host:port
安全security_modeMessageSecurityModeinNONE / Sign / SignAndEncrypt, 默认 NONE (Python 用大写 NONE 避开关键字 None)
security_policy_uriOptional[str]in不传按 mode 自动选 (NONE → #None, 其他 → #Basic256Sha256)
client_cert_pathOptional[str]in客户端 PFX 路径 (Sign / SignAndEncrypt 必填), 内含私钥
client_key_pathOptional[str]in私钥路径 (当前从 PFX 自带, 此参数保留以对齐 Stack)
server_cert_pathOptional[str]in服务端 DER, 不传则 TOFU (Trust On First Use)
登录usernameOptional[str]inNone = Anonymous, 非空 = UsernameIdentityToken
passwordOptional[str]in配 username 用
超时 (毫秒)session_timeout_msintin服务端 Session 超时, 默认 600 000 ms (10 分钟)
request_timeout_msintin单 RPC 超时, 默认 10 000 ms (10 秒)
connect_timeout_msintin建链超时, 默认 10 000 ms
keepalive_interval_msintinKeepAlive 心跳周期, 默认 10 000 ms (0 = 禁用)
行为auto_reconnectboolin断线后自动重连, 默认 True
auto_publish_on_connectboolinconnect() 成功后自动启 Publish 后台线程, 默认 True

例子​

1. 匿名 + 明文 (开发期)​

from darra_opcua import OpcUaSession

with OpcUaSession("opc.tcp://localhost:4840") as ua:
    ua.connect()
    # ...

2. 匿名 + Sign 加密​

from darra_opcua import OpcUaSession, MessageSecurityMode

with OpcUaSession(
        "opc.tcp://server:4840",
        security_mode=MessageSecurityMode.Sign,
        client_cert_path=r"C:\certs\client.pfx") as ua:
    ua.connect()

3. 用户名 + SignAndEncrypt (典型生产)​

with OpcUaSession(
        "opc.tcp://server:4840",
        security_mode=MessageSecurityMode.SignAndEncrypt,
        username="operator",
        password="secret",
        client_cert_path=r"C:\certs\client.pfx") as ua:
    ua.connect()

4. 按 GetEndpoints 结果工厂构造​

from darra_opcua import OpcUaSession, OpcUaDiscovery, MessageSecurityMode

eps = OpcUaDiscovery.get_endpoints("opc.tcp://server:4840")
ep = next(e for e in eps if e.security_mode == MessageSecurityMode.SignAndEncrypt)

with OpcUaSession(
        endpoint_url=ep.endpoint_url,
        security_mode=ep.security_mode,
        security_policy_uri=ep.security_policy_uri,
        username="operator",
        password="secret",
        client_cert_path=r"C:\certs\client.pfx") as ua:
    ua.connect()

异常​

异常原因
ValueError("endpoint_url is required")endpoint_url 为空
OpcUaException(BadConfigurationError)Stack 初始化失败
OpcUaException(BadCertificateInvalid)客户端证书加载失败

OpcUaException.status_code 是 StatusCode 枚举:

from darra_opcua import OpcUaException, StatusCode

try:
    ua = OpcUaSession("opc.tcp://bad-host:4840", connect_timeout_ms=1000)
    ua.connect()
except OpcUaException as ex:
    if ex.status_code == StatusCode.BadCommunicationError:
        print("Network down")
    else:
        print(f"Connect failed: {ex}")

资源管理​

OpcUaSession 实现 context manager, 内部维护引用计数的 native Stack 初始化. 强烈推荐用 with:

# 推荐: with 自动 close
with OpcUaSession(...) as ua:
    ua.connect()
    # ...

# 或显式
ua = OpcUaSession(...)
try:
    ua.connect()
    # ...
finally:
    ua.close()

close() (= dispose()) 内部自动调用 disconnect(), 一般 with 即可.

下一步​