Subvolume User Interface (#1032)

* Deflate the user interactions file

* Fix flake8

* GlobalMenu split from selection_menu.py

* Upgrades to ListManager:
Can now show an empty list if there is no null action.
More information to the user at the header

* Put only_hd.py  and swiss.py to use new config printing mechanism
Solved a couple of bugs at
ListManager adding a str and a DeferredTranslation
ManageUser was missing an self argument in _check ...

* Create list and menus to manage subvolumes in btrfs partitions
Needed to modify manage_new_and_existing_partitions
Added a new parameter filter to select_partition, to allow filtering there

* Update internationalization strings

Co-authored-by: Daniel Girtler <girtler.daniel@gmail.com>
Co-authored-by: Anton Hvornum <anton@hvornum.se>

5 files changed, 461 insertions, 4 deletions

diff --git a/archinstall/lib/user_interaction/__init__.py b/archinstall/lib/user_interaction/__init__.py
index b0174d94..f1ef5d91 100644
--- a/archinstall/lib/user_interaction/__init__.py
+++ b/archinstall/lib/user_interaction/__init__.py
@@ -10,3 +10,4 @@ from .general_conf import (ask_ntp, ask_for_a_timezone, ask_for_audio_selection,
 							select_additional_repositories, ask_hostname)
 from .disk_conf import ask_for_main_filesystem_format, select_individual_blockdevice_usage, select_disk_layout, select_disk
 from .utils import get_password, do_countdown
+from .global_menu import GlobalMenu
diff --git a/archinstall/lib/user_interaction/global_menu.py b/archinstall/lib/user_interaction/global_menu.py
new file mode 100644
index 00000000..2001103a
--- /dev/null
+++ b/archinstall/lib/user_interaction/global_menu.py
@@ -0,0 +1,293 @@
+from __future__ import annotations
+
+from typing import Any, List, Optional
+
+from ..menu import Menu
+from ..menu.selection_menu import Selector, GeneralMenu
+from ..general import SysCommand, secret
+from ..hardware import has_uefi
+from ..storage import storage
+from ..output import log
+from ..profiles import is_desktop_profile
+from ..disk import encrypted_partitions
+
+from ..user_interaction import get_password, ask_for_a_timezone, save_config
+from ..user_interaction import ask_ntp
+from ..user_interaction import ask_for_swap
+from ..user_interaction import ask_for_bootloader
+from ..user_interaction import ask_hostname
+from ..user_interaction import ask_for_audio_selection
+from ..user_interaction import ask_additional_packages_to_install
+from ..user_interaction import ask_to_configure_network
+from ..user_interaction import ask_for_superuser_account
+from ..user_interaction import ask_for_additional_users
+from ..user_interaction import select_language
+from ..user_interaction import select_mirror_regions
+from ..user_interaction import select_locale_lang
+from ..user_interaction import select_locale_enc
+from ..user_interaction import select_disk_layout
+from ..user_interaction import select_kernel
+from ..user_interaction import select_encrypted_partitions
+from ..user_interaction import select_harddrives
+from ..user_interaction import select_profile
+from ..user_interaction import select_additional_repositories
+
+class GlobalMenu(GeneralMenu):
+	def __init__(self,data_store):
+		super().__init__(data_store=data_store, auto_cursor=True)
+
+	def _setup_selection_menu_options(self):
+		# archinstall.Language will not use preset values
+		self._menu_options['archinstall-language'] = \
+			Selector(
+				_('Select Archinstall language'),
+				lambda x: self._select_archinstall_language('English'),
+				default='English',
+				enabled=True)
+		self._menu_options['keyboard-layout'] = \
+			Selector(_('Select keyboard layout'), lambda preset: select_language('us',preset), default='us')
+		self._menu_options['mirror-region'] = \
+			Selector(
+				_('Select mirror region'),
+				select_mirror_regions,
+				display_func=lambda x: list(x.keys()) if x else '[]',
+				default={})
+		self._menu_options['sys-language'] = \
+			Selector(_('Select locale language'), lambda preset: select_locale_lang('en_US',preset), default='en_US')
+		self._menu_options['sys-encoding'] = \
+			Selector(_('Select locale encoding'), lambda preset: select_locale_enc('utf-8',preset), default='utf-8')
+		self._menu_options['harddrives'] = \
+			Selector(
+				_('Select harddrives'),
+				self._select_harddrives)
+		self._menu_options['disk_layouts'] = \
+			Selector(
+				_('Select disk layout'),
+				lambda x: select_disk_layout(
+					storage['arguments'].get('harddrives', []),
+					storage['arguments'].get('advanced', False)
+				),
+				dependencies=['harddrives'])
+		self._menu_options['!encryption-password'] = \
+			Selector(
+				_('Set encryption password'),
+				lambda x: self._select_encrypted_password(),
+				display_func=lambda x: secret(x) if x else 'None',
+				dependencies=['harddrives'])
+		self._menu_options['swap'] = \
+			Selector(
+				_('Use swap'),
+				lambda preset: ask_for_swap(preset),
+				default=True)
+		self._menu_options['bootloader'] = \
+			Selector(
+				_('Select bootloader'),
+				lambda preset: ask_for_bootloader(storage['arguments'].get('advanced', False),preset),
+				default="systemd-bootctl" if has_uefi() else "grub-install")
+		self._menu_options['hostname'] = \
+			Selector(
+				_('Specify hostname'),
+				ask_hostname,
+				default='archlinux')
+		# root password won't have preset value
+		self._menu_options['!root-password'] = \
+			Selector(
+				_('Set root password'),
+				lambda preset:self._set_root_password(),
+				display_func=lambda x: secret(x) if x else 'None')
+		self._menu_options['!superusers'] = \
+			Selector(
+				_('Specify superuser account'),
+				lambda preset: self._create_superuser_account(),
+				exec_func=lambda n,v:self._users_resynch(),
+				dependencies_not=['!root-password'],
+				display_func=lambda x: self._display_superusers())
+		self._menu_options['!users'] = \
+			Selector(
+				_('Specify user account'),
+				lambda x: self._create_user_account(),
+				default={},
+				exec_func=lambda n,v:self._users_resynch(),
+				display_func=lambda x: list(x.keys()) if x else '[]')
+		self._menu_options['profile'] = \
+			Selector(
+				_('Specify profile'),
+				lambda x: self._select_profile(),
+				display_func=lambda x: x if x else 'None')
+		self._menu_options['audio'] = \
+			Selector(
+				_('Select audio'),
+				lambda preset: ask_for_audio_selection(is_desktop_profile(storage['arguments'].get('profile', None)),preset))
+		self._menu_options['kernels'] = \
+			Selector(
+				_('Select kernels'),
+				lambda preset: select_kernel(preset),
+				default=['linux'])
+		self._menu_options['packages'] = \
+			Selector(
+				_('Additional packages to install'),
+				# lambda x: ask_additional_packages_to_install(storage['arguments'].get('packages', None)),
+				ask_additional_packages_to_install,
+				default=[])
+		self._menu_options['additional-repositories'] = \
+			Selector(
+				_('Additional repositories to enable'),
+				select_additional_repositories,
+				default=[])
+		self._menu_options['nic'] = \
+			Selector(
+				_('Configure network'),
+				ask_to_configure_network,
+				display_func=lambda x: x if x else _('Not configured, unavailable unless setup manually'),
+				default={})
+		self._menu_options['timezone'] = \
+			Selector(
+				_('Select timezone'),
+				lambda preset: ask_for_a_timezone(preset),
+				default='UTC')
+		self._menu_options['ntp'] = \
+			Selector(
+				_('Set automatic time sync (NTP)'),
+				lambda preset: self._select_ntp(preset),
+				default=True)
+		self._menu_options['save_config'] = \
+			Selector(
+				_('Save configuration'),
+				lambda preset: save_config(self._data_store),
+				enabled=True,
+				no_store=True)
+		self._menu_options['install'] = \
+			Selector(
+				self._install_text(),
+				exec_func=lambda n,v: True if len(self._missing_configs()) == 0 else False,
+				preview_func=self._prev_install_missing_config,
+				enabled=True,
+				no_store=True)
+
+		self._menu_options['abort'] = Selector(_('Abort'), exec_func=lambda n,v:exit(1), enabled=True)
+
+	def _update_install_text(self, name :str = None, result :Any = None):
+		text = self._install_text()
+		self._menu_options.get('install').update_description(text)
+
+	def post_callback(self,name :str = None ,result :Any = None):
+		self._update_install_text(name, result)
+
+	def exit_callback(self):
+		if self._data_store.get('harddrives', None) and self._data_store.get('!encryption-password', None):
+			# If no partitions was marked as encrypted, but a password was supplied and we have some disks to format..
+			# Then we need to identify which partitions to encrypt. This will default to / (root).
+			if len(list(encrypted_partitions(storage['arguments'].get('disk_layouts', [])))) == 0:
+				storage['arguments']['disk_layouts'] = select_encrypted_partitions(
+					storage['arguments']['disk_layouts'], storage['arguments']['!encryption-password'])
+
+	def _install_text(self):
+		missing = len(self._missing_configs())
+		if missing > 0:
+			return _('Install ({} config(s) missing)').format(missing)
+		return 'Install'
+
+	def _prev_install_missing_config(self) -> Optional[str]:
+		if missing := self._missing_configs():
+			text = str(_('Missing configurations:\n'))
+			for m in missing:
+				text += f'- {m}\n'
+			return text[:-1]  # remove last new line
+		return None
+
+	def _missing_configs(self) -> List[str]:
+		def check(s):
+			return self._menu_options.get(s).has_selection()
+
+		missing = []
+		if not check('bootloader'):
+			missing += ['Bootloader']
+		if not check('hostname'):
+			missing += ['Hostname']
+		if not check('audio'):
+			missing += ['Audio']
+		if not check('!root-password') and not check('!superusers'):
+			missing += [str(_('Either root-password or at least 1 superuser must be specified'))]
+		if not check('harddrives'):
+			missing += ['Hard drives']
+		if check('harddrives'):
+			if not self._menu_options.get('harddrives').is_empty() and not check('disk_layouts'):
+				missing += ['Disk layout']
+
+		return missing
+
+	def _set_root_password(self):
+		prompt = str(_('Enter root password (leave blank to disable root): '))
+		password = get_password(prompt=prompt)
+		return password
+
+	def _select_encrypted_password(self):
+		if passwd := get_password(prompt=str(_('Enter disk encryption password (leave blank for no encryption): '))):
+			return passwd
+		else:
+			return None
+
+	def _select_ntp(self, preset :bool = True) -> bool:
+		ntp = ask_ntp(preset)
+
+		value = str(ntp).lower()
+		SysCommand(f'timedatectl set-ntp {value}')
+
+		return ntp
+
+	def _select_harddrives(self, old_harddrives : list) -> list:
+		# old_haddrives = storage['arguments'].get('harddrives', [])
+		harddrives = select_harddrives(old_harddrives)
+
+		# in case the harddrives got changed we have to reset the disk layout as well
+		if old_harddrives != harddrives:
+			self._menu_options.get('disk_layouts').set_current_selection(None)
+			storage['arguments']['disk_layouts'] = {}
+
+		if not harddrives:
+			prompt = _(
+				"You decided to skip harddrive selection\nand will use whatever drive-setup is mounted at {} (experimental)\n"
+				"WARNING: Archinstall won't check the suitability of this setup\n"
+				"Do you wish to continue?"
+			).format(storage['MOUNT_POINT'])
+
+			choice = Menu(prompt, ['yes', 'no'], default_option='yes').run()
+
+			if choice == 'no':
+				return self._select_harddrives()
+
+		return harddrives
+
+	def _select_profile(self):
+		profile = select_profile()
+
+		# Check the potentially selected profiles preparations to get early checks if some additional questions are needed.
+		if profile and profile.has_prep_function():
+			namespace = f'{profile.namespace}.py'
+			with profile.load_instructions(namespace=namespace) as imported:
+				if not imported._prep_function():
+					log(' * Profile\'s preparation requirements was not fulfilled.', fg='red')
+					exit(1)
+
+		return profile
+
+	def _create_superuser_account(self):
+		superusers = ask_for_superuser_account(str(_('Manage superuser accounts: ')))
+		return superusers if superusers else None
+
+	def _create_user_account(self):
+		users = ask_for_additional_users(str(_('Manage ordinary user accounts: ')))
+		return users
+
+	def _display_superusers(self):
+		superusers = self._data_store.get('!superusers', {})
+
+		if self._menu_options.get('!root-password').has_selection():
+			return list(superusers.keys()) if superusers else '[]'
+		else:
+			return list(superusers.keys()) if superusers else ''
+
+	def _users_resynch(self):
+		self.synch('!superusers')
+		self.synch('!users')
+		return False
diff --git a/archinstall/lib/user_interaction/manage_users_conf.py b/archinstall/lib/user_interaction/manage_users_conf.py
index 0af0d776..6985a8eb 100644
--- a/archinstall/lib/user_interaction/manage_users_conf.py
+++ b/archinstall/lib/user_interaction/manage_users_conf.py
@@ -89,7 +89,7 @@ class UserList(ListManager):
 		elif self.action == self.actions[3]:  # delete
 			del self.data[active_user]
 
-	def _check_for_correct_username(username: str) -> bool:
+	def _check_for_correct_username(self, username: str) -> bool:
 		if re.match(r'^[a-z_][a-z0-9_-]*\$?$', username) and len(username) <= 32:
 			return True
 		log("The username you entered is invalid. Try again", level=logging.WARNING, fg='red')
diff --git a/archinstall/lib/user_interaction/partitioning_conf.py b/archinstall/lib/user_interaction/partitioning_conf.py
index 8c5d1375..ef4ba885 100644
--- a/archinstall/lib/user_interaction/partitioning_conf.py
+++ b/archinstall/lib/user_interaction/partitioning_conf.py
@@ -1,12 +1,13 @@
 from __future__ import annotations
 
-from typing import List, Any, Dict, Union, TYPE_CHECKING
+from typing import List, Any, Dict, Union, TYPE_CHECKING, Callable
 
 from ..disk import BlockDevice, suggest_single_disk_layout, suggest_multi_disk_layout, valid_parted_position
 from ..menu import Menu
 from ..output import log
 
 from ..disk.validators import fs_types
+from .subvolume_config import SubvolumeList
 
 if TYPE_CHECKING:
 	from ..disk.partition import Partition
@@ -64,8 +65,22 @@ def _current_partition_layout(partitions: List[Partition], with_idx: bool = Fals
 	return f'\n\n{title}:\n\n{current_layout}'
 
 
-def select_partition(title: str, partitions: List[Partition], multiple: bool = False) -> Union[int, List[int], None]:
-	partition_indexes = list(map(str, range(len(partitions))))
+def select_partition(title :str, partitions :List[Partition], multiple :bool = False, filter :Callable = None) -> Union[int, List[int], None]:
+	"""
+	filter allows to filter out the indexes once they are set. Should return True if element is to be included
+	"""
+	partition_indexes = []
+	for i in range(len(partitions)):
+		if filter:
+			if filter(partitions[i]):
+				partition_indexes.append(str(i))
+		else:
+			partition_indexes.append(str(i))
+	if len(partition_indexes) == 0:
+		return None
+	# old code without filter
+	# partition_indexes = list(map(str, range(len(partitions))))
+
 	partition = Menu(title, partition_indexes, multi=multiple).run()
 
 	if partition is not None:
@@ -111,6 +126,7 @@ def manage_new_and_existing_partitions(block_device: BlockDevice) -> Dict[str, A
 	mark_encrypted = str(_('Mark/Unmark a partition as encrypted'))
 	mark_bootable = str(_('Mark/Unmark a partition as bootable (automatic for /boot)'))
 	set_filesystem_partition = str(_('Set desired filesystem for a partition'))
+	set_btrfs_subvolumes = str(_('Set desired subvolumes on a btrfs partition'))
 
 	while True:
 		modes = [new_partition, suggest_partition_layout]
@@ -124,6 +140,7 @@ def manage_new_and_existing_partitions(block_device: BlockDevice) -> Dict[str, A
 				mark_encrypted,
 				mark_bootable,
 				set_filesystem_partition,
+				set_btrfs_subvolumes,
 			]
 
 		title = _('Select what to do with\n{}').format(block_device)
@@ -275,6 +292,23 @@ def manage_new_and_existing_partitions(block_device: BlockDevice) -> Dict[str, A
 
 					block_device_struct["partitions"][partition]['filesystem']['format'] = fstype
 
+			elif task == set_btrfs_subvolumes:
+				# TODO get preexisting partitions
+				title = _('{}\n\nSelect which partition to set subvolumes on').format(current_layout)
+				partition = select_partition(title, block_device_struct["partitions"],filter=lambda x:True if x.get('filesystem',{}).get('format') == 'btrfs' else False)
+				if partition is not None:
+					if not block_device_struct["partitions"][partition].get('btrfs', {}):
+						block_device_struct["partitions"][partition]['btrfs'] = {}
+					if not block_device_struct["partitions"][partition]['btrfs'].get('subvolumes', {}):
+						block_device_struct["partitions"][partition]['btrfs']['subvolumes'] = {}
+
+					prev = block_device_struct["partitions"][partition]['btrfs']['subvolumes']
+					result = SubvolumeList(_("Manage btrfs subvolumes for current partition"),prev).run()
+					if result:
+						block_device_struct["partitions"][partition]['btrfs']['subvolumes'] = result
+					else:
+						del block_device_struct["partitions"][partition]['btrfs']
+
 	return block_device_struct
 
 
diff --git a/archinstall/lib/user_interaction/subvolume_config.py b/archinstall/lib/user_interaction/subvolume_config.py
new file mode 100644
index 00000000..6de8d0ef
--- /dev/null
+++ b/archinstall/lib/user_interaction/subvolume_config.py
@@ -0,0 +1,129 @@
+from ..menu.list_manager import ListManager
+from ..menu.selection_menu import Selector, GeneralMenu
+from ..menu.text_input import TextInput
+from ..menu import Menu
+"""
+UI classes
+"""
+
+class SubvolumeList(ListManager):
+	def __init__(self,prompt,list):
+		self.ObjectNullAction = None # str(_('Add'))
+		self.ObjectDefaultAction = str(_('Add'))
+		super().__init__(prompt,list,None,self.ObjectNullAction,self.ObjectDefaultAction)
+
+	def reformat(self):
+		def presentation(key,value):
+			text = _(" Subvolume :{:16}").format(key)
+			if isinstance(value,str):
+				text += _(" mounted at {:16}").format(value)
+			else:
+				if value.get('mountpoint'):
+					text += _(" mounted at {:16}").format(value['mountpoint'])
+				else:
+					text += (' ' * 28)
+				if value.get('options',[]):
+					text += _(" with option {}").format(', '.join(value['options']))
+			return text
+
+		return sorted(list(map(lambda x:presentation(x,self.data[x]),self.data)))
+
+	def action_list(self):
+		return super().action_list()
+
+	def exec_action(self):
+		if self.target:
+			origkey,origval = list(self.target.items())[0]
+		else:
+			origkey = None
+
+		if self.action == str(_('Delete')):
+			del self.data[origkey]
+			return True
+
+		if self.action == str(_('Add')):
+			self.target = {}
+			print(_('\n Fill the desired values for a new subvolume \n'))
+			with SubvolumeMenu(self.target,self.action) as add_menu:
+				for data in ['name','mountpoint','options']:
+					add_menu.exec_option(data)
+		else:
+			SubvolumeMenu(self.target,self.action).run()
+		self.data.update(self.target)
+
+		return True
+
+class SubvolumeMenu(GeneralMenu):
+	def __init__(self,parameters,action=None):
+		self.data = parameters
+		self.action = action
+		self.ds = {}
+		self.ds['name'] = None
+		self.ds['mountpoint'] = None
+		self.ds['options'] = None
+		if self.data:
+			origkey,origval = list(self.data.items())[0]
+			self.ds['name'] = origkey
+			if isinstance(origval,str):
+				self.ds['mountpoint'] = origval
+			else:
+				self.ds['mountpoint'] = self.data[origkey].get('mountpoint')
+				self.ds['options'] = self.data[origkey].get('options')
+
+		super().__init__(data_store=self.ds)
+
+	def _setup_selection_menu_options(self):
+		# [str(_('Add')),str(_('Copy')),str(_('Edit')),str(_('Delete'))]
+		self._menu_options['name'] = Selector(str(_('Subvolume name ')),
+			self._select_subvolume_name if not self.action or self.action in (str(_('Add')),str(_('Copy'))) else None,
+			mandatory=True,
+			enabled=True)
+		self._menu_options['mountpoint'] = Selector(str(_('Subvolume mountpoint')),
+			self._select_subvolume_mount_point if not self.action or self.action in (str(_('Add')),str(_('Edit'))) else None,
+			enabled=True)
+		self._menu_options['options'] = Selector(str(_('Subvolume options')),
+			self._select_subvolume_options if not self.action or self.action in (str(_('Add')),str(_('Edit'))) else None,
+			enabled=True)
+		self._menu_options['save'] = Selector(str(_('Save')),
+													exec_func=lambda n,v:True,
+													enabled=True)
+		self._menu_options['cancel'] = Selector(str(_('Cancel')),
+													# func = lambda pre:True,
+													exec_func=lambda n,v:self.fast_exit(n),
+													enabled=True)
+		self.cancel_action = 'cancel'
+		self.save_action = 'save'
+		self.bottom_list = [self.save_action,self.cancel_action]
+
+	def fast_exit(self,accion):
+		if self.option(accion).get_selection():
+			for item in self.list_options():
+				if self.option(item).is_mandatory():
+					self.option(item).set_mandatory(False)
+		return True
+
+	def exit_callback(self):
+		# we exit without moving data
+		if self.option(self.cancel_action).get_selection():
+			return
+		if not self.ds['name']:
+			return
+		else:
+			key = self.ds['name']
+			value = {}
+			if self.ds['mountpoint']:
+				value['mountpoint'] = self.ds['mountpoint']
+			if self.ds['options']:
+				value['options'] = self.ds['options']
+			self.data.update({key : value})
+
+	def _select_subvolume_name(self,value):
+		return TextInput(str(_("Subvolume name :")),value).run()
+
+	def _select_subvolume_mount_point(self,value):
+		return TextInput(str(_("Select a mount point :")),value).run()
+
+	def _select_subvolume_options(self,value):
+		# def __init__(self, title, p_options, skip=True, multi=False, default_option=None, sort=True):
+		return Menu(str(_("Select the desired subvolume options ")),['nodatacow','compress'],
+			skip=True,preset_values=value,multi=True).run()