Proxmox hot-plug USB script, requires `python3-pyudev`

gistfile1.txt

#!/usr/bin/python3
import json
import pyudev
import socket

vm_id = 100

def send_monitor_command(command, **kwargs):
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
        sock.connect("/var/run/qemu-server/{}.qmp".format(vm_id))
        sock.recv(1024) # Discard hello
        sock.sendall(json.dumps({"execute": "qmp_capabilities"}).encode("utf-8"))
        sock.recv(1024) # Discard capability return
        sock.sendall(json.dumps({"execute": command, "arguments": kwargs}).encode("utf-8"))
        return json.loads(sock.recv(4096).decode("utf-8"))

def handle_bus_event(device):
    action = device.action
    vendor_id, model_id = device.get("PRODUCT").split("/")[:2]
    fullpath = device.get("DEVPATH").split("/")[-1]
    bus_id, port_id = fullpath.split("-", 1)

    if action == "bind":
        print("Connecting USB device {}:{} at bus {} port {}".format(vendor_id, model_id, bus_id, port_id))
        send_monitor_command("device_add", driver="usb-host", bus="xhci.0", hostbus=bus_id, hostport=port_id, id="hotplug-" + fullpath)
    elif action == "unbind":
        print("Disconnecting USB device {}:{} at bus {} port {}".format(vendor_id, model_id, bus_id, port_id))
        send_monitor_command("device_del", id="hotplug-" + fullpath)

def handle_udev_event(device):
    if device.get("SUBSYSTEM") == "usb":
        if "BUSNUM" in device and "DEVNUM" in device:
            handle_bus_event(device)

def main_loop():
    print("Opening udev context")
    context = pyudev.Context()

    print("Opening event monitor")
    monitor = pyudev.Monitor.from_netlink(context)

    print("Adding XHCI bus")
    send_monitor_command("device_add", driver="nec-usb-xhci", id="xhci", addr="0x1B", bus="pci.1")

    print("Watching for USB devices")
    for device in iter(monitor.poll, None):
        handle_udev_event(device)

if __name__ == "__main__":
    main_loop()