Subscribe to OPC UA data changes and log them to CSV (Python)

"""Subscribe to an OPC UA node's data changes and log them to CSV (asyncua).

Usage:  python opcua_subscription_logger.py opc.tcp://192.168.0.20:4840 "ns=2;i=2" out.csv 30
"""

import asyncio
import csv
import sys
from datetime import datetime, timezone

from asyncua import Client

PUBLISHING_INTERVAL_MS = 500


class ChangeLogger:
    """asyncua calls datachange_notification() on the event loop for every update."""

    def __init__(self, writer):
        self.writer = writer

    def datachange_notification(self, node, value, data):
        timestamp = datetime.now(timezone.utc).isoformat()
        print(f"{timestamp}  {node}  {value}")
        self.writer.writerow([timestamp, str(node), value])


async def run(url, node_id, csv_path, seconds):
    async with Client(url=url) as client:
        node = client.get_node(node_id)

        with open(csv_path, "w", newline="") as fh:
            writer = csv.writer(fh)
            writer.writerow(["timestamp", "node", "value"])

            handler = ChangeLogger(writer)
            sub = await client.create_subscription(PUBLISHING_INTERVAL_MS, handler)
            sub_handle = await sub.subscribe_data_change(node)

            print(f"Subscribed to {node_id} - logging for {seconds}s")
            await asyncio.sleep(seconds)

            await sub.unsubscribe(sub_handle)
            await sub.delete()


def main():
    if len(sys.argv) < 4:
        sys.exit(
            "Usage: python opcua_subscription_logger.py "
            "<endpoint-url> <node-id> <csv-path> [seconds]"
        )
    url, node_id, csv_path = sys.argv[1:4]
    seconds = int(sys.argv[4]) if len(sys.argv) > 4 else 30
    asyncio.run(run(url, node_id, csv_path, seconds))


if __name__ == "__main__":
    main()