Total Complexity | 3 |
Total Lines | 43 |
Duplicated Lines | 0 % |
Changes | 0 |
1 | import asyncio |
||
2 | import logging |
||
3 | import sys |
||
4 | sys.path.insert(0, "..") |
||
5 | from asyncua import Client, Node, ua |
||
6 | from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256 |
||
7 | |||
8 | logging.basicConfig(level=logging.INFO) |
||
9 | _logger = logging.getLogger("asyncua") |
||
10 | |||
11 | cert = "certificates/peer-certificate-example-1.der" |
||
12 | private_key = "certificates/peer-private-key-example-1.pem" |
||
13 | |||
14 | |||
15 | async def task(loop): |
||
16 | url = "opc.tcp://0.0.0.0:4840/freeopcua/server/" |
||
17 | try: |
||
18 | client = Client(url=url) |
||
19 | await client.set_security( |
||
20 | SecurityPolicyBasic256Sha256, |
||
21 | certificate_path=cert, |
||
22 | private_key_path=private_key |
||
23 | ) |
||
24 | await client.connect() |
||
25 | root = client.nodes.root |
||
26 | print(await root.get_children()) |
||
27 | |||
28 | except Exception: |
||
29 | _logger.exception('error') |
||
30 | finally: |
||
31 | await client.disconnect() |
||
32 | |||
33 | |||
34 | def main(): |
||
35 | loop = asyncio.get_event_loop() |
||
36 | loop.set_debug(True) |
||
37 | loop.run_until_complete(task(loop)) |
||
38 | loop.close() |
||
39 | |||
40 | |||
41 | if __name__ == "__main__": |
||
42 | main() |
||
43 |