-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathopc_server_asyncio.py
More file actions
38 lines (31 loc) · 1.2 KB
/
opc_server_asyncio.py
File metadata and controls
38 lines (31 loc) · 1.2 KB
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
import time
from asyncua.sync import Server
if __name__ == "__main__":
# setup our server
server = Server()
server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
# setup our own namespace, not really necessary but should as spec
uri = "sample_namespace"
server_namespace = server.register_namespace(uri)
# get Objects node, this is where we should put our nodes
root_objects = server.nodes.objects
# populating our address space
my_object = root_objects.add_object(server_namespace, "MyObject")
my_variable = my_object.add_variable(server_namespace, "MyVariable", 6.7)
my_variable_2 = my_object.add_variable(server_namespace, "MyVariable_2", 7.6)
my_variable.set_writable() # Set MyVariable to be writable by clients
# starting!
server.start()
try:
count = 0
while True:
print(my_object.get_children()[0].get_value())
#print(my_object.get_children()[1].get_value())
time.sleep(1)
count += 1.0
my_variable.write_value(count)
except KeyboardInterrupt:
print('Interrupted')
finally:
# close connection, remove subcsriptions, etc
server.stop()