Control Changes¶

Control Change Data Messages¶

When control changes are initiated by any Application on the Kelvin Platform, they can be processed by any Application.

Stream Decorators (Preferred)¶

The preferred and simplest method for consuming incoming control change messages is to create a Stream Decorator function and add a filter for Control Changes.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from kelvin.application import KelvinApp
from kelvin.message.typing import AssetDataMessage

app = KelvinApp()

@app.stream()
async def handle(msg: AssetDataMessage):
    if app.msg_is_control_change(msg):
        print(f"Control change: {msg.payload}")

app.run()

Callbacks¶

Event Callbacks are functions that are triggered when a control change is initiated.

Video Tutorial¶

You can watch this Event Callback video tutorial which will show you how to program and test it on you local machine with the Kelvin test data generator Python script.

Control Change Data Event¶

When any new control change is initiated, a callback event is created.

The full code is given here for the main.py script.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from kelvin.application import KelvinApp # (1)!
from kelvin.message.typing import AssetDataMessage # (2)!

async def on_control_change(msg: AssetDataMessage): # (3)!
    # Get Asset and Value
    asset = msg.resource.asset
    value = msg.payload

    print("Received Control Message: ", msg)
    print(f"Asset: {asset}, Value: {value}")

app.run()

1. The KelvinApp class is used to initlize and connect the Application to the Kelvin Platform.
2. The AssetDataMessage is a class representing the structure and properties of the asset data messages which will be held in an instance called msg.
3. The on_control_change callback function is triggered automatically whenever a new control change object is initiated to an Asset associated with the Application. It serves as an event handler to process incoming control change data in real time, ensuring that relevant review, actions or data updates occur immediately upon receipt.

When you run this Python script, the following output will be view in the terminal or logs;

1
2
3

Received Control Message:  id=UUID('44ba34f7-6689-464f-9804-7bcc0e347f68') type=KMessageTypeData('data', 'pt=number') trace_id=None source=None timestamp=datetime.datetime(2024, 10, 28, 14, 29, 41, 553346, tzinfo=datetime.timezone(datetime.timedelta(seconds=25200), '+07')) resource=KRNAssetDataStream(asset='test-asset-1', data_stream='sa-cc-number') payload=20.0

Asset: test-asset-1, Value: 20.0

App to App¶

Control Changes can be sent from one Kelvin Application to another.

Producing App¶

To set this up, the app.yaml file of the Application that produces the control change object must be defined under the control_change key.

1
2
3
4

control_changes:
  outputs:
    - name: motor_speed_set_point
      data_type: number

Then you can produce a Control Change to be sent to another Application.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from datetime import timedelta

from kelvin.application import KelvinApp
from kelvin.message import ControlChange
from kelvin.krn import KRNAssetDataStream

(...)

# Create and Publish a Control Change
await app.publish(
    ControlChange(
        resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
        payload=1000,
        expiration_date=timedelta(minutes=5)
    )
)

Consuming App¶

and the Application that consumes the control change object must define the input Data Stream with the control_change key.

1
2
3
4

control_changes:
  inputs:
    - name: motor_speed_set_point
      data_type: number

Then you can consume a Control Change that has been sent by another Application.

1
2
3
4

async def on_control_change(cc: AssetDataMessage) -> None:

    """Callback when a Control Change is received."""
    print("Received Control Change: ", cc)

You can also see the control change status with this code;

1
2
3
4
5
6
7

from kelvin.message import ControlChangeStatus

(...)

async def on_control_status(cc_status: ControlChangeStatus) -> None:
    """Callback when a Control Status is received."""
    print("Received Control Status: ", cc_status)

Data Generator for Local Application Testing¶

Easily test your Application locally on your computer.

Comprehensive documentation is available for the Generator Tool. Click here to learn how to use this event callback script with the Test Generator on the "Test an Application" ⟹ "Generator" page.