Writing Python Scripts

Writing Python Scripts

With the introduction of Eliona Version 12.2, important changes were made that require scripts to be adapted to use the new API design.

Script Definition

When a new script is created in the Eliona system, a function header is automatically generated. This header contains two parameters:

• id: The unique ID of the script.

• eliona: An object through which the user can interact with the Eliona system.

The function header looks like this:

# you may add imports only inside the function
# don't change or delete the function definition
def UserFunction(id, eliona):
	# add your code here

Importing Modules

All necessary modules must be imported within the function. It is important that modules are only imported within the function and no global imports are used.

Example:

def UserFunction(id, eliona):
    import random
    # Weitere Codezeilen folgen

---

Methods of the Eliona Object

The eliona object provides a set of methods that allow users to interact with the Eliona system. Here are the most important methods along with their parameters and descriptions:

Methode	Parameter	Beschreibung
GetHeap	gai: str, subtype: str, attribute: str	Retrieves the current value of an attribute or the entire data JSON of an asset from the heap.
SetHeap	gai: str, subtype: str, data: dict, source: str	Sends data to the calculator to process it and store it in the heap.
GetAssetIDByGAI	gai: str	Returns the numerical asset ID for the specified GAI.
GetAll	`ids: list[int	str], subtype: str`	Retrieves all data points of a specific subtype for a list of assets and returns them as a dictionary (GAI → data).
SQLQuery	query: str	Executes an arbitrary SQL query and returns the results as a list of tuples.
MakeSource	id: int	Generates a source string for heap operations based on the function ID.
OpenFile	name: str, mode: str	Opens a file in the Eliona working directory (analogous to Python’s open()).
AddAssetTag	gai: str, tag: str	Adds a tag to an asset (if not already present).
RemoveAssetTag	gai: str, tag: str	Removes a tag from an asset (if present).
GetTrendRecords	asset_id: int, subtype: str, attr: str, begin: datetime	Retrieves historical measured values as a list of TrendRecord(ts, value) from the specified time.
WriteTrendRecords	asset_id: int, subtype: str, attr: str, `recs: TrendRecord	Writes one or more TrendRecord data sets to the system.
GetAggregate	agg: Aggregate, asset_id: int, subtype: str, attr: str,raster: str, start: datetime,end: datetime = now()	Calculates aggregated values (Avg, Sum, Cusum) in a fixed raster over a period.
GetLastAggregate	agg: Aggregate, asset_id: int, subtype: str, attr: str, raster: str	Returns the latest cumulative aggregate in the selected raster (incl. last_ts).

---

Example Scripts

GetHeap

Description: Retrieves the current value of an attribute or the entire JSON of an asset from the heap.

Parameter:

• gai: str (required) – GAI of the asset, e.g., "K86_WP01"

• subtype: str (required) – Data subtype, e.g., "input"

• attribute: str (required) – Attributename, e.g., "Aussentemperatur"

Example-Code:

value = eliona.GetHeap("K86_WP01", "input", "Aussentemperatur")

Example-Output:

23.5

---

SetHeap

Description: Sends a dictionary of values to the calculator to process and store in the heap.

Parameter:

• gai: str (required)

• subtype: str (required)

• data: dict (required) – e.g., {"Aussentemperatur": 23.5}

• source: str (required) – typically eliona.MakeSource(id)

Code-Example:

data   = {"Aussentemperatur": 23.5}
source = eliona.MakeSource(id)
eliona.SetHeap("K86_WP01", "input", data, source)

---

GetAssetIDByGAI

Description: Returns the numerical asset ID for the specified GAI.

Parameter:

• gai: str (required)

Code-Example:

asset_id = eliona.GetAssetIDByGAI("K86_WP01")

---

GetAll

Description: Retrieves all data points of a specific subtype for a list of assets and returns them as a dictionary (GAI → data).

Parameter:

• ids: list[int|str] (required) – e.g., ["K86_WP01", 1073]

• subtype: str (required)

Code-Example:

all_data = eliona.GetAll(["K86_WP01"], "input")

---

SQLQuery

Description: Executes an arbitrary SQL query and returns the results as a list of tuples.

Parameter:

• query: str (required)

Code-Example:

res = eliona.SQLQuery(
    "SELECT data->>'Aussentemperatur' "
    "FROM public.heap "
    "WHERE gai='K86_WP01' AND subtype='input' "
    "ORDER BY timestamp DESC LIMIT 1;"
)

Example-Output:

[('23.5',)]

---

MakeSource

Description: Generates a source string (z. B. "ssr:123") for heap operations based on the function ID.

Parameter:

• id: int (required)

Code-Example:

src = eliona.MakeSource(id)

---

OpenFile

Description: Opens a file in the Eliona working directory (analogous to Python’s open()).

Parameter:

• name: str (required) – Filename, e.g., "log.txt"

• mode: str (required) – e.g., "w", "r"

Code-Example:

f = eliona.OpenFile("aussentemp_log.txt", "w")
f.write("Temperatur: 23.5\n")
f.close()

---

AddAssetTag

Description: Adds a tag to an asset if it is not yet assigned.

Parameter:

• gai: str (required)

• tag: str (required)

Code-Example:

eliona.AddAssetTag("K86_WP01", "has-aussentemp")

---

RemoveAssetTag

Description: Removes a tag from an asset, if present.

Parameter:

• gai: str (required)

• tag: str (required)

Code-Example:

eliona.RemoveAssetTag("K86_WP01", "has-aussentemp")

---

GetTrendRecords

Description: Loads a list of historical measured values (TrendRecord(ts, value)) from the specified start time.

Parameter:

• asset_id: int (required)

• subtype: str (required)

• attr: str (required)

• begin: datetime (required)

Code-Example:

from datetime import datetime, timedelta

aid   = eliona.GetAssetIDByGAI("K86_WP01")
begin = datetime.now() - timedelta(hours=1)
records = eliona.GetTrendRecords(aid, "input", "Aussentemperatur", begin)

---

WriteTrendRecords

Description: Writes one or more TrendRecord data sets to the system.

Parameter:

• asset_id: int (required)

• subtype: str (required)

• attr: str (required)

• recs: TrendRecord | list[TrendRecord] (required)

Code-Example:

from datetime import datetime, timedelta
import eliona_types

aid = eliona.GetAssetIDByGAI("K86_WP01")
now = datetime.now()
r1  = eliona_types.TrendRecord(now, 18.7)
r2  = eliona_types.TrendRecord(now + timedelta(minutes=30), 19.2)
eliona.WriteTrendRecords(aid, "input", "Aussentemperatur", [r1, r2])

---

GetAggregate

Description: Calculates aggregated values (Average, Sum, cumulative sum) in a fixed raster over a period.

Parameter:

• agg: Aggregate (required) – e.g., eliona_types.Aggregate.Avg

• asset_id: int (required)

• subtype: str (required)

• attr: str (required)

• raster: str (required) – ISO 8601-Duration, e.g., "PT1H" for hourly

• start: datetime (required)

• end: datetime (optional, default: now())

• recs: TrendRecord | list[TrendRecord] (required)

Code-Example:

from datetime import datetime, timedelta
import eliona_types

aid   = eliona.GetAssetIDByGAI("K86_WP01")
end   = datetime.now()
start = end - timedelta(days=1)
aggs  = eliona.GetAggregate(
    eliona_types.Aggregate.Avg,
    aid, "input", "Aussentemperatur",
    "PT1H", start, end
)

Example-Output-Struktur:

[
  TrendAggregate(ts=<datetime>, cnt=<int>, avg=<float>,
                 sum=<float>, first=<float>,
                 min=<float>, max=<float>, last=<float>),
  ...
]

---

GetLastAggregate

Description: Returns the latest cumulative aggregate in the selected raster (including last_ts).

Parameter:

• agg: Aggregate (required)

• asset_id: int (required)

• subtype: str (required)

• attr: str (required)

• raster: str (required)

Code-Example:

import eliona_types

aid  = eliona.GetAssetIDByGAI("K86_WP01")
last = eliona.GetLastAggregate(
    eliona_types.Aggregate.Cusum,
    aid, "input", "Aussentemperatur", "P1D"
)

Example-Output-Struktur:

RunningAggregate(
  ts=<datetime>, cnt=<int>, avg=<float>,
  sum=<float>, first=<float>,
  min=<float>, max=<float>, last=<float>,
  last_ts=<datetime>
)

---

Simple Script for Data Manipulation

This example shows how a script generates a random value and writes it to the heap of an asset:

def UserFunction(id, eliona):
    import random

    data = {
        "power_val": random.randint(10, 100)
    }

    eliona.SetHeap("TestInactIn", "input", data, eliona.MakeSource(id))

Description:

• The script imports the random module.

• A dictionary data is created that contains a key power_val, to which a random value between 10 and 100 is assigned.

• The generated value is written to the heap of the TestInactIn asset using the SetHeap method.

---

Get and Modify Data

In this example, the value of an attribute is retrieved from one asset, modified, and saved in another asset:

def UserFunction(id, eliona):
    from datetime import datetime

    heap = eliona.GetHeap("TestInactIn", "input")

    data = {
        "energy_val": heap["power_val"] * 3
    }

    eliona.SetHeap("TestInactIn", "input", data, eliona.MakeSource(id))

Description:

• The script imports the datetime module.

• The current value of the power_val attribute is retrieved from the heap of the TestInactIn asset.

• This value is tripled and written to the same asset as energy_val.

---

Advanced Functions and SQL Queries

With the SQLQuery method, any SQL queries can be executed directly from the script:

def UserFunction(id, eliona):
    result = eliona.SQLQuery("SELECT * FROM public.asset WHERE gai = 'TestInactIn';")
    # Weitere Verarbeitung des Ergebnisses

Description:

• The SQL query retrieves all information about the asset with the GAI TestInactIn.

• The result can then be processed further.