-
Notifications
You must be signed in to change notification settings - Fork 0
/
opcua_server.py
143 lines (121 loc) · 4.75 KB
/
opcua_server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import sys
from opcua.ua.uaprotocol_auto import UadpWriterGroupMessageDataType, UserIdentityToken
from opcua.ua.uatypes import VariantType
sys.path.insert(0, "..")
import time
import json
import random
from opcua import ua, Server
from opcua.common.type_dictionary_buider import DataTypeDictionaryBuilder, get_ua_class
value_update_frequency = 5
def getdelta(type, function):
delta_value = 0
if type.startswith("Int"):
delta_value = 1
elif type.startswith("Float"):
delta_value = random.random()
if function == "inc":
delta_value = delta_value * 1
elif function == "rnd_inc":
if random.randint(0, 1) == 1:
delta_value = delta_value * 1
else:
delta_value = 0
elif function == "dec":
delta_value = delta_value * -1
elif function == "rnd_dec":
if random.randint(0, 1) == 1:
delta_value = delta_value * -1
else:
delta_value = 0
elif function == "rnd_walk":
if random.randint(0, 1) == 1:
delta_value = delta_value * -1
return delta_value
def process_children(children, parent):
for child in children:
if child["type"] == "object":
new_obj = parent.add_object(idx, child["name"])
process_children(child["children"], new_obj)
else:
new_variable_state = child
variable_states.append(new_variable_state)
new_variable_state["var"] = None
datatype = ua.VariantType.__dict__.get(child["type"])
if datatype:
new_variable_state["var"] = parent.add_variable(
idx, child["name"], child["value"], datatype
)
else:
# assume this is a custom datatype/struct
new_variable_state["var"] = parent.add_variable(
idx,
child["name"],
ua.Variant(None, ua.VariantType.Null),
datatype=custom_types[child["type"]].data_type,
)
if child["writable"]:
new_variable_state["var"].set_writable()
if __name__ == "__main__":
# load the config file
f = open(
"opcua_server.json",
)
config = json.load(f)
f.close()
# setup our server
server = Server()
server.set_endpoint(config["server"]["endpoint"])
# setup our own namespace, not really necessary but should as spec
uri = config["server"]["uri"]
idx = server.register_namespace(uri)
server.set_server_name(config["server"]["name"])
# custom structure storage
datatype_builder = DataTypeDictionaryBuilder(server, idx, uri, "customStructures")
# get Objects node, this is where we should put our nodes
objects = server.get_objects_node()
# build the custom structures needed
custom_types = {}
for struct in config["structures"]:
custom_types[struct["name"]] = datatype_builder.create_data_type(struct["name"])
for field in struct["fields"]:
custom_types[struct["name"]].add_field(
field["name"], ua.VariantType.__dict__.get(field["type"])
)
# save and load the custom structures into the server
datatype_builder.set_dict_byte_string()
server.load_type_definitions()
# populating our address space
variable_states = []
for object in config["objects"]:
new_obj = objects.add_object(idx, object["name"])
if object["children"]:
process_children(object["children"], new_obj)
# start the server
server.start()
try:
count = 0
while True:
for variable_state in variable_states:
delta_value = getdelta(
variable_state["type"], variable_state["function"]
)
datatype = ua.VariantType.__dict__.get(variable_state["type"])
if datatype:
variable_state["value"] = variable_state["value"] + delta_value
else:
variable_state["value"] = get_ua_class(variable_state["type"])()
for sub_vars in variable_state["value"].ua_types:
current_val = getattr(variable_state["value"], sub_vars[0])
delta_value = getdelta(sub_vars[1], variable_state["function"])
setattr(
variable_state["value"],
sub_vars[0][0],
current_val + delta_value,
)
if variable_state["writable"]:
variable_state["var"].set_value(variable_state["value"])
time.sleep(value_update_frequency)
finally:
# close connection, remove subcsriptions, etc
server.stop()