Kelvin Python SDK

A framework to build Kelvin applications in Python. This package helps you connect and interact with the Kelvin platform.

It provides tools for building, publishing, receiving, and filtering Kelvin messages (the primary communication method with Kelvin) and easier access to the resources (Assets) that the application is deployed to.

Quickstart

Here's a simple example receiving a data Message and publishing another.

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

app = KelvinApp()


@app.stream()
async def handle_input(msg):
    """Process all incoming messages"""
    print(f"Received: {msg.payload} from {msg.resource}")

    # Publish a response
    await app.publish(Number(resource=KRNAssetDataStream(msg.resource.asset, "output"), payload=msg.payload * 2))


# Run the application
app.run()

Table of Contents

• Installation
• Core Concepts
  • Kelvin Messages
  • Kelvin Resource Names (KRN)
• Application Deployment Context
• Message Handling
  • Stream Decorators
  • Filters
  • Callbacks
• Background Tasks
  • Tasks
  • Timers
  • Schedules
• Data Windows
  • Tumbling Window
  • Hopping Window
  • Rolling Window
• Publishing Messages
• Application Health
• Leader Election (HA)
• Testing with KelvinPublisher
• Testing with KelvinAppTest
• Complete Example

Installation

pip install kelvin-python-sdk

For AI/Data Science features (pandas support):

pip install kelvin-python-sdk[ai]

For development and testing with the publisher tool:

pip install kelvin-python-sdk[publisher]

Run kelvin-publisher --help to see available testing options.

Core Concepts

Kelvin Messages

Messages are the primary interface for exchanging data with the Kelvin platform. Applications send and receive messages to communicate with assets and other components.

Message Types

The SDK provides pre-built message types for common data:

from kelvin.message import Number, String, Boolean
from kelvin.krn import KRNAssetDataStream

# Create primitive data messages
number_msg = Number(resource=KRNAssetDataStream("asset-name", "datastream-name"), payload=42.5)

string_msg = String(resource=KRNAssetDataStream("asset-name", "datastream-name"), payload="Hello Kelvin")

boolean_msg = Boolean(resource=KRNAssetDataStream("asset-name", "datastream-name"), payload=True)

Message Builders

For more complex messages, use the provided MessageBuilder helpers:

from kelvin.message import (
    ControlChange,
    ControlAck,
    Recommendation,
    Insight,
    AssetParameter,
    AssetParameters,
    StateEnum,
)
from kelvin.krn import KRNAsset, KRNAssetDataStream, KRNAssetParameter
from datetime import datetime, timedelta

# Control Change - request a change in asset behavior
control_change = ControlChange(
    resource=KRNAssetDataStream("asset-name", "setpoint"),
    payload=75.0,
    expiration_date=timedelta(minutes=10),
    timeout=60,
    retries=3,
)

# Control Acknowledgement - respond to control changes
ack = ControlAck(
    resource=KRNAssetDataStream("asset-name", "setpoint"),
    state=StateEnum.applied,
    message="Control change successfully applied",
)

# Recommendation - suggest multiple control changes
recommendation = Recommendation(
    resource=KRNAsset("asset-name"),
    type="optimization",
    control_changes=[control_change],
    expiration_date=timedelta(hours=1),
    auto_accepted=False,
)

# Insight (formerly Data Tag) - add metadata to data
insight = Insight(
    resource=KRNAsset("asset-name"),
    start_date=datetime.now(),
    tag_name="anomaly_detected",
    description="Temperature spike detected",
)

# Asset Parameters - update asset configuration
param = AssetParameter(resource=KRNAssetParameter("asset-name", "threshold"), value=100)
params = AssetParameters(parameters=[param])

Custom Actions

A custom action is a request for something to happen outside the app - raising a work order, dispatching a crew, calling an external system. The app that executes the action answers with a CustomActionResult, and the app that created it receives a CustomActionReport once the action reaches a terminal state.

from kelvin.krn import KRNAsset
from kelvin.message import CustomAction
from datetime import timedelta

# Create an action
action = CustomAction(
    resource=KRNAsset("beam-pump-01"),
    type="create-work-order",
    title="Inspect excessive pump vibration",
    expiration_date=timedelta(hours=1),
    payload={"priority": "high"},
)
await app.publish(action)

Executing an action and answering with a response:

@app.on_custom_action
async def handle_action(action: CustomAction):
    work_order_id = await create_work_order(action.payload)
    # `response` travels back to the app that created the action
    await app.publish(action.result(success=True, response={"work_order_id": work_order_id}))

Deprecated: metadata on CustomActionResult and on CustomAction.result() is deprecated in favour of response. Both are still accepted: when only metadata is given, its value is used as the response.

Custom Action Reports

Reports are automatic for every custom action type the app declares as an output - there is nothing to opt into. Exactly one report is delivered per action, when it reaches a terminal state (completed, failed or cancelled).

from kelvin.message import CustomActionReport, CustomActionState


@app.on_custom_action_report
async def handle_report(report: CustomActionReport):
    if report.state == CustomActionState.COMPLETED:
        print(f"{report.action_id} completed: {report.response}")
    else:
        print(f"{report.action_id} ended as {report.state}: {report.message}")

response is optional even on a completed action - an executor may legitimately report none - so treat None as "no response", not as an empty one.

Evidences

Evidences provide context to recommendations by including supporting data and visual information. They help users understand the reasoning behind a recommendation, presenting analysis results, charts, and explanations that justify the suggested actions.

from kelvin.message import Recommendation
from kelvin.message.evidences import DataExplorer, DataExplorerSelector, Markdown
from kelvin.krn import KRNAsset, KRNAssetDataStream
from datetime import datetime, timedelta

# Markdown evidence - provide textual explanations
markdown_evidence = Markdown(
    title="Analysis Summary",
    markdown="""
## Temperature Anomaly Detected

The system detected unusually high temperature readings:
- Average temperature: 85°C
- Normal range: 60-75°C
- Duration: 2 hours

**Recommended action**: Reduce speed to allow cooling.
""",
)

# Data Explorer evidence - visualize time-series data
now = datetime.now()
data_explorer = DataExplorer(
    title="Temperature Trend Analysis",
    start_time=now - timedelta(hours=6),
    end_time=now,
    selectors=[
        DataExplorerSelector(resource=KRNAssetDataStream("asset-name", "temperature")),
        DataExplorerSelector(resource=KRNAssetDataStream("asset-name", "pressure"), agg="mean", time_bucket="5m"),
    ],
)

# Include evidences in recommendation
recommendation = Recommendation(
    resource=KRNAsset("asset-name"), type="temperature_optimization", evidences=[markdown_evidence, data_explorer]
)

The SDK supports multiple evidence types including Markdown, DataExplorer, LineChart, BarChart, Image, and IFrame. For a complete list of available evidence types and their configurations, see kelvin.message.evidences.

Kelvin Resource Names (KRN)

KRNs are used to uniquely identify Kelvin resources. They follow a structured format to reference assets, datastreams, parameters, and other entities.

from kelvin.krn import KRN, KRNAsset, KRNAssetDataStream, KRNAssetParameter

# Generic KRN (parse from string)
generic_krn = KRN.from_string("krn:asset:my-asset")

# Asset KRN
asset_krn = KRNAsset("my-asset")

# Asset DataStream KRN
datastream_krn = KRNAssetDataStream("my-asset", "temperature")

Application Deployment Context

When your application connects to Kelvin, it receives runtime information about its deployment environment. The KelvinApp instance provides access to:

• app.app_configuration: Application-specific configuration settings
• app.inputs: Input datastreams configured for your application
• app.outputs: Output datastreams configured for your application
• app.assets: The assets your application is deployed to (most important)

Working with Assets

The app.assets dictionary contains all assets available to your application. Each asset provides access to its properties, parameters, and datastreams:

from kelvin.application import KelvinApp

app = KelvinApp()


async def on_connect():
    # Access all available assets
    for asset_name, asset in app.assets.items():
        print(f"Asset: {asset_name}")

        # Access asset properties
        print(f"  Properties: {asset.properties}")

        # Access asset parameters
        for param_name, param_value in asset.parameters.items():
            print(f"  Parameter {param_name}: {param_value}")

        # Access asset datastreams
        for ds_name, datastream in asset.datastreams.items():
            print(f"  Datastream {ds_name}: {datastream}")


app.on_connect = on_connect
app.run()

Assets are the core resources in Kelvin, representing physical or logical entities (machines, sensors, systems, etc.). Your application interacts with assets by reading from their datastreams and publishing data or control changes back to them.

Message Handling

Stream Decorators

Stream decorators are the recommended way to handle incoming messages. They allow you to process messages based on specific criteria:

Use decorators to process messages based on specific criteria:

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

app = KelvinApp()


# Process all messages
@app.stream()
async def handle_all(msg: AssetDataMessage):
    print(f"All messages: {msg.payload}")


# Filter by asset
@app.stream(assets=["asset-1", "asset-2"])
async def handle_specific_assets(msg: AssetDataMessage):
    print(f"From specific assets: {msg.payload}")


# Filter by input datastream
@app.stream(inputs=["temperature", "pressure"])
async def handle_specific_inputs(msg: AssetDataMessage):
    print(f"From specific inputs: {msg.payload}")


# Can also register functions directly
def my_handler(msg: AssetDataMessage):
    print(f"Handler: {msg.payload}")


app.stream(my_handler, inputs=["humidity"])

app.run()

Filters

Filters allow you to selectively process messages using queues or async streams:

from kelvin.application import KelvinApp, filters

app = KelvinApp()


async def main():
    await app.connect()

    # Using a queue with filters
    queue = app.filter(filters.input_equals("temperature"))
    while True:
        msg = await queue.get()
        print(f"Temperature: {msg.payload}")

    # Using async stream with filters
    stream = app.stream_filter(filters.asset_equals("asset-1"))
    async for msg in stream:
        print(f"Asset 1 data: {msg.payload}")


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Built-in Filters

from kelvin.application import filters

# Filter by input datastream name(s)
filters.input_equals("temperature")
filters.input_equals(["temperature", "pressure"])

# Filter by asset name(s)
filters.asset_equals("my-asset")
filters.asset_equals(["asset-1", "asset-2"])

# Filter by resource KRN
filters.resource_equals(krn_instance)
filters.resource_equals([krn1, krn2])

# Filter by message type
filters.is_data_message(msg)
filters.is_asset_data_message(msg)
filters.is_control_status_message(msg)
filters.is_custom_action(msg)
filters.is_custom_action_report(msg)
filters.is_data_quality_message(msg)

Custom Filters

You can create custom filter functions:

def custom_filter(msg: Message) -> bool:
    """Filter for high-value readings"""
    return msg.payload > 100


queue = app.filter(custom_filter)

Callbacks

For advanced scenarios, you can define callbacks for specific lifecycle events. However, stream decorators are generally preferred for message processing.

Callbacks can be registered using decorators or direct assignment. Both sync and async functions are supported — sync functions are automatically wrapped to run in a thread pool.

Using Decorators

from kelvin.application import KelvinApp, AssetInfo
from kelvin.message.typing import AssetDataMessage
from typing import Optional

app = KelvinApp()


@app.on_connect
async def handle_connect():
    """Called when the app connects to Kelvin"""
    print("Connected to Kelvin platform")
    print(f"Configuration: {app.app_configuration}")
    print(f"Assets: {app.assets}")


@app.on_asset_input
async def handle_asset_input(msg: AssetDataMessage):
    """Called for data messages from asset inputs"""
    print(f"Data from {msg.resource}: {msg.payload}")


@app.on_control_change
async def handle_control_change(msg: AssetDataMessage):
    """Called when control changes are received"""
    print(f"Control change for {msg.resource}: {msg.payload}")


@app.on_asset_change
async def handle_asset_change(new_asset: Optional[AssetInfo], old_asset: Optional[AssetInfo]):
    """Called when assets are added, removed, or modified"""
    if new_asset is None:
        print(f"Asset removed: {old_asset.name}")
    else:
        print(f"Asset changed: {new_asset.name}")


@app.on_app_configuration
async def handle_app_configuration(config: dict):
    """Called when app configuration changes"""
    print(f"New configuration: {config}")


app.run()

Using Direct Assignment

app = KelvinApp()


async def on_connect():
    print("Connected to Kelvin platform")


app.on_connect = on_connect
app.on_asset_input = on_asset_input
app.on_control_change = on_control_change
app.on_asset_change = on_asset_change
app.on_app_configuration = on_app_configuration

app.run()

Available Callbacks

Background Tasks

Tasks

Tasks are functions that run in the background. They're started when the application connects:

from kelvin.application import KelvinApp
import asyncio

app = KelvinApp()


@app.task
async def background_task():
    """Runs once when the app starts"""
    print("Task started")
    # Perform initialization or one-time operations


@app.task
async def continuous_task():
    """Runs continuously in the background"""
    while True:
        print("Processing...")
        await asyncio.sleep(10)


# Can also register functions directly
async def another_task():
    print("Another task")


app.task(another_task, name="my_task")

app.run()

Timers

Timers execute functions at regular intervals:

from kelvin.application import KelvinApp

app = KelvinApp()


@app.timer(interval=30)  # seconds
async def periodic_check():
    """Runs every 30 seconds"""
    print("Periodic check executed")


@app.timer(interval=60)
async def publish_metrics():
    """Runs every 60 seconds"""
    # Publish periodic metrics
    await app.publish(...)


# Register timers directly
def sync_timer():
    print("Sync timer")


app.timer(sync_timer, interval=10, name="sync_timer")

app.run()

Schedules

Schedules execute functions on cron-like schedules with timezone support. They are ideal for tasks that need to run at specific times of day or on specific days of the week.

Human-readable schedules

Use every and at parameters for readable schedule definitions:

from kelvin.application import KelvinApp

app = KelvinApp()


@app.schedule(every="day", at="19:00", timezone="Australia/Sydney")
async def daily_report():
    """Runs every day at 7pm Sydney time"""
    print("Daily report")


@app.schedule(every="monday", at="09:00")
async def weekly_standup():
    """Runs every Monday at 09:00 UTC"""
    print("Weekly standup reminder")


@app.schedule(every="weekday", at="08:00", timezone="Europe/London")
async def weekday_morning():
    """Runs Monday-Friday at 08:00 London time"""
    print("Good morning!")


# Multiple times per day
@app.schedule(every="day", at=["09:00", "17:00"])
async def twice_daily():
    """Runs at 09:00 and 17:00 UTC"""
    print("Running twice daily")


app.run()

The every parameter accepts: "day", "monday" through "sunday", "weekday", or "weekend".

Cron expressions

For more complex schedules, use standard 5-field cron expressions:

@app.schedule(cron="*/5 * * * *")
def every_five_minutes():
    """Runs every 5 minutes (sync functions are also supported)"""
    print("Running every 5 minutes")


@app.schedule(cron="0 9 * * MON-FRI", timezone="Europe/London")
async def weekday_job():
    """Runs at 09:00 London time on weekdays"""
    print("Weekday job")

Interval scheduling

Use interval to fire every Nth occurrence. Combined with start_time, the count alignment is deterministic:

from datetime import datetime, timezone


@app.schedule(
    every="monday",
    at="09:00",
    interval=2,
    start_time=datetime(2026, 4, 6, tzinfo=timezone.utc),
)
async def biweekly_sync():
    """Runs every other Monday at 09:00 UTC, starting from April 6, 2026"""
    print("Biweekly sync")

Parameters

Schedules can also be registered directly without the decorator syntax:

async def my_scheduled_task():
    print("Scheduled task")


app.schedule(my_scheduled_task, every="day", at="12:00", name="noon_task")

Data Windows

Data windows aggregate incoming data over time or count, returning pandas DataFrames for analysis. All window operations require the ai optional dependency.

Tumbling Window

Non-overlapping, fixed-size time windows:

import asyncio
from datetime import datetime, timedelta
from kelvin.application import KelvinApp

app = KelvinApp()


async def main():
    await app.connect()

    # Process data in 10-second non-overlapping windows
    window_start = datetime.now()
    async for asset_name, df in app.tumbling_window(window_size=timedelta(seconds=10)).stream(window_start):
        print(f"Asset: {asset_name}")
        print(df)  # pandas DataFrame with all data in window


if __name__ == "__main__":
    asyncio.run(main())

Use case: Aggregate data every N seconds for batch processing (e.g., computing averages every 10 seconds).

Hopping Window

Overlapping, fixed-size windows with a configurable hop interval:

import asyncio
from datetime import datetime, timedelta
from kelvin.application import KelvinApp

app = KelvinApp()


async def main():
    await app.connect()

    # 10-second windows, moving forward by 5 seconds each time
    window_start = datetime.now()
    async for asset_name, df in app.hopping_window(
        window_size=timedelta(seconds=10), hop_size=timedelta(seconds=5)
    ).stream(window_start=window_start):
        print(f"Asset: {asset_name}")
        print(df)  # pandas DataFrame with overlapping data


if __name__ == "__main__":
    asyncio.run(main())

Use case: Sliding window analysis where you want overlapping data (e.g., moving averages with 50% overlap).

Rolling Window

Count-based windows that slide with each new message:

import asyncio
from kelvin.application import KelvinApp

app = KelvinApp()


async def main():
    await app.connect()

    # Window of last 5 messages, slides by 2 messages
    async for asset_name, df in app.rolling_window(count_size=5, slide=2).stream():
        print(f"Asset: {asset_name}")
        print(df)  # pandas DataFrame with last 5 messages


if __name__ == "__main__":
    asyncio.run(main())

Use case: Process the last N data points (e.g., calculate trend over the last 10 readings).

Publishing Messages

Use app.publish() to send messages to the Kelvin platform:

from kelvin.application import KelvinApp
from kelvin.message import Number, ControlChange, Recommendation
from kelvin.krn import KRNAssetDataStream, KRNAsset
from datetime import timedelta

app = KelvinApp()


@app.timer(interval=10)
async def publish_data():
    # Publish data
    await app.publish(Number(resource=KRNAssetDataStream("asset-1", "output"), payload=42.0))

    # Publish control change
    await app.publish(
        ControlChange(
            resource=KRNAssetDataStream("asset-1", "setpoint"), payload=75.0, expiration_date=timedelta(minutes=5)
        )
    )

    # Publish recommendation
    await app.publish(
        Recommendation(
            resource=KRNAsset("asset-1"), type="optimization", control_changes=[...], expiration_date=timedelta(hours=1)
        )
    )


app.run()

Application Health

kelvin.health lets the app report liveness and readiness from anywhere in the code. The state is served over HTTP to the orchestrator's probes (GET /health/live and GET /health/ready), so Kubernetes restarts hung or broken workloads and keeps traffic away from ones that aren't ready yet.

On the platform, health.start() binds the port injected as KELVIN_HEALTH_PORT; off the platform (local runs, tests, CI) it is a no-op, so the same code runs everywhere. From start() until your code says otherwise, /health/live answers 200 and /health/ready answers 503 — starting slowly is not being broken.

from kelvin import health
from kelvin.application import KelvinApp

app = KelvinApp()


@app.on_connect
async def on_connect():
    # Connected and configured: route traffic to the app
    health.ready()


@app.on_disconnect
async def on_disconnect():
    # Out of rotation while disconnected; no restart needed
    health.not_ready("disconnected from the Kelvin stream")


@app.timer(interval=5)
async def process_batch():
    # An escaping exception marks the app unhealthy (sticky) with the
    # exception as the probe body, so the orchestrator restarts the container
    with health.critical("process_batch"):
        ...


health.start()  # bind the health port; safe to call from anywhere
app.run()

Report a caught exception directly; the reason becomes the probe body:

try:
    reconnect()
except ConnectionError as e:
    health.unhealthy(e)  # fail liveness -> restart

Opt into hung-app detection with a watchdog. Arming never starts anything: something must call health.heartbeat() at least once per window, and health.heartbeat_loop is the ready-made petter. In the standard decorator app, register it as a background task:

app.task(health.heartbeat_loop)  # petter mounts when the app connects

health.start(watchdog=30.0)  # arms the watchdog, starts nothing
app.run()

Managing the event loop manually? Run the petter yourself with asyncio.create_task(health.heartbeat_loop()) and cancel it on shutdown.

Sync apps — and async apps that want the watchdog to prove work progresses rather than "loop alive" — heartbeat from the main loop instead:

health.start(watchdog=30.0)

while True:
    process_batch()
    health.heartbeat()

Pick a window longer than the app's worst-case startup/connect time: the petter only starts on first connect, so a shorter window trips the watchdog while the app is still starting up.

Tests (or apps needing a custom port) can instantiate kelvin.health.AppHealth directly instead of using the module-level functions.

See examples/health for a complete runnable app.

Leader Election (HA)

Run N replicas of the same app at the edge with exactly one acting at a time. kelvin.ha elects against the cluster-local Leader API — calling elect is the heartbeat, so there's no separate keep-alive to manage. It's fail-closed: if the API becomes unreachable, is_leader drops to False within one lease duration.

Zero-Config Setup

LeaderElection() / AsyncLeaderElection() derive their identity and target from platform-injected environment variables — no constructor arguments required in production:

All five are required. Running locally (e.g. against a port-forwarded Leader API) means exporting all of them yourself.

Sync Usage

import time
from kelvin.ha import LeaderElection

leader = LeaderElection()  # identity from platform-injected env vars
leader.on_promoted = resume_publishing
leader.on_demoted = drain_queues
leader.start()

leader.wait_until_leader(timeout=30)  # -> bool, False on timeout

while True:
    if leader.is_leader:
        publish()
    time.sleep(1)

leader.stop()

Async Usage

from kelvin.ha import AsyncLeaderElection

leader = AsyncLeaderElection()


@leader.on_demoted
async def drain() -> None:
    await publisher.drain()


await leader.start()
await leader.wait_until_leader()

Callbacks may be registered by assignment (leader.on_promoted = fn, shown above) or as a decorator (@leader.on_demoted, shown above). AsyncLeaderElection accepts sync or async callbacks in either style; LeaderElection (the thread-based, sync class) callbacks must be sync — registering an async def there is logged as an error and dropped rather than run.

Notes

• If the required environment variables are missing (e.g. running outside a Kelvin workload), construction does not raise: a warning is logged naming the missing variables, start() is a no-op, and is_leader stays False.
• Pass a kelvin.krn type to contend a resource other than the default krn:app:<KELVIN_APP_NAME> — e.g. resource=KRNAppVersion("my-app", "1.2.0"). The Leader API rejects non-KRN resources (err_invalid_krn), so the parameter is typed as KRN.
• current_leader reports the last observed leader identity — visible to followers too — and is not cleared on demotion or stop.
• wait_until_leader(timeout) blocks until this replica becomes leader, returning False on timeout (omit timeout, or pass None, to wait indefinitely).
• Leadership is advisory: kelvin.ha faithfully reflects what the Leader API reports, but does not fence stale writes (e.g. from a replica that paused past its lease and hasn't yet observed its own demotion). Apps performing correctness-critical writes should add fencing downstream.

Testing with KelvinPublisher

kelvin-publisher is a CLI tool for testing applications during development. It's not meant to be imported in your code, but used as a standalone testing utility.

Installation

pip install kelvin-python-sdk[publisher]

Run kelvin-publisher --help to see all available commands and options.

Usage

The publisher has three modes for simulating data:

1. Simulator - Random Data

Generate random data to your application's inputs:

kelvin-publisher simulator --help
kelvin-publisher simulator

This automatically discovers your application's inputs and publishes random data.

2. CSV - File-based Data

Publish data from a CSV file:

kelvin-publisher csv --help
kelvin-publisher csv --csv data.csv

Replays test data from CSV files.

3. Generator - Custom Data

Use a custom Python class to generate data:

kelvin-publisher generator --help
kelvin-publisher generator --entrypoint action_generator.py:CustomActionGenerator

Example generator:

import asyncio
from typing import AsyncGenerator

from kelvin.message import Number
from kelvin.krn import KRNAssetDataStream
from kelvin.publisher import DataGenerator


class OtherGenerator(DataGenerator):
    def __init__(self) -> None:
        print("Hello from OtherGenerator")

    async def run(self) -> AsyncGenerator[Number, None]:
        print("Running OtherGenerator")
        for i in range(20, 30):
            yield Number(
                resource=KRNAssetDataStream("test-asset-1", "input-number"),
                payload=i,
            )
            await asyncio.sleep(1)

This mode allows you to implement sophisticated test scenarios with custom logic.

Testing with KelvinAppTest

The kelvin.testing module provides a deterministic, in-memory test harness for KelvinApp instances. Tests run without network dependencies and use a virtual clock for time control.

Quick Start

Add a pytest.ini alongside your tests:

[pytest]
pythonpath = .

Write tests against your app singleton:

import pytest
from main import app

from kelvin.krn import KRNAssetDataStream
from kelvin.message import Number
from kelvin.testing import KelvinAppTest, ManifestBuilder


def _build_manifest():
    return ManifestBuilder().add_input("temperature").add_output("alert").add_asset("pump-001")


class TestMyApp:
    @pytest.mark.asyncio
    async def test_processes_input(self):
        harness = KelvinAppTest(app, manifest=_build_manifest().build())

        async with harness:
            await harness.publish(
                Number(
                    resource=KRNAssetDataStream("pump-001", "temperature"),
                    payload=42.0,
                )
            )
            await harness.run_until_idle()

        outputs = harness.outputs
        assert len(outputs) > 0

The pattern is always:

1. Build a manifest with ManifestBuilder (assets, datastreams, configuration)
2. Create KelvinAppTest wrapping your app
3. Use async with to connect/disconnect
4. Publish inputs and/or advance time with run_until_idle()
5. Assert on harness.outputs

Testing Timers

Timers fire automatically when virtual time advances. Set the timeout parameter past the timer interval:

@pytest.mark.asyncio
async def test_timer_fires(self, capsys):
    harness = KelvinAppTest(app, manifest=ManifestBuilder().build())

    async with harness:
        # Timer interval is 5s - advance past the first firing
        await harness.run_until_idle(timeout=6.0)

    captured = capsys.readouterr().out
    assert "timer fired" in captured

Inspecting Outputs

Filter outputs by message type:

from kelvin.message.base_messages import RecommendationMsg, ControlChangeMsg

recs = [m for m in harness.outputs if isinstance(m, RecommendationMsg)]
ccs = [m for m in harness.outputs if isinstance(m, ControlChangeMsg)]

Data Sources

Automate input generation with built-in data sources:

from datetime import timedelta
from kelvin.testing import RandomSource, CSVSource, SyntheticSource, SineWave

# Random values
source = RandomSource(datastreams=["temperature"], min_value=0, max_value=100).with_asset("pump-001")

# Replay from CSV
source = CSVSource("data.csv", playback=True).with_asset("pump-001")

# Synthetic wave patterns
source = SyntheticSource(
    pattern=SineWave(amplitude=10, period=timedelta(minutes=1)),
    datastream="temperature",
).with_asset("pump-001")

harness.add_source(source)  # add before connecting

For the full testing guide including ManifestBuilder options, all data source types, and detailed testing patterns, see docs/modules/testing.rst.

Complete Example

Here's a complete example that demonstrates multiple features working together:

import asyncio
from datetime import timedelta
from kelvin.application import KelvinApp, filters
from kelvin.message import Number, ControlChange, ControlAck, StateEnum
from kelvin.krn import KRNAssetDataStream

app = KelvinApp()


# Lifecycle callbacks
@app.on_connect
async def handle_connect():
    print("Connected to Kelvin platform")
    print(f"Configuration: {app.app_configuration}")
    print(f"Available assets: {list(app.assets.keys())}")


@app.on_control_change
async def handle_control_change(msg):
    print(f"Control change received for {msg.resource}")
    # Acknowledge the control change
    await app.publish(
        ControlAck(resource=msg.resource, state=StateEnum.applied, message="Control change successfully applied")
    )


# Stream decorators for processing inputs
@app.stream(inputs=["temperature"])
async def process_temperature(msg):
    """Process temperature readings"""
    temp = msg.payload
    print(f"Temperature from {msg.resource.asset}: {temp}°C")

    # Publish processed data
    await app.publish(
        Number(resource=KRNAssetDataStream(msg.resource.asset, "temp_fahrenheit"), payload=temp * 9 / 5 + 32)
    )


@app.stream(inputs=["pressure"])
async def process_pressure(msg):
    """Process pressure readings"""
    print(f"Pressure from {msg.resource.asset}: {msg.payload} Pa")


# Background task
@app.task
async def monitor_system():
    """Continuous monitoring task"""
    await asyncio.sleep(5)  # Wait for initialization

    while True:
        print("Monitoring system health...")
        # Add your monitoring logic here
        await asyncio.sleep(30)


# Periodic timer
@app.timer(interval=60)
async def publish_heartbeat():
    """Publish heartbeat every minute"""
    for asset_name in app.assets.keys():
        await app.publish(Number(resource=KRNAssetDataStream(asset_name, "heartbeat"), payload=1))


# Run the application
if __name__ == "__main__":
    app.run()