From ddd243c2763dfecac6970b039b43b1a1f00df35c Mon Sep 17 00:00:00 2001 From: Erox-02 Date: Fri, 31 Jul 2026 11:22:22 +0530 Subject: [PATCH] add compression scheme selection for subvolumes --- archinstall/lib/disk/device_handler.py | 72 ++++---- archinstall/lib/disk/subvolume_menu.py | 224 +++++++++++++++---------- archinstall/lib/models/__init__.py | 138 +++++++-------- archinstall/lib/models/device.py | 213 ++++++++++++++--------- 4 files changed, 364 insertions(+), 283 deletions(-) diff --git a/archinstall/lib/disk/device_handler.py b/archinstall/lib/disk/device_handler.py index bb8636e456..e95af4f2da 100644 --- a/archinstall/lib/disk/device_handler.py +++ b/archinstall/lib/disk/device_handler.py @@ -21,6 +21,7 @@ from archinstall.lib.models.device import ( DEFAULT_ITER_TIME, BDevice, + BtrfsCompression, BtrfsMountOption, DeviceModification, DiskEncryption, @@ -28,6 +29,7 @@ LsblkInfo, ModificationStatus, PartitionFlag, + PartitionGUID, PartitionModification, PartitionTable, SubvolumeModification, @@ -74,7 +76,6 @@ def load_devices(self) -> None: if dev_lsblk_info.type == 'rom': continue - # exclude archiso loop device if dev_lsblk_info.mountpoint == ARCHISO_MOUNTPOINT: continue @@ -194,10 +195,6 @@ def get_btrfs_info( mount(dev_path, self._TMP_BTRFS_MOUNT, create_target_mountpoint=True) mountpoint = self._TMP_BTRFS_MOUNT else: - # when multiple subvolumes are mounted then the lsblk output may look like - # "mountpoint": "/mnt/archinstall/var/log" - # "mountpoints": ["/mnt/archinstall/var/log", "/mnt/archinstall/home", ..] - # so we'll determine the minimum common path and assume that's the root try: common_path = os.path.commonpath(lsblk_info.mountpoints) except ValueError: @@ -211,17 +208,9 @@ def get_btrfs_info( debug(f'Failed to read btrfs subvolume information: {err}') return subvol_infos - # It is assumed that lsblk will contain the fields as - # "mountpoints": ["/mnt/archinstall/log", "/mnt/archinstall/home", "/mnt/archinstall", ...] - # "fsroots": ["/@log", "/@home", "/@"...] - # we'll thereby map the fsroot, which are the mounted filesystem roots - # to the corresponding mountpoints btrfs_subvol_info = dict(zip(lsblk_info.fsroots, lsblk_info.mountpoints)) - # ID 256 gen 16 top level 5 path @ for line in result.splitlines(): - # expected output format: - # ID 257 gen 8 top level 5 path @home name = Path(line.split(' ')[-1]) sub_vol_mountpoint = btrfs_subvol_info.get('/' / name, None) subvol_infos.append(_BtrfsSubvolumeInfo(name, sub_vol_mountpoint)) @@ -243,17 +232,14 @@ def format( match fs_type: case FilesystemType.BTRFS | FilesystemType.XFS: - # Force overwrite options.append('-f') case FilesystemType.F2FS: options.append('-f') options.extend(('-O', 'extra_attr')) case FilesystemType.EXT2 | FilesystemType.EXT3 | FilesystemType.EXT4: - # Force create options.append('-F') case _ if fs_type.is_fat(): mkfs_type = 'fat' - # Set FAT size options.extend(('-F', fs_type.value.removeprefix(mkfs_type))) case FilesystemType.LINUX_SWAP: command = 'mkswap' @@ -342,8 +328,6 @@ def _setup_partition( requires_delete: bool, arch: str | None = None, ) -> None: - # when we require a delete and the partition to be (re)created - # already exists then we have to delete it first if requires_delete and part_mod.status in [ModificationStatus.MODIFY, ModificationStatus.DELETE]: info(f'Delete existing partition: {part_mod.safe_dev_path}') part_info = self.find_partition(part_mod.safe_dev_path) @@ -400,7 +384,6 @@ def _setup_partition( elif PartitionFlag.LINUX_HOME not in part_mod.flags and part_mod.is_home(): partition.setFlag(PartitionFlag.LINUX_HOME.flag_id) - # the partition has a path now that it has been added part_mod.dev_path = Path(partition.path) def fetch_part_info(self, path: Path) -> LsblkInfo: @@ -430,7 +413,16 @@ def create_lvm_btrfs_subvolumes( ) -> None: info(f'Creating subvolumes: {path}') - mount(path, self._TMP_BTRFS_MOUNT, create_target_mountpoint=True) + all_mount_options = mount_options.copy() + for subvol in btrfs_subvols: + if subvol.compression != BtrfsCompression.NONE: + comp_opt = subvol.compression.mount_option + if comp_opt and comp_opt not in all_mount_options: + # Remove any existing compress= options to avoid conflicts + all_mount_options = [o for o in all_mount_options if not o.startswith("compress=")] + all_mount_options.append(comp_opt) + + mount(path, self._TMP_BTRFS_MOUNT, create_target_mountpoint=True, options=all_mount_options) for sub_vol in sorted(btrfs_subvols, key=lambda x: x.name): debug(f'Creating subvolume: {sub_vol.name}') @@ -445,7 +437,7 @@ def create_lvm_btrfs_subvolumes( except SysCallError as err: raise DiskError(f'Could not set nodatacow attribute at {subvol_path}: {err}') - if BtrfsMountOption.compress.value in mount_options: + if sub_vol.compression != BtrfsCompression.NONE: try: SysCommand(f'chattr +c {subvol_path}') except SysCallError as err: @@ -460,7 +452,6 @@ def create_btrfs_volumes( ) -> None: info(f'Creating subvolumes: {part_mod.safe_dev_path}') - # unlock the partition first if it's encrypted if enc_conf is not None and part_mod in enc_conf.partitions: if not part_mod.mapper_name: raise ValueError('No device path specified for modification') @@ -479,11 +470,19 @@ def create_btrfs_volumes( luks_handler = None dev_path = part_mod.safe_dev_path + mount_options = part_mod.mount_options.copy() + for subvol in part_mod.btrfs_subvols: + if subvol.compression != BtrfsCompression.NONE: + comp_opt = subvol.compression.mount_option + if comp_opt and comp_opt not in mount_options: + mount_options = [o for o in mount_options if not o.startswith("compress=")] + mount_options.append(comp_opt) + mount( dev_path, self._TMP_BTRFS_MOUNT, create_target_mountpoint=True, - options=part_mod.mount_options, + options=mount_options, ) for sub_vol in sorted(part_mod.btrfs_subvols, key=lambda x: x.name): @@ -493,6 +492,12 @@ def create_btrfs_volumes( SysCommand(f'btrfs subvolume create -p {subvol_path}') + if sub_vol.compression != BtrfsCompression.NONE: + try: + SysCommand(f'chattr +c {subvol_path}') + except SysCallError as err: + raise DiskError(f'Could not set compress attribute at {subvol_path}: {err}') + umount(dev_path) if luks_handler is not None and luks_handler.mapper_dev is not None: @@ -506,7 +511,6 @@ def umount_all_existing(self, device_path: Path) -> None: for partition in existing_partitions: debug(f'Unmounting: {partition.path}') - # un-mount for existing encrypted partitions if partition.fs_type == FilesystemType.CRYPTO_LUKS: Luks2(partition.path).lock() else: @@ -517,12 +521,8 @@ def partition( modification: DeviceModification, partition_table: PartitionTable | None = None, ) -> None: - """ - Create a partition table on the block device and create all partitions. - """ partition_table = partition_table or self.partition_table - # WARNING: the entire device will be wiped and all data lost if modification.wipe: if partition_table.is_mbr() and len(modification.partitions) > 3: raise DiskError('Too many partitions on disk, MBR disks can only have 3 primary partitions') @@ -535,14 +535,11 @@ def partition( info(f'Creating partitions: {modification.device_path}') - # don't touch existing partitions filtered_part = [p for p in modification.partitions if not p.exists()] arch = platform.machine() for part_mod in filtered_part: - # if the entire disk got nuked then we don't have to delete - # any existing partitions anymore because they're all gone already requires_delete = modification.wipe is False self._setup_partition( part_mod, @@ -554,14 +551,11 @@ def partition( disk.commit() - # Wipe filesystem/LVM signatures from newly created partitions - # to prevent "signature detected" errors for part_mod in filtered_part: if part_mod.dev_path: debug(f'Wiping signatures from: {part_mod.dev_path}') SysCommand(f'wipefs --all {part_mod.dev_path}') - # Sync with udev after wiping signatures if filtered_part: udev_sync() @@ -607,20 +601,10 @@ def partprobe(self, path: Path | None = None) -> None: error(f'"{command}" failed to run (continuing anyway): {err}') def _wipe(self, dev_path: Path) -> None: - """ - Wipe a device (partition or otherwise) of meta-data, be it file system, LVM, etc. - @param dev_path: Device path of the partition to be wiped. - @type dev_path: str - """ with open(dev_path, 'wb') as p: p.write(bytearray(1024)) def wipe_dev(self, block_device: BDevice) -> None: - """ - Wipe the block device of meta-data, be it file system, LVM, etc. - This is not intended to be secure, but rather to ensure that - auto-discovery tools don't recognize anything here. - """ info(f'Wiping partitions and metadata: {block_device.device_info.path}') for partition in block_device.partition_infos: diff --git a/archinstall/lib/disk/subvolume_menu.py b/archinstall/lib/disk/subvolume_menu.py index 94854b3c93..8b0d4171fa 100644 --- a/archinstall/lib/disk/subvolume_menu.py +++ b/archinstall/lib/disk/subvolume_menu.py @@ -1,102 +1,142 @@ from pathlib import Path from typing import assert_never, override - -from archinstall.lib.menu.helpers import Input +from archinstall.lib.models.device import BtrfsCompression, SubvolumeModification +from archinstall.lib.menu.helpers import Input, Selection from archinstall.lib.menu.list_manager import ListManager from archinstall.lib.menu.util import prompt_dir -from archinstall.lib.models.device import SubvolumeModification from archinstall.lib.translationhandler import tr +from archinstall.tui.menu_item import MenuItem, MenuItemGroup from archinstall.tui.result import ResultType class SubvolumeMenu(ListManager[SubvolumeModification]): def __init__( - self, - btrfs_subvols: list[SubvolumeModification], - prompt: str | None = None, - ): - self._actions = [ - tr('Add subvolume'), - tr('Edit subvolume'), - tr('Delete subvolume'), - ] - - super().__init__( - btrfs_subvols, - [self._actions[0]], - self._actions[1:], - prompt, - ) - - async def show(self) -> list[SubvolumeModification] | None: - return await super()._run() - - @override - def selected_action_display(self, selection: SubvolumeModification) -> str: - return str(selection.name) - - async def _add_subvolume(self, preset: SubvolumeModification | None = None) -> SubvolumeModification | None: - def validate(value: str | None) -> str | None: - if value: - return None - return tr('Value cannot be empty') - - result = await Input( - header=tr('Enter subvolume name'), - allow_skip=True, - default_value=str(preset.name) if preset else None, - validator_callback=validate, - ).show() - - match result.type_: - case ResultType.Skip: - return preset - case ResultType.Selection: - name = result.get_value() - case ResultType.Reset: - raise ValueError('Unhandled result type') - case _: - assert_never(result.type_) - - header = f'{tr("Subvolume name")}: {name}\n\n' - header += tr('Enter subvolume mountpoint') - - path = await prompt_dir( - header=header, - allow_skip=True, - validate=True, - must_exist=False, - ) - - if not path: - return preset - - return SubvolumeModification(Path(name), path) - - @override - async def handle_action( - self, - action: str, - entry: SubvolumeModification | None, - data: list[SubvolumeModification], - ) -> list[SubvolumeModification]: - if action == self._actions[0]: - new_subvolume = await self._add_subvolume() - - if new_subvolume is not None: - # in case a user with the same username as an existing user - # was created we'll replace the existing one - data = [d for d in data if d.name != new_subvolume.name] - data += [new_subvolume] - elif entry is not None: - if action == self._actions[1]: - new_subvolume = await self._add_subvolume(entry) - - if new_subvolume is not None: - # we'll remove the original subvolume and add the modified version - data = [d for d in data if d.name != entry.name and d.name != new_subvolume.name] - data += [new_subvolume] - elif action == self._actions[2]: - data = [d for d in data if d != entry] - - return data + self, + btrfs_subvols: list[SubvolumeModification], + prompt: str | None = None, + ): + self._actions = [ + tr('Add subvolume'), + tr('Edit subvolume'), + tr('Delete subvolume'), + ] + + super().__init__( + btrfs_subvols, + [self._actions[0]], + self._actions[1:], + prompt, + ) + + async def show(self) -> list[SubvolumeModification] | None: + return await super()._run() + + @override + def selected_action_display(self, selection: SubvolumeModification) -> str: + base = str(selection.name) + if selection.compression != BtrfsCompression.ZSTD_3: + base += f" [{selection.compression.value}]" + return base + + async def _add_subvolume(self, preset: SubvolumeModification | None = None) -> SubvolumeModification | None: + def validate(value: str | None) -> str | None: + if value: + return None + return tr('Value cannot be empty') + + result = await Input( + header=tr('Enter subvolume name'), + allow_skip=True, + default_value=str(preset.name) if preset else None, + validator_callback=validate, + ).show() + + match result.type_: + case ResultType.Skip: + return preset + case ResultType.Selection: + name = result.get_value() + case ResultType.Reset: + raise ValueError('Unhandled result type') + case _: + assert_never(result.type_) + + header = f'{tr("Subvolume name")}: {name}\n\n' + header += tr('Enter subvolume mountpoint') + + path = await prompt_dir( + header=header, + allow_skip=True, + validate=True, + must_exist=False, + ) + + if not path: + return preset + + default_compression = preset.compression if preset else BtrfsCompression.ZSTD_3 + compression = await self._select_compression(default_compression) + + if compression is None: + if preset: + return preset + compression = BtrfsCompression.ZSTD_3 + + return SubvolumeModification( + Path(name), + path, + compression + ) + async def _select_compression(self, default: BtrfsCompression) -> BtrfsCompression | None: + header = tr('Select compression algorithm for this subvolume') + '\n\n' + header += tr('Higher compression levels save more space but are slower') + '\n' + header += tr('ZSTD is generally recommended for most use cases') + '\n\n' + header += tr('Selection') + ':' + + items = [] + for display_name, comp_value in BtrfsCompression.get_ui_options(): + label = display_name + if comp_value == default: + label = f"* {label} (default)" + items.append(MenuItem(label, value=comp_value)) + + group = MenuItemGroup(items, sort_items=False) + + result = await Selection[BtrfsCompression]( + group, + header=header, + allow_skip=True, + ).show() + + match result.type_: + case ResultType.Selection: + return result.get_value() + case ResultType.Skip: + return default + case _: + return None + + @override + async def handle_action( + self, + action: str, + entry: SubvolumeModification | None, + data: list[SubvolumeModification], + ) -> list[SubvolumeModification]: + if action == self._actions[0]: + new_subvolume = await self._add_subvolume() + + if new_subvolume is not None: + data = [d for d in data if d.name != new_subvolume.name] + data += [new_subvolume] + elif entry is not None: + if action == self._actions[1]: + new_subvolume = await self._add_subvolume(entry) + + if new_subvolume is not None: + data = [d for d in data if d.name != entry.name and d.name != new_subvolume.name] + data += [new_subvolume] + elif action == self._actions[2]: + data = [d for d in data if d != entry] + + return data diff --git a/archinstall/lib/models/__init__.py b/archinstall/lib/models/__init__.py index 5f808c96f2..e951639efc 100644 --- a/archinstall/lib/models/__init__.py +++ b/archinstall/lib/models/__init__.py @@ -1,30 +1,31 @@ from archinstall.lib.models.application import ApplicationConfiguration, Audio, AudioConfiguration, BluetoothConfiguration, PrintServiceConfiguration from archinstall.lib.models.bootloader import Bootloader from archinstall.lib.models.device import ( - BDevice, - DeviceGeometry, - DeviceModification, - DiskEncryption, - DiskLayoutConfiguration, - DiskLayoutType, - EncryptionType, - Fido2Device, - FilesystemType, - LsblkInfo, - LvmConfiguration, - LvmLayoutType, - LvmVolume, - LvmVolumeGroup, - ModificationStatus, - PartitionFlag, - PartitionModification, - PartitionTable, - PartitionType, - SectorSize, - Size, - SubvolumeModification, - Unit, - _DeviceInfo, + BDevice, + BtrfsCompression, + DeviceGeometry, + DeviceModification, + DiskEncryption, + DiskLayoutConfiguration, + DiskLayoutType, + EncryptionType, + Fido2Device, + FilesystemType, + LsblkInfo, + LvmConfiguration, + LvmLayoutType, + LvmVolume, + LvmVolumeGroup, + ModificationStatus, + PartitionFlag, + PartitionModification, + PartitionTable, + PartitionType, + SectorSize, + Size, + SubvolumeModification, + Unit, + _DeviceInfo, ) from archinstall.lib.models.locale import LocaleConfiguration from archinstall.lib.models.mirrors import CustomRepository, MirrorConfiguration, MirrorRegion @@ -34,48 +35,49 @@ from archinstall.lib.models.users import PasswordStrength, User __all__ = [ - 'ApplicationConfiguration', - 'Audio', - 'AudioConfiguration', - 'BDevice', - 'BluetoothConfiguration', - 'Bootloader', - 'CustomRepository', - 'DeviceGeometry', - 'DeviceModification', - 'DiskEncryption', - 'DiskLayoutConfiguration', - 'DiskLayoutType', - 'EncryptionType', - 'Fido2Device', - 'FilesystemType', - 'LocalPackage', - 'LocaleConfiguration', - 'LsblkInfo', - 'LvmConfiguration', - 'LvmLayoutType', - 'LvmVolume', - 'LvmVolumeGroup', - 'MirrorConfiguration', - 'MirrorRegion', - 'ModificationStatus', - 'NetworkConfiguration', - 'Nic', - 'NicType', - 'PackageSearch', - 'PackageSearchResult', - 'PartitionFlag', - 'PartitionModification', - 'PartitionTable', - 'PartitionType', - 'PasswordStrength', - 'PrintServiceConfiguration', - 'ProfileConfiguration', - 'Repository', - 'SectorSize', - 'Size', - 'SubvolumeModification', - 'Unit', - 'User', - '_DeviceInfo', + 'ApplicationConfiguration', + 'Audio', + 'AudioConfiguration', + 'BDevice', + 'BluetoothConfiguration', + 'Bootloader', + 'BtrfsCompression', + 'CustomRepository', + 'DeviceGeometry', + 'DeviceModification', + 'DiskEncryption', + 'DiskLayoutConfiguration', + 'DiskLayoutType', + 'EncryptionType', + 'Fido2Device', + 'FilesystemType', + 'LocalPackage', + 'LocaleConfiguration', + 'LsblkInfo', + 'LvmConfiguration', + 'LvmLayoutType', + 'LvmVolume', + 'LvmVolumeGroup', + 'MirrorConfiguration', + 'MirrorRegion', + 'ModificationStatus', + 'NetworkConfiguration', + 'Nic', + 'NicType', + 'PackageSearch', + 'PackageSearchResult', + 'PartitionFlag', + 'PartitionModification', + 'PartitionTable', + 'PartitionType', + 'PasswordStrength', + 'PrintServiceConfiguration', + 'ProfileConfiguration', + 'Repository', + 'SectorSize', + 'Size', + 'SubvolumeModification', + 'Unit', + 'User', + '_DeviceInfo', ] diff --git a/archinstall/lib/models/device.py b/archinstall/lib/models/device.py index 20559d70d9..cbcf407aaf 100644 --- a/archinstall/lib/models/device.py +++ b/archinstall/lib/models/device.py @@ -1,10 +1,9 @@ -import builtins import math import uuid from dataclasses import dataclass, field from enum import Enum, StrEnum, auto from pathlib import Path -from typing import Any, NotRequired, Self, TypedDict, override +from typing import Any, NotRequired, Self, TypedDict, override, Optional from uuid import UUID import parted @@ -62,7 +61,6 @@ class DiskLayoutConfiguration(SubConfig): disk_encryption: DiskEncryption | None = None btrfs_options: BtrfsOptions | None = None - # used for pre-mounted config mountpoint: Path | None = None @override @@ -181,7 +179,6 @@ def parse_arg( flags=flags, btrfs_subvols=SubvolumeModification.parse_args(partition.get('btrfs', [])), ) - # special 'invisible' attr to internally identify the part mod device_partition._obj_id = partition['obj_id'] device_partitions.append(device_partition) @@ -222,7 +219,6 @@ def parse_arg( elif last.end > total_size.align(): raise ValueError('Partition too large for device') - # Parse LVM configuration from settings if (lvm_arg := disk_config.get('lvm_config', None)) is not None: config.lvm_config = LvmConfiguration.parse_arg(lvm_arg, config) @@ -268,26 +264,26 @@ class Units(Enum): class Unit(Enum): - B = 1 # byte - kB = 1000**1 # kilobyte - MB = 1000**2 # megabyte - GB = 1000**3 # gigabyte - TB = 1000**4 # terabyte - PB = 1000**5 # petabyte - EB = 1000**6 # exabyte - ZB = 1000**7 # zettabyte - YB = 1000**8 # yottabyte - - KiB = 1024**1 # kibibyte - MiB = 1024**2 # mebibyte - GiB = 1024**3 # gibibyte - TiB = 1024**4 # tebibyte - PiB = 1024**5 # pebibyte - EiB = 1024**6 # exbibyte - ZiB = 1024**7 # zebibyte - YiB = 1024**8 # yobibyte - - sectors = 'sectors' # size in sector + B = 1 + kB = 1000**1 + MB = 1000**2 + GB = 1000**3 + TB = 1000**4 + PB = 1000**5 + EB = 1000**6 + ZB = 1000**7 + YB = 1000**8 + + KiB = 1024**1 + MiB = 1024**2 + GiB = 1024**3 + TiB = 1024**4 + PiB = 1024**5 + EiB = 1024**6 + ZiB = 1024**7 + YiB = 1024**8 + + sectors = 'sectors' @classmethod def get_all_units(cls) -> list[str]: @@ -335,9 +331,6 @@ def parse_args(cls, arg: _SectorSizeSerialization) -> Self: ) def normalize(self) -> int: - """ - will normalize the value of the unit to Byte - """ return int(self.value * self.unit.value) @@ -439,8 +432,6 @@ def si_unit_highest(self, include_unit: bool = True) -> str: all_si_values = [self.convert(si) for si in si_units] filtered = filter(lambda x: x.value >= 1, all_si_values) - # we have to get the max by the unit value as we're interested - # in getting the value in the highest possible unit without floats si_value = max(filtered, key=lambda x: x.unit.value) if include_unit: @@ -465,9 +456,6 @@ def gpt_end(self) -> Size: return self - Size(1, Unit.MiB, self.sector_size) def _normalize(self) -> int: - """ - will normalize the value of the unit to Byte - """ if self.unit == Unit.sectors and self.sector_size is not None: return self.value * self.sector_size.normalize() return int(self.value * self.unit.value) @@ -514,6 +502,96 @@ class BtrfsMountOption(Enum): nodatacow = 'nodatacow' + + +class BtrfsCompression(StrEnum): + NONE = "none" + LZO = "lzo" + ZSTD_1 = "zstd:1" + ZSTD_2 = "zstd:2" + ZSTD_3 = "zstd:3" # def + ZSTD_4 = "zstd:4" + ZSTD_5 = "zstd:5" + ZSTD_6 = "zstd:6" + ZSTD_7 = "zstd:7" + ZSTD_8 = "zstd:8" + ZSTD_9 = "zstd:9" + ZSTD_10 = "zstd:10" + ZSTD_11 = "zstd:11" + ZSTD_12 = "zstd:12" + ZSTD_13 = "zstd:13" + ZSTD_14 = "zstd:14" + ZSTD_15 = "zstd:15" + ZSTD_16 = "zstd:16" + ZSTD_17 = "zstd:17" + ZSTD_18 = "zstd:18" + ZSTD_19 = "zstd:19" + + @property + def mount_option(self) -> str: + """Convert to mount option string for fstab""" + if self == BtrfsCompression.NONE: + return "" + return f"compress={self.value}" + + @property + def display_name(self) -> str: + """Get human-readable display name for UI""" + mapping = { + BtrfsCompression.NONE: "None (no compression)", + BtrfsCompression.LZO: "LZO (fast, moderate compression)", + BtrfsCompression.ZSTD_1: "ZSTD:1 (fastest)", + BtrfsCompression.ZSTD_2: "ZSTD:2 (very fast)", + BtrfsCompression.ZSTD_3: "ZSTD:3 (default, balanced)", + BtrfsCompression.ZSTD_4: "ZSTD:4", + BtrfsCompression.ZSTD_5: "ZSTD:5", + BtrfsCompression.ZSTD_6: "ZSTD:6", + BtrfsCompression.ZSTD_7: "ZSTD:7 (better compression)", + BtrfsCompression.ZSTD_8: "ZSTD:8", + BtrfsCompression.ZSTD_9: "ZSTD:9", + BtrfsCompression.ZSTD_10: "ZSTD:10", + BtrfsCompression.ZSTD_11: "ZSTD:11", + BtrfsCompression.ZSTD_12: "ZSTD:12", + BtrfsCompression.ZSTD_13: "ZSTD:13", + BtrfsCompression.ZSTD_14: "ZSTD:14", + BtrfsCompression.ZSTD_15: "ZSTD:15 (excellent compression)", + BtrfsCompression.ZSTD_16: "ZSTD:16", + BtrfsCompression.ZSTD_17: "ZSTD:17", + BtrfsCompression.ZSTD_18: "ZSTD:18", + BtrfsCompression.ZSTD_19: "ZSTD:19 (maximum compression)", + } + return mapping[self] + + @classmethod + def from_string(cls, value: str) -> "BtrfsCompression": + if not value or value == "none": + return BtrfsCompression.NONE + + if value.startswith("zstd:"): + level = value.split(":")[1] + try: + return getattr(BtrfsCompression, f"ZSTD_{level}") + except AttributeError: + return BtrfsCompression.ZSTD_3 + + if value == "lzo": + return BtrfsCompression.LZO + + return BtrfsCompression.ZSTD_3 + + @classmethod + def get_ui_options(cls) -> list[tuple[str, "BtrfsCompression"]]: + return [ + ("None (no compression)", BtrfsCompression.NONE), + ("LZO (fast, moderate compression)", BtrfsCompression.LZO), + ("ZSTD:1 (fastest)", BtrfsCompression.ZSTD_1), + ("ZSTD:3 (default, balanced)", BtrfsCompression.ZSTD_3), + ("ZSTD:7 (better compression, slower)", BtrfsCompression.ZSTD_7), + ("ZSTD:15 (excellent compression, slow)", BtrfsCompression.ZSTD_15), + ("ZSTD:19 (maximum compression, very slow)", BtrfsCompression.ZSTD_19), + ] + + @dataclass class _BtrfsSubvolumeInfo: name: Path @@ -658,12 +736,13 @@ def from_disk(cls, disk: Disk) -> Self: class _SubvolumeModificationSerialization(TypedDict): name: str mountpoint: str - + compression: NotRequired[str] @dataclass class SubvolumeModification: name: Path | str mountpoint: Path | None = None + compression: BtrfsCompression = BtrfsCompression.ZSTD_3 @classmethod def from_existing_subvol_info(cls, info: _BtrfsSubvolumeInfo) -> Self: @@ -679,16 +758,15 @@ def parse_args(cls, subvol_args: list[_SubvolumeModificationSerialization]) -> l mountpoint = Path(entry['mountpoint']) if entry['mountpoint'] else None - mods.append(cls(entry['name'], mountpoint)) + compression_str = entry.get('compression', 'zstd:3') + compression = BtrfsCompression.from_string(compression_str) + + mods.append(cls(entry['name'], mountpoint, compression)) return mods @property def relative_mountpoint(self) -> Path: - """ - Will return the relative path based on the anchor - e.g. Path('/mnt/test') -> Path('mnt/test') - """ if self.mountpoint is not None: return self.mountpoint.relative_to(self.mountpoint.anchor) @@ -702,11 +780,22 @@ def is_root(self) -> bool: def is_default_root(self) -> bool: return self.name == Path('@') and self.is_root() + def get_mount_options(self) -> str: + return self.compression.mount_option + def json(self) -> _SubvolumeModificationSerialization: - return {'name': str(self.name), 'mountpoint': str(self.mountpoint)} + return { + 'name': str(self.name), + 'mountpoint': str(self.mountpoint), + 'compression': self.compression.value + } - def table_data(self) -> _SubvolumeModificationSerialization: - return self.json() + def table_data(self) -> dict[str, str]: + return { + 'name': str(self.name), + 'mountpoint': str(self.mountpoint), + 'compression': self.compression.value + } class DeviceGeometry: @@ -804,11 +893,6 @@ def from_string(cls, s: str) -> Self | None: class PartitionGUID(Enum): - """ - A list of Partition type GUIDs (lsblk -o+PARTTYPE) can be found here: - https://en.wikipedia.org/wiki/GUID_Partition_Table#Partition_type_GUIDs - """ - LINUX_ROOT_AARCH64 = 'B921B045-1DF0-41C3-AF44-4C6F280D3FAE' LINUX_ROOT_X86_64 = '4F68BCE3-E8CD-4DB1-96E7-FBCAF984B709' @@ -830,8 +914,6 @@ class FilesystemType(StrEnum): XFS = auto() LINUX_SWAP = 'linux-swap' - # this is not a FS known to parted, so be careful - # with the usage from this enum CRYPTO_LUKS = 'crypto_LUKS' def is_crypto(self) -> bool: @@ -890,7 +972,6 @@ class PartitionModification: flags: list[PartitionFlag] = field(default_factory=list) btrfs_subvols: list[SubvolumeModification] = field(default_factory=list) - # only set if the device was created or exists dev_path: Path | None = None partn: int | None = None partuuid: str | None = None @@ -899,7 +980,6 @@ class PartitionModification: _obj_id: UUID | str = field(init=False) def __post_init__(self) -> None: - # needed to use the object as a dictionary key due to hash func if not hasattr(self, '_obj_id'): self._obj_id = uuid.uuid4() @@ -965,10 +1045,6 @@ def from_existing_partition(cls, partition_info: _PartitionInfo) -> Self: @property def relative_mountpoint(self) -> Path: - """ - Will return the relative path based on the anchor - e.g. Path('/mnt/test') -> Path('mnt/test') - """ if self.mountpoint: return self.mountpoint.relative_to(self.mountpoint.anchor) @@ -1038,9 +1114,6 @@ def invert_flag(self, flag: PartitionFlag) -> None: self.set_flag(flag) def json(self) -> _PartitionModificationSerialization: - """ - Called for configuration settings - """ return { 'obj_id': self.obj_id, 'status': self.status.value, @@ -1056,9 +1129,6 @@ def json(self) -> _PartitionModificationSerialization: } def table_data(self) -> dict[str, str]: - """ - Called for displaying data in table format - """ part_mod = { 'Status': self.status.value, 'Device': str(self.dev_path) if self.dev_path else '', @@ -1080,9 +1150,6 @@ def table_data(self) -> dict[str, str]: class LvmLayoutType(Enum): Default = 'default' - - # Manual = 'manual_lvm' - def display_msg(self) -> str: match self: case LvmLayoutType.Default: @@ -1147,15 +1214,12 @@ class LvmVolume: mount_options: list[str] = field(default_factory=list) btrfs_subvols: list[SubvolumeModification] = field(default_factory=list) - # volume group name vg_name: str | None = None - # mapper device path /dev// dev_path: Path | None = None _obj_id: uuid.UUID | str = field(init=False) def __post_init__(self) -> None: - # needed to use the object as a dictionary key due to hash func if not hasattr(self, '_obj_id'): self._obj_id = uuid.uuid4() @@ -1196,10 +1260,6 @@ def safe_fs_type(self) -> FilesystemType: @property def relative_mountpoint(self) -> Path: - """ - Will return the relative path based on the anchor - e.g. Path('/mnt/test') -> Path('mnt/test') - """ if self.mountpoint is not None: return self.mountpoint.relative_to(self.mountpoint.anchor) @@ -1296,7 +1356,6 @@ class LvmConfiguration: vol_groups: list[LvmVolumeGroup] def __post_init__(self) -> None: - # make sure all volume groups have unique PVs pvs = [] for group in self.vol_groups: for pv in group.pvs: @@ -1315,8 +1374,7 @@ def parse_arg(cls, arg: _LvmConfigurationSerialization, disk_config: DiskLayoutC lvm_pvs = [] for mod in disk_config.device_modifications: for part in mod.partitions: - # FIXME: 'lvm_pvs' does not seem like it can ever exist in the 'arg' serialization - if part.obj_id in arg.get('lvm_pvs', []): # type: ignore[operator] + if part.obj_id in arg.get('lvm_pvs', []): lvm_pvs.append(part) return cls( @@ -1428,9 +1486,6 @@ def get_root_partition(self) -> PartitionModification | None: return next(filtered, None) def json(self) -> _DeviceModificationSerialization: - """ - Called when generating configuration files - """ return { 'device': str(self.device.device_info.path), 'wipe': self.wipe, @@ -1504,7 +1559,7 @@ def json(self) -> _DiskEncryptionSerialization: if self.hsm_device: obj['hsm_device'] = self.hsm_device.json() - if self.iter_time != DEFAULT_ITER_TIME: # Only include if not default + if self.iter_time != DEFAULT_ITER_TIME: obj['iter_time'] = self.iter_time return obj @@ -1520,7 +1575,7 @@ def validate_enc( for part in mod.partitions: partitions.append(part) - if len(partitions) > 2: # assume one boot and at least 2 additional + if len(partitions) > 2: if lvm_config: return False