Hi everyone,
I'm a beginner with this so any help would make me happy lol. I imported this project and im trying to view the alarms. However, Every time, I go to press the button. I get this error.
I checked if the key data exists in the dictionary and it doesn't. not sure what to do now.
Any help would be much appreciated.
Here is the error code:
Traceback (most recent call last):
File "<event:actionPerformed>", line 222, in doLotsOfStuff
File "<event:actionPerformed>", line 49, in call
File "<event:actionPerformed>", line 18, in filter_by_udt_tag
KeyError: 'data'
at java.base/java.util.concurrent.FutureTask.report(Unknown Source)
at java.base/java.util.concurrent.FutureTask.get(Unknown Source)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.base/java.lang.reflect.Method.invoke(Unknown Source)
java.util.concurrent.ExecutionException: java.util.concurrent.ExecutionException: Traceback (most recent call last):
File "<event:actionPerformed>", line 49, in call
File "<event:actionPerformed>", line 18, in filter_by_udt_tag
KeyError: 'data'
at org.python.core.Py.JavaError(Py.java:547)
at org.python.core.Py.JavaError(Py.java:538)
at org.python.core.PyReflectedFunction.__call__(PyReflectedFunction.java:192)
at org.python.core.PyReflectedFunction.__call__(PyReflectedFunction.java:208)
at org.python.core.PyObject.__call__(PyObject.java:461)
at org.python.core.PyObject.__call__(PyObject.java:465)
at org.python.core.PyMethod.__call__(PyMethod.java:126)
at org.python.pycode._pyx109.doLotsOfStuff$8(<event:actionPerformed>:268)
at org.python.pycode._pyx109.call_function(<event:actionPerformed>)
at org.python.core.PyTableCode.call(PyTableCode.java:173)
at org.python.core.PyBaseCode.call(PyBaseCode.java:306)
at org.python.core.PyFunction.function___call__(PyFunction.java:474)
at org.python.core.PyFunction.__call__(PyFunction.java:469)
at org.python.core.PyFunction.__call__(PyFunction.java:464)
at com.inductiveautomation.ignition.common.script.ScriptManager.runFunction(ScriptManager.java:846)
at com.inductiveautomation.ignition.client.script.DesignerSystemUtilities.lambda$_invokeAsyncImpl$1(DesignerSystemUtilities.java:140)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.util.concurrent.ExecutionException: Traceback (most recent call last):
File "<event:actionPerformed>", line 49, in call
File "<event:actionPerformed>", line 18, in filter_by_udt_tag
KeyError: 'data'
at java.base/java.util.concurrent.FutureTask.report(Unknown Source)
at java.base/java.util.concurrent.FutureTask.get(Unknown Source)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.base/java.lang.reflect.Method.invoke(Unknown Source)
at org.python.core.PyReflectedFunction.__call__(PyReflectedFunction.java:190)
... 14 more
Caused by: Traceback (most recent call last):
File "<event:actionPerformed>", line 49, in call
File "<event:actionPerformed>", line 18, in filter_by_udt_tag
KeyError: 'data'
at org.python.core.Py.KeyError(Py.java:226)
at org.python.core.PyObject.__getitem__(PyObject.java:719)
at org.python.pycode._pyx109.filter_by_udt_tag$1(<event:actionPerformed>:19)
at org.python.pycode._pyx109.call_function(<event:actionPerformed>)
at org.python.core.PyTableCode.call(PyTableCode.java:173)
at org.python.core.PyBaseCode.call(PyBaseCode.java:150)
at org.python.core.PyFunction.__call__(PyFunction.java:426)
at org.python.pycode._pyx109.call$4(<event:actionPerformed>:50)
at org.python.pycode._pyx109.call_function(<event:actionPerformed>)
at org.python.core.PyTableCode.call(PyTableCode.java:173)
at org.python.core.PyBaseCode.call(PyBaseCode.java:306)
at org.python.core.PyBaseCode.call(PyBaseCode.java:197)
at org.python.core.PyFunction.__call__(PyFunction.java:485)
at org.python.core.PyMethod.instancemethod___call__(PyMethod.java:237)
at org.python.core.PyMethod.__call__(PyMethod.java:228)
at org.python.core.PyMethod.__call__(PyMethod.java:218)
at org.python.core.PyMethod.__call__(PyMethod.java:213)
at org.python.core.PyObject._jcallexc(PyObject.java:3565)
at org.python.proxies.__builtin__$APICallable$18.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
... 1 more
Ignition v8.1.31 (b2023081007)
Java: Azul Systems, Inc. 11.0.18
and here is the script:
import time
import sys
from urllib2 import urlopen
from datetime import datetime
from java.util.concurrent import Executors, Callable
from java.text import SimpleDateFormat
def filter_by_udt_tag(data, udt_tag):
"""
Filters the data objects based on the UDT_TAG key.
Parameters:
data (dict): A dictionary containing a list of data objects.
udt_tag (str): The UDT_TAG value to filter by.
Returns:
list: A list of filtered data objects.
"""
filtered_data = [item for item in data["data"] if item.get("udt_tag") == udt_tag]
return filtered_data
class APICallable(Callable):
"""
A class that represents an API callable object.
This class encapsulates the arguments required for an API call and provides a method to execute the call.
Attributes:
args (dict): The arguments required for the API call.
Methods:
call():
Sends an API request with the stored arguments.
Returns the response from the API request.
"""
def __init__(self, args):
"""
Initializes the APICallable object with the given arguments.
Args:
args (dict): A dictionary of arguments required for the API call.
"""
self.args = args
def call(self):
"""
Sends an API request using the stored arguments.
Returns:
Response: The response from the API request.
"""
response = send_api_request(self.args)
filtered_response = filter_by_udt_tag(response, self.args[3])
return filtered_response
def send_api_request(args):
"""
Sends an API request with the given arguments.
This function extracts the start date, end date, and whid from the arguments,
formats the dates into ISO 8601 format, and makes a common API request.
Args:
args (tuple): A tuple containing the start date (str), end date (str), and whid (str).
Returns:
Response: The response from the API request.
If an exception occurs, returns a dictionary with the error message.
"""
start_date, end_date, whid, udt = args
try:
iso8601_dates = [start_date, end_date]
response = make_api_request_common(iso8601_dates, None, whid, udt)
return response
except Exception as e:
return {"error": str(e)}
def update_ui():
"""
Update the user interface with specific flags and messages.
"""
event.source.parent.computationInProgress = False
event.source.parent.update = False
def make_api_request_common(iso8601_dates, params=None, whid=None, udt=None):
"""
Make a common API request for historical alarms based on the specified timestamp range and parameters.
Args:
- iso8601_dates (list): A list containing the start and end timestamps in ISO8601 format.
- params (str, optional): A string representing the next token for paginated requests (default is None).
- whid (str, optional): A string representing the warehouse ID (default is None).
Returns:
dict: A dictionary containing the API response for historical alarms.
Note:
This function handles paginated requests and extends the response data if a 'next_token' is present.
"""
mapRegion = Global.iStream.gatewayRegion.get("FE")
if mapRegion == "AU" or mapRegion == "AU-NAWS":
default_region = "ap-southeast-2"
default_aws_service = "execute-api"
default_host = "https://sap-service-au.sap.rme.amazon.dev/"
else:
default_region = "eu-west-1"
default_aws_service = "execute-api"
default_host = "https://sap-service-eu.sap.rme.amazon.dev/"
# Get AWS credentials
access_key, secret_key = shared.EURME_SCADA.SAP.sap_scripts.retrieve_ssm()
json_data = {
"whid": whid,
"engine": "SCADA",
"t_start": iso8601_dates[0],
"t_end": iso8601_dates[1],
"udt_tag": udt,
}
if params is not None:
json_data["next_token"] = params
api_request = Global.iStream.aws4auth.IamRequest(
url=default_host,
path="data/historicalAlarms",
json=json_data,
access_key=access_key,
secret_key=secret_key,
aws_region=default_region,
aws_service=default_aws_service,
)
api_response = system.util.jsonDecode(str(urlopen(api_request).readlines()[0]))
if "next_token" in api_response:
params = api_response["next_token"]
udt = event.source.parent.tagPath
new_api_response = make_api_request_common(iso8601_dates, params, whid, udt)
api_response["data"].extend(new_api_response["data"])
return api_response
if event.source.parent.fillColors:
try:
tagColorConfig = event.source.parent.colorsConfig
for row in range(tagColorConfig.getRowCount()):
JSONString = {}
for col in range(tagColorConfig.getColumnCount()):
if not tagColorConfig.getColumnName(col) == "condition":
JSONString[str(tagColorConfig.getColumnName(col))] = str(tagColorConfig.getValueAt(row, col))
alarmDescription = tagColorConfig.getValueAt(row, 0)
if "Diagnostic" in alarmDescription:
event.source.parent.Diagnostic = JSONString
if "ActiveUnacked" in alarmDescription:
if "{priority}=1" in alarmDescription:
event.source.parent.ActiveUnackedSev1 = JSONString
if "{priority}=2" in alarmDescription:
event.source.parent.ActiveUnackedSev2 = JSONString
if "{priority}=3" in alarmDescription:
event.source.parent.ActiveUnackedSev3 = JSONString
if "{priority}=4" in alarmDescription:
event.source.parent.ActiveUnackedSev4 = JSONString
if "ClearUnacked" in alarmDescription:
if "{priority}=1" in alarmDescription:
event.source.parent.ClearUnackedSev1 = JSONString
if "{priority}=2" in alarmDescription:
event.source.parent.ClearUnackedSev2 = JSONString
if "{priority}=3" in alarmDescription:
event.source.parent.ClearUnackedSev3 = JSONString
if "{priority}=4" in alarmDescription:
event.source.parent.ClearUnackedSev4 = JSONString
if "ActiveAcked" in alarmDescription:
if "{priority}=1" in alarmDescription:
event.source.parent.ActiveAckedSev1 = JSONString
if "{priority}=2" in alarmDescription:
event.source.parent.ActiveAckedSev2 = JSONString
if "{priority}=3" in alarmDescription:
event.source.parent.ActiveAckedSev3 = JSONString
if "{priority}=4" in alarmDescription:
event.source.parent.ActiveAckedSev4 = JSONString
if "ClearAcked" in alarmDescription:
event.source.parent.ClearAckedSev4 = JSONString
if "{priority}=1" in alarmDescription:
event.source.parent.ClearAckedSev1 = JSONString
if "{priority}=2" in alarmDescription:
event.source.parent.ClearAckedSev2 = JSONString
if "{priority}=3" in alarmDescription:
event.source.parent.ClearAckedSev3 = JSONString
if "{priority}=4" in alarmDescription:
event.source.parent.ClearAckedSev4 = JSONString
except BaseException:
event.source.parent.loadData = False
exc_type, exc_obj, tb = sys.exc_info()
lineno = tb.tb_lineno
error_description = str(lineno) + " , " + str(exc_type) + " , " + str(exc_obj)
system.gui.messageBox(error_description)
def doLotsOfStuff():
if event.source.parent.update:
tags = ["Configuration/FC"]
values = system.tag.readBlocking(tags)
oWHID = values[0].value
oStartDT = event.source.parent.getComponent("StartDate").text
oStopDT = event.source.parent.getComponent("EndDate").text
iso8601_dates = shared.EURME_SCADA.SAP.sap_scripts.convert_to_iso8601(oStartDT, oStopDT)
udt = event.source.parent.tagPath
start_date = datetime.strptime(iso8601_dates[0], "%Y-%m-%dT%H:%M:%S.%fZ")
end_date = datetime.strptime(iso8601_dates[1], "%Y-%m-%dT%H:%M:%S.%fZ")
interval_minutes = 720
intervals = shared.EURME_SCADA.SAP.sap_scripts.generate_time_intervals(start_date, end_date, interval_minutes)
max_concurrent_threads = len(intervals)
args = [(interval[0], interval[1], oWHID, udt) for interval in intervals]
start_time = time.time()
executor = Executors.newFixedThreadPool(max_concurrent_threads)
futures = [executor.submit(APICallable(arg)) for arg in args]
results = [future.get() for future in futures]
executor.shutdown()
end_time = time.time()
api_response = {"data": []}
for result in results:
if isinstance(result, list):
api_response["data"].extend(result)
else:
print("Unexpected response format: {}".format(result))
print("Time taken for all API requests: {:.2f} seconds".format(end_time - start_time))
print("Number of items in combined 'data': {}".format(len(api_response["data"])))
data_sap = []
if not "No data found" in api_response["data"]:
seconds_to_hours_conv = {item["duration_millis"]: shared.EURME_SCADA.SAP.sap_scripts.seconds_to_hours(item["duration_millis"] / 1000) for item in api_response["data"]}
for item in api_response["data"]:
timestampUTC = item["time_start"]
timestamp = shared.EURME_SCADA.SAP.sap_scripts.convert_to_local_time_with_fractional_seconds(timestampUTC)
dt = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S.%f")
date_without_milliseconds = dt.strftime("%Y-%m-%d %H:%M:%S")
duration = seconds_to_hours_conv[item["duration_millis"]]
data_sap.append([date_without_milliseconds, duration, item["name"], item["priority"], item["udt_tag"], item["displaypath"]])
else:
system.gui.warningBox("No data found.")
event.source.parent.computationInProgress = False
headers_sap = ["TimeStamp", "Duration", "Name", "Priority", "UDT_tag", "DisplayPath"]
def parse_timestamp(ts):
dateFormat = SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
return dateFormat.parse(ts)
data_sap.sort(
key=lambda x: (
-x[3], # Sort by column 3 in descending order
-parse_timestamp(x[0]).getTime(), # Sort by column 0 (timestamp) in descending order, getting time in milliseconds
)
)
historical_data = system.dataset.toDataSet(headers_sap, data_sap)
event.source.parent.getComponent("table_ts").data = historical_data
system.util.invokeLater(update_ui)
if not event.source.parent.computationInProgress:
event.source.parent.computationInProgress = True
event.source.parent.runTimerForScripting = False
system.util.invokeAsynchronous(doLotsOfStuff)
Here is a screenshot of me see if the key data is present
Best wishes