Check if iommu_group devices path exists, before listing it.

.gitignore

@@ -160,3 +160,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+/*util.sh

src/pci_passthrough_assist/__main__.py

@@ -1,5 +1,10 @@
 from sys import platform
+from pci_passthrough_assist.permissions import is_ran_by_root
 
 if platform != "linux":
     print("This tool will only work on Linux based OS.")
     exit(1)
+
+if not is_ran_by_root():
+    print("This script needs to run as root.")
+    exit(1)

src/pci_passthrough_assist/pci.py

@@ -0,0 +1,9 @@
+from os import listdir
+
+
+def all_pci_device_ids() -> list[str]:
+    return listdir("/sys/bus/pci/devices")
+
+
+def all_pci_driver_ids() -> list[str]:
+    return listdir("/sys/bus/pci/drivers")

src/pci_passthrough_assist/pci_device.py

@@ -0,0 +1,66 @@
+from os.path import exists, realpath, basename
+from os import listdir
+
+
+class PciDevice:
+
+    def __init__(self, device_id: str):
+        assert exists(f"/sys/bus/pci/devices/{device_id}")
+        self.device_id = device_id
+
+    def driver_name(self) -> str:
+        pci_driver_path = f"/sys/bus/pci/devices/{self.device_id}/driver"
+        if not exists(pci_driver_path):
+            return "NO-DRIVER-BOUND"
+        driver_directory: str = realpath(pci_driver_path)
+        return basename(driver_directory)
+
+    def is_vga(self) -> bool:
+        return exists(f"/sys/bus/pci/devices/{self.device_id}/boot_vga")
+
+    def vendor_code(self, remove_prefix: bool = True) -> str:
+        with open(f"/sys/bus/pci/devices/{self.device_id}/vendor") as vendor:
+            return vendor.read() if not remove_prefix else vendor.read(
+            ).lstrip("0x")
+
+    def device_code(self, remove_prefix: bool = True) -> str:
+        with open(f"/sys/bus/pci/devices/{self.device_id}/vendor") as device:
+            return device.read() if not remove_prefix else device.read(
+            ).lstrip("0x")
+
+    def unbind_driver(self):
+        driver_unbind_path = f"/sys/bus/pci/devices/{self.device_id}/driver/unbind"
+        if not exists(driver_unbind_path):
+            print("Device is not bound to any driver.")
+            return
+
+        with open(driver_unbind_path, "w") as device_driver:
+            device_driver.write(self.device_id)
+
+    def set_driver_override(self, reserved_for_driver: str):
+        with open(f"/sys/bus/pci/devices/{self.device_id}/driver_override",
+                  "w") as driver_override:
+            driver_override.write(reserved_for_driver)
+
+    def bind_to_driver(self, driver_to_bind: str, unbind_first: bool = True):
+        if unbind_first:
+            self.unbind_driver()
+
+        driver_bind_path = f"/sys/bus/pci/drivers/{driver_to_bind}/bind"
+        if not exists(driver_bind_path):
+            print(f"Can't bind to driver: {driver_bind_path}.")
+            return
+
+        with open(driver_bind_path, "w") as driver:
+            driver.write(self.device_id)
+
+    def devices_in_iommu_group(self) -> list['PciDevice']:
+        iommu_group_device_path = f"/sys/bus/pci/devices/{self.device_id}/iommu_group/devices"
+        if not exists(iommu_group_device_path):
+            print("Device does not have iommu_group devices.")
+            return []
+        device_ids: list[str] = listdir(iommu_group_device_path)
+        return [PciDevice(device_id) for device_id in device_ids]
+
+    def __str__(self) -> str:
+        return f"{self.device_id} driver: {self.driver_name()} VGA: {self.is_vga()}"

src/pci_passthrough_assist/permissions.py

@@ -0,0 +1,5 @@
+from pci_passthrough_assist.process_runner import sh
+
+
+def is_ran_by_root() -> bool:
+    return sh(["whoami"]) == "root"

src/pci_passthrough_assist/process_runner.py

@@ -0,0 +1,14 @@
+from subprocess import run
+
+
+def sh_binary(args: list[str], ) -> bytes:
+    return run(args, capture_output=True).stdout
+
+
+def sh(args: list[str], rstrip_newline=True):
+    string_output: str = sh_binary(args).decode()
+    return string_output if not rstrip_newline else string_output.rstrip("\n")
+
+
+def sh_lines(args: list[str]) -> list[str]:
+    return sh(args, rstrip_newline=True).splitlines()