RomanNum3ral
/
archinstall
mirror of https://github.com/archlinux/archinstall.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add compression scheme selection for subvolumes

This commit is contained in:
Erox-02 2026-07-31 11:22:22 +05:30
parent 3576eec234
commit ddd243c276
4 changed files with 349 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
archinstall/lib/disk/device_handler.py
View File

@ -21,6 +21,7 @@ from archinstall.lib.log import debug, error, info, log
from archinstall.lib.models.device import (
DEFAULT_ITER_TIME,
BDevice,
BtrfsCompression,
BtrfsMountOption,
DeviceModification,
DiskEncryption,
@ -28,6 +29,7 @@ from archinstall.lib.models.device import (
LsblkInfo,
ModificationStatus,
PartitionFlag,
PartitionGUID,
PartitionModification,
PartitionTable,
SubvolumeModification,
@ -74,7 +76,6 @@ class DeviceHandler:
if dev_lsblk_info.type == 'rom':
continue
# exclude archiso loop device
if dev_lsblk_info.mountpoint == ARCHISO_MOUNTPOINT:
continue
@ -194,10 +195,6 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
) -> 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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
) -> 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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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 @@ class DeviceHandler:
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:

196
archinstall/lib/disk/subvolume_menu.py
View File

@ -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'),
]
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,
)
super().__init__(
btrfs_subvols,
[self._actions[0]],
self._actions[1:],
prompt,
)
async def show(self) -> list[SubvolumeModification] | None:
return await super()._run()
async def show(self) -> list[SubvolumeModification] | None:
return await super()._run()
@override
def selected_action_display(self, selection: SubvolumeModification) -> str:
return str(selection.name)
@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')
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()
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_)
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')
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,
)
path = await prompt_dir(
header=header,
allow_skip=True,
validate=True,
must_exist=False,
)
if not path:
return preset
if not path:
return preset
return SubvolumeModification(Path(name), path)
default_compression = preset.compression if preset else BtrfsCompression.ZSTD_3
compression = await self._select_compression(default_compression)
@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 compression is None:
if preset:
return preset
compression = BtrfsCompression.ZSTD_3
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)
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') + ':'
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]
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))
return data
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

138
archinstall/lib/models/__init__.py
View File

@ -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.profile import ProfileConfiguration
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',
]

211
archinstall/lib/models/device.py
View File

@ -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 @@ class DiskLayoutConfiguration(SubConfig):
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 @@ class DiskLayoutConfiguration(SubConfig):
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
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 # 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
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' # size in sector
sectors = 'sectors'
@classmethod
def get_all_units(cls) -> list[str]:
@ -335,9 +331,6 @@ class SectorSize:
)
def normalize(self) -> int:
"""
will normalize the value of the unit to Byte
"""
return int(self.value * self.unit.value)
@ -439,8 +432,6 @@ class Size:
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 @@ class 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 @@ class _DeviceInfo:
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 @@ class SubvolumeModification:
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 @@ class SubvolumeModification:
def is_default_root(self) -> bool:
return self.name == Path('@') and self.is_root()
def json(self) -> _SubvolumeModificationSerialization:
return {'name': str(self.name), 'mountpoint': str(self.mountpoint)}
def get_mount_options(self) -> str:
return self.compression.mount_option
def table_data(self) -> _SubvolumeModificationSerialization:
return self.json()
def json(self) -> _SubvolumeModificationSerialization:
return {
'name': str(self.name),
'mountpoint': str(self.mountpoint),
'compression': self.compression.value
}
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 @@ class PartitionFlag(PartitionFlagDataMixin, Enum):
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 @@ class PartitionModification:
@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 @@ class PartitionModification:
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 @@ class PartitionModification:
}
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 @@ class PartitionModification:
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/<vg>/<vol>
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 @@ class LvmVolume:
@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 @@ class LvmConfiguration:
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 @@ class DeviceModification:
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 @@ class DiskEncryption:
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 @@ class DiskEncryption:
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