| 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 |