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

  • 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).

  • OPCUAServerConfig: Configuration for the OPC UA server, including the endpoint URI, namespace URI, and optional default subscription parameters. If subscription is omitted, default subscription values are used.

  • OPCUAVariableConfig: Configuration for a single variable. Contains a browse_name plus optional overrides for queue size and sampling interval. If overrides are not provided, server-level defaults (or the automatic defaults if server subscription is omitted) are applied.

  • OPCUADeviceConfig: Configuration for a device on the OPC UA server. Devices may be specified either by a hierarchical path or a node_id. Variables can be defined as simple strings (which are normalized to OPCUAVariableConfig using server defaults) or as full OPCUAVariableConfig objects with overrides. Methods are mapped by local name to OPC UA BrowseNames. If node_id is given, its namespace index, identifier type, and identifier are parsed out.

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

Validation Features:#

  • Ensures exactly one of path or node_id is provided for a device.

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

  • Variables and methods are optional, providing flexibility for different server setups.

  • Forbids unknown fields to ensure strict schema conformance.

YAML Example:#

type: opcua

# ---------------------------------------------------------
# Example 1: Server subscription omitted → defaults applied
# ---------------------------------------------------------
server:
  uri: opc.tcp://127.0.0.1:4840/freeopcua/server/
  namespace_uri: http://examples.openfactory.local/opcua
  # subscription omitted → defaults will be used:
  #   publishing_interval: 100
  #   queue_size: 1
  #   sampling_interval: 0

device:
  path: Sensors/TemperatureSensor_1
  variables:
    temp: Temperature         # simple string → inherits server defaults
    hum:                      # explicit variable config → overrides defaults
      browse_name: Humidity
      queue_size: 5
      sampling_interval: 50
  methods:
    calibrate: Calibrate

# ---------------------------------------------------------
# Example 2: Server subscription explicitly provided
# ---------------------------------------------------------
server:
  uri: opc.tcp://127.0.0.1:4840/freeopcua/server/
  namespace_uri: http://examples.openfactory.local/opcua
  subscription:
    publishing_interval: 200
    queue_size: 10
    sampling_interval: 25

device:
  path: Sensors/TemperatureSensor_2
  variables:
    temp: Temperature          # inherits server subscription values
    hum:
      browse_name: Humidity    # overrides server subscription values
      queue_size: 5
      sampling_interval: 50
  methods:
    calibrate: Calibrate

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 and device configuration.

During initialization, all device 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.

device: OPCUADeviceConfig#
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 device variables to OPCUAVariableConfig with defaults applied.

Return type:

None

server: OPCUAServerConfig#
type: Literal['opcua']#
class openfactory.schemas.connectors.opcua.OPCUADeviceConfig(**data)[source]#

Bases: BaseModel

OPC UA Device configuration.

identifier: Optional[str]#
identifier_type: Optional[str]#
methods: Optional[Dict[str, 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: Optional[str]#
path: Optional[str]#
classmethod validate_and_parse_node_id(values)[source]#

Ensure that exactly one of ‘path’ or ‘node_id’ is provided. If node_id is provided, validate its format and parse namespace_index, identifier_type, and identifier.

Return type:

dict

variables: Optional[Dict[str, Union[str, OPCUAVariableConfig]]]#
class openfactory.schemas.connectors.opcua.OPCUAServerConfig(**data)[source]#

Bases: BaseModel

OPC UA Server configuration.

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

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

namespace_uri: str#
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.

All device variables are normalized into this model during initialization. If the YAML contains only a string (BrowseName), it is automatically expanded into an OPCUAVariableConfig using server-level subscription defaults. If overrides are provided here, they take precedence over server defaults.

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

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

queue_size: Optional[int]#
sampling_interval: Optional[float]#