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

dynamic generation of new server through using jdonparse w errors

parent ab6bda5f
Branches
No related tags found
No related merge requests found
import json
import os
import sys
import writefile as wf
from collections import namedtuple
......@@ -170,6 +170,10 @@ def main():
print(functionobj)
print(parameterobj)
#exec(open("writefile.py").read())
wf.write(measurementobj)
if __name__ == '__main__':
......
import jsonparse
import os
def defFunction(functionname, file, indent, contents = None):
writeindent = ""
for i in range(indent):
writeindent = writeindent + "\t"
file.write(indent + "async def " + functionname + ":")
def write(measurementobj):
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"])
myobjtype = await server.nodes.base_object_type.add_object_type(idx, "Robot")
vartype = await server.nodes.base_data_type.add_data_type(idx, "Measurement")
""")
indent = 1
writeindent = ""
for j in range(indent):
writeindent = writeindent + " "
for i in measurementobj:
varianttype = "ua.VariantType.Int64"
if i.datatype == "float" or i.datatype == "double":
varianttype = "ua.VariantType.Double"
elif i.datatype == "Boolean":
varianttype = "ua.VariantType.Boolean"
datatype = "vartype.nodeid"
varname = ((i.name).lower()).replace(" ", "") + "var"
server.write(writeindent + varname + " = " + "await myobjtype.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]:
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 + "await " + rangename + ".set_modelling_rule(True)\n")
if i.unit is not None:
unitname = "unitprop"
server.write(writeindent + unitname + " = await " + varname + ".add_property(idx, \"unit\"," + "\"" + i.unit + "\"" + ")\n")
server.write(writeindent + "await " + unitname + ".set_modelling_rule(True)\n")
server.write(writeindent + "description = await name.add_property(idx, \"Description\", " + "\"" + i.description + "\"" + "\n")
server.write(writeindent + "await description.set_modelling_rule(True)\n")
server.write("""
async with server:
print("Server is running!")
while True:
await asyncio.sleep(1)
if name == 'main':
asyncio.run(main())""")
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment