simple udp logger for unique messages. specify port.

udpLogger.py

import argparse
import socket
import datetime
from collections import defaultdict

# --------------------------------------------------------------------------- #
# Data structure
# --------------------------------------------------------------------------- #
#   { "192.168.0.2" : { "01" : {"count":2, "first_seen":datetime, "last_seen":datetime},
#                       "02" : {"count":1, ... } },
#     "192.168.0.3" : { "ff" : {"count":2, ... } } }
# --------------------------------------------------------------------------- #
collected_data: dict[str, dict[str, dict[str, object]]] = defaultdict(dict)


def save_to_file(host, port, data_to_save: dict[str, dict[str, dict[str, object]]]) -> None:
    ts = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
    filename = f"udp_log_{host}_{port}_{ts}.txt"

    try:
        with open(filename, "w", encoding="utf-8") as file:
            for address, payload_map in data_to_save.items():
                file.write(f"{address} says:\n")
                for data, meta in payload_map.items():
                    first = meta["first_seen"].strftime("%Y-%m-%d %H:%M:%S")
                    last  = meta["last_seen"].strftime("%Y-%m-%d %H:%M:%S")
                    file.write(f"{data}\ncount: {meta['count']} - first seen: {first} - last seen: {last}\n\n")
                file.write("\n")          # blank line between devices

        print(f"[+] Snapshot written to {filename}")

    except Exception as exc:
        print(f"[!] Could not write snapshot to {filename}: {exc}")


def start_udp_listener(host, port):
    # Create a UDP socket, and bind
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_address = (host, port)
    sock.bind(server_address)
    
    print(f"UDP Listener started on {host}:{port}\n")
    packets_since_last_save = 0
    auto_save_count = 500 # sd card has low tolerance for frequent writes

    try:
        while True:
            # Receive data from the client
            data, address = sock.recvfrom(4096)  # Buffer size is 4096 bytes
            hex_data = ' '.join(f"{byte:02x}" for byte in data)
            ip = address[0]
            now = datetime.datetime.now()
            #message = data.decode('latin-1')

            print(f"{now} - {address} says:\n{hex_data}\n")
            
            # Get the nested map for this IP; create if missing
            device_map = collected_data[ip]

            if hex_data in device_map:
                # Existing payload – bump counter and update last_seen
                meta = device_map[hex_data]
                meta["count"] += 1
                meta["last_seen"] = now
            else:
                # New payload – initialise counters and timestamps
                device_map[hex_data] = {
                    "count": 1,
                    "first_seen": now,
                    "last_seen": now,
                }

            packets_since_last_save += 1
            if packets_since_last_save >= auto_save_count:
                save_to_file(host, port, collected_data)
                packets_since_last_save = 0
    
    except KeyboardInterrupt:
        print("UDP Listener stopped.")
    except Exception as exc:
        print(f"\nError: {exc}")
    finally:
        sock.close()
        save_to_file(host, port, collected_data)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="UDP Packet Logger")
    parser.add_argument("--host", type=str, default="0.0.0.0", 
                        help="Host interface to bind (default: 0.0.0.0)")
    parser.add_argument("--port", type=int, default=57621, 
                        help="Port to listen on (default: 57621)")
    args = parser.parse_args()
    
    start_udp_listener(args.host, args.port)

usage:

python udpLogger.py --host <ip> --port <port>

host can be omitted to default any 0.0.0.0
omitting port will default to what ever I left it hard coded to at the time of writing...