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 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 variables, events, and method sources. During initialization, all variables, events, and methods are normalized into their respective config models. Server-level subscription defaults are applied to variables where necessary.

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 variables, events, and methods.

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

  • Ensures uniqueness of node_id and browse_path within variables, within events, and within methods.

  • Prevents local name conflicts between variables, events, and methods.

  • 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:
        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

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.

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: Dict[str, OPCUAEventConfig] | None#
methods: Dict[str, OPCUAMethodConfig] | None#
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: Dict[str, OPCUAVariableConfig] | None#
class openfactory.schemas.connectors.opcua.OPCUAEventConfig(**data)[source]#

Bases: OPCUANodeConfig

Configuration for an OPC UA event source.

model_config: ClassVar[ConfigDict] = {'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: ClassVar[ConfigDict] = {'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 class for configs that can have node_id or path to identify an OPC UA node.

browse_path: str | None#
identifier: str | None#
identifier_type: str | None#
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}#

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

model_dump(*args, **kwargs)[source]#
!!! abstract “Usage Documentation”

[model_dump](../concepts/serialization.md#python-mode)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

  • mode – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.

  • include – A set of fields to include in the output.

  • exclude – A set of fields to exclude from the output.

  • context – Additional context to pass to the serializer.

  • by_alias – Whether to use the field’s alias in the dictionary key if defined.

  • exclude_unset – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults – Whether to exclude fields that are set to their default value.

  • exclude_none – Whether to exclude fields that have a value of None.

  • exclude_computed_fields – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.

  • round_trip – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any – Whether to serialize fields with duck-typing serialization behavior.

Returns:

A dictionary representation of the model.

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

dict

classmethod validate_browse_path_format(values)[source]#
Return type:

dict

classmethod validate_node_id_or_browse_path(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: OPCUASubscriptionConfig | None#
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: 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: ClassVar[ConfigDict] = {'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#