Control change messages

Control Change Messages¶

Control Changes are a special type of output Message which is a more rigorous fault tolerant process to write any data to an Asset.

Note

It also has failure protocols should it be impossible to write to the Asset, for example if it is offline.

To learn more about control changes and the detailed workflow, check out the full documentation in the overview concepts here.

To ensure the output is treated as a control change:

Warning

You do not put a Control Change output definitions under the data_streams/outputs key.

app.yaml Example
control_changes:
  outputs:
    - name: motor_speed_set_point
      data_type: number

The Control Change Object supports the following attributes :

Attribute	Required	Default Value	Description
`resource`	required	N/A	The KRNAssetDatastream that this Control Change is meant for.
`expiration_date`	required	N/A	Absolute datetime or a timedelta (from now) when the Control Change will expire.
`payload`	required	N/A	The desired target value for the control change (Boolean, Integer, Float or String).
`retries`	optional	0	Number of additional write requests sent to the Connection after the initial write. Retrying stops when an Applied confirmation is received or when the Asset returns an explicit failure (relayed by the Connection). Exhausting retries does not fail or expire the Control Change.
`timeout`	optional	300s	Interval in seconds between write requests to the Connection (retry interval).
`control_change_id`	optional	Random UUID	Sets a user specific ID for the control change (UUID).
`from_value`	optional		Initial (trigger) value of the control change.

Offline Edge Operations¶

Control Changes can operate even if the edge device does not have any Internet connection.

The only stipulation is that the Application that produces the control change object and the Application or Connector that consumes the control change object are hosted on the same Cluster and have local communications if they are hosted on different Nodes.

To set this up, the app.yaml file of the Application that produces the control change object must define the output Data Stream with the control_change key.

app.yaml Example
control_changes:
  outputs:
    - name: motor_speed_set_point
      data_type: number

If the consumer is a Connector, then this key is automatically set using the Kelvin Connector setup process.

The Application or Connector will now automatically receive the control change object directly from the Application without requiring any connection to the Kelvin Cloud.

Examples¶

Basic UsageFrom ValuesRetriesExpiration DateTimeout

Here is a minimal Control Change.

Basic Publish Control Change Python Example
from datetime import timedelta

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

app = KelvinApp()

# Create and Publish a Control Change every 10 seconds
@app.timer(interval=10)
async def publish_data():
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
            payload=1000,
            expiration_date=timedelta(minutes=5)
        )
    )

app.run()

In many cases you want to show what the current value is when the control change request is initiated.

This allows users to make informed decisions when the Control Change is embedded in a Recommendation or for Data Scientists when reviewing their model's historical performance.

Note

This code assumes the last_value variable is declared somewhere else.

Basic Publish Control Change Python Example
from datetime import datetime, timedelta, timezone

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

app = KelvinApp()

# Create and Publish a Control Change every 10 seconds
@app.timer(interval=10)
async def publish_data():
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
            from_value={
                        "value": last_value,
                        "timestamp": datetime.now(timezone.utc)
                        },
            payload=1000,
            expiration_date=timedelta(minutes=5)
        )
    )

app.run()

In some cases you may want a cool-off period between retry attempts, or to limit the number of retries. After the last retry, the Control Change remains pending until the expiration_date is reached.

Note

Exhausting retries does not abort the Control Change early.

Retries control how many additional write requests the Control Change Manager sends to the Connection after the initial write. The Connection writes to the Asset on each attempt. Retrying stops when an Applied confirmation is received or when the Asset returns an explicit failure (relayed by the Connection). If all retries are exhausted without Applied, the Control Change Manager stops sending write requests but the Control Change remains pending until expiration_date.

Publish Control Change with Retries Python Example
from datetime import timedelta

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

app = KelvinApp()

# Create and Publish a Control Change every 10 seconds
@app.timer(interval=10)
async def publish_data():
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
            payload=1000,
            expiration_date=timedelta(minutes=10),
            retries=3, # only attempt 3 extra times to write the data if the first attempt fails
            timeout=60 # retry every minute if the previous attempt failed
        )
    )

app.run()

If the "Applied" status is not received from the Connection to the Control Change Manager by the expiration_date, the process is deemed unsuccessful.

No more attempts will be made to apply the new value to the Asset.

In this example the expiration date is made ten minutes from the time the control change is created.

Control Change with Expiration Date Python Example
from datetime import timedelta

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

app = KelvinApp()

# Create and Publish a Control Change every 10 seconds
@app.timer(interval=10)
async def publish_data():
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
            payload=1000,
            expiration_date=timedelta(minutes=10),
            retries=3, # only attempt 3 extra times to write the data if the first attempt fails
            timeout=60 # retry every minute if the previous attempt failed
        )
    )

app.run()

The timeout (retry interval) parameter sets the interval in seconds between write requests to the Connection. The Control Change Manager waits this long after each write request before sending the next retry.

In this example timeout (retry interval) is set to 60 seconds.

Control Change with Timeout Python Example
from datetime import timedelta

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

app = KelvinApp()

# Create and Publish a Control Change every 10 seconds
@app.timer(interval=10)
async def publish_data():
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("my-motor-asset", "motor_speed_set_point"),
            payload=1000,
            expiration_date=timedelta(minutes=10),
            retries=3, # only attempt 3 extra times to write the data if the first attempt fails
            timeout=60 # retry every minute if the previous attempt failed
        )
    )

app.run()