Insights

Insights & Data Tags¶

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

In the Kelvin SDK, the Data Tag names are deprecated in favor of the Insight names. Use Insight, app.on_insight and filters.is_insight instead of DataTag, app.on_data_tag and filters.is_data_tag in new code.

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.

Insight Messages¶

Applications can receive Insight messages published by other Applications or created manually through the Kelvin UI. This allows your Application to react to labeled events on Asset time series data.

Each received Insight has an acceptance attribute, set by the platform. It is accepted, or snoozed while the Insight type is snoozed for the Asset.

Live¶

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

app.yaml setup¶

To consume insights, declare the tag names your Application 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_insight decorator. You can filter by tag name using app.app_configuration to limit processing to only the tags your Application 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 Insight

app = KelvinApp()

@app.on_insight
async def on_insight(insight: Insight):
    configured_tags = app.app_configuration.get("consume_tag_names", [])
    if configured_tags and insight.tag_name not in configured_tags:
        return

    logger.info(
        "Insight received",
        asset=getattr(insight.resource, "asset", None),
        tag_name=insight.tag_name,
        resource=str(insight.resource),
    )

app.run()

Callback¶

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from kelvin.application import KelvinApp
from kelvin.message import Insight

app = KelvinApp()

async def on_insight(insight: Insight) -> None:
    logger.info("Received insight", tag_name=insight.tag_name, resource=str(insight.resource))

app.on_insight = on_insight

app.run()

Historical¶

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

Historical Data Tag Queries¶

In addition to receiving live insights via on_insight, you can query historical insights over a time range using app.api.data_tag.list_data_tag(). The Kelvin API Client (Python) keeps the Data Tag names. This is useful for catching events that occurred before your Application 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,
        )