OpenFactory Devices Schemas#

OpenFactory Device Schemas

This module defines Pydantic models and utility functions to parse and validate device configuration files in OpenFactory. Device definitions include connection details, UNS metadata, ksql table mappings, and supervisor configurations.

Key Components:#

Device: Defines a single device including its UUID, connector, optional supervisor, UNS metadata, and supported Kafka ksqlDB tables.
DevicesConfig: Validates a dictionary of device entries and ensures UUID uniqueness.
get_devices_from_config_file: Loads, validates, and enriches devices from a YAML file.

Features:#

Supports UNS (Unified Namespace) enrichment through the AttachUNSMixin.
Restricts configuration fields with extra=”forbid” to ensure strict schema conformance.
Includes validation logic to ensure all device UUIDs are unique.
Enforces allowed values for ksql_tables.

YAML Example:#

devices:
  press-1:
    uuid: "press-001"
    uns:
      location: building-a
      workcenter: press
    connector:
      type: mtconnect
      ip: 192.168.1.100
    supervisor:
      image: ghcr.io/openfactoryio/opcua-supervisor:v4.0.1
      adapter:
        ip: 192.168.0.201
        port: 4840
        environment:
        - NAMESPACE_URI=openfactory
        - BROWSE_NAME=PRESS

Usage:#

Use get_devices_from_config_file(path, uns_schema) to load and validate a device configuration YAML file, with automatic UNS enrichment.

This module is used by OpenFactory agents and deployment tools for runtime configuration.

class openfactory.schemas.devices.Device(**data)[source]#

Bases: AttachUNSMixin, BaseModel

OpenFactory Device Schema.

connector: Annotated[MTConnectConnectorSchema | OPCUAConnectorSchema]#

ksql_tables: Optional[List[str]]#

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

supervisor: Optional[Supervisor]#

uns: Optional[Dict[str, Any]]#

uuid: str#

classmethod validate_ksql_tables(value)[source]#

Validates the ksql_tables field.

Return type:: Optional[List[str]]
Parameters:: value (List[str]) – List of ksql tables.
Returns:: List[str] – Validated list of ksql tables.
Raises:: ValueError – If the provided ksql tables contain invalid entries.

class openfactory.schemas.devices.DevicesConfig(**data)[source]#

Bases: BaseModel

Schema for OpenFactory device configurations loaded from YAML files.

This schema validates the structure of device configurations.

Example usage:

devices = DevicesConfig(devices=yaml_data['devices'])
# or
devices = DevicesConfig(**yaml_data)

Parameters:: devices (dict) – Dictionary of device configurations.
Raises:: pydantic.ValidationError – If the input data does not conform to the expected schema.

devices: Dict[str, Device]#

property devices_dict: Dict[str, Any]#: Return plain dict of devices suitable for serialization.

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

validate_devices(uns_schema)[source]#

Validates the devices configuration at the collection level.

Checks that all device UUIDs are unique and attaches UNS metadata using the provided UNS schema.

Return type:: None
Parameters:: uns_schema (UNSSchema) – Schema instance used to extract and validate UNS metadata for each device.
Raises:: ValueError – If the devices configuration is invalid or UUID are not unique.

openfactory.schemas.devices.get_devices_from_config_file(devices_yaml_config_file, uns_schema)[source]#

Load, validate, and enrich device configurations from a YAML file using UNS metadata.

Return type:

Optional[Dict[str, Device]]

Parameters:

devices_yaml_config_file (str) – Path to the YAML file defining device configurations.
uns_schema (UNSSchema) – Schema instance used to extract and validate UNS metadata for each device.

Returns:

Optional[Dict[str, Device]] – A dictionary of validated and enriched device configurations, or None if validation fails.

Note

In case of validation errors, user notifications will be triggered and None will be returned.