OPC UA Connector Schemas#

OPC UA Connector Schemas

This module provides Pydantic models to define and validate configuration schemas for OPC UA devices within OpenFactory.

Key Models:#

  • OPCUAServerConfig: Configuration for the OPC UA server, including the endpoint URI and optional default subscription parameters.

  • OPCUASubscriptionConfig: Optional subscription parameters (publishing interval, queue size, sampling interval). Can be defined at the server level or overridden per variable. If not provided at the server level, default values are applied (publishing_interval=100 ms, queue_size=1, sampling_interval=0 ms).

  • OPCUAVariableConfig: Configuration for a single variable. Contains node_id to identify the OPC UA node and a tag used by OpenFactory to label data. Optional overrides for queue size and sampling interval can be provided.

  • OPCUAConnectorSchema: Wrapper schema that encapsulates the server configuration. During initialization, all variables are normalized into OPCUAVariableConfig instances, with server-level subscription defaults applied where necessary.

Validation Features:#

  • Validates node_id format and parses it into namespace_index, identifier_type, and identifier fields.

  • Normalizes all variables into OPCUAVariableConfig, applying server-level subscription defaults when not overridden. If the server subscription is omitted, default values are applied.

  • Forbids unknown fields to ensure strict schema conformance.

YAML Example:#

# ---------------------------------------------------------
# Example 1: Server subscription omitted → defaults applied
# ---------------------------------------------------------

type: opcua

server:
    uri: opc.tcp://127.0.0.1:4840/freeopcua/server/

    # subscription omitted → defaults will be used:
    #   publishing_interval: 100
    #   queue_size: 1
    #   sampling_interval: 0

variables:
    temp:
        node_id: ns=3;i=1050
        tag: Temperature
        deadband: 0.1
    hum:
        node_id: ns=2;i=10
        tag: Humidity

events:
    iolinkmaster: ns=6;i=43

# ---------------------------------------------------------
# Example 2: Server subscription explicitly provided
# ---------------------------------------------------------

type: opcua

server:
    uri: opc.tcp://127.0.0.1:4840/freeopcua/server/

    subscription:
        publishing_interval: 200
        queue_size: 10
        sampling_interval: 25

variables:
    temp:
        node_id: ns=3;i=1050
        tag: Temperature
    hum:
        node_id: ns=2;i=10
        tag: Humidity
        queue_size: 5          # overrides server subscription
        sampling_interval: 50  # overrides server subscription

See also

The runtime class of the OPCUAConnectorSchema schema is openfactory.connectors.opcua.opcua_connector.OPCUAConnector.

class openfactory.schemas.connectors.opcua.OPCUAConnectorSchema(**data)[source]#

Bases: BaseModel

OPC UA Connector schema wrapping the server configuration.

During initialization, all variables are normalized into OPCUAVariableConfig instances, inheriting server-level subscription defaults where no overrides are given.

The type field is a discriminator for Pydantic to select this schema.

See also

The runtime class of the class OPCUAConnectorSchema schema is openfactory.connectors.opcua.opcua_connector.OPCUAConnector.

events: Optional[Dict[str, OPCUAEventConfig]]#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_OPCUAConnectorSchema__context)[source]#

Normalize all server variables with subscription defaults and enforce unique keys.

Return type:

None

server: OPCUAServerConfig#
type: Literal['opcua']#
variables: Optional[Dict[str, OPCUAVariableConfig]]#
class openfactory.schemas.connectors.opcua.OPCUAEventConfig(**data)[source]#

Bases: BaseModel

Configuration for an OPC UA event source.

identifier: Optional[str]#
identifier_type: Optional[str]#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

namespace_index: Optional[int]#
node_id: str#
classmethod validate_and_parse_node_id(values)[source]#
Return type:

dict

class openfactory.schemas.connectors.opcua.OPCUAServerConfig(**data)[source]#

Bases: BaseModel

OPC UA Server configuration with variables.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

subscription: Optional[OPCUASubscriptionConfig]#
uri: str#
class openfactory.schemas.connectors.opcua.OPCUASubscriptionConfig(**data)[source]#

Bases: BaseModel

Optional subscription parameters for server or individual variables.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

publishing_interval: Optional[float]#
queue_size: Optional[int]#
sampling_interval: Optional[float]#
class openfactory.schemas.connectors.opcua.OPCUAVariableConfig(**data)[source]#

Bases: BaseModel

Configuration for a single OPC UA variable.

deadband: float#
identifier: Optional[str]#
identifier_type: Optional[str]#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

namespace_index: Optional[int]#
node_id: str#
queue_size: Optional[int]#
sampling_interval: Optional[float]#
tag: str#
classmethod validate_and_parse_node_id(values)[source]#

Validate node_id format and parse it into namespace_index, identifier_type, and identifier.

Return type:

dict