Started to remove subprocessees, starting 207 and 171.

 Please enter the commit message for your changes. Lines starting
 with '' will be ignored, and an empty message aborts the commit.

 On branch dev
 Your branch is up to date with 'origin/dev'.

 Changes to be committed:
	new file:   src/lufus/block_ops.py
	modified:   src/lufus/drives/find_usb.py
	modified:   src/lufus/drives/formatting.py
	modified:   src/lufus/drives/get_usb_info.py
	modified:   src/lufus/gui/gui.py
	new file:   src/lufus/iso9660.py
	modified:   src/lufus/writing/flash_usb.py
	modified:   src/lufus/writing/install_ventoy.py
	modified:   src/lufus/writing/windows/detect.py
	modified:   src/lufus/writing/windows/flash.py
	modified:   src/lufus/writing/windows/tweaks.py
	modified:   tests/test_find_usb.py
	modified:   tests/test_flash_usb_and_find_usb_fixes.py
	modified:   tests/test_formatting.py
	modified:   tests/test_get_usb_info.py
	modified:   tests/test_get_usb_info_and_detect_windows_fixes.py
	modified:   tests/test_linux_iso_detection.py

Author: R-Larocque

src/lufus/block_ops.py

@@ -1,9 +1,9 @@
 import psutil
 import os
-import subprocess
 import getpass
 from lufus import state
 from lufus.lufus_logging import get_logger
+from lufus.block_ops import get_device_label
 
 log = get_logger(__name__)
 
@@ -51,16 +51,7 @@ def find_usb() -> dict[str, str]:
         if not device_node:
             continue
 
-        label = None
-        try:
-            label = subprocess.check_output(
-                ["lsblk", "-d", "-n", "-o", "LABEL", device_node],
-                text=True,
-                timeout=5,
-            ).strip()
-        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
-            pass
-
+        label = get_device_label(device_node)
         if not label:
             label = os.path.basename(mount_path)
 

src/lufus/drives/find_usb.py

@@ -10,6 +10,14 @@
 from lufus.drives import find_usb as fu
 from lufus.utils import strip_partition_suffix, require_root, get_mount_and_drive
 from lufus.lufus_logging import get_logger
+from lufus.block_ops import (
+    mount as block_mount,
+    umount as block_umount,
+    umount_lazy,
+    get_logical_block_size,
+    write_single_partition_table,
+    reread_partitions,
+)
 
 log = get_logger(__name__)
 
@@ -59,23 +67,12 @@ def unmount(drive: str = None) -> bool:
     targets = glob.glob(f"{drive}*")
     log.info("Unmounting %s...", drive)
     for target in targets:
-        try:
-            subprocess.run(["umount", "-l", target], check=True)
+        if umount_lazy(target):
             time.sleep(0.5)
             log.info("Unmounted %s successfully.", target)
-        except subprocess.CalledProcessError as e:
-            # exit 32 = "not mounted" on Linux — acceptable when clearing a drive
-            if e.returncode == 32:
-                log.info("Unmounted %s (was already unmounted).", target)
-                continue
-            log.error("umount -l %s exited with code %d", target, e.returncode)
-            unmount_fail()
-            return False
-        except Exception as e:
-            log.error("(UMNTFUNC) Unexpected error type: %s — %s", type(e).__name__, e)
-            log_unexpected_error()
-            return False
-    subprocess.run(["udevadm", "settle"])
+        else:
+            # Target may already be unmounted — not a fatal error
+            log.info("Unmounted %s (was already unmounted or not a mount).", target)
     time.sleep(0.5)
     return True
 
@@ -98,10 +95,9 @@ def remount(drive: str = None) -> bool:
         return False
     log.info("Remounting %s -> %s...", drive, mount)
     try:
-        subprocess.run(["mount", drive, mount], check=True)
-        log.info("Remounted %s -> %s successfully.", drive, mount)
-        return True
-    except subprocess.CalledProcessError:
+        if block_mount(drive, mount):
+            log.info("Remounted %s -> %s successfully.", drive, mount)
+            return True
         format_fail()
         return False
     except Exception as e:
@@ -212,27 +208,13 @@ def check_device_bad_blocks() -> bool:
     # device geometry. Fall back to 4096 bytes if detection fails.
     logical_block_size = 4096
     try:
-        probe = subprocess.run(
-            [_find_tool("blockdev"), "--getss", drive],
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if probe.returncode == 0:
-            probed = probe.stdout.strip()
-            if probed.isdigit():
-                logical_block_size = int(probed)
-            else:
-                log.warning(
-                    "Unexpected blockdev output for %r: %r. Using default block size.",
-                    drive,
-                    probed,
-                )
+        probed = get_logical_block_size(drive)
+        if probed is not None:
+            logical_block_size = probed
         else:
             log.warning(
-                "blockdev failed for %s (exit %d). Using default block size.",
+                "Could not probe sector size for %s. Using default block size.",
                 drive,
-                probe.returncode,
             )
     except Exception as exc:
         log.warning(
@@ -388,41 +370,13 @@ def _apply_partition_scheme(drive: str) -> None:
     scheme_name = "GPT" if scheme == 0 else "MBR"
     log.info("Applying %s partition scheme to %s...", scheme_name, raw_device)
     try:
-        if scheme == 0:
-            # GPT — used for UEFI targets
-            subprocess.run([_find_tool("parted"), "-s", raw_device, "mklabel", "gpt"], check=True)
-            subprocess.run(
-                [
-                    _find_tool("parted"),
-                    "-s",
-                    raw_device,
-                    "mkpart",
-                    "primary",
-                    "1MiB",
-                    "100%",
-                ],
-                check=True,
-            )
+        scheme_key = "gpt" if scheme == 0 else "mbr"
+        if write_single_partition_table(raw_device, scheme=scheme_key):
+            log.info("Partition scheme %s applied to %s.", scheme_name, raw_device)
         else:
-            # MBR — used for BIOS/legacy targets
-            subprocess.run([_find_tool("parted"), "-s", raw_device, "mklabel", "msdos"], check=True)
-            subprocess.run(
-                [
-                    _find_tool("parted"),
-                    "-s",
-                    raw_device,
-                    "mkpart",
-                    "primary",
-                    "1MiB",
-                    "100%",
-                ],
-                check=True,
-            )
-        log.info("Partition scheme %s applied to %s.", scheme_name, raw_device)
+            log.error("(PARTITION) Failed to apply partition scheme.")
     except FileNotFoundError:
         log.error("'parted' not found. Install parted.")
-    except subprocess.CalledProcessError as e:
-        log.error("(PARTITION) Failed to apply partition scheme: %s", e)
     except Exception as e:
         log.error("(PARTITION) Unexpected error: %s: %s", type(e).__name__, e)
         log_unexpected_error()
@@ -433,17 +387,15 @@ def drive_repair() -> None:
     # add smartctl check if possible
     # use fsck to prevent deletion of files for repair
     # use testdisk for partition recovery if possible
-    # do dd if=/dev/zero of=/dev/sdX bs=1M count=10 conv=notrunc before sfdisk use
     _, drive, _ = _get_mount_and_drive()
     if not drive:
         log.error("No drive node found. Cannot repair.")
         return
     raw_device = strip_partition_suffix(drive)
-    cmd = [_find_tool("sfdisk"), raw_device]
     log.info("Attempting drive repair on %s (raw: %s)...", drive, raw_device)
     try:
-        subprocess.run(["umount", drive], check=True)
-        subprocess.run(cmd, input=b",,0c;\n", check=True)
+        block_umount(drive)
+        write_single_partition_table(raw_device, scheme="mbr")
         subprocess.run([_find_tool("mkfs.vfat"), "-F", "32", "-n", "REPAIRED", drive], check=True)
         log.info("Successfully repaired drive %s (FAT32).", drive)
     except Exception as e:

src/lufus/drives/formatting.py

@@ -1,8 +1,8 @@
 import psutil
 import os
-import subprocess
 from typing import TypedDict
 from lufus.lufus_logging import get_logger
+from lufus.block_ops import get_device_size, get_device_label
 
 log = get_logger(__name__)
 
@@ -25,20 +25,12 @@ def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
             log.warning("Could not find device node for USB path: %s", usb_path)
             return None
 
-        size_output = subprocess.check_output(
-            ["lsblk", "-d", "-n", "-b", "-o", "SIZE", device_node],
-            text=True,
-            timeout=5,
-        ).strip()
-
-        usb_size = int(size_output) if size_output.isdigit() else 0
-        if not size_output.isdigit():
-            log.warning("Could not parse device size: %r", size_output)
+        usb_size = get_device_size(device_node) or 0
 
         if usb_size > 32 * 1024**3:
             log.warning("USB device is large (%d bytes); confirm before flashing.", usb_size)
 
-        label = subprocess.check_output(["lsblk", "-d", "-n", "-o", "LABEL", device_node], text=True, timeout=5).strip()
+        label = get_device_label(device_node)
         if not label:
             label = os.path.basename(usb_path)
 
@@ -49,15 +41,9 @@ def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
         }
         log.info("USB Info: %s", usb_info)
         return usb_info
-    except subprocess.TimeoutExpired as e:
-        log.error("Timed out getting USB info for %s: %s", usb_path, e)
-        return None
     except PermissionError:
         log.error("Permission denied when trying to get USB info: %s", usb_path)
         return None
-    except subprocess.CalledProcessError as e:
-        log.error("Error getting USB info: %s", e)
-        return None
     except Exception as err:
         log.error("Unexpected error getting USB info: %s", err)
         return None