Skip to content
Snippets Groups Projects
Commit 980cb97e authored by Susanna Weber's avatar Susanna Weber
Browse files

generating server file with components with errors

parent d68f28dd
Branches
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
jsonparse.py
+ 64
− 53
View file @ 980cb97e
......@@ -118,6 +118,63 @@ def getobjnames(argarray):
else:
return None
def createcomponent(jsonobj, iscomponent):
measurements = jsonobj["measurements"]
measurementobj = []
if len(measurements) > 0:
for i in measurements:
obj = i
# del obj["value"]
del obj["uuid"]
dimensionchange(obj)
# if i["range"]:
changerange(obj)
obj = json.dumps(obj)
measurementobji = json.loads(obj, object_hook=measurementJsonDecod)
measurementobj.append(measurementobji)
print(measurementobj)
parameters = jsonobj["parameters"]
parameterobj = []
if len(parameters) > 0:
for i in parameters:
obj = i
# del obj["value"]
del obj["uuid"]
dimensionchange(obj)
if i["range"]:
changerange(obj)
obj = json.dumps(obj)
parameterobji = json.loads(obj, object_hook=parameterJsonDecod)
parameterobj.append(parameterobji)
functions = jsonobj["functions"]
functionobj = []
if len(functions) > 0:
for i in functions:
obj = i
# del obj["value"]
obj["inargsnames"] = getobjnames(obj["arguments"])
obj["outargsnames"] = getobjnames(obj["returns"])
obj["inargsdatatypes"] = getdatatypes(obj["arguments"])
obj["outargsdatatypes"] = getdatatypes(obj["returns"])
del obj["uuid"]
del obj["arguments"]
del obj["returns"]
obj = json.dumps(obj)
functionobji = json.loads(obj, object_hook=functionJsonDecod)
functionobj.append(functionobji)
print(functionobj)
print(parameterobj)
# exec(open("writefile.py").read())
wf.writeinserver(jsonobj["name"], jsonobj["description"], measurementobj, parameterobj, functionobj, iscomponent)
def main():
try:
JSON_MODEL = sys.argv[1][:-5]
......@@ -131,59 +188,13 @@ def main():
#robot = json.dumps(file.read())
robotload = json.loads(file.read())
measurements = robotload["measurements"]
if len(measurements) > 0:
measurementobj = []
for i in measurements:
obj = i
#del obj["value"]
del obj["uuid"]
dimensionchange(obj)
#if i["range"]:
changerange(obj)
obj = json.dumps(obj)
measurementobji = json.loads(obj, object_hook=measurementJsonDecod)
measurementobj.append(measurementobji)
print(measurementobj)
parameters = robotload["parameters"]
if len(parameters) > 0:
parameterobj = []
for i in parameters:
obj = i
# del obj["value"]
del obj["uuid"]
dimensionchange(obj)
if i["range"]:
changerange(obj)
obj = json.dumps(obj)
parameterobji = json.loads(obj, object_hook=parameterJsonDecod)
parameterobj.append(parameterobji)
functions = robotload["functions"]
if len(functions) > 0:
functionobj = []
for i in functions:
obj = i
# del obj["value"]
obj["inargsnames"] = getobjnames(obj["arguments"])
obj["outargsnames"] = getobjnames(obj["returns"])
obj["inargsdatatypes"] = getdatatypes(obj["arguments"])
obj["outargsdatatypes"] = getdatatypes(obj["returns"])
del obj["uuid"]
del obj["arguments"]
del obj["returns"]
obj = json.dumps(obj)
functionobji = json.loads(obj, object_hook=functionJsonDecod)
functionobj.append(functionobji)
print(functionobj)
print(parameterobj)
#exec(open("writefile.py").read())
wf.write(measurementobj, parameterobj, functionobj)
wf.writebeginning()
createcomponent(robotload, False)
for i in robotload["components"]:
createcomponent(i, True)
wf.writend()
......
writefile.py
+ 82
− 59
View file @ 980cb97e
......@@ -5,12 +5,14 @@ def defFunction(functionname, file, indent, inargs):
writeindent = ""
for i in range(indent):
writeindent = writeindent + " "
file.write(writeindent + "async def " + functionname + "(parent, ")
for i in range (len(inargs)):
if i < (len(inargs) - 1):
file.write(inargs[i] + ", ")
else:
file.write(inargs[i])
file.write(writeindent + "async def " + functionname + "(parent")
if inargs is not None:
file.write(", ")
for i in range (len(inargs)):
if i < (len(inargs) - 1):
file.write(inargs[i] + ", ")
else:
file.write(inargs[i])
file.write("):\n")
writeindent = writeindent + " "
file.write(writeindent + """#ToDo: Write an algorithm that fits the description of the function.
......@@ -24,48 +26,16 @@ def getVariantType(datatype):
elif datatype == "Boolean":
varianttype = "ua.VariantType.Boolean"
return varianttype
def write(measurementobj, parameterobj, functionobj):
if os.path.exists("opcua-server.py"):
os.remove("opcua-server.py")
def writeinserver(objname, objdescription, measurementobj, parameterobj, functionobj, iscomponent):
server = open("opcua-server.py", "a")
server.write("""import logging
import asyncio
from asyncua import ua, Server
from asyncua.common.methods import uamethod
from asyncua.common.structures104 import new_struct, new_enum, new_struct_field
from asyncua.common import node
logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger('asyncua')
""")
server.write("\nasync def main():")
#äinten = "\tab"
server.write("""
# setup our server
server = Server()
await server.init()
server.set_endpoint('opc.tcp://0.0.0.0:4840/freeopcua/server/')
# setup our own namespace, not really necessary but should as spec
uri = 'http://examples.freeopcua.github.io/'
idx = await server.register_namespace(uri)
#id of rangetype
rangeid = await server.nodes.base_data_type.get_child(["0:Structure", "0:Range"])
#creating the objecttype
myobjtype = await server.nodes.base_object_type.add_object_type(idx, "Robot")
#creating dataypes for parameters and measurements
measurementtype = await server.nodes.base_data_type.add_data_type(idx, "Measurement")
parametertype = await server.nodes.base_data_type.add_data_type(idx, "Parameter")
""")
indent = 1
writeindent = ""
for j in range(indent):
writeindent = writeindent + " "
server.write(writeindent + "#creating the objecttype\n")
objtypename = (objname.lower()).replace(" ", "") + "type"
server.write(writeindent + objtypename + "= await server.nodes.base_object_type.add_object_type(idx, \"" + objname + "\" )\n")
server.write("\n" + writeindent + "# creating all measurements with their properterties\n")
......@@ -75,18 +45,23 @@ _logger = logging.getLogger('asyncua')
datatype = "measurementtype.nodeid"
varname = ((i.name).lower()).replace(" ", "") + "var"
measurementdict[i.name] = varname
server.write(writeindent + varname + " = " + "await myobjtype.add_variable(idx, \"" + str(i.name) + "\", " + str(i.value) + ", " + varianttype + ", " + datatype + ")\n")
server.write(writeindent + varname + " = " + "await " + objtypename + ".add_variable(idx, \"" + str(i.name) + "\", " + str(i.value) + ", " + varianttype + ", " + datatype + ")\n")
server.write(writeindent + "await " + varname + "." + "set_modelling_rule(True)\n")
if i.range is not None and i.range is not [None, None]:
headnode = ""
if iscomponent:
headnode = objtypename
else:
headnode = varname
if i.range is not None and i.range is not "(None, None)":
rangename = "rangeprop"
server.write(writeindent + rangename + " = await " + varname + ".add_property(idx, \"range\", ua.Range(" + str(i.range[0]) + "," + str(i.range[1]) + "))\n")
server.write(writeindent + rangename + " = await " + headnode + ".add_property(idx, \"range\", ua.Range(" + str(i.range[0]) + "," + str(i.range[1]) + "))\n")
server.write(writeindent + "await " + rangename + ".set_modelling_rule(True)\n")
if i.unit is not None:
if i.unit != "UNITLESS":
unitname = "unitprop"
server.write(writeindent + unitname + " = await " + varname + ".add_property(idx, \"unit\"," + "\"" + i.unit + "\")\n")
server.write(writeindent + unitname + " = await " + headnode + ".add_property(idx, \"unit\"," + "\"" + i.unit + "\")\n")
server.write(writeindent + "await " + unitname + ".set_modelling_rule(True)\n")
server.write(writeindent + "description = await " + varname + ".add_property(idx, \"Description\", " + "\"" + i.description + "\")" + "\n")
server.write(writeindent + "description = await " + headnode + ".add_property(idx, \"Description\", " + "\"" + i.description + "\")" + "\n")
server.write(writeindent + "await description.set_modelling_rule(True)\n\n")
server.write("\n" + writeindent + "#creating all parameters with their properties\n")
......@@ -97,21 +72,30 @@ _logger = logging.getLogger('asyncua')
datatype = "parametertype.nodeid"
varname = ((i.name).lower()).replace(" ", "") + "var"
parameterdict[i.name] = varname
server.write(writeindent + varname + " = " + "await myobjtype.add_variable(idx, \"" + str(i.name) + "\", " + str(i.value) + ", " + varianttype + ", " + datatype + ")\n")
server.write(writeindent + varname + " = " + "await " + objtypename + ".add_variable(idx, \"" + str(i.name) + "\", " + str(i.value) + ", " + varianttype + ", " + datatype + ")\n")
server.write(writeindent + "await " + varname + "." + "set_modelling_rule(True)\n")
if i.range is not None and i.range is not [None, None]:
if iscomponent:
headnode = objtypename
else:
headnode = varname
if i.range is not None and i.range is not (None, None):
rangename = "rangeprop"
server.write(writeindent + rangename + " = await " + varname + ".add_property(idx, \"range\", ua.Range(" + str(i.range[0]) + "," + str(i.range[1]) + "))\n")
server.write(writeindent + rangename + " = await " + headnode + ".add_property(idx, \"range\", ua.Range(" + str(i.range[0]) + "," + str(i.range[1]) + "))\n")
server.write(writeindent + "await " + rangename + ".set_modelling_rule(True)\n")
if i.unit is not None:
if i.unit is not "UNITLESS":
unitname = "unitprop"
server.write(writeindent + unitname + " = await " + varname + ".add_property(idx, \"unit\"," + "\"" + i.unit + "\")\n")
server.write(writeindent + unitname + " = await " + headnode + ".add_property(idx, \"unit\"," + "\"" + i.unit + "\")\n")
server.write(writeindent + "await " + unitname + ".set_modelling_rule(True)\n")
server.write(writeindent + "description = await " + varname + ".add_property(idx, \"Description\", " + "\"" + i.description + "\")" + "\n")
server.write(writeindent + "description = await " + headnode + ".add_property(idx, \"Description\", " + "\"" + i.description + "\")" + "\n")
server.write(writeindent + "await description.set_modelling_rule(True)\n\n")
server.write(writeindent + "mobilerobot = await server.nodes.objects.add_object(idx, \"Mobile Robot \", myobjtype.nodeid)\n")
objnamevar = (objname.lower()).replace(" ", "")
if(not iscomponent):
server.write(writeindent + objnamevar + "= await server.nodes.objects.add_object(idx, \"" + objname + "\"," + objtypename + ".nodeid)\n")
else:
server.write(writeindent + "headnodeid = await server.nodes.objects.get_child([\"2:Mobile Robot\"])\n")
server.write(writeindent + objnamevar + "= await headnodeid.add_object(idx,\"" + objname + "\")\n")
server.write(writeindent + "#adding functions to our robot\n")
......@@ -120,7 +104,7 @@ _logger = logging.getLogger('asyncua')
defFunction(functionname, server, indent, i.inargsnames)
server.write("\n")
inargslist = []
if i.inargsdatatypes != 0:
if i.inargsdatatypes != None:
for j in i.inargsdatatypes:
inargslist.append(getVariantType(j))
outargslist = []
......@@ -128,14 +112,53 @@ _logger = logging.getLogger('asyncua')
for j in i.outargsdataypes:
inargslist.append(getVariantType(j))
server.write(writeindent + "await mobilerobot.add_method(idx, \"" + i.name +"\", " + functionname + ", " + str(inargslist) + ", "+ str(outargslist) + ")\n\n")
def writebeginning():
if os.path.exists("opcua-server.py"):
os.remove("opcua-server.py")
server = open("opcua-server.py", "a")
server.write("""import logging
import asyncio
from asyncua import ua, Server
from asyncua.common.methods import uamethod
from asyncua.common.structures104 import new_struct, new_enum, new_struct_field
from asyncua.common import node
logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger('asyncua')
""")
server.write("\nasync def main():")
#äinten = "\tab"
server.write("""
# setup our server
server = Server()
await server.init()
server.set_endpoint('opc.tcp://0.0.0.0:4840/freeopcua/server/')
# setup our own namespace, not really necessary but should as spec
uri = 'http://examples.freeopcua.github.io/'
idx = await server.register_namespace(uri)
#id of rangetype
rangeid = await server.nodes.base_data_type.get_child(["0:Structure", "0:Range"])
#creating dataypes for parameters and measurements
measurementtype = await server.nodes.base_data_type.add_data_type(idx, "Measurement")
parametertype = await server.nodes.base_data_type.add_data_type(idx, "Parameter")
""")
def writend():
server = open("opcua-server.py", "a")
server.write("""
async with server:
print("Server is running!")
while True:
await asyncio.sleep(1)
if __name__ == '__main__':
asyncio.run(main())""")
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment