OpenFactory Devices Schemas

Pydantic schemas for validating OpenFactory OpenFactory Adapter, Agent, Supervisor and Device definitions.

class openfactory.schemas.devices.Adapter(**data)

Bases: BaseModel

OpenFactory Adapter Schema.

deploy: Optional[Deploy]

environment: List[str]

image: str

ip: str

model_config: ClassVar[ConfigDict] = {}

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

port: int

classmethod validate_adapter(values)

Validates the adapter configuration.

Return type:

Dict

Parameters:

values (Dict) – Dictionary of values to validate.

Returns:

Dict – Validated values.

Raises:

ValueError – If ‘ip’ or ‘image’ is missing or incorrectly defined.

class openfactory.schemas.devices.Agent(**data)

Bases: BaseModel

OpenFactory Agent Schema.

adapter: Optional[Adapter]

deploy: Optional[Deploy]

device_xml: str

ip: str

model_config: ClassVar[ConfigDict] = {}

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

port: int

classmethod validate_agent(values)

Validates the agent configuration.

Return type:

Dict

Parameters:

values (Dict) – Dictionary of values to validate.

Returns:

Dict – Validated values.

Raises:

ValueError – If ‘device_xml’ or ‘adapter’ is missing or incorrectly defined.

class openfactory.schemas.devices.Deploy(**data)

Bases: BaseModel

Deploy Schema.

model_config: ClassVar[ConfigDict] = {}

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

placement: Placement

replicas: Optional[int]

resources: Resources

class openfactory.schemas.devices.Device(**data)

Bases: BaseModel

OpenFactory Device Schema.

__init__(**data)

Initialize the Device model.

Parameters:

**data (Dict) – Keyword arguments to initialize the model.

agent: Agent

influxdb: Optional[InfluxDB]

ksql_tables: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {'extra': 'ignore'}

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

supervisor: Optional[Supervisor]

uuid: str

classmethod validate_ksql_tables(value)

Validates the ksql_tables field.

Return type:

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)

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

Dictionary with all configured devices.

model_config: ClassVar[ConfigDict] = {}

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

validate_devices()

Validates the devices configuration.

Return type:

None

Raises:

ValueError – If the devices configuration is invalid.

class openfactory.schemas.devices.InfluxDB(**kwargs)

Bases: BaseModel

InfluxDB configuration.

__init__(**kwargs)

Initialize the InfluxDB model.

Parameters:

**kwargs (Dict) – Keyword arguments to initialize the model.

Raises:

ValueError – If ‘url’ or ‘token’ is not provided and not defined in the configuration.

bucket: str

model_config: ClassVar[ConfigDict] = {}

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

organisation: str

token: str

url: str

class openfactory.schemas.devices.Placement(**data)

Bases: BaseModel

Placement Schema.

constraints: List[str]

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.devices.Resources(**data)

Bases: BaseModel

Resources Schema.

limits: ResourcesDefinition

model_config: ClassVar[ConfigDict] = {}

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

reservations: ResourcesDefinition

class openfactory.schemas.devices.ResourcesDefinition(**data)

Bases: BaseModel

Resources Definition Schema.

cpus: float

memory: str

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.devices.Supervisor(**data)

Bases: BaseModel

OpenFactory Supervisor Schema.

adapter: Adapter

deploy: Optional[Deploy]

image: str

model_config: ClassVar[ConfigDict] = {}

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

openfactory.schemas.devices.get_devices_from_config_file(devices_yaml_config_file, uns_schema)

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

This function reads a YAML file containing device definitions, validates its content using the DevicesConfig Pydantic model, and augments each validated device entry with Unified Namespace (UNS) metadata derived from the provided schema.

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.