OpenFactory Application Framework#

This module provides the core building blocks for implementing an OpenFactory application.

It combines three main concepts:

  1. Declarative attributes (via AttributeField and subclasses)

  2. Callable methods (via @ofa_method decorator)

  3. Runtime integration with OpenFactory infrastructure (Kafka, ksqlDB, Asset model)

Core Concepts#

An OpenFactoryApp is an Asset in OpenFactory.

As such, it exposes:

  • Attributes — representing state, telemetry, or metadata of the application

  • Methods — representing actions that can be triggered by other Assets

This module provides a declarative and structured way to define both.

Declarative Attributes#

Attributes are defined as class-level fields using subclasses of AttributeField (e.g., EventAttribute, SampleAttribute).

These are automatically collected by the OpenFactoryAppMeta metaclass and registered as OpenFactory AssetAttribute instances during initialization.

Declarative Attributes Usage Example

class MyApp(OpenFactoryApp):
   status = EventAttribute(value="idle", tag="App.Status")
   temperature = SampleAttribute(tag="Sensor.Temperature")

These become runtime attributes of the OpenFactory asset.

Callable Methods#

Instance methods can be exposed as remotely callable OpenFactory commands using the @ofa_method decorator.

The decorator:

  • Validates the method signature

  • Extracts parameter metadata

  • Registers a command interface automatically

Callable Methods Usage Example

class MyApp(OpenFactoryApp):

   @ofa_method(description="Move axis to position")
   def move(self, x: float, y: float):
         self.logger.info(f"Moving to ({x}, {y})")

At runtime:

  • A command attribute move is created

  • A corresponding move_CMD attribute is subscribed to

  • Incoming messages trigger method execution

Application Lifecycle#

An OpenFactory app is a subclass of OpenFactoryApp.

Typical lifecycle:

  1. Instantiate the app with Kafka + ksqlDB configuration

  2. Attributes and methods are automatically registered

  3. Call OpenFactoryApp.run or OpenFactoryApp.async_run

  4. Implement OpenFactoryApp.main_loop (or OpenFactoryApp.async_main_loop)

Running an OpenFactoryApp

app = MyApp(
   ksqlClient=KSQLDBClient("http://localhost:8088"),
   bootstrap_servers="localhost:9092"
)
app.run()

What Happens Automatically#

When an app is instantiated:

  • All declarative attributes are converted into AssetAttribute objects

  • All @ofa_method methods are:

    • Registered as callable OpenFactory methods

    • Subscribed to their corresponding *_CMD attributes

  • Logging is configured with the app UUID

  • Storage backend is mounted (if configured via STORAGE env)

  • Signal handlers are installed for graceful shutdown

Minimal Example

import os
from openfactory.apps import OpenFactoryApp, EventAttribute, ofa_method

class DemoApp(OpenFactoryApp):

   status = EventAttribute(value="idle", tag="App.Status")

   @ofa_method()
   def ping(self):
         """Simple health check."""
         self.logger.info("pong")

   def main_loop(self):
         self.status = 'Running'
         while True:
            self.logger.info("Running...")
            time.sleep(5)

app = DemoApp(
   ksqlClient=KSQLDBClient(os.getenv("KSQLDB_URL", "http://localhost:8088")),
   bootstrap_servers=os.getenv("KAFKA_BROKER", "localhost:9092")
)
app.run()

Note

  • The application UUID is injected via the APP_UUID environment variable during deployment.

  • The environment variables KSQLDB_URL and KAFKA_BROKER are automatically provided in cluster deployments. The default values can be used for local development.

  • Either main_loop or async_main_loop must be implemented by subclasses.

  • Attribute and method registration is fully declarative and happens at class creation time.

See Also#