Testing with Generator

In the Kelvin SDK there is a test tool to help you ingest data to the inputs and debug the outputs to test the performance of your Kelvin SmartApp™.

The Kelvin SDK test generator tool is designed to allow testers to ingest messages from another Python script for testing purposes.

The general command to run the Test Generator tool is;

kelvin app test generator --entrypoint <python script>:<class name>

Generator Tool Overview

The test generator tool supports a variety of options to customize its behavior. Below is a list of the available options:

Parameter	Required	Description	Note
--config TEXT	optional	Path to the app.yaml	Default: Current directory
`--entrypoint	required	Path to the entrypoint (setuptools style) of the generator python class. It can be path a .py file or path to a module. Eg: mygenerator.py:MyGenerator
--asset-count INTEGER	optional	Number of Assets (test-asset-N)	Default: 1

Video Tutorial

You can watch this Event Callback video tutorial which will show you how to setup a Kelvin SmartApp™ and create a Generator Test Python script to test the app all on you local machine with the Kelvin Test Tools.

Copy the code in the Video Tutorial

In the following chapters after the video tutorial you can see and copy all the scripts shown in the video tutorial.

The Generator Tool section starts at minute 2:23.

Generator Python Script

This is a local Python script you use to create the data and send it to the Kelvin SmartApp™ using the command line kelvin app test tool.

Detailed Code Explanations

Click on the for details about the code

import asyncio # (1)!
from typing import AsyncGenerator
import random
import string
import json
from kelvin.krn import KRNAssetDataStream # (2)!
from kelvin.publisher import DataGenerator, MessageData # (3)!
from kelvin.message import Number

# $ kelvin app test generator --help
#
# Usage: kelvin app test generator [OPTIONS]
#
#   Publishes data generated by a custom generator class
#
# Options:
#   --config TEXT          Path to the app config file  [default: app.yaml;
#                          required]
#   --entrypoint TEXT      Path to the entrypoint (setuptools style) of the
#                          generator python class. It can be path a .py file or
#                          path to a module. Eg: mygenerator.py:MyGenerator
#                          [required]
#   --asset-count INTEGER  Number of test assets from 'test-asset-1' to 'test-
#                          asset-N'
#   --help                 Show this message and exit.
#
#
# i.e.:
#
# $ kelvin app test generator --entrypoint generator.py:MyNumberGenerator

class MyNumberGenerator(DataGenerator): # (4)!
    def __init__(self):
        print("Hello from MyNumberGenerator")
    async def run(self) -> AsyncGenerator[MessageData, None]:
        print("Running MyNumberGenerator")
        for i in range(10):
            yield MessageData(
                resource=KRNAssetDataStream("", "sa-rw-number"),
                value=i,
            )
            await asyncio.sleep(1)

class MyBooleanGenerator: # (6)!
    def __init__(self):
        print("Hello from MyBooleanGenerator")
    async def run(self) -> AsyncGenerator[MessageData, None]:
        print("Running MyBooleanGenerator")
        while True:
            i=bool(random.getrandbits(1))
            yield MessageData(
                resource=KRNAssetDataStream("", "sa-rw-bool"),
                timestamp=None,
                value=i,
            )
            await asyncio.sleep(1)

class MyStringGenerator: # (7)!
    def __init__(self):
        print("Hello from MyStringGenerator")
    async def run(self) -> AsyncGenerator[MessageData, None]:
        print("Running MyStringGenerator")
        while True:
            i=''.join(random.choices(string.ascii_letters,k=7)) # initializing size of string
            yield MessageData(
                resource=KRNAssetDataStream("", "sa-rw-text"),
                timestamp=None,
                value=str(i),
            )
            await asyncio.sleep(1)

class MyObjectGenerator: # (8)!
    def __init__(self):
        print("Hello from MyObjectGenerator")
    async def run(self) -> AsyncGenerator[MessageData, None]:
        print("Running MyObjectGenerator")
        file = open('object.json')
        # returns JSON object as
        # a dictionary
        data = json.load(file)
        while True:
            yield MessageData(
                resource=KRNAssetDataStream("", "sa-object"),
                timestamp=None,
                value=data,
            )
            await asyncio.sleep(1)

class CCGenerator: # (9)!
    def __init__(self):
        print("Hello from CCGenerator")
    async def run(self) -> AsyncGenerator[MessageData, None]:
        print("Running CCGenerator")
        for i in range(20, 30):
            yield MessageData(
                resource=KRNAssetDataStream("", "sa-cc-number"),
                timestamp=None,
                value=i
            )
            await asyncio.sleep(1)

1. The asyncio module is used to ensure the program is running in asynchronous mode. This allows code, especially that which typically waits such as I/O-bound operations, to effectively execute in parallel thereby making the most efficient use of processor time.
2. The KRNAssetDataStream class is used to define the Asset / Data Stream pair names into the Kelvin format. KRN stands for Kelvin Resource Name.
3. The DataGenerator and MessageData are two classes used for declaring the Generator and creating the Message to be sent as the test data.
4. The MyNumberGenerator and MyOtherNumberGenerator class creates a number from 1 to 10 and sends it as test number data once per second for 10 seconds.
5. The MyNumberGenerator and MyOtherNumberGenerator class creates a number from 1 to 10 and sends it as test number data once per second for 10 seconds.
6. The MyBooleanGenerator class randomly choose either True or False and sends it as test boolean data once per second.
7. The MyStringGenerator class creates a 7 letter word with random letters and sends it as test string data once per second.
8. The MyObjectGenerator class load data from a file (it can be generated by the Python script to) and sends it as test object data once per second. See below for a sample of the object data as a dynacard, though in reality it can be anything.
9. The CCGenerator class creates a control change for an Asset / Data Stream pair and sends it as control change object data once per second for 10 seconds.

Test your SmartApp

1. Requirements

Normally you will use the kelvin app test on Kelvin SmartApps™, you will be working with a complete complimentary set of files created by kelvin app create.

As we are only interesting in showing the abilities of the Generator Tool we will just create two files, main.py and app.yaml for this demo purpose.

Info

You can learn more about how to create a Kelvin SmartApp™ in the "Development" ⟹ "Create a SmartApp" page.

main.py

In order to show how to use the generator, we will use it with the example event callback app in the menu Development ⟹ Consume.

[Event Callbacks Example Script](../consume/event-callbacks.md){ .md-button .md-button--primary }

If you are not interested in the code, you can simply copy and paste this script into a file called main.py.

import asyncio

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

async def on_control_change(msg: AssetDataMessage):
    # Get Asset and Value
    asset = msg.resource.asset
    value = msg.payload

    print("Received Control Message: ", msg)
    print(f"Asset: {asset}, Value: {value}")

async def main() -> None:

    # Initialize the SmartApp
    app = KelvinApp()

    # Define the function to call when event callback is triggered
    app.on_control_change = on_control_change

    # Connect the SmartApp with the Kelvin Platform
    await app.connect()

    # Force the Application to run forever
    while True:
        await asyncio.sleep(1)

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

app.yaml

For the app.yaml file, simply copy this and paste into a file called app.yaml.

spec_version: 4.13.0

info:
  name: sa-e2e-consumer-app
  title: sa-e2e consumer App Generator
  version: 1.0.0
  description: Consumer App

app:
  type: kelvin
  kelvin:
    language:
      type: python
      python:
        entry_point: kelvin_python_sdk

    inputs:
    - data_type: number
      name: sa-rw-number
    - data_type: string
      name: sa-rw-text
    - data_type: boolean
      name: sa-rw-bool
    - data_type: object
      name: sa-object
    - data_type: number
      name: sa-cc-number
      control_change: true # control change input

system:
  environment_vars:
  - name: KELVIN_GW_MODE
    value: SOCKETS

1. Run the Generator Test Tool:

   To test by sending numbers to your Kelvin SmartApp™

   kelvin app test generator --entrypoint generator.py:MyNumberGenerator
   

   To test by sending control changes to your Kelvin SmartApp™

   kelvin app test generator --entrypoint generator.py:MyCCGenerator
   

   and you should see the following output which includes any print statement from the generator Python script and the Kelvin Publisher class itself:

   Hello from <Class that is called>
   Publisher started. Listening on 0.0.0.0:49167
   

2. Open a NEW terminal window and run the Kelvin SmartApp™:

   python3 main.py
   

   and you should see the following output:

   Received Data Message:  id=UUID('720f9e8d-1dd2-4054-afb4-5d579b31e998') type=KMessageTypeData('data', 'pt=number') trace_id=None source=None timestamp=datetime.datetime(2024, 10, 28, 21, 30, 40, 440062, tzinfo=datetime.timezone(datetime.timedelta(seconds=25200), '+07')) resource=KRNAssetDataStream(asset='test-asset-1', data_stream='sa-rw-number') payload=0.0
   
   Asset: test-asset-1, Value: 0.0
   
   Received Data Message:  id=UUID('ca147a01-2f88-4ba6-8d93-5bb186967574') type=KMessageTypeData('data', 'pt=number') trace_id=None source=None timestamp=datetime.datetime(2024, 10, 28, 21, 30, 41, 441811, tzinfo=datetime.timezone(datetime.timedelta(seconds=25200), '+07')) resource=KRNAssetDataStream(asset='test-asset-1', data_stream='sa-rw-number') payload=1.0
   
   Asset: test-asset-1, Value: 1.0
   
   (...)
   

3. Go to the Generator Tool terminal and check the logs:

   (...)
   
   Connected
   Running MyNumberGenerator