Asset Utils

OpenFactory Assets helper methods.

Utility Classes

class openfactory.assets.utils.AssetAttribute(id, value, type, tag, timestamp=<factory>)

Bases: object

Represents a single attribute of an asset, including its value, type, tag, and timestamp.

id

the ID of the AssetAttribute

Type:

str

value

The actual value of the attribute. Can be a string or float.

Type:

Union[str, float]

type

The category/type of the attribute, must be one of the allowed literal strings.

Type:

Literal[‘Samples’, ‘Condition’, ‘Events’, ‘Method’, ‘OpenFactory’, ‘UNAVAILABLE’]

tag

The tag or identifier associated with this attribute.

Type:

str

timestamp

Timestamp when the attribute was recorded, in OpenFactory format. Defaults to the current timestamp if not provided.

Type:

str

__init__(id, value, type, tag, timestamp=<factory>)

class openfactory.assets.utils.AsyncLoopThread

Bases: object

Runs an asyncio event loop in a dedicated background thread.

This allows running asynchronous coroutines in a synchronous context, without blocking the main thread.

Example usage:

loop_thread = AsyncLoopThread() future = loop_thread.run_coro(some_async_func()) result = future.result() # blocks until coroutine completes loop_thread.stop()

__init__()

Initializes the asyncio loop and starts the background thread.

run_coro(coro)

Schedules a coroutine to run in the background loop.

Return type:

Future

Parameters:

coro (Coroutine) – The coroutine to run in the loop.

Returns:

asyncio.Future – A Future representing the execution of the coroutine.

stop()

Stops the background event loop and joins the thread.

This blocks until the thread has fully terminated.

Return type:

None

NATS Classes and Helpers

class openfactory.assets.utils.AssetNATSCallback(*args, **kwargs)

Bases: Protocol

Interface for callback used to handle NATS asset messages.

Parameters:

• msg_subject (str) – The NATS subject of the message (e.g., ‘OPCUA-SENSOR-001.*’).

• msg_value (dict) – The JSON-decoded payload of the message.

__init__(*args, **kwargs)

class openfactory.assets.utils.NATSSubscriber(loop_thread, servers, subject, on_message)

Bases: object

Represents a NATS subscriber that runs in a dedicated asyncio event loop thread.

This class connects to a NATS server, subscribes to a subject, and calls the provided callback for each received message.

__init__(loop_thread, servers, subject, on_message)

Initializes the NATS subscriber.

Parameters:

• loop_thread (AsyncLoopThread) – Event loop thread used to run async operations.

• servers (str | list[str]) – NATS server URL(s).

• subject (str) – NATS subject to subscribe to.

• on_message (AssetNATSCallback) – Callback to invoke when a message is received.

start()

Starts the NATS subscriber in the background event loop.

Return type:

None

stop()

Stops the NATS subscription and closes the connection.

Return type:

None

openfactory.assets.utils.get_nats_cluster_url(asset_uuid)

Fetch the NATS URL for a given asset UUID.

Return type:

str

Parameters:

asset_uuid (str) – The asset identifier (e.g., “toto”).

Returns:

str – The nats_url of the asset.

Raises:

• requests.exceptions.RequestException – If the HTTP request fails.

• ValueError – If the response is not valid JSON or nats_url is missing.

Time Helpers

openfactory.assets.utils.openfactory_timestamp(timestamp)

Convert a datetime to OpenFactory timestamp format.

The format is ISO 8601 with milliseconds precision and a ‘Z’ to indicate UTC time, e.g., ‘2025-05-04T12:34:56.789Z’.

Return type:

str

Parameters:

timestamp (datetime.datetime) – A datetime object (UTC recommended).

Returns:

str – Timestamp formatted in OpenFactory style.

openfactory.assets.utils.current_timestamp()

Returns the current timestamp in OpenFactory format.

The format is ISO 8601 with milliseconds precision and a ‘Z’ to indicate UTC time, e.g., ‘2025-05-04T12:34:56.789Z’.

Return type:

str

Returns:

str – The current UTC timestamp formatted in OpenFactory style.