-
Notifications
You must be signed in to change notification settings - Fork 0
/
server.py
35 lines (35 loc) · 1.44 KB
/
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
import logging
import asyncio
import sys
sys.path.insert(0, "..")
from asyncua import ua, Server
from asyncua.common.methods import uamethod
@uamethod
def func(parent, value):
return value * 2
async def main():
_logger = logging.getLogger('asyncua')
# setup our server
server = Server()
await server.init()
server.set_endpoint('opc.tcp://localhost:62541/Quickstarts/ReferenceServer')
# setup our own namespace, not really necessary but should as spec
uri = 'http://examples.freeopcua.github.io'
idx = await server.register_namespace(uri)
# populating our address space
# server.nodes, contains links to very common nodes like objects and root
myobj = await server.nodes.objects.add_object(idx, 'MyObject')
myvar = await myobj.add_variable(idx, 'MyVariable', 6.7)
# Set MyVariable to be writable by clients
await myvar.set_writable()
await server.nodes.objects.add_method(ua.NodeId('ServerMethod', 2), ua.QualifiedName('ServerMethod', 2), func, [ua.VariantType.Int64], [ua.VariantType.Int64])
_logger.info('Starting server!')
async with server:
while True:
await asyncio.sleep(1)
new_val = await myvar.get_value() + 0.1
_logger.info('Set value of %s to %.1f', myvar, new_val)
await myvar.write_value(new_val)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
asyncio.run(main(), debug=True)