Publish Control Changes¶

A control change is a setpoint command — a value your app sends to a physical asset to change its operating state, such as adjusting a valve position or a speed setpoint. When your app receives a control change, it handles it and responds with an acknowledgement.

Testing control changes locally lets you confirm that your app's handler fires correctly, produces the right acknowledgement, and handles both applied and failed outcomes before you deploy.

What you need¶

Read Test Harness & Setup first if you have not already. You need pytest and pytest-asyncio installed and your app = KelvinApp() instance exposed in main.py.

Setup: ManifestBuilder¶

Declare the control change input and output channels. These must match the names in your app.yaml:

from kelvin.testing import ManifestBuilder

manifest = (
    ManifestBuilder()
    .add_control_change_input("setpoint")          # the channel your app receives control changes on
    .add_control_change_output("setpoint-ack")     # the channel your app sends acknowledgements on
    .add_asset("doc_demo_plunger_01")
    .add_asset("doc_demo_plunger_02")
    .build()
)

The ControlChange message¶

A ControlChange carries a value, a target resource, and an expiration window. Your app receives this and is expected to apply it and respond:

from datetime import timedelta
from kelvin.message import ControlChange
from kelvin.krn import KRNAssetDataStream

ControlChange(
    resource=KRNAssetDataStream("doc_demo_plunger_01", "setpoint"),   # which asset and stream
    payload=75.0,                                                      # the new setpoint value
    expiration_date=timedelta(minutes=10),                             # how long before it expires
    timeout=60,                                                        # seconds to wait for ack
    retries=3,                                                         # retry attempts on failure
)

Field	Description
`resource`	`KRNAssetDataStream(asset, stream)` — the asset and control stream to update
`payload`	The new value to apply
`expiration_date`	A `timedelta` for how long this change remains valid
`timeout`	Seconds to wait for an acknowledgement before timing out
`retries`	Number of retry attempts if the initial attempt fails

Full test example¶

import pytest
from datetime import timedelta
from main import app
from kelvin.testing import KelvinAppTest, ManifestBuilder
from kelvin.message import ControlChange, ControlChangeMsg
from kelvin.krn import KRNAssetDataStream

manifest = (
    ManifestBuilder()
    .add_control_change_input("setpoint")
    .add_control_change_output("setpoint-ack")
    .add_asset("doc_demo_plunger_01")
    .build()
)

@pytest.mark.asyncio
async def test_setpoint_change_acknowledged():
    harness = KelvinAppTest(app, manifest=manifest)

    async with harness:
        await harness.publish(
            ControlChange(
                resource=KRNAssetDataStream("doc_demo_plunger_01", "setpoint"),
                payload=75.0,
                expiration_date=timedelta(minutes=10),
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    acks = [o for o in outputs if isinstance(o, ControlChangeMsg)]
    assert len(acks) == 1
    assert acks[0].resource.asset == "doc_demo_plunger_01"

Testing failure paths¶

Test what happens when your app cannot apply the control change. Your handler should still publish a response — with an appropriate failure state:

@pytest.mark.asyncio
async def test_setpoint_rejected_when_out_of_range():
    harness = KelvinAppTest(app, manifest=manifest)

    async with harness:
        await harness.publish(
            ControlChange(
                resource=KRNAssetDataStream("doc_demo_plunger_01", "setpoint"),
                payload=999.0,                     # out of valid operating range
                expiration_date=timedelta(minutes=10),
            )
        )
        await harness.run_until_idle()
        outputs = harness.outputs

    acks = [o for o in outputs if isinstance(o, ControlChangeMsg)]
    assert len(acks) == 1
    assert acks[0].payload.state != "applied"

Run the tests¶

pytest

pytest exits with code 0 if all assertions pass, or non-zero if any fail.