Select Git revision
newmodeljsonparse.py
newmodeljsonparse.py 11.20 KiB
import json
import os
import sys
import writefile as wf
from collections import namedtuple
def measurementJsonDecod(measurementdict):
return namedtuple('Measurement', measurementdict.keys())(*measurementdict.values())
def parameterJsonDecod(parameterdict):
return namedtuple('Parameter', parameterdict.keys())(*parameterdict.values())
def functionJsonDecod(functiondict):
return namedtuple('Function', functiondict.keys())(*functiondict.values())
class Measurement:
def __init__(self, name, description, datatype, value, dimension, unit=None, range=None):
self.name = name
self.description = description
self.datatype = datatype
self.range = range
self.dimension = dimension
self.unit = unit
array = []
if self.datatype == "float":
self.value = float(value)
else:
self.value = value
class Parameter:
def __init__(self, name, description, datatype, dimension, value, rangevar=None, unit=None, default=None, constant=False):
self.name = name
self.description = description
self.datatype = datatype
# if range == [None, None]:
# self.range = None
# else:
self.range = rangevar
self.dimension = dimension
self.unit = unit
self.default = default
self.constant = constant
self.value = value
# array = []
# calculate a start value
# if self.default == False:
# if self.dimension != 0:
# for i in range(self.dimension):
# array.append(self.datatype((self.range[0] + self.range[1]) / 2))
# self.value = array
# else:
# self.value = self.datatype(((self.range[0] + self.range[1]) / 2))
# else:
# self.value = self.datatype(default)
class Function:
def __init__(self, name, description, inargsdatatypes = None, outargsdatatypes = None):
self.name = name
self.description = description
self.inargsdatatypes = inargsdatatypes
self.outargsdatatypes = outargsdatatypes
def dimensionchange(string):
dimarr = string["dimension"]
if type(dimarr) == list:
if dimarr:
value = dimarr[0]
else:
value = 0
newdimension = {"dimension": value}
string.update(newdimension)
def getdatatypes(argarray):
returnarr = []
if len(argarray) > 0:
for i in argarray:
returnarr.append(i["datatype"])
return returnarr
else:
return None
def changerange(string):
if iskeythere(string, "range"):
rangeval = string["range"]
rangetuple = (rangeval[0], rangeval[1])
string.update({"range": rangetuple})
else:
string.update({"range": (None, None)})
# print(string)
def getobjnames(argarray):
returnarr = []
if len(argarray) > 0:
for i in argarray:
name = (i["name"].lower()).replace(" ", "")
returnarr.append(name)
return returnarr
else:
return None
def getParameters(parametersDict):
# extracts the measurements of a Json-Model
parameterobj = []
if len(parametersDict) > 0:
for i in parametersDict:
obj = i
dimensionchange(obj)
obj["value"]=getdefaultvalue(obj["dimension"], obj["datatype"])
obj = json.dumps(i)
parameterobji = json.loads(obj, object_hook=parameterJsonDecod)
parameterobj.append(parameterobji)
return parameterobj
def getMeasurements(measurementsDict):
measurementobj = []
if len(measurementsDict) > 0:
for i in measurementsDict:
obj = i
dimensionchange(obj)
obj["value"] = getdefaultvalue(obj["dimension"], obj["datatype"])
changerange(obj)
if not iskeythere(obj, "unit"):
obj["unit"] = "UNITLESS"
obj = json.dumps(i)
measurementobji = json.loads(obj, object_hook=measurementJsonDecod)
measurementobj.append(measurementobji)
return measurementobj
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 = []
if len(obj["returns"]) > 0:
returns = searchlists(obj["returns"], jsonmeasurements) + searchlists(obj["returns"], jsonparameters)
obj["inargsnames"] = getobjnames(arguments)
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)
return functionobj
def searchlists(list1, list2):
print("I am searching")
objmeasurementlist = []
if len(list1) <= 0:
return []
for input in list1:
for searchobj in list2:
if input["value"] == searchobj["uuid"]:
objmeasurementlist.append(searchobj)
return objmeasurementlist
def getdefaultvalue(dimension, datatype):
if datatype == "float" or datatype == "double" or datatype == "int":
if dimension > 1:
value = []
for i in range(dimension):
value.append(0)
return value
else:
return 0
else:
return True
def iskeythere(list, key):
for keysearch in list:
if(keysearch == key):
return True
return False
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"])
def getInterfaceparam(list1, interface, searched, jsoncomponents, searchinglist):
for inputsearch in interface[searched]:
for searchobj in searchinglist:
if inputsearch["value"] == searchobj["uuid"]:
list1.append(searchobj)
if iskeythere(list(interface.keys()), "baseComponent"):
for component in jsoncomponents:
if component["uuid"] == interface["baseComponent"]:
return getInterfaceparam(list1, component, searched, jsoncomponents, searchinglist)
else:
return list1
def buildComponents(interfacesreturn, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, objcomponentsclone):
for obj in jsoninterfaces:
rootobj = {}
if obj["elementType"] == "interface":
for search in jsoncomponents:
if obj["rootComponent"] == search["uuid"]:
rootobj = search
else:
rootobj = obj
if obj["name"] in list(interfacesreturn.keys()):
continue
objcomponentslist = getInterfaceparam([], rootobj, "components", jsoncomponents, jsoncomponents)
if obj["name"] not in interfacesreturn.keys():
interfacesreturn[str(obj["name"])] = wf.openobject(obj["name"], rootobj["description"])
objmeasurementlist = getInterfaceparam([], rootobj, "measurements", jsoncomponents, jsonmeasurements)
print(objmeasurementlist)
measurementsobjs = getMeasurements(objmeasurementlist)
wf.writemeasurement(measurementsobjs, interfacesreturn[str(obj["name"])])
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
def instantiateComponents(typesdict, interfaces, jsoncomponents, headnode, instantiatedict):
count = 0
countcomponent = 0
for interface in interfaces:
root = interface
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)
if len(components) > 0:
instantiateComponents(typesdict,components , jsoncomponents, interface, instantiatedict)
return instantiatedict
def instantiate(typesdict, interfaces, jsoncomponents):
instantiatedict = {}
for interface in interfaces:
root = {}
count = 0
for component in jsoncomponents:
if interface["rootComponent"] == component["uuid"]:
root = component
components = getInterfaceparam([], root, "components", jsoncomponents, jsoncomponents)
instantiatedict[str(interface["name"])] = wf.writeinterface(typesdict[str(interface["name"])], interface["name"], count)
instantiateComponents(typesdict, components, jsoncomponents, interface, instantiatedict)
def main():
try:
JSON_MODEL = sys.argv[1][:-5]
if not os.path.exists(os.path.join('..', 'soil-opc-ua', f'{JSON_MODEL}.json')):
raise Exception()
except Exception as e:
print('You must provide a file from the folder to be used as root file of the parsing, e.g., "python jsonparser.py model.json".')
exit()
with open(f'{JSON_MODEL}.json') as file:
wf.writebeginning()
jsonobj = json.loads(file.read())
elements = jsonobj["elements"]
if len(elements) <= 0:
exit()
jsonmeasurements = []
jsonparameters = []
jsonfunctions = []
jsoncomponents = []
jsoninterfaces = []
for obj in elements:
#print(obj)
if obj["elementType"] == "measurement":
jsonmeasurements.append(obj)
elif obj["elementType"] == "parameter":
jsonparameters.append(obj)
elif obj["elementType"] == "function":
jsonfunctions.append(obj)
elif obj["elementType"] == "component":
jsoncomponents.append(obj)
elif obj["elementType"] == "interface":
jsoninterfaces.append(obj)
typesdict = buildComponents({}, jsoninterfaces, jsoncomponents, jsonmeasurements, jsonparameters, jsonfunctions, [])
intantiateddict = instantiate(typesdict, jsoninterfaces, jsoncomponents)
wf.writend()
if __name__ == '__main__':
main()