AssetUNS#

class openfactory.assets.AssetUNS(uns_id, ksqlClient, bootstrap_servers=None)[source]#

Bases: BaseAsset

Represents an OpenFactory Asset using the UNS identifier.

This class encapsulates Asset metadata and a Kafka producer responsible for sending Asset data. It uses the ksqlDB topology based on the ASSETS_STREAM_UNS stream to handle Asset data.

Note

All write operations to the asset take place in the assets stream.

KSQL_ASSET_TABLE#

Name of ksqlDB table of Asset states (assets_uns)

Type:

str

KSQL_ASSET_ID#

ksqlDB ID used to identify the Asset (uns_id) in the KSQL_ASSET_TABLE

Type:

str

ASSET_ID#

value of the identifer of the Asset (uns_id) used in the KSQL_ASSET_TABLE

Type:

str

ksql#

Client for interacting with ksqlDB.

Type:

KSQLDBClient

bootstrap_servers#

Kafka bootstrap server address.

Type:

str

ASSET_CONSUMER_CLASS#

Kafka consumer class for reading messages from Asset strean.

Type:

KafkaAssetUNSConsumer

producer#

Kafka producer instance for sending Asset messages.

Type:

AssetProducer

Example usage:
import time
from openfactory.assets import AssetUNS
from openfactory.kafka import KSQLDBClient

ksql = KSQLDBClient('http://localhost:8088')
cnc = AssetUNS('cnc-003', ksqlClient=ksql)

# list samples
print(cnc.samples())
print(cnc.Zact.value)
print(cnc.Zact.type)
print(cnc.Zact.timestamp)

# redefine an attribute value
cnc.Zact = 10.0
print(cnc.Zact.value)

# callbacks for subscriptions
def on_messages(msg_key, msg_value):
    print(f"[Message] [{msg_key}] {msg_value}")

def on_sample(msg_key, msg_value):
    print(f"[Sample] [{msg_key}] {msg_value}")

def on_event(msg_key, msg_value):
    print(f"[Event] [{msg_key}] {msg_value}")

def on_condition(msg_key, msg_value):
    print(f"[Condition] [{msg_key}] {msg_value}")

cnc.subscribe_to_messages(on_messages, 'demo_messages_group')
cnc.subscribe_to_samples(on_sample, 'demo_samples_group')
cnc.subscribe_to_events(on_event, 'demo_events_group')
cnc.subscribe_to_conditions(on_condition, 'demo_conditions_group')

# run a main loop while subscriptions remain active
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("Stopping consumer threads ...")
    cnc.stop_messages_subscription()
    cnc.stop_samples_subscription()
    cnc.stop_events_subscription()
    cnc.stop_conditions_subscription()
    print("Consumers stopped")
finally:
    ksql.close()
__init__(uns_id, ksqlClient, bootstrap_servers=None)[source]#

Initializes the Asset with metadata and a Kafka producer.

Parameters:

  • uns_id (str) – UNS identifier of the asset.

  • ksqlClient (KSQLDBClient) – Client for interacting with ksqlDB.

  • bootstrap_servers (str) – Kafka bootstrap server address. Defaults to config setting.

add_attribute(asset_attribute)#

Adds a new attribute to the asset.

Return type:

None

Parameters:

asset_attribute (AssetAttribute) – The attribute to be added.

add_reference_above(above_asset_reference)#

Adds a reference to an asset above the current asset.

Return type:

None

Parameters:

above_asset_reference (str) – The asset-reference of the asset above the current one to be added.

add_reference_below(below_asset_reference)#

Adds a reference to an asset below the current asset.

Return type:

None

Parameters:

below_asset_reference (str) – The asset-reference of the asset below the current one to be added.

property asset_uuid: str#

Returns the asset UUID based on runtime state.

Returns:

str – The asset’s UUID.

attributes()#

Returns all non-‘Method’ attribute IDs associated with this asset.

Queries KSQL_ASSET_TABLE for all attribute IDs of the asset where the type is not ‘Method’.

Return type:

List[str]

Returns:

List[str] – A list of attribute IDs.

conditions()#

Returns all condition-type attributes for this asset.

Return type:

List[Dict[str, Any]]

Returns:

List[Dict] –

A list of dictionaries, each containing:

  • ”ID” (str): The attribute ID.

  • ”VALUE” (Any): The value of the condition.

  • ”TAG” (str): The condition tag (‘Normal’, ‘Warning’, ‘Fault’)

events()#

Returns all event-type attributes for this asset.

Return type:

List[Dict[str, Any]]

Returns:

List[Dict] –

A list of dictionaries, each containing:

  • ”ID” (str): The attribute ID.

  • ”VALUE” (Any): The value of the event.

  • ”TAG” (str): The cleaned tag name with placeholders removed.

method(method, args='')#

Requests the execution of a method for the asset by sending a command to the Kafka stream.

Constructs a message with the provided method name and optional arguments, and sends it to the CMDS_STREAM Kafka topic for processing.

Return type:

None

Parameters:

  • method (str) – The name of the method to be executed.

  • args (str) – Arguments for the method, if any. Defaults to an empty string.

methods()#

Returns method-type attributes for this asset.

Queries KSQL_ASSET_TABLE for entries where TYPE = ‘Method’ for the asset.

Return type:

Dict[str, Any]

Returns:

Dict – A dictionary where keys are method attribute IDs and values are the corresponding method values.

references_above()#

Retrieves a list of assets above the current asset.

Return type:

List[Self]

Returns:

List[Self] – A list of asset objects that are above the current asset.

references_above_uuid()#

Retrieves a list of asset-references of assets above the current asset.

Return type:

List[str]

Returns:

List[str] – A list of asset-references (as strings) that are above the current asset.

references_below()#

Retrieves a list of assets below the current asset.

Return type:

List[Self]

Returns:

List[Self] – A list of asset objects that are below the current asset.

references_below_uuid()#

Retrieves a list of asset-references below the current asset.

Return type:

List[str]

Returns:

List[str] – A list of asset-references (as strings) that are below the current asset.

samples()#

Returns all sample-type attributes for this asset.

Return type:

List[Dict[str, Any]]

Returns:

List[Dict] –

A list of dictionaries, each containing:

  • ”ID” (str): The attribute ID.

  • ”VALUE” (Any): The value of the sample.

  • ”TAG” (str): The cleaned tag name with placeholders removed.

stop_conditions_subscription()#

Stops the NATS consumer and gracefully shuts down the subscription for conditions.

Return type:

None

stop_events_subscription()#

Stops the NATS consumer and gracefully shuts down the subscription for events.

Return type:

None

stop_messages_subscription()#

Stops the NATS consumer and gracefully shuts down the subscription.

Return type:

None

stop_samples_subscription()#

Stops the NATS consumer and gracefully shuts down the subscription for samples.

Return type:

None

subscribe_to_conditions(on_condition)#

Subscribes to asset conditions using a NATS consumer. Only messages with TYPE == ‘Condition’ are forwarded to the callback.

Return type:

None

Parameters:

on_condition (AssetNATSCallback) – Callable that takes (msg_subject: str, msg_value: dict).

subscribe_to_events(on_event)#

Subscribes to asset events using a NATS consumer. Only messages with TYPE == ‘Events’ are forwarded to the callback.

Return type:

None

Parameters:

on_event (AssetNATSCallback) – Callable that takes (msg_subject: str, msg_value: dict).

subscribe_to_messages(on_message)#

Subscribes to asset messages using a NATS consumer.

Return type:

None

Parameters:

on_message (AssetNATSCallback) – Callable that takes (msg_subject: str, msg_value: dict) and handles messages.

subscribe_to_samples(on_sample)#

Subscribes to asset samples using a NATS consumer. Only messages with TYPE == ‘Samples’ are forwarded to the callback.

Return type:

None

Parameters:

on_meon_samplessage (AssetNATSCallback) – Callable that takes (msg_subject: str, msg_value: dict).

property type: Literal['Samples', 'Condition', 'Events', 'Method', 'OpenFactory', 'UNAVAILABLE']#

Retrieves the type of the asset from ksqlDB.

Executes a SQL query to fetch the asset type. If the query returns no result, the method defaults to ‘UNAVAILABLE’.

Returns:

Literal[‘Samples’, ‘Condition’, ‘Events’, ‘Method’, ‘OpenFactory’, ‘UNAVAILABLE’] – The asset type as stored in the assets_type table, or ‘UNAVAILABLE’ if not found.

wait_until(attribute_id, value, timeout=30, use_ksqlDB=False)#

Waits until the asset attribute has a specific value or times out.

Monitors either the NATS cluster or ksqlDB to check if the attribute value matches the expected value. Returns True if the value is found within the given timeout, False otherwise.

Return type:

bool

Parameters:

  • attribute_id (str) – The attribute ID of the asset to monitor.

  • value (Any) – The value to wait for the attribute to match.

  • timeout (int) – The maximum time to wait, in seconds. Default is 30 seconds.

  • use_ksqlDB (bool) – If True, uses ksqlDB instead of NATS to check the attribute value. Default is False.

Returns:

bool – True if the attribute value matches the expected value within the timeout, False otherwise.