Move LVM helpers to dedicated module

archinstall/lib/disk/device_handler.py

@@ -1,10 +1,7 @@
-import json
 import logging
 import os
-import time
 from collections.abc import Iterable
 from pathlib import Path
-from typing import Literal, overload
 
 from parted import Device, Disk, DiskException, FileSystem, Geometry, IOException, Partition, PartitionException, freshDisk, getAllDevices, getDevice, newDisk
 
@@ -19,17 +16,13 @@
 	DiskEncryption,
 	FilesystemType,
 	LsblkInfo,
-	LvmGroupInfo,
-	LvmPVInfo,
 	LvmVolume,
 	LvmVolumeGroup,
-	LvmVolumeInfo,
 	ModificationStatus,
 	PartitionFlag,
 	PartitionGUID,
 	PartitionModification,
 	PartitionTable,
-	SectorSize,
 	Size,
 	SubvolumeModification,
 	Unit,
@@ -348,123 +341,12 @@ def format_encrypted(
 		info(f'luks2 locking device: {dev_path}')
 		luks_handler.lock()
 
-	def _lvm_info(
-		self,
-		cmd: str,
-		info_type: Literal['lv', 'vg', 'pvseg'],
-	) -> LvmVolumeInfo | LvmGroupInfo | LvmPVInfo | None:
-		raw_info = SysCommand(cmd).decode().split('\n')
-
-		# for whatever reason the output sometimes contains
-		# "File descriptor X leaked leaked on vgs invocation
-		data = '\n'.join(raw for raw in raw_info if 'File descriptor' not in raw)
-
-		debug(f'LVM info: {data}')
-
-		reports = json.loads(data)
-
-		for report in reports['report']:
-			if len(report[info_type]) != 1:
-				raise ValueError('Report does not contain any entry')
-
-			entry = report[info_type][0]
-
-			match info_type:
-				case 'pvseg':
-					return LvmPVInfo(
-						pv_name=Path(entry['pv_name']),
-						lv_name=entry['lv_name'],
-						vg_name=entry['vg_name'],
-					)
-				case 'lv':
-					return LvmVolumeInfo(
-						lv_name=entry['lv_name'],
-						vg_name=entry['vg_name'],
-						lv_size=Size(int(entry['lv_size'][:-1]), Unit.B, SectorSize.default()),
-					)
-				case 'vg':
-					return LvmGroupInfo(
-						vg_uuid=entry['vg_uuid'],
-						vg_size=Size(int(entry['vg_size'][:-1]), Unit.B, SectorSize.default()),
-					)
-
-		return None
-
-	@overload
-	def _lvm_info_with_retry(self, cmd: str, info_type: Literal['lv']) -> LvmVolumeInfo | None: ...
-
-	@overload
-	def _lvm_info_with_retry(self, cmd: str, info_type: Literal['vg']) -> LvmGroupInfo | None: ...
-
-	@overload
-	def _lvm_info_with_retry(self, cmd: str, info_type: Literal['pvseg']) -> LvmPVInfo | None: ...
-
-	def _lvm_info_with_retry(
-		self,
-		cmd: str,
-		info_type: Literal['lv', 'vg', 'pvseg'],
-	) -> LvmVolumeInfo | LvmGroupInfo | LvmPVInfo | None:
-		# Retry for up to 5 mins
-		max_retries = 100
-		for attempt in range(max_retries):
-			try:
-				return self._lvm_info(cmd, info_type)
-			except ValueError:
-				if attempt < max_retries - 1:
-					debug(f'LVM info query failed (attempt {attempt + 1}/{max_retries}), retrying in 3 seconds...')
-					time.sleep(3)
-
-		debug(f'LVM info query failed after {max_retries} attempts')
-		return None
-
-	def lvm_vol_info(self, lv_name: str) -> LvmVolumeInfo | None:
-		cmd = f'lvs --reportformat json --unit B -S lv_name={lv_name}'
-
-		return self._lvm_info_with_retry(cmd, 'lv')
-
-	def lvm_group_info(self, vg_name: str) -> LvmGroupInfo | None:
-		cmd = f'vgs --reportformat json --unit B -o vg_name,vg_uuid,vg_size -S vg_name={vg_name}'
-
-		return self._lvm_info_with_retry(cmd, 'vg')
-
-	def lvm_pvseg_info(self, vg_name: str, lv_name: str) -> LvmPVInfo | None:
-		cmd = f'pvs --segments -o+lv_name,vg_name -S vg_name={vg_name},lv_name={lv_name} --reportformat json '
-
-		return self._lvm_info_with_retry(cmd, 'pvseg')
-
-	def lvm_vol_change(self, vol: LvmVolume, activate: bool) -> None:
-		active_flag = 'y' if activate else 'n'
-		cmd = f'lvchange -a {active_flag} {vol.safe_dev_path}'
-
-		debug(f'lvchange volume: {cmd}')
-		SysCommand(cmd)
-
 	def lvm_export_vg(self, vg: LvmVolumeGroup) -> None:
 		cmd = f'vgexport {vg.name}'
 
 		debug(f'vgexport: {cmd}')
 		SysCommand(cmd)
 
-	def lvm_import_vg(self, vg: LvmVolumeGroup) -> None:
-		# Check if the VG is actually exported before trying to import it
-		check_cmd = f'vgs --noheadings -o vg_exported {vg.name}'
-
-		try:
-			result = SysCommand(check_cmd)
-			is_exported = result.decode().strip() == 'exported'
-		except SysCallError:
-			# VG might not exist yet, skip import
-			debug(f'Volume group {vg.name} not found, skipping import')
-			return
-
-		if not is_exported:
-			debug(f'Volume group {vg.name} is already active (not exported), skipping import')
-			return
-
-		cmd = f'vgimport {vg.name}'
-		debug(f'vgimport: {cmd}')
-		SysCommand(cmd)
-
 	def lvm_vol_reduce(self, vol_path: Path, amount: Size) -> None:
 		val = amount.format_size(Unit.B, include_unit=False)
 		cmd = f'lvreduce -L -{val}B {vol_path}'

archinstall/lib/disk/filesystem.py

@@ -22,6 +22,7 @@
 )
 from ..output import debug, info
 from .device_handler import device_handler
+from .lvm import lvm_group_info, lvm_vol_info
 
 
 class FilesystemHandler:
@@ -168,7 +169,7 @@ def _setup_lvm(
 			device_handler.lvm_vg_create(pv_dev_paths, vg.name)
 
 			# figure out what the actual available size in the group is
-			vg_info = device_handler.lvm_group_info(vg.name)
+			vg_info = lvm_group_info(vg.name)
 
 			if not vg_info:
 				raise ValueError('Unable to fetch VG info')
@@ -202,7 +203,7 @@ def _setup_lvm(
 
 				while True:
 					debug('Fetching LVM volume info')
-					lv_info = device_handler.lvm_vol_info(lv.name)
+					lv_info = lvm_vol_info(lv.name)
 					if lv_info is not None:
 						break
 

archinstall/lib/disk/lvm.py

@@ -0,0 +1,137 @@
+import json
+import time
+from pathlib import Path
+from typing import Literal, overload
+
+from archinstall.lib.command import SysCommand
+from archinstall.lib.exceptions import SysCallError
+from archinstall.lib.models.device import (
+	LvmGroupInfo,
+	LvmPVInfo,
+	LvmVolume,
+	LvmVolumeGroup,
+	LvmVolumeInfo,
+	SectorSize,
+	Size,
+	Unit,
+)
+from archinstall.lib.output import debug
+
+
+def _lvm_info(
+	cmd: str,
+	info_type: Literal['lv', 'vg', 'pvseg'],
+) -> LvmVolumeInfo | LvmGroupInfo | LvmPVInfo | None:
+	raw_info = SysCommand(cmd).decode().split('\n')
+
+	# for whatever reason the output sometimes contains
+	# "File descriptor X leaked leaked on vgs invocation
+	data = '\n'.join(raw for raw in raw_info if 'File descriptor' not in raw)
+
+	debug(f'LVM info: {data}')
+
+	reports = json.loads(data)
+
+	for report in reports['report']:
+		if len(report[info_type]) != 1:
+			raise ValueError('Report does not contain any entry')
+
+		entry = report[info_type][0]
+
+		match info_type:
+			case 'pvseg':
+				return LvmPVInfo(
+					pv_name=Path(entry['pv_name']),
+					lv_name=entry['lv_name'],
+					vg_name=entry['vg_name'],
+				)
+			case 'lv':
+				return LvmVolumeInfo(
+					lv_name=entry['lv_name'],
+					vg_name=entry['vg_name'],
+					lv_size=Size(int(entry['lv_size'][:-1]), Unit.B, SectorSize.default()),
+				)
+			case 'vg':
+				return LvmGroupInfo(
+					vg_uuid=entry['vg_uuid'],
+					vg_size=Size(int(entry['vg_size'][:-1]), Unit.B, SectorSize.default()),
+				)
+
+	return None
+
+
+@overload
+def _lvm_info_with_retry(cmd: str, info_type: Literal['lv']) -> LvmVolumeInfo | None: ...
+
+
+@overload
+def _lvm_info_with_retry(cmd: str, info_type: Literal['vg']) -> LvmGroupInfo | None: ...
+
+
+@overload
+def _lvm_info_with_retry(cmd: str, info_type: Literal['pvseg']) -> LvmPVInfo | None: ...
+
+
+def _lvm_info_with_retry(
+	cmd: str,
+	info_type: Literal['lv', 'vg', 'pvseg'],
+) -> LvmVolumeInfo | LvmGroupInfo | LvmPVInfo | None:
+	# Retry for up to 5 mins
+	max_retries = 100
+	for attempt in range(max_retries):
+		try:
+			return _lvm_info(cmd, info_type)
+		except ValueError:
+			if attempt < max_retries - 1:
+				debug(f'LVM info query failed (attempt {attempt + 1}/{max_retries}), retrying in 3 seconds...')
+				time.sleep(3)
+
+	debug(f'LVM info query failed after {max_retries} attempts')
+	return None
+
+
+def lvm_vol_info(lv_name: str) -> LvmVolumeInfo | None:
+	cmd = f'lvs --reportformat json --unit B -S lv_name={lv_name}'
+
+	return _lvm_info_with_retry(cmd, 'lv')
+
+
+def lvm_group_info(vg_name: str) -> LvmGroupInfo | None:
+	cmd = f'vgs --reportformat json --unit B -o vg_name,vg_uuid,vg_size -S vg_name={vg_name}'
+
+	return _lvm_info_with_retry(cmd, 'vg')
+
+
+def lvm_pvseg_info(vg_name: str, lv_name: str) -> LvmPVInfo | None:
+	cmd = f'pvs --segments -o+lv_name,vg_name -S vg_name={vg_name},lv_name={lv_name} --reportformat json '
+
+	return _lvm_info_with_retry(cmd, 'pvseg')
+
+
+def lvm_vol_change(vol: LvmVolume, activate: bool) -> None:
+	active_flag = 'y' if activate else 'n'
+	cmd = f'lvchange -a {active_flag} {vol.safe_dev_path}'
+
+	debug(f'lvchange volume: {cmd}')
+	SysCommand(cmd)
+
+
+def lvm_import_vg(vg: LvmVolumeGroup) -> None:
+	# Check if the VG is actually exported before trying to import it
+	check_cmd = f'vgs --noheadings -o vg_exported {vg.name}'
+
+	try:
+		result = SysCommand(check_cmd)
+		is_exported = result.decode().strip() == 'exported'
+	except SysCallError:
+		# VG might not exist yet, skip import
+		debug(f'Volume group {vg.name} not found, skipping import')
+		return
+
+	if not is_exported:
+		debug(f'Volume group {vg.name} is already active (not exported), skipping import')
+		return
+
+	cmd = f'vgimport {vg.name}'
+	debug(f'vgimport: {cmd}')
+	SysCommand(cmd)

archinstall/lib/installer.py

@@ -14,8 +14,8 @@
 from types import TracebackType
 from typing import Any, Self
 
-from archinstall.lib.disk.device_handler import device_handler
 from archinstall.lib.disk.fido import Fido2
+from archinstall.lib.disk.lvm import lvm_import_vg, lvm_pvseg_info, lvm_vol_change
 from archinstall.lib.disk.utils import (
 	get_lsblk_by_mountpoint,
 	get_lsblk_info,
@@ -341,10 +341,10 @@ def _import_lvm(self) -> None:
 			return
 
 		for vg in lvm_config.vol_groups:
-			device_handler.lvm_import_vg(vg)
+			lvm_import_vg(vg)
 
 			for vol in vg.volumes:
-				device_handler.lvm_vol_change(vol, True)
+				lvm_vol_change(vol, True)
 
 	def _prepare_luks_lvm(
 		self,
@@ -1147,7 +1147,7 @@ def _get_kernel_params_lvm(
 				if not lvm.vg_name:
 					raise ValueError(f'Unable to determine VG name for {lvm.name}')
 
-				pv_seg_info = device_handler.lvm_pvseg_info(lvm.vg_name, lvm.name)
+				pv_seg_info = lvm_pvseg_info(lvm.vg_name, lvm.name)
 
 				if not pv_seg_info:
 					raise ValueError(f'Unable to determine PV segment info for {lvm.vg_name}/{lvm.name}')