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).
OPCUAConstantConfig:

Configuration for a constant OPC UA value. Contains node_id or browse_path to identify the OPC UA node and a tag used by OpenFactory. Constants are read once at deployment time and are not subscribed.
OPCUAVariableConfig:

Configuration for a single variable. Contains node_id or browse_path to identify the OPC UA node and a tag used by OpenFactory to label data. Variables may also declare an access_level (ro or rw) describing how OpenFactory is allowed to interact with the node. Optional overrides for queue size and sampling interval can be provided.
OPCUAEventConfig:

Configuration for an event source. Contains node_id or browse_path identifying the OPC UA node that emits events. Events do not use tags or subscription overrides.
OPCUAMethodConfig:

Configuration for a method source. Each entry references an OPC UA node (node_id or browse_path) from which all available OPC UA methods will be discovered at runtime. Methods are not individually declared in the schema; instead, the node acts as a method container.
OPCUAConnectorSchema:

Wrapper schema that encapsulates the server configuration along with constants, variables, events, and method sources. During initialization:
- variables are normalized into OPCUAVariableConfig with server-level defaults
- constants, events, and methods are validated and normalized
- cross-section conflicts are checked

Validation Features:#

Validates node_id format and parses it into namespace_index, identifier_type, and identifier fields.
Enforces that exactly one of node_id or browse_path is provided for constants, variables, events, and methods.
Normalizes variables into OPCUAVariableConfig, applying server-level subscription defaults when not overridden.
Ensures uniqueness of node_id and browse_path within each section (constants, variables, events, methods).
Prevents local name conflicts across sections (constants, variables, events, methods).
Allows reuse of the same OPC UA node across sections (e.g. a variable and an event can reference the same node).
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/

constants:
    device_model:
        node_id: ns=2;i=1001
        tag: Model

variables:
    temp:
        node_id: ns=3;i=1050
        tag: Temperature
        deadband: 0.1
    hum:
        browse_path: 0:Root/0:Objects/2:Sensors/2:Humidity
        tag: Humidity

events:
    iolinkmaster:
        node_id: ns=6;i=43

methods:
    tempSensor:
        browse_path: 0:Root/0:Objects/2:Sensors/2:TemperatureSensor
    humiSensor:
        node_id: ns=5;i=12

# ---------------------------------------------------------
# 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

constants:
    serial:
        browse_path: 0:Root/0:Objects/2:Device/2:SerialNumber
        tag: SerialNumber

variables:
    temp:
        node_id: ns=3;i=1050
        tag: Temperature
        queue_size: 5          # overrides server subscription
        sampling_interval: 50  # overrides server subscription
    setpoint:
        node_id: ns=2;i=10
        tag: Setpoint
        access_level: rw       # read-write variable

See also

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

constants: Dict[str, OPCUAConstantConfig] | None#

events: Dict[str, OPCUAEventConfig] | None#

methods: Dict[str, OPCUAMethodConfig] | None#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_OPCUAConnectorSchema__context)[source]#

Normalize configuration sections and enforce uniqueness constraints.

Return type:: None

Ensures:

Unique local names within each section.
No cross-section name conflicts.
No duplicate node identifiers (node_id or browse_path).
Variables inherit server-level subscription defaults.

Parameters:: __context – Pydantic initialization context.
Raises:: ValueError – If uniqueness or validation constraints are violated.

server: OPCUAServerConfig#

type: Literal['opcua']#

variables: Dict[str, OPCUAVariableConfig] | None#

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

Bases: OPCUANodeConfig

Configuration for a constant OPC UA value (read once at deployment).

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tag: str#

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

Bases: OPCUANodeConfig

Configuration for an OPC UA event source.

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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

Bases: OPCUANodeConfig

Configuration for an OPC UA method source.

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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

Bases: BaseModel

Base configuration for OPC UA nodes identified by either a node_id or a browse_path.

Exactly one of node_id or browse_path must be provided.

browse_path: str | None#

identifier: str | None#

identifier_type: str | None#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_dump(*args, **kwargs)[source]#

Serialize the model while excluding fields with None values.

Return type:

Dict[str, Any]

Parameters:

*args – Positional arguments forwarded to BaseModel.model_dump.
**kwargs – Keyword arguments forwarded to BaseModel.model_dump.

Returns:

Serialized model dictionary with None fields excluded.

namespace_index: int | None#

node_id: str | None#

classmethod validate_and_parse_node_id(values)[source]#

Validate and parse node_id into structured components.

Return type:: Dict[str, Any]

Extracts:

namespace_index
identifier_type
identifier

Parameters:: values – Raw input values.
Returns:: Updated values including parsed fields.
Raises:: ValueError – If node_id format is invalid.

classmethod validate_browse_path_format(values)[source]#

Validate the format of browse_path.

Return type:: Dict[str, Any]

Ensures:

Path starts with "0:Root".
Each segment follows ns_index:identifier format.
No leading/trailing whitespace.
Namespace index is numeric.
Identifier is non-empty.

Parameters:: values – Raw input values.
Returns:: Validated values.
Raises:: ValueError – If the browse path format is invalid.

classmethod validate_node_id_or_browse_path(values)[source]#

Ensure exactly one of node_id or browse_path is provided.

Return type:: Dict[str, Any]
Parameters:: values – Raw input values.
Returns:: Validated values.
Raises:: ValueError – If neither or both fields are provided.

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

Bases: BaseModel

OPC UA Server configuration with variables.

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

subscription: OPCUASubscriptionConfig | None#

uri: str#

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

Bases: BaseModel

Optional subscription parameters for server or individual variables.

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

publishing_interval: float | None#

queue_size: int | None#

sampling_interval: float | None#

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

Bases: OPCUANodeConfig

Configuration for a single OPC UA variable.

access_level: Literal['ro', 'rw']#

deadband: float#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

queue_size: int | None#

sampling_interval: float | None#

tag: str#