Read `dpinger` data to monitor the health of a network interface on a UniFi UDM-Pro.

unifi-monitor-dpinger.py

#!/usr/bin/env python3

# Original code by travis.vitek at https://bit.ly/3WPIu4h
import argparse
import json
import os
import socket
import sys
import time
from datetime import datetime

def find_socket(interface):
    """Find the first monitoring socket for the given interface."""
    run_dir = '/run'
    sockets = [f for f in os.listdir(run_dir) if f.endswith('.sock') and f.startswith(interface + '-mon')]
    if not sockets:
        return None
    sockets.sort()
    return os.path.join(run_dir, sockets[0])

def _main():
    parser = argparse.ArgumentParser()
    parser.add_argument('-d', '--delay', type=int, default=3, choices=[1,3,5,10,15,30,60])
    parser.add_argument('-i', '--interface', type=str, required=True)

    options = parser.parse_args()

    if options.delay <= 0:
        raise argparse.ArgumentTypeError(f'{options.delay} must be positive.')

    socket_path = find_socket(options.interface)
    if socket_path is None:
        print(f"No socket found for interface {options.interface}", file=sys.stderr)
        return

    while True:
        try:
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
                sock.connect(socket_path)

                data = sock.recv(128).decode().strip()
                when = datetime.now()

                # Parse the data: <Socket> <Average Latency in μs> <Standard Deviation in μs> <Percentage of Loss>
                parts = data.split()
                if len(parts) == 4:
                    socket_name, avg_str, std_str, loss_str = parts
                    avg = int(avg_str)
                    std = int(std_str)
                    loss = float(loss_str)
                    record = {
                        "timestamp": when.isoformat(),
                        "avg_latency_us": avg,
                        "std_dev_us": std,
                        "loss_percent": loss
                    }
                    print(json.dumps(record), flush=True)
        except FileNotFoundError:
            when = datetime.now()
            record = {
                "timestamp": when.isoformat(),
                "status": "interface is down"
            }
            print(json.dumps(record), file=sys.stderr, flush=True)

        time.sleep(options.delay)


if __name__ == "__main__":
    _main()