Merge remote-tracking branch 'origin/master' into disk-config-submenu

archinstall/lib/disk/device_handler.py

@@ -134,6 +134,20 @@ def get_parent_device_path(self, dev_path: Path) -> Path:
 		lsblk = get_lsblk_info(dev_path)
 		return Path(f'/dev/{lsblk.pkname}')
 
+	def get_unique_path_for_device(self, dev_path: Path) -> Optional[Path]:
+		paths = Path('/dev/disk/by-id').glob('*')
+		linked_targets = {p.resolve(): p for p in paths}
+		linked_wwn_targets = {p: linked_targets[p] for p in linked_targets
+			if p.name.startswith('wwn-') or p.name.startswith('nvme-eui.')}
+
+		if dev_path in linked_wwn_targets:
+			return linked_wwn_targets[dev_path]
+
+		if dev_path in linked_targets:
+			return linked_targets[dev_path]
+
+		return None
+
 	def get_uuid_for_path(self, path: Path) -> Optional[str]:
 		partition = self.find_partition(path)
 		return partition.partuuid if partition else None
@@ -380,16 +394,13 @@ def fetch_part_info(self, path: Path) -> LsblkInfo:
 		attempts = 3
 		lsblk_info: Optional[LsblkInfo] = None
 
-		self.partprobe(path)
 		for attempt_nr in range(attempts):
 			time.sleep(attempt_nr + 1)
 			lsblk_info = get_lsblk_info(path)
 
 			if lsblk_info.partn and lsblk_info.partuuid and lsblk_info.uuid:
 				break
 
-			self.partprobe(path)
-
 		if not lsblk_info:
 			debug(f'Unable to get partition information: {path}')
 			raise DiskError(f'Unable to get partition information: {path}')
@@ -528,8 +539,6 @@ def partition(
 			requires_delete = modification.wipe is False
 			self._perform_partitioning(part_mod, modification.device, disk, requires_delete=requires_delete)
 
-		self.partprobe(modification.device.device_info.path)
-
 	def mount(
 		self,
 		dev_path: Path,

archinstall/lib/disk/device_model.py

@@ -26,6 +26,122 @@
 	_: Any
 
 
+class DiskLayoutType(Enum):
+	Default = 'default_layout'
+	Manual = 'manual_partitioning'
+	Pre_mount = 'pre_mounted_config'
+
+	def display_msg(self) -> str:
+		match self:
+			case DiskLayoutType.Default:
+				return str(_('Use a best-effort default partition layout'))
+			case DiskLayoutType.Manual:
+				return str(_('Manual Partitioning'))
+			case DiskLayoutType.Pre_mount:
+				return str(_('Pre-mounted configuration'))
+
+
+@dataclass
+class DiskLayoutConfiguration:
+	config_type: DiskLayoutType
+	device_modifications: List[DeviceModification] = field(default_factory=list)
+	lvm_config: Optional[LvmConfiguration] = None
+
+	# used for pre-mounted config
+	mountpoint: Optional[Path] = None
+
+	def json(self) -> Dict[str, Any]:
+		if self.config_type == DiskLayoutType.Pre_mount:
+			return {
+				'config_type': self.config_type.value,
+				'mountpoint': str(self.mountpoint)
+			}
+		else:
+			config: Dict[str, Any] = {
+				'config_type': self.config_type.value,
+				'device_modifications': [mod.json() for mod in self.device_modifications],
+			}
+
+			if self.lvm_config:
+				config['lvm_config'] = self.lvm_config.json()
+
+			return config
+
+	@classmethod
+	def parse_arg(cls, disk_config: Dict[str, Dict[str, Any]]) -> Optional[DiskLayoutConfiguration]:
+		from .device_handler import device_handler
+
+		device_modifications: List[DeviceModification] = []
+		config_type = disk_config.get('config_type', None)
+
+		if not config_type:
+			raise ValueError('Missing disk layout configuration: config_type')
+
+		config = DiskLayoutConfiguration(
+			config_type=DiskLayoutType(config_type),
+			device_modifications=device_modifications
+		)
+
+		if config_type == DiskLayoutType.Pre_mount.value:
+			if not (mountpoint := disk_config.get('mountpoint')):
+				raise ValueError('Must set a mountpoint when layout type is pre-mount')
+
+			path = Path(str(mountpoint))
+
+			mods = device_handler.detect_pre_mounted_mods(path)
+			device_modifications.extend(mods)
+
+			storage['MOUNT_POINT'] = path
+
+			config.mountpoint = path
+
+			return config
+
+		for entry in disk_config.get('device_modifications', []):
+			device_path = Path(entry.get('device', None)) if entry.get('device', None) else None
+
+			if not device_path:
+				continue
+
+			device = device_handler.get_device(device_path)
+
+			if not device:
+				continue
+
+			device_modification = DeviceModification(
+				wipe=entry.get('wipe', False),
+				device=device
+			)
+
+			device_partitions: List[PartitionModification] = []
+
+			for partition in entry.get('partitions', []):
+				device_partition = PartitionModification(
+					status=ModificationStatus(partition['status']),
+					fs_type=FilesystemType(partition['fs_type']),
+					start=Size.parse_args(partition['start']),
+					length=Size.parse_args(partition['size']),
+					mount_options=partition['mount_options'],
+					mountpoint=Path(partition['mountpoint']) if partition['mountpoint'] else None,
+					dev_path=Path(partition['dev_path']) if partition['dev_path'] else None,
+					type=PartitionType(partition['type']),
+					flags=[PartitionFlag[f] for f in partition.get('flags', [])],
+					btrfs_subvols=SubvolumeModification.parse_args(partition.get('btrfs', [])),
+				)
+				# special 'invisible attr to internally identify the part mod
+				setattr(device_partition, '_obj_id', partition['obj_id'])
+				device_partitions.append(device_partition)
+
+			device_modification.partitions = device_partitions
+			device_modifications.append(device_modification)
+
+		# Parse LVM configuration from settings
+		if lvm_arg := disk_config.get('lvm_config', None):
+			config.lvm_config = LvmConfiguration.parse_arg(lvm_arg, config)
+
+		return config
+
+
 class PartitionTable(Enum):
 	GPT = 'gpt'
 	MBR = 'msdos'
@@ -352,97 +468,6 @@ def from_disk(cls, disk: Disk) -> _DeviceInfo:
 		)
 
 
-class DiskLayoutType(Enum):
-	Default = 'default_layout'
-	Manual = 'manual_partitioning'
-	Pre_mount = 'pre_mounted_config'
-
-	def display_msg(self) -> str:
-		match self:
-			case DiskLayoutType.Default:
-				return str(_('Use a best-effort default partition layout'))
-			case DiskLayoutType.Manual:
-				return str(_('Manual Partitioning'))
-			case DiskLayoutType.Pre_mount:
-				return str(_('Pre-mounted configuration'))
-
-
-@dataclass
-class DiskLayoutConfiguration:
-	config_type: DiskLayoutType
-	device_modifications: List[DeviceModification] = field(default_factory=list)
-	lvm_config: Optional[LvmConfiguration] = None
-
-	def json(self) -> Dict[str, Any]:
-		config: Dict[str, Any] = {
-			'config_type': self.config_type.value,
-			'device_modifications': [mod.json() for mod in self.device_modifications],
-		}
-
-		if self.lvm_config:
-			config['lvm_config'] = self.lvm_config.json()
-
-		return config
-
-	@classmethod
-	def parse_arg(cls, arg: Dict[str, Any]) -> Optional[DiskLayoutConfiguration]:
-		from .device_handler import device_handler
-
-		device_modifications: List[DeviceModification] = []
-		config_type = arg.get('config_type', None)
-
-		if not config_type:
-			raise ValueError('Missing disk layout configuration: config_type')
-
-		config = DiskLayoutConfiguration(
-			config_type=DiskLayoutType(config_type),
-			device_modifications=device_modifications
-		)
-
-		for entry in arg.get('device_modifications', []):
-			device_path = Path(entry.get('device', None)) if entry.get('device', None) else None
-
-			if not device_path:
-				continue
-
-			device = device_handler.get_device(device_path)
-
-			if not device:
-				continue
-
-			device_modification = DeviceModification(
-				wipe=entry.get('wipe', False),
-				device=device
-			)
-
-			device_partitions: List[PartitionModification] = []
-
-			for partition in entry.get('partitions', []):
-				device_partition = PartitionModification(
-					status=ModificationStatus(partition['status']),
-					fs_type=FilesystemType(partition['fs_type']),
-					start=Size.parse_args(partition['start']),
-					length=Size.parse_args(partition['size']),
-					mount_options=partition['mount_options'],
-					mountpoint=Path(partition['mountpoint']) if partition['mountpoint'] else None,
-					type=PartitionType(partition['type']),
-					flags=[PartitionFlag[f] for f in partition.get('flags', [])],
-					btrfs_subvols=SubvolumeModification.parse_args(partition.get('btrfs', [])),
-				)
-				# special 'invisible attr to internally identify the part mod
-				setattr(device_partition, '_obj_id', partition['obj_id'])
-				device_partitions.append(device_partition)
-
-			device_modification.partitions = device_partitions
-			device_modifications.append(device_modification)
-
-		# Parse LVM configuration from settings
-		if lvm_arg := arg.get('lvm_config', None):
-			config.lvm_config = LvmConfiguration.parse_arg(lvm_arg, config)
-
-		return config
-
-
 @dataclass
 class SubvolumeModification:
 	name: Path
@@ -1331,10 +1356,7 @@ def _fetch_lsblk_info(dev_path: Optional[Union[Path, str]] = None, retry: int =
 	if not dev_path:
 		dev_path = ''
 
-	if retry == 0:
-		retry = 1
-
-	for retry_attempt in range(retry):
+	for retry_attempt in range(retry + 1):
 		try:
 			result = SysCommand(f'lsblk --json -b -o+{lsblk_fields} {dev_path}').decode()
 			break
@@ -1346,7 +1368,7 @@ def _fetch_lsblk_info(dev_path: Optional[Union[Path, str]] = None, retry: int =
 			else:
 				raise err
 
-			if retry_attempt == retry - 1:
+			if retry_attempt == retry:
 				raise err
 
 			time.sleep(1)

archinstall/lib/global_menu.py

@@ -345,8 +345,11 @@ def _validate_bootloader(self) -> Optional[str]:
 		if boot_partition is None:
 			return "Boot partition not found"
 
-		if bootloader == Bootloader.Limine and boot_partition.fs_type == disk.FilesystemType.Btrfs:
-			return "Limine bootloader does not support booting from BTRFS filesystem"
+		if bootloader == Bootloader.Limine:
+			if boot_partition.fs_type != disk.FilesystemType.Fat32:
+				return "Limine does not support booting from filesystems other than FAT32"
+			elif self._menu_options['uki'].current_selection:
+				return "Limine does not support booting UKIs"
 
 		return None