Data tag messages

Insights & Data Tags¶

In the Kelvin UI, Data Tags are now known as Insights.

Data Tags may be depreciated in the Kelvin SDK and Kelvin API in future releases.

Keep an eye on future releases for updated announcements.

Insights Management¶

You can also manage your insights using the Kelvin SDK and Kelvin API.

You can read all the details in the Platform Administration in the Insights section.

Data Tag Messages¶

Kelvin SmartApps™ can receive DataTag messages published by other SmartApps or created manually through the Kelvin UI. This allows your SmartApp to react to labeled events on Asset time series data.

Live¶

Receive insights in real time as they are published. Use this for event-driven reactions within a running SmartApp.

app.yaml setup¶

To consume insights, declare the tag names your SmartApp should receive in the data_tags.inputs section of your app.yaml. Each entry is the tag name as a string:

1
2
3
4

data_tags:
  inputs:
    - cycle_detected
    - temperature_too_high

Decorator (Preferred)¶

The preferred method is to use the @app.on_data_tag decorator. You can filter by tag name using app.app_configuration to limit processing to only the tags your SmartApp cares about.

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

from kelvin.application import KelvinApp
from kelvin.message import DataTag

app = KelvinApp()

@app.on_data_tag
async def on_data_tag(tag: DataTag):
    configured_tags = app.app_configuration.get("consume_tag_names", [])
    if configured_tags and tag.tag_name not in configured_tags:
        return

    logger.info(
        "Data tag received",
        asset=getattr(tag.resource, "asset", None),
        tag_name=tag.tag_name,
        resource=str(tag.resource),
    )

app.run()

Callback¶

For advanced scenarios, assign a handler function directly to app.on_data_tag:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from kelvin.application import KelvinApp
from kelvin.message import DataTag

app = KelvinApp()

async def on_data_tag(tag: DataTag) -> None:
    logger.info("Received data tag", tag_name=tag.tag_name, resource=str(tag.resource))

app.on_data_tag = on_data_tag

app.run()

Historical¶

Query insights over a past time range using the Kelvin API. Use this for catching events that occurred before your SmartApp started, or for periodic reconciliation.

Historical Data Tag Queries¶

In addition to receiving live insights via on_data_tag, you can query historical insights over a time range using app.api.data_tag.list_data_tag(). This is useful for catching events that occurred before your SmartApp started, or for periodic reconciliation.

1
2

api_permissions:
  - <permission-string-from-docstring>

The example below fetches all "Temperature Too High" tags for a specific Asset over the past 3 days:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from datetime import datetime, timedelta, timezone
from kelvin.logs import logger

TARGET_TAG = "Temperature Too High"

async def fetch_recent_temperature_high_tags(app, asset_name: str):
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=3)
    asset_krn = f"krn:asset:{asset_name}"

    tags = await app.api.data_tag.list_data_tag(
        data={
            "start_date": start_time,
            "end_date": end_time,
        },
    )

    matching = [
        tag for tag in tags
        if tag.tag_name == TARGET_TAG and str(tag.resource) == asset_krn
    ]

    for tag in matching:
        logger.info(
            "Historical data tag",
            tag_name=tag.tag_name,
            resource=str(tag.resource),
            start_date=tag.start_date,
        )