22 ) << EOF & |
22 ) << EOF & |
23 |
23 |
24 import sys |
24 import sys |
25 import os |
25 import os |
26 import time |
26 import time |
|
27 import asyncio |
27 |
28 |
28 from opcua import ua, Server |
29 from asyncua import ua, Server |
|
30 from asyncua.server.users import User, UserRole |
29 |
31 |
30 server = Server() |
32 # Asyncua can't work without (over)simple shared cerificates/privkey. |
31 host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1") |
33 # No user is involved in that case, but asyncua needs it. |
32 endpoint = "opc.tcp://"+host+":4840/freeopcua/server/" |
34 # Over permessive User Manager hereafter helps cuting that corner. |
33 server.set_endpoint(endpoint) |
35 class AllAdminUserManager: |
|
36 def get_user(self, iserver, username=None, password=None, certificate=None): |
|
37 return User(role=UserRole.Admin) |
34 |
38 |
35 server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt]) |
39 async def main(): |
36 server.load_certificate("my_cert.der") |
40 server = Server(user_manager=AllAdminUserManager()) |
37 server.load_private_key("my_private_key.pem") |
41 host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1") |
|
42 endpoint = "opc.tcp://"+host+":4840/freeopcua/server/" |
|
43 await server.init() |
|
44 server.set_endpoint(endpoint) |
38 |
45 |
39 uri = "http://beremiz.github.io" |
46 server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt]) |
40 idx = server.register_namespace(uri) |
47 await server.load_certificate("my_cert.der") |
|
48 await server.load_private_key("my_private_key.pem") |
41 |
49 |
42 objects = server.get_objects_node() |
50 uri = "http://beremiz.github.io" |
|
51 idx = await server.register_namespace(uri) |
43 |
52 |
44 testobj = objects.add_object(idx, "TestObject") |
53 objects = server.get_objects_node() |
45 testvarout = testobj.add_variable(idx, "TestOut", 1.2) |
|
46 testvar = testobj.add_variable(idx, "TestIn", 5.6) |
|
47 testvar.set_writable() |
|
48 |
54 |
49 server.start() |
55 testobj = await objects.add_object(idx, "TestObject") |
|
56 testvarout = await testobj.add_variable(idx, "TestOut", 1.2) |
|
57 testvar = await testobj.add_variable(idx, "TestIn", 5.6) |
|
58 await testvar.set_writable() |
50 |
59 |
51 try: |
60 await server.start() |
52 while True: |
61 |
53 time.sleep(1) |
62 try: |
54 print testvar.get_value() |
63 while True: |
55 sys.stdout.flush() |
64 await asyncio.sleep(1) |
56 finally: |
65 print(await testvar.get_value()) |
57 server.stop() |
66 sys.stdout.flush() |
|
67 finally: |
|
68 await server.stop() |
|
69 |
|
70 asyncio.run(main()) |
|
71 |
58 EOF |
72 EOF |
59 SERVER_PID=$! |
73 SERVER_PID=$! |
60 |
74 |
61 PROJECT_FILES_DIR=$BEREMIZPATH/tests/projects/opcua_client_encrypted/project_files |
75 PROJECT_FILES_DIR=$BEREMIZPATH/tests/projects/opcua_client_encrypted/project_files |
62 mkdir $PROJECT_FILES_DIR |
76 mkdir $PROJECT_FILES_DIR |