Read and watch a TwinCAT PLC symbol for changes (pyads)

"""Read a TwinCAT PLC symbol once, then log every change via a device
notification (pyads).

Usage:  python pyads_symbol_logger.py 5.24.34.100.1.1 MAIN.spindleSpeed out.csv 30
"""

import csv
import sys
import time
from ctypes import sizeof

import pyads

PORT = pyads.PORT_TC3PLC1
PLC_TYPE = pyads.PLCTYPE_LREAL


def main():
    if len(sys.argv) < 4:
        sys.exit(
            "Usage: python pyads_symbol_logger.py <ams-net-id> <symbol-name> "
            "<csv-path> [seconds]"
        )
    ams_net_id, symbol, csv_path = sys.argv[1:4]
    seconds = int(sys.argv[4]) if len(sys.argv) > 4 else 30

    plc = pyads.Connection(ams_net_id, PORT)

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

        initial = plc.read_by_name(symbol, PLC_TYPE)
        print(f"initial {symbol} = {initial}")
        writer.writerow([time.time(), symbol, initial])

        @plc.notification(PLC_TYPE)
        def on_change(handle, name, timestamp, value):
            print(f"{timestamp}  {name} = {value}")
            writer.writerow([timestamp, name, value])
            fh.flush()

        # ADSTRANS_SERVERONCHA (the default) fires only when the value
        # actually changes, not on a fixed poll interval.
        attr = pyads.NotificationAttrib(sizeof(PLC_TYPE))
        handles = plc.add_device_notification(symbol, attr, on_change)

        print(f"Watching {symbol} for {seconds}s - Ctrl+C to stop early.")
        try:
            time.sleep(seconds)
        finally:
            plc.del_device_notification(*handles)


if __name__ == "__main__":
    main()