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

disk: rework manual partitioning (#3071) 

* disk: rework manual partitioning

* disk: fix end alignment

* disk: validate config

* test: fix disk config partition overlap
This commit is contained in:
codefiles 2025-01-07 15:44:03 -05:00 committed by GitHub
parent 25b7bbb0ed
commit 9163e8c84c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 338 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
archinstall/__init__.py
View File

@ -274,7 +274,11 @@ def post_process_arguments(args: dict[str, Any]) -> None:
path = args['plugin']
load_plugin(path)
load_config()
try:
load_config()
except ValueError as err:
warn(str(err))
exit(1)
define_arguments()

53
archinstall/lib/disk/device_model.py
View File

@ -13,6 +13,7 @@ from pydantic import BaseModel, Field, ValidationInfo, field_serializer, field_v
from ..exceptions import DiskError, SysCallError
from ..general import SysCommand
from ..hardware import SysInfo
from ..output import debug
from ..storage import storage
@ -148,6 +149,41 @@ class DiskLayoutConfiguration:
device_modification.partitions = device_partitions
device_modifications.append(device_modification)
using_gpt = SysInfo.has_uefi()
for dev_mod in device_modifications:
partitions = sorted(dev_mod.partitions, key=lambda p: p.start)
for i, current_partition in enumerate(partitions[1:], start=1):
previous_partition = partitions[i - 1]
if (
current_partition.status == ModificationStatus.Create
and current_partition.start < previous_partition.end
):
raise ValueError('Partitions overlap')
partitions = [
part_mod for part_mod in dev_mod.partitions
if part_mod.status == ModificationStatus.Create
]
if not partitions:
continue
for part in partitions:
if (
part.start != part.start.align()
or part.length != part.length.align()
):
raise ValueError('Partition is misaligned')
total_size = dev_mod.device.device_info.total_size
if using_gpt and partitions[-1].end > total_size.gpt_end():
raise ValueError('Partition overlaps backup GPT header')
elif partitions[-1].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)
@ -355,6 +391,14 @@ class Size:
else:
return self.si_unit_highest(include_unit)
def align(self) -> Size:
align_norm = Size(1, Unit.MiB, self.sector_size)._normalize()
src_norm = self._normalize()
return self - Size(abs(src_norm % align_norm), Unit.B, self.sector_size)
def gpt_end(self) -> Size:
return self - Size(33, Unit.sectors, self.sector_size)
def _normalize(self) -> int:
"""
will normalize the value of the unit to Byte
@ -907,11 +951,18 @@ class PartitionModification:
def is_modify(self) -> bool:
return self.status == ModificationStatus.Modify
def is_delete(self) -> bool:
return self.status == ModificationStatus.Delete
def exists(self) -> bool:
return self.status == ModificationStatus.Exist
def is_exists_or_modify(self) -> bool:
return self.status in [ModificationStatus.Exist, ModificationStatus.Modify]
return self.status in [
ModificationStatus.Exist,
ModificationStatus.Delete,
ModificationStatus.Modify
]
def is_create_or_modify(self) -> bool:
return self.status in [ModificationStatus.Create, ModificationStatus.Modify]

477
archinstall/lib/disk/partitioning_menu.py
View File

@ -1,7 +1,6 @@
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, override
@ -14,7 +13,6 @@ from ..utils.util import prompt_dir
from .device_model import (
BDevice,
BtrfsMountOption,
DeviceGeometry,
FilesystemType,
ModificationStatus,
PartitionFlag,
@ -34,17 +32,55 @@ if TYPE_CHECKING:
_: Callable[[str], DeferredTranslation]
@dataclass
class DefaultFreeSector:
start: Size
end: Size
class FreeSpace:
def __init__(self, start: Size, end: Size) -> None:
self.start = start
self.end = end
@property
def length(self) -> Size:
return self.end - self.start
def table_data(self) -> dict[str, str]:
"""
Called for displaying data in table format
"""
return {
'Start': self.start.format_size(Unit.sectors, self.start.sector_size, include_unit=False),
'End': self.end.format_size(Unit.sectors, self.start.sector_size, include_unit=False),
'Size': self.length.format_highest(),
}
class DiskSegment:
def __init__(self, segment: PartitionModification | FreeSpace) -> None:
self.segment = segment
def table_data(self) -> dict[str, str]:
"""
Called for displaying data in table format
"""
if isinstance(self.segment, PartitionModification):
return self.segment.table_data()
part_mod = PartitionModification(
status=ModificationStatus.Create,
type=PartitionType._Unknown,
start=self.segment.start,
length=self.segment.length,
)
data = part_mod.table_data()
data.update({'Status': 'free', 'Type': '', 'FS type': ''})
return data
class PartitioningList(ListManager):
def __init__(self, prompt: str, device: BDevice, device_partitions: list[PartitionModification]):
self._device = device
self._buffer = Size(1, Unit.MiB, device.device_info.sector_size)
self._using_gpt = SysInfo.has_uefi()
self._actions = {
'create_new_partition': str(_('Create a new partition')),
'suggest_partition_layout': str(_('Suggest partition layout')),
'remove_added_partitions': str(_('Remove all newly added partitions')),
'assign_mountpoint': str(_('Assign mountpoint')),
@ -59,49 +95,137 @@ class PartitioningList(ListManager):
display_actions = list(self._actions.values())
super().__init__(
device_partitions,
display_actions[:2],
display_actions[3:],
self.as_segments(device_partitions),
display_actions[:1],
display_actions[2:],
prompt
)
@override
def selected_action_display(self, selection: PartitionModification) -> str:
if selection.status == ModificationStatus.Create:
return str(_('Partition - New'))
else:
return str(selection.dev_path)
def as_segments(self, device_partitions: list[PartitionModification]) -> list[DiskSegment]:
end = self._device.device_info.total_size
if self._using_gpt:
end = end.gpt_end()
end = end.align()
# Reorder device_partitions to move all deleted partitions to the top
device_partitions.sort(key=lambda p: p.is_delete(), reverse=True)
partitions = [DiskSegment(p) for p in device_partitions if not p.is_delete()]
segments = [DiskSegment(p) for p in device_partitions]
if not partitions:
free_space = FreeSpace(self._buffer, end)
if free_space.length > self._buffer:
return segments + [DiskSegment(free_space)]
return segments
first_part_index, first_partition = next(
(i, disk_segment) for i, disk_segment in enumerate(segments)
if isinstance(disk_segment.segment, PartitionModification)
and not disk_segment.segment.is_delete()
)
prev_partition = first_partition
index = 0
for partition in segments[1:]:
index += 1
if isinstance(partition.segment, PartitionModification) and partition.segment.is_delete():
continue
if prev_partition.segment.end < partition.segment.start:
free_space = FreeSpace(prev_partition.segment.end, partition.segment.start)
if free_space.length > self._buffer:
segments.insert(index, DiskSegment(free_space))
index += 1
prev_partition = partition
if first_partition.segment.start > self._buffer:
free_space = FreeSpace(self._buffer, first_partition.segment.start)
if free_space.length > self._buffer:
segments.insert(first_part_index, DiskSegment(free_space))
if partitions[-1].segment.end < end:
free_space = FreeSpace(partitions[-1].segment.end, end)
if free_space.length > self._buffer:
segments.append(DiskSegment(free_space))
return segments
@staticmethod
def get_part_mods(disk_segments: list[DiskSegment]) -> list[PartitionModification]:
return [
s.segment for s in disk_segments
if isinstance(s.segment, PartitionModification)
]
@override
def filter_options(self, selection: PartitionModification, options: list[str]) -> list[str]:
def run(self) -> list[PartitionModification]:
disk_segments = super().run()
return self.get_part_mods(disk_segments)
@override
def _run_actions_on_entry(self, entry: DiskSegment) -> None:
# Do not create a menu when the segment is free space
if isinstance(entry.segment, FreeSpace):
self._data = self.handle_action('', entry, self._data)
else:
super()._run_actions_on_entry(entry)
@override
def selected_action_display(self, selection: DiskSegment) -> str:
if isinstance(selection.segment, PartitionModification):
if selection.segment.status == ModificationStatus.Create:
return str(_('Partition - New'))
elif selection.segment.is_delete() and selection.segment.dev_path:
title = str(_('Partition')) + '\n\n'
title += 'status: delete\n'
title += f'device: {selection.segment.dev_path}\n'
for part in self._device.partition_infos:
if part.path == selection.segment.dev_path:
if part.partuuid:
title += f'partuuid: {part.partuuid}'
return title
return str(selection.segment.dev_path)
return ''
@override
def filter_options(self, selection: DiskSegment, options: list[str]) -> list[str]:
not_filter = []
# only display formatting if the partition exists already
if not selection.exists():
not_filter += [self._actions['mark_formatting']]
else:
# only allow options if the existing partition
# was marked as formatting, otherwise we run into issues where
# 1. select a new fs -> potentially mark as wipe now
# 2. Switch back to old filesystem -> should unmark wipe now, but
# how do we know it was the original one?
not_filter += [
self._actions['set_filesystem'],
self._actions['mark_bootable'],
self._actions['btrfs_mark_compressed'],
self._actions['btrfs_mark_nodatacow'],
self._actions['btrfs_set_subvolumes']
]
if isinstance(selection.segment, PartitionModification):
if selection.segment.is_delete():
not_filter = list(self._actions.values())
# only display formatting if the partition exists already
elif not selection.segment.exists():
not_filter += [self._actions['mark_formatting']]
else:
# only allow options if the existing partition
# was marked as formatting, otherwise we run into issues where
# 1. select a new fs -> potentially mark as wipe now
# 2. Switch back to old filesystem -> should unmark wipe now, but
# how do we know it was the original one?
not_filter += [
self._actions['set_filesystem'],
self._actions['mark_bootable'],
self._actions['btrfs_mark_compressed'],
self._actions['btrfs_mark_nodatacow'],
self._actions['btrfs_set_subvolumes']
]
# non btrfs partitions shouldn't get btrfs options
if selection.fs_type != FilesystemType.Btrfs:
not_filter += [
self._actions['btrfs_mark_compressed'],
self._actions['btrfs_mark_nodatacow'],
self._actions['btrfs_set_subvolumes']
]
else:
not_filter += [self._actions['assign_mountpoint']]
# non btrfs partitions shouldn't get btrfs options
if selection.segment.fs_type != FilesystemType.Btrfs:
not_filter += [
self._actions['btrfs_mark_compressed'],
self._actions['btrfs_mark_nodatacow'],
self._actions['btrfs_set_subvolumes']
]
else:
not_filter += [self._actions['assign_mountpoint']]
return [o for o in options if o not in not_filter]
@ -109,62 +233,79 @@ class PartitioningList(ListManager):
def handle_action(
self,
action: str,
entry: PartitionModification | None,
data: list[PartitionModification]
) -> list[PartitionModification]:
action_key = [k for k, v in self._actions.items() if v == action][0]
match action_key:
case 'create_new_partition':
new_partition = self._create_new_partition()
data += [new_partition]
case 'suggest_partition_layout':
new_partitions = self._suggest_partition_layout(data)
if len(new_partitions) > 0:
data = new_partitions
case 'remove_added_partitions':
if self._reset_confirmation():
data = [part for part in data if part.is_exists_or_modify()]
case 'assign_mountpoint' if entry:
entry.mountpoint = self._prompt_mountpoint()
if entry.mountpoint == Path('/boot'):
entry.set_flag(PartitionFlag.BOOT)
if SysInfo.has_uefi():
entry.set_flag(PartitionFlag.ESP)
case 'mark_formatting' if entry:
self._prompt_formatting(entry)
case 'mark_bootable' if entry:
entry.invert_flag(PartitionFlag.BOOT)
if SysInfo.has_uefi():
entry.invert_flag(PartitionFlag.ESP)
case 'set_filesystem' if entry:
fs_type = self._prompt_partition_fs_type()
if fs_type:
entry.fs_type = fs_type
# btrfs subvolumes will define mountpoints
if fs_type == FilesystemType.Btrfs:
entry.mountpoint = None
case 'btrfs_mark_compressed' if entry:
self._toggle_mount_option(entry, BtrfsMountOption.compress)
case 'btrfs_mark_nodatacow' if entry:
self._toggle_mount_option(entry, BtrfsMountOption.nodatacow)
case 'btrfs_set_subvolumes' if entry:
self._set_btrfs_subvolumes(entry)
case 'delete_partition' if entry:
data = self._delete_partition(entry, data)
entry: DiskSegment | None,
data: list[DiskSegment]
) -> list[DiskSegment]:
if not entry:
action_key = [k for k, v in self._actions.items() if v == action][0]
match action_key:
case 'suggest_partition_layout':
part_mods = self.get_part_mods(data)
new_partitions = self._suggest_partition_layout(part_mods)
if len(new_partitions) > 0:
data = self.as_segments(new_partitions)
case 'remove_added_partitions':
if self._reset_confirmation():
data = [
s for s in data
if isinstance(s.segment, PartitionModification)
and s.segment.is_exists_or_modify()
]
elif isinstance(entry.segment, PartitionModification):
partition = entry.segment
action_key = [k for k, v in self._actions.items() if v == action][0]
match action_key:
case 'assign_mountpoint':
partition.mountpoint = self._prompt_mountpoint()
if partition.mountpoint == Path('/boot'):
partition.set_flag(PartitionFlag.BOOT)
if self._using_gpt:
partition.set_flag(PartitionFlag.ESP)
case 'mark_formatting':
self._prompt_formatting(partition)
case 'mark_bootable':
partition.invert_flag(PartitionFlag.BOOT)
if self._using_gpt:
partition.invert_flag(PartitionFlag.ESP)
case 'set_filesystem':
fs_type = self._prompt_partition_fs_type()
if fs_type:
partition.fs_type = fs_type
# btrfs subvolumes will define mountpoints
if fs_type == FilesystemType.Btrfs:
partition.mountpoint = None
case 'btrfs_mark_compressed':
self._toggle_mount_option(partition, BtrfsMountOption.compress)
case 'btrfs_mark_nodatacow':
self._toggle_mount_option(partition, BtrfsMountOption.nodatacow)
case 'btrfs_set_subvolumes':
self._set_btrfs_subvolumes(partition)
case 'delete_partition':
data = self._delete_partition(partition, data)
else:
part_mods = self.get_part_mods(data)
index = data.index(entry)
part_mods.insert(index, self._create_new_partition(entry.segment))
data = self.as_segments(part_mods)
return data
def _delete_partition(
self,
entry: PartitionModification,
data: list[PartitionModification]
) -> list[PartitionModification]:
data: list[DiskSegment]
) -> list[DiskSegment]:
if entry.is_exists_or_modify():
entry.status = ModificationStatus.Delete
return data
part_mods = self.get_part_mods(data)
else:
return [d for d in data if d != entry]
part_mods = [
d.segment for d in data
if isinstance(d.segment, PartitionModification)
and d.segment != entry
]
return self.as_segments(part_mods)
def _toggle_mount_option(
self,
@ -246,9 +387,8 @@ class PartitioningList(ListManager):
def _validate_value(
self,
sector_size: SectorSize,
total_size: Size,
text: str,
start: Size | None
max_size: Size,
text: str
) -> Size | None:
match = re.match(r'([0-9]+)([a-zA-Z|%]*)', text, re.I)
@ -257,10 +397,9 @@ class PartitioningList(ListManager):
str_value, unit = match.groups()
if unit == '%' and start:
available = total_size - start
value = int(available.value * (int(str_value) / 100))
unit = available.unit.name
if unit == '%':
value = int(max_size.value * (int(str_value) / 100))
unit = max_size.unit.name
else:
value = int(str_value)
@ -270,29 +409,41 @@ class PartitioningList(ListManager):
unit = Unit[unit] if unit else Unit.sectors
size = Size(value, unit, sector_size)
if start and size <= start:
if size.format_highest() == max_size.format_highest():
return max_size
elif size > max_size or size < self._buffer:
return None
return size
def _enter_size(
self,
sector_size: SectorSize,
total_size: Size,
text: str,
header: str,
default: Size,
start: Size | None,
) -> Size:
def _prompt_size(self, free_space: FreeSpace) -> Size:
def validate(value: str) -> str | None:
size = self._validate_value(sector_size, total_size, value, start)
size = self._validate_value(sector_size, max_size, value)
if not size:
return str(_('Invalid size'))
return None
device_info = self._device.device_info
sector_size = device_info.sector_size
text = str(_('Selected free space segment on device {}:')).format(device_info.path) + '\n\n'
free_space_table = FormattedOutput.as_table([free_space])
prompt = text + free_space_table + '\n'
max_sectors = free_space.length.format_size(Unit.sectors, sector_size)
max_bytes = free_space.length.format_size(Unit.B)
prompt += str(_('Size: {} / {}')).format(max_sectors, max_bytes) + '\n\n'
prompt += str(_('All entered values can be suffixed with a unit: %, B, KB, KiB, MB, MiB...')) + '\n'
prompt += str(_('If no unit is provided, the value is interpreted as sectors')) + '\n'
max_size = free_space.length
title = str(_('Size (default: {}): ')).format(max_size.format_highest())
result = EditMenu(
text,
header=f'{header}\b',
title,
header=f'{prompt}\b',
allow_skip=True,
validator=validate
).input()
@ -301,117 +452,23 @@ class PartitioningList(ListManager):
match result.type_:
case ResultType.Skip:
size = default
size = max_size
case ResultType.Selection:
value = result.text()
if value:
size = self._validate_value(sector_size, total_size, value, start)
size = self._validate_value(sector_size, max_size, value)
else:
size = default
size = max_size
assert size
return size
def _prompt_size(self) -> tuple[Size, Size]:
device_info = self._device.device_info
def _create_new_partition(self, free_space: FreeSpace) -> PartitionModification:
length = self._prompt_size(free_space)
text = str(_('Current free sectors on device {}:')).format(device_info.path) + '\n\n'
free_space_table = FormattedOutput.as_table(device_info.free_space_regions)
prompt = text + free_space_table + '\n'
total_sectors = device_info.total_size.format_size(Unit.sectors, device_info.sector_size)
total_bytes = device_info.total_size.format_size(Unit.B)
prompt += str(_('Total: {} / {}')).format(total_sectors, total_bytes) + '\n\n'
prompt += str(_('All entered values can be suffixed with a unit: %, B, KB, KiB, MB, MiB...')) + '\n'
prompt += str(_('If no unit is provided, the value is interpreted as sectors')) + '\n'
default_free_sector = self._find_default_free_space()
if not default_free_sector:
default_free_sector = DefaultFreeSector(
Size(0, Unit.sectors, self._device.device_info.sector_size),
Size(0, Unit.sectors, self._device.device_info.sector_size)
)
# prompt until a valid start sector was entered
start_text = str(_('Start (default: sector {}): ')).format(default_free_sector.start.value)
start_size = self._enter_size(
device_info.sector_size,
device_info.total_size,
start_text,
prompt,
default_free_sector.start,
None
)
prompt += f'\nStart: {start_size.as_text()}\n'
if start_size.value == default_free_sector.start.value and default_free_sector.end.value != 0:
end_size = default_free_sector.end
else:
end_size = device_info.total_size
# prompt until valid end sector was entered
end_text = str(_('End (default: {}): ')).format(end_size.as_text())
end_size = self._enter_size(
device_info.sector_size,
device_info.total_size,
end_text,
prompt,
end_size,
start_size
)
return start_size, end_size
def _find_default_free_space(self) -> DefaultFreeSector | None:
device_info = self._device.device_info
largest_free_area: DeviceGeometry | None = None
largest_deleted_area: PartitionModification | None = None
if len(device_info.free_space_regions) > 0:
largest_free_area = max(device_info.free_space_regions, key=lambda r: r.get_length())
deleted_partitions = list(filter(lambda x: x.status == ModificationStatus.Delete, self._data))
if len(deleted_partitions) > 0:
largest_deleted_area = max(deleted_partitions, key=lambda p: p.length)
def _free_space(space: DeviceGeometry) -> DefaultFreeSector:
start = Size(space.start, Unit.sectors, device_info.sector_size)
end = Size(space.end, Unit.sectors, device_info.sector_size)
return DefaultFreeSector(start, end)
def _free_deleted(space: PartitionModification) -> DefaultFreeSector:
start = space.start.convert(Unit.sectors, self._device.device_info.sector_size)
end = space.end.convert(Unit.sectors, self._device.device_info.sector_size)
return DefaultFreeSector(start, end)
if not largest_deleted_area and largest_free_area:
return _free_space(largest_free_area)
elif not largest_free_area and largest_deleted_area:
return _free_deleted(largest_deleted_area)
elif not largest_deleted_area and not largest_free_area:
return None
elif largest_free_area and largest_deleted_area:
free_space = _free_space(largest_free_area)
if free_space.start > largest_deleted_area.start:
return free_space
else:
return _free_deleted(largest_deleted_area)
return None
def _create_new_partition(self) -> PartitionModification:
fs_type = self._prompt_partition_fs_type()
start_size, end_size = self._prompt_size()
length = end_size - start_size
mountpoint = None
if fs_type != FilesystemType.Btrfs:
mountpoint = self._prompt_mountpoint()
@ -419,7 +476,7 @@ class PartitioningList(ListManager):
partition = PartitionModification(
status=ModificationStatus.Create,
type=PartitionType.Primary,
start=start_size,
start=free_space.start,
length=length,
fs_type=fs_type,
mountpoint=mountpoint
@ -427,7 +484,7 @@ class PartitioningList(ListManager):
if partition.mountpoint == Path('/boot'):
partition.set_flag(PartitionFlag.BOOT)
if SysInfo.has_uefi():
if self._using_gpt:
partition.set_flag(PartitionFlag.ESP)
return partition

18
archinstall/lib/interactions/disk_conf.py
View File

@ -342,8 +342,9 @@ def suggest_single_disk_layout(
using_gpt = SysInfo.has_uefi()
if using_gpt:
# Remove space for end alignment buffer
available_space -= disk.Size(1, disk.Unit.MiB, sector_size)
available_space = available_space.gpt_end()
available_space = available_space.align()
# Used for reference: https://wiki.archlinux.org/title/partitioning
# 2 MiB is unallocated for GRUB on BIOS. Potentially unneeded for other bootloaders?
@ -500,9 +501,6 @@ def suggest_multi_disk_layout(
root_device_sector_size = root_device_modification.device.device_info.sector_size
home_device_sector_size = home_device_modification.device.device_info.sector_size
root_align_buffer = disk.Size(1, disk.Unit.MiB, root_device_sector_size)
home_align_buffer = disk.Size(1, disk.Unit.MiB, home_device_sector_size)
using_gpt = SysInfo.has_uefi()
# add boot partition to the root device
@ -513,7 +511,9 @@ def suggest_multi_disk_layout(
root_length = root_device.device_info.total_size - root_start
if using_gpt:
root_length -= root_align_buffer
root_length = root_length.gpt_end()
root_length = root_length.align()
# add root partition to the root device
root_partition = disk.PartitionModification(
@ -527,14 +527,16 @@ def suggest_multi_disk_layout(
)
root_device_modification.add_partition(root_partition)
home_start = home_align_buffer
home_start = disk.Size(1, disk.Unit.MiB, home_device_sector_size)
home_length = home_device.device_info.total_size - home_start
flags = []
if using_gpt:
home_length -= home_align_buffer
home_length = home_length.gpt_end()
flags.append(disk.PartitionFlag.LINUX_HOME)
home_length = home_length.align()
# add home partition to home device
home_partition = disk.PartitionModification(
status=disk.ModificationStatus.Create,

8
tests/data/test_config.json
View File

@ -49,7 +49,7 @@
"value": 512
},
"unit": "GiB",
"value": 20
"value": 32
},
"mount_options": [],
"mountpoint": "/",
@ -76,7 +76,7 @@
"value": 512
},
"unit": "GiB",
"value": 100
"value": 32
},
"mount_options": [],
"mountpoint": "/home",
@ -86,8 +86,8 @@
"unit": "B",
"value": 512
},
"unit": "GiB",
"value": 20
"unit": "MiB",
"value": 33281
},
"status": "create",
"type": "primary"