From 8dd8ed3a2deae70ceed1a3452491d4dbf2063ddf Mon Sep 17 00:00:00 2001
From: Susa Giogoli <su.giogoli@icloud.com>
Date: Mon, 10 Jul 2023 14:38:41 +0200
Subject: [PATCH] events can be generated on robot.json
---
newmodeljsonparse.py | 76 ++++++++++++++++++++++++++++++--------------
writefile.py | 50 ++++++++++++++++++++++++++---
2 files changed, 98 insertions(+), 28 deletions(-)
diff --git a/newmodeljsonparse.py b/newmodeljsonparse.py
index 5daea21..20cb058 100644
--- a/newmodeljsonparse.py
+++ b/newmodeljsonparse.py
@@ -4,14 +4,17 @@ import sys
import writefile as wf
from collections import namedtuple
+
#creates the constructor for the measurement-object with corresponding parameters
def measurementJsonDecod(measurementdict):
return namedtuple('Measurement', measurementdict.keys())(*measurementdict.values())
+
#creates the constructor for the parameter-object with corresponding parameters
def parameterJsonDecod(parameterdict):
return namedtuple('Parameter', parameterdict.keys())(*parameterdict.values())
+
#creates the constructor for the function-object with corresponding parameters
def functionJsonDecod(functiondict):
return namedtuple('Function', functiondict.keys())(*functiondict.values())
@@ -32,6 +35,7 @@ class Measurement:
else:
self.value = value
+
#Parameter class that can be instantiated using JsonDecod functions
class Parameter:
def __init__(self, name, description, datatype, dimension, value, rangevar=None, unit=None, default=None, constant=False):
@@ -60,8 +64,6 @@ class Parameter:
# self.value = self.datatype(default)
-
-
#Function class that can be instantiated using JsonDecod functions
class Function:
def __init__(self, name, description, inargsdatatypes = None, outargsdatatypes = None):
@@ -71,7 +73,6 @@ class Function:
self.outargsdatatypes = outargsdatatypes
-
#switches the dimension into the right format
def dimensionchange(string):
dimarr = string["dimension"]
@@ -84,6 +85,7 @@ def dimensionchange(string):
string.update(newdimension)
return string
+
#gets the datatypes of functions to set for opcua
def getdatatypes(argarray):
returnarr = []
@@ -94,6 +96,7 @@ def getdatatypes(argarray):
else:
return None
+
#switches range into the right format
def changerange(obj):
if iskeythere(obj, "range"):
@@ -103,11 +106,13 @@ def changerange(obj):
obj["range"] = [None, None]
return obj
+
def getunit(string):
if not iskeythere(string, "unit"):
string["unit"] = "UNITLESS"
return string
+
#gets the objectnames for arguments of the function
def getobjnames(argarray):
returnarr = []
@@ -119,6 +124,7 @@ def getobjnames(argarray):
else:
return None
+
#returns a list of all the parameters as objects
def getParameters(parametersDict):
# extracts the measurements of a Json-Mode
@@ -133,10 +139,9 @@ def getParameters(parametersDict):
obj = json.dumps(obj)
parameterobji = json.loads(obj, object_hook=parameterJsonDecod)
parameterobj.append(parameterobji)
- print("parameterobj")
- print(parameterobj)
return parameterobj
+
#returns a list of all the measurements as objects
def getMeasurements(measurementsDict):
measurementobj = []
@@ -153,13 +158,13 @@ def getMeasurements(measurementsDict):
measurementobj.append(measurementobji)
return measurementobj
+
#returns a list of all the functions as objects
def getFunctions(functionsDict, jsonmeasurements, jsonparameters):
functionobj = []
if len(functionsDict) > 0:
for i in functionsDict:
obj = i
- # del obj["value"]
arguments = searchlists(obj["arguments"], jsonmeasurements)
arguments = arguments + searchlists(obj["arguments"], jsonparameters)
returns = []
@@ -169,9 +174,6 @@ def getFunctions(functionsDict, jsonmeasurements, jsonparameters):
obj["outargsnames"] = getobjnames(returns)
obj["inargsdatatypes"] = getdatatypes(arguments)
obj["outargsdatatypes"] = getdatatypes(returns)
- #del obj["uuid"]
- #del obj["arguments"]
- #del obj["returns"]
obj = json.dumps(obj)
functionobji = json.loads(obj, object_hook=functionJsonDecod)
functionobj.append(functionobji)
@@ -189,6 +191,7 @@ def searchlists(list1, list2):
objmeasurementlist.append(searchobj)
return objmeasurementlist
+
def getdefaultvalue(obj):
if iskeythere(obj, "default"):
if obj["datatype"] == "string" or obj["datatype"] == "enum":
@@ -214,7 +217,7 @@ def getdefaultvalue(obj):
def iskeythere(list, key):
for keysearch in list:
- if(keysearch == key):
+ if keysearch == key:
return True
return False
@@ -223,7 +226,7 @@ def iteratingInterface(obj, dict, componentslist):
if iskeythere(list(obj.keys()), "baseComponent"):
for searchobj in componentslist:
if searchobj["uuid"] == obj["baseComponent"]:
- wf.writeinterface(componentslist[searchobj["name"]], searchobj["name"])
+ wf.writeinterface(componentslist[searchobj["name"]], searchobj["name"], 0)
def getInterfaceparam(list1, interface, searched, jsoncomponents, searchinglist):
@@ -234,7 +237,18 @@ def getInterfaceparam(list1, interface, searched, jsoncomponents, searchinglist)
if iskeythere(list(interface.keys()), "baseComponent"):
for component in jsoncomponents:
if component["uuid"] == interface["baseComponent"]:
- return getInterfaceparam(list1, component, searched, jsoncomponents, searchinglist)
+ return getInterfaceparam(list1, component, searched, jsoncomponents, searchinglist)
+ else:
+ return list1
+
+
+def getVariables(list1, interface, jsoncomponents):
+ variablelist = interface["parameters"] + interface["measurements"]
+ list1 = list1 + variablelist
+ if iskeythere(list(interface.keys()), "baseComponent"):
+ for component in jsoncomponents:
+ if component["uuid"] == interface["baseComponent"]:
+ return getVariables(list1, component, jsoncomponents)
else:
return list1
@@ -250,8 +264,7 @@ def getEvents(list1, component, jsoncomponents):
return list1
-
-def buildComponents(interfacesreturn, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, objcomponentsclone):
+def buildComponents(interfacesreturn, variabledict, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, objcomponentsclone):
for obj in jsoninterfaces:
rootobj = {}
if obj["elementType"] == "interface":
@@ -269,14 +282,16 @@ def buildComponents(interfacesreturn, jsoninterfaces, jsoncomponents, jsonmeasur
#print(objmeasurementlist)
measurementsobjs = getMeasurements(objmeasurementlist)
wf.writemeasurement(measurementsobjs, interfacesreturn[str(obj["name"])])
+ variablelist = getVariables([], rootobj, jsoncomponents)
+ variabledict[obj["uuid"]] = variablelist
objparameterlist = getInterfaceparam([], rootobj, "parameters", jsoncomponents, jsonparameters)
parameterobjs = getParameters(objparameterlist)
wf.writeparameter(parameterobjs, interfacesreturn[str(obj["name"])])
objfunctionlist = getInterfaceparam([], rootobj, "functions", jsoncomponents, jsonfunctions)
functionobj = getFunctions(objfunctionlist, jsonparameters, jsonmeasurements)
wf.writefunction(functionobj, interfacesreturn[str(obj["name"])])
- buildComponents(interfacesreturn, objcomponentslist, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, objcomponentsclone)
- return interfacesreturn
+ buildComponents(interfacesreturn, variabledict, objcomponentslist, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, objcomponentsclone)
+ return interfacesreturn, variabledict
def instantiateComponents(typesdict, interfaces, jsoncomponents, headnode, instantiatedict, eventsdict):
@@ -288,10 +303,10 @@ def instantiateComponents(typesdict, interfaces, jsoncomponents, headnode, insta
components = getInterfaceparam([], root, "components", jsoncomponents, jsoncomponents)
if interface["name"] in list(instantiatedict.keys()):
count += 1
- instantiatedict[str(interface["name"])] = wf.writeobjects(typesdict[str(interface["name"])], interface["name"], instantiatedict[str(headnode["name"])], count)
- eventsdict[str(instantiatedict[str(interface["name"])])] = events
+ instantiatedict[str(interface["uuid"])] = wf.writeobjects(typesdict[str(interface["name"])], interface["name"], instantiatedict[str(headnode["uuid"])], count)
+ eventsdict[str(interface["uuid"])] = events
if len(components) > 0:
- instantiateComponents(typesdict,components , jsoncomponents, interface, instantiatedict, eventsdict)
+ instantiateComponents(typesdict, components, jsoncomponents, interface, instantiatedict, eventsdict)
return instantiatedict, eventsdict
@@ -307,8 +322,8 @@ def instantiate(typesdict, interfaces, jsoncomponents):
root = component
events = getEvents([], root, jsoncomponents)
components = getInterfaceparam([], root, "components", jsoncomponents, jsoncomponents)
- instantiatedict[str(interface["name"])] = wf.writeinterface(typesdict[str(interface["name"])], interface["name"], count)
- eventsdict[str(instantiatedict[str(interface["name"])])] = events
+ instantiatedict[str(interface["uuid"])] = wf.writeinterface(typesdict[str(interface["name"])], interface["name"], count)
+ eventsdict[str(interface["uuid"])] = events
return instantiateComponents(typesdict, components, jsoncomponents, interface, instantiatedict, eventsdict)
@@ -349,13 +364,26 @@ def main():
elif obj["elementType"] == "enum":
jsonenums.append(obj)
- typesdict = buildComponents({}, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, [])
+ returnComponents = buildComponents({}, {}, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, [])
+ typesdict = returnComponents[0]
+ variabledict = returnComponents[1]
+ print("Variabledict: ")
+ print(variabledict)
returnval = instantiate(typesdict, jsoninterfaces, jsoncomponents)
instantiatedict = returnval[0]
eventsdict = returnval[1]
+ print("Instantiatedict: ")
+ print(instantiatedict)
+ print("Eventsdict: ")
print(eventsdict)
-
- wf.writend()
+ generatordict = {}
+ for key in list(eventsdict.keys()):
+ eventlist = wf.writeevents(eventsdict[key], instantiatedict[key], {}, {})
+ generatordict.update(eventlist)
+ wf.writendpt1()
+ for key in list(eventsdict.keys()):
+ wf.triggerevents(generatordict, eventsdict[key], instantiatedict[key], variabledict[key], jsonmeasurements, jsonparameters)
+ wf.writendpt2()
if __name__ == '__main__':
diff --git a/writefile.py b/writefile.py
index 00a9427..2fda739 100644
--- a/writefile.py
+++ b/writefile.py
@@ -200,9 +200,19 @@ def writeinterface(objtype, objname, count):
return objtypename
-def writeevents(events, component):
+def writeevents(events, component, eventlist, eventgenlist):
server = open("opcua-server.py", "a")
- server.write(")
+ writeindent = " "
+ #eventelements = []
+ for event in events:
+ #eventelements.append(event["element"])
+ eventname = (event["element"].lower()).replace(" ", "") + component + str(event["severity"])
+ eventgenname = eventname + "generator"
+ eventlist[eventname] = eventgenname
+ #print(eventname)
+ server.write(writeindent + eventname + " = await server.create_custom_event_type(idx, \"" + str(event["element"]) + " " + str(event["severity"]) + "\", ua.ObjectIds.SystemEventType)\n")
+ server.write(writeindent + str(eventlist[eventname]) + " = await server.get_event_generator(" + eventname + ", " + component + ")\n")
+ return eventlist
def writendpt1():
@@ -212,9 +222,40 @@ def writendpt1():
print("Server is running!")
count = 0
- while True:
- """)
+ while True:\n""")
+
+def triggerevents(generatordict, eventslist, component, variablelist, jsonmeasurements, jsonparameters):
+ server = open("opcua-server.py", "a")
+ indent = 3
+ print(variablelist)
+ for event in eventslist:
+ eventsearchname = ""
+ for variable in variablelist:
+ if variable["name"] == event["element"]:
+ for measurement in jsonmeasurements:
+ if variable["value"] == measurement["uuid"]:
+ eventsearchname = measurement["name"]
+ for parameter in jsonparameters:
+ if variable["value"] == parameter["uuid"]:
+ eventsearchname = parameter["name"]
+
+ writeindent = ""
+ for i in range(indent):
+ writeindent += " "
+ dictname = (event["element"].lower()).replace(" ", "") + component + event["severity"]
+ generatorname = generatordict[dictname]
+ valuename = event["element"] + component
+ trigger = str(event["target"])
+ if type(event["target"]) == str:
+ trigger = "\"" + str(event["target"] + "\"")
+ server.write(writeindent + valuename + "= await " + component + ".get_child([\"2:" + eventsearchname + "\"])\n")
+ server.write(writeindent + "if (await " + valuename + ".get_value()) " + event["trigger"] + " " + trigger + ":\n")
+ writeindent += " "
+ server.write(writeindent + "await " + generatorname+ ".trigger(message=\"" + event["message"] + "\")\n")
+ writeindent = ""
+ for i in range(indent):
+ writeindent += " "
@@ -222,6 +263,7 @@ def writendpt2():
server = open("opcua-server.py", "a")
server.write("""
await asyncio.sleep(1)
+ count += 1
if __name__ == '__main__':
asyncio.run(main())""")
--
GitLab