Removed LSBLK dependency, fixed progress bar. (Issue #178, partially #171)

README.md

@@ -11,7 +11,7 @@
 
 ## Beta Release Disclaimer
 
-**Lufus** is currently in **Beta**. 
+**Lufus** is currently in **Beta**! 
 
 Lufus is a physical drive imaging and formatting utility written in Python, inspired by **Rufus** on Windows, with the goal of delivering a greater experience for Linux users.
 While core functionality has been implemented, the project is still under active development. Users should expect bugs, incomplete features, and ongoing structural changes.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "lufus"
-version = "1.0.0"  # keep synced with [tool.briefcase] version
+version = "1.0.1b1"  # keep synced with [tool.briefcase] version
 description = "A rufus clone written in Python and designed to work with Linux."
 readme = "README.md"
 license = "MIT"
@@ -29,7 +29,7 @@ lufus = "lufus.__main__:main"
 [tool.briefcase]
 project_name = "Lufus"
 bundle = "com.github.hog185"
-version = "1.0.0"  # keep synced with [project] version
+version = "1.0.1b1"  # keep synced with [project] version
 url = "https://github.com/Hog185/Lufus"
 license.file = "LICENSE"
 author = "Hog185"
@@ -50,6 +50,7 @@ requires = [
     "psutil>=7.2",
     "pyqt6>=6.8.0",
     "pyudev>=0.24.4",
+    "requests",
     "packaging",
     "platformdirs>=4.2"
 ]

scripts/build-appimage.sh

@@ -4,17 +4,16 @@ set -euo pipefail
 ls -la
 
 # Install system dependencies
-## COMMENTED BECAUSE ALREADY HANDELED IN YAML
-# echo "------------ Installing system libraries ------------"
-# apt-get update && apt-get upgrade -y
-# INSTALLER="apt-get install -y"
-# if [[ -f requirements-system.txt ]]; then
-#     $INSTALLER $(cat requirements-system.txt) >> appimage-setup.log
-#     echo "System libraries installed."
-# else
-#     echo "requirements-system.txt not found!"
-#     exit 1
-# fi
+echo "------------ Installing system libraries ------------"
+apt-get update && apt-get upgrade -y
+INSTALLER="apt-get install -y"
+if [[ -f requirements-system.txt ]]; then
+    $INSTALLER $(cat requirements-system.txt) >> appimage-setup.log
+    echo "System libraries installed."
+else
+    echo "requirements-system.txt not found!"
+    exit 1
+fi
 
 if command -v python3 &>/dev/null; then
     PYTHON=python3

src/lufus/drives/find_usb.py

@@ -1,6 +1,6 @@
+import pyudev
 import psutil
 import os
-import subprocess
 import getpass
 from lufus import state
 from lufus.lufus_logging import get_logger
@@ -42,30 +42,30 @@ def find_usb() -> dict[str, str]:
     all_directories = _media_directories()
     dir_set = set(all_directories)
 
+    context = pyudev.Context()
+
     # Check each partition to see if it matches our potential mount points
     for part in psutil.disk_partitions(all=True):
         if part.mountpoint not in dir_set:
             continue
         mount_path = part.mountpoint
         device_node = part.device
-        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
-
-        if not label:
-            label = os.path.basename(mount_path)
-
-        usbdict[mount_path] = label
-        log.info("Found USB: %s -> %s", mount_path, label)
+        if device_node:
+            label = os.path.basename(mount_path)  # fallback
+            try:
+                st = os.stat(device_node)
+                udev_device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+                udev_label = udev_device.get("ID_FS_LABEL")
+                if udev_label:
+                    label = udev_label
+            except Exception:
+                log.debug(
+                    "Failed to get label for %s via udev; using mountpoint basename",
+                    device_node,
+                    exc_info=True,
+                )
+            usbdict[mount_path] = label
+            log.info("Found USB: %s -> %s", mount_path, label)
 
     return usbdict
 

src/lufus/drives/formatting.py

@@ -26,6 +26,26 @@ def _find_tool(name: str) -> str:
     return name
 
 
+def _get_logical_block_size(drive: str) -> int:
+    """Return the logical sector size of *drive* by reading sysfs.
+
+    Uses ``/sys/class/block/<dev>/queue/logical_block_size`` so no external
+    tool is required.  Falls back to 512 bytes if the value cannot be read.
+    """
+    dev_name = os.path.basename(drive)
+    sysfs_path = f"/sys/class/block/{dev_name}/queue/logical_block_size"
+    try:
+        with open(sysfs_path) as fh:
+            return int(fh.read().strip())
+    except Exception as exc:
+        log.warning(
+            "Could not read sector size from %s: %s. Using 512-byte default.",
+            sysfs_path,
+            exc,
+        )
+        return 512
+
+
 #######
 
 
@@ -82,18 +102,12 @@ def unmount(drive: str = None) -> bool:
 
 # mountain
 def remount(drive: str = None) -> bool:
-    mount = None
     if not drive:
         mount, drive, _ = _get_mount_and_drive()
-    else:
-        # drive was supplied by caller; resolve mount point from current state
-        _, _, mount_dict = _get_mount_and_drive()
-        # find the mount point whose device node matches the given drive
-        mount = next((mp for mp, _label in mount_dict.items()), None)
     if not drive:
         log.error("No drive node found. Cannot unmount.")
         return False
-    if not mount:
+    if not drive or not mount:
         log.error("No drive node or mount point found. Cannot remount.")
         return False
     log.info("Remounting %s -> %s...", drive, mount)
@@ -209,37 +223,8 @@ def check_device_bad_blocks() -> bool:
     passes = 2 if state.check_bad else 1
 
     # Probe the device's logical sector size so badblocks uses the real
-    # 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,
-                )
-        else:
-            log.warning(
-                "blockdev failed for %s (exit %d). Using default block size.",
-                drive,
-                probe.returncode,
-            )
-    except Exception as exc:
-        log.warning(
-            "Could not probe sector size for %s: %s. Using default block size.",
-            drive,
-            exc,
-        )
+    # device geometry. Fall back to 512 bytes if detection fails.
+    logical_block_size = _get_logical_block_size(drive)
 
     # -s = show progress, -v = verbose output
     # -n = non-destructive read-write test (safe default)
@@ -320,30 +305,10 @@ def _status(msg: str) -> None:
             "NTFS",
             "ntfs-3g",
         ),
-        1: (
-            "mkfs.vfat",
-            lambda: ["-I", "-s", str(sectors_per_cluster), "-F", "32", raw_device],
-            "FAT32",
-            "dosfstools",
-        ),
-        2: (
-            "mkfs.exfat",
-            lambda: ["-b", str(block_size), raw_device],
-            "exFAT",
-            "exfatprogs or exfat-utils",
-        ),
-        3: (
-            "mkfs.ext4",
-            lambda: ["-b", str(block_size), raw_device],
-            "ext4",
-            "e2fsprogs",
-        ),
-        4: (
-            "mkudffs",
-            lambda: ["--blocksize=" + str(sector_size), raw_device],
-            "UDF",
-            "udftools",
-        ),
+        1: ("mkfs.vfat", lambda: ["-I", "-s", str(sectors_per_cluster), "-F", "32", raw_device], "FAT32", "dosfstools"),
+        2: ("mkfs.exfat", lambda: ["-b", str(block_size), raw_device], "exFAT", "exfatprogs or exfat-utils"),
+        3: ("mkfs.ext4", lambda: ["-b", str(block_size), raw_device], "ext4", "e2fsprogs"),
+        4: ("mkudffs", lambda: ["--blocksize=" + str(sector_size), raw_device], "UDF", "udftools"),
     }
 
     if fs_type not in fs_configs:
@@ -391,30 +356,14 @@ def _apply_partition_scheme(drive: str) -> None:
             # 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%",
-                ],
+                [_find_tool("parted"), "-s", raw_device, "mkpart", "primary", "1MiB", "100%"],
                 check=True,
             )
         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%",
-                ],
+                [_find_tool("parted"), "-s", raw_device, "mkpart", "primary", "1MiB", "100%"],
                 check=True,
             )
         log.info("Partition scheme %s applied to %s.", scheme_name, raw_device)
@@ -428,11 +377,6 @@ def _apply_partition_scheme(drive: str) -> None:
 
 
 def drive_repair() -> None:
-    # todo:
-    # 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.")

src/lufus/drives/get_usb_info.py

@@ -1,6 +1,6 @@
+import pyudev
 import psutil
 import os
-import subprocess
 from typing import TypedDict
 from lufus.lufus_logging import get_logger
 
@@ -25,20 +25,25 @@ 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)
+        context = pyudev.Context()
+        st = os.stat(device_node)
+        device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+
+        # Size in bytes: udev attributes 'size' is in 512-byte sectors
+        size_attr = device.attributes.get("size")
+        try:
+            usb_size = int(size_attr) * 512 if size_attr is not None else 0
+        except (ValueError, TypeError):
+            log.warning(
+                "Unexpected non-numeric udev size attribute %r; defaulting USB size to 0",
+                size_attr,
+            )
+            usb_size = 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 = device.get("ID_FS_LABEL")
         if not label:
             label = os.path.basename(usb_path)
 
@@ -49,15 +54,6 @@ 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