Dynamic USB passthrough using virsh

{
    "devices": [
        {
            "vid": "0483",
            "pid": "3748",
            "vm": "RDPWindows"
        }
    ]
}

#!/usr/bin/python3

import json
import subprocess
import pyudev

config = None
current_devices = {}
known_devices = {}
connected_devices = {}

with open("config.json", 'r') as f:
    config = json.load(f)

if config is None:
    print("ERROR: config.json not found!")
    exit(-1)

def updateKnownDevices():
    global known_devices
    global current_devices

    current_devices = {}

    try:
        lsusb_output = subprocess.check_output(['lsusb']).decode('utf-8').strip().split("\n")
        for line in lsusb_output:
            vid, pid = line.split(" ")[5].split(':')
            vendor_name = line.split(" ")[6]
            product_name = line.split(" ")[7]

            known_devices[f'{vendor_name}:{product_name}'] = f'{vid}:{pid}'
            current_devices[f'{vendor_name}:{product_name}'] = f'{vid}:{pid}'
    except subprocess.CalledProcessError as e:
        pass

def get_target_vm_name_from_config(vid, pid):
    global config

    for entry in config["devices"]:
        if entry["vid"] == vid and entry["pid"] == pid:
            return entry["vm"]

    return None

def check_if_vm_is_running(name):
    vm_running = False
    try:
        if "running" in str(subprocess.check_output(['virsh', 'domstate', name])):
            vm_running = True
    except subprocess.CalledProcessError as e:
        pass

    return vm_running

def process_device_to_vm(vid, pid, is_connect):
    vm = get_target_vm_name_from_config(vid, pid)

    if vm is None:
        print(f'VM not found for device {vid}:{pid}...')
        return
    
    if not check_if_vm_is_running(vm):
        print(f'VM {vm} is not running, skipping device {vid}:{pid}...')
        return
    
    command = None
    if is_connect:
        print(f'Connecting device {vid}:{pid} to VM {vm}...')
        command = f"""bash -c 'echo "<hostdev mode=\\"subsystem\\" type=\\"usb\\" managed=\\"yes\\"><source><vendor id=\\"0x{vid}\\"/><product id=\\"0x{pid}\\"/></source></hostdev>" | sudo virsh attach-device {vm} /dev/stdin'"""
    else:
        print(f'Disconnecting device {vid}:{pid} from VM {vm}...')
        command = f"""bash -c 'echo "<hostdev mode=\\"subsystem\\" type=\\"usb\\" managed=\\"yes\\"><source><vendor id=\\"0x{vid}\\"/><product id=\\"0x{pid}\\"/></source></hostdev>" | sudo virsh detach-device {vm} /dev/stdin'"""

    if command is not None:
        try:
            subprocess.check_output(command, shell=True)
        except subprocess.CalledProcessError as e:
            pass

context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by("usb")

for action, device in monitor:
    updateKnownDevices()

    vendor_name = device.get("ID_VENDOR_FROM_DATABASE", None)
    product_name = device.get("ID_MODEL_FROM_DATABASE", None)

    if (vendor_name is None or product_name is None) and action != "remove":
        continue

    vid = None
    pid = None
    if vendor_name is not None and product_name is not None:
        for key in known_devices:
            value = known_devices[key]
            if vendor_name in key and product_name in key:
                vid, pid = value.split(":")

    if vid is None or pid is None:
        if action == "remove":
            for device in connected_devices:
                if connected_devices[device] == 1:
                    vid, pid = device.split(":")
                    if not (vid in current_devices.values() and pid in current_devices.values()):
                        connected_devices[device] = 0
                        process_device_to_vm(vid, pid, False)
                    break
        continue

    has_connected = False
    has_disconnected = False

    if f'{vid}:{pid}' not in connected_devices:
        connected_devices[f'{vid}:{pid}'] = -1

    if action == "bind":
        if connected_devices[f'{vid}:{pid}'] != 1:
            connected_devices[f'{vid}:{pid}'] = 1
            has_connected = True
    elif action == "remove":
        if connected_devices[f'{vid}:{pid}'] != 0:
            connected_devices[f'{vid}:{pid}'] = 0
            has_disconnected = True
    
    if has_connected:
        process_device_to_vm(vid, pid, True)
    elif has_disconnected:
        process_device_to_vm(vid, pid, False)