merged with upstreamHEADmaster

1 files changed, 175 insertions, 143 deletions

diff --git a/archinstall/__init__.py b/archinstall/__init__.py
index 9de4a3ec..1c980390 100644
--- a/archinstall/__init__.py
+++ b/archinstall/__init__.py
@@ -1,59 +1,70 @@
 """Arch Linux installer - guided, templates etc."""
-import typing
+import importlib
+import os
+import sys
+import time
+import traceback
 from argparse import ArgumentParser, Namespace
-
-from .lib.disk import *
-from .lib.exceptions import *
-from .lib.general import *
-from .lib.hardware import *
-from .lib.installer import __packages__, Installer, accessibility_tools_in_use
-from .lib.locale_helpers import *
-from .lib.luks import *
-from .lib.mirrors import *
-from .lib.models.network_configuration import NetworkConfigurationHandler
-from .lib.models.users import User
-from .lib.networking import *
-from .lib.output import *
-from .lib.models.dataclasses import (
-	VersionDef,
-	PackageSearchResult,
-	PackageSearch,
-	LocalPackage
-)
-from .lib.packages.packages import (
-	group_search,
-	package_search,
-	find_package,
-	find_packages,
-	installed_package,
-	validate_package_list,
-)
-from .lib.profiles import *
-from .lib.services import *
-from .lib.storage import *
-from .lib.systemd import *
-from .lib.user_interaction import *
-from .lib.menu import Menu
-from .lib.menu.list_manager import ListManager
-from .lib.menu.text_input import TextInput
-from .lib.menu.global_menu import GlobalMenu
-from .lib.menu.abstract_menu import (
-	Selector,
-	AbstractMenu
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, Union
+
+from .lib import disk
+from .lib import menu
+from .lib import models
+from .lib import packages
+from .lib import exceptions
+from .lib import luks
+from .lib import locale
+from .lib import mirrors
+from .lib import networking
+from .lib import profile
+from .lib import interactions
+from . import default_profiles
+
+from .lib.hardware import SysInfo, GfxDriver
+from .lib.installer import Installer, accessibility_tools_in_use
+from .lib.output import FormattedOutput, log, error, debug, warn, info
+from .lib.pacman import Pacman
+from .lib.storage import storage
+from .lib.global_menu import GlobalMenu
+from .lib.boot import Boot
+from .lib.translationhandler import TranslationHandler, Language, DeferredTranslation
+from .lib.plugins import plugins, load_plugin
+from .lib.configuration import ConfigurationOutput
+
+from .lib.general import (
+	generate_password, locate_binary, clear_vt100_escape_codes,
+	JSON, UNSAFE_JSON, SysCommandWorker, SysCommand,
+	run_custom_user_commands, json_stream_to_structure, secret
 )
-from .lib.translationhandler import TranslationHandler, DeferredTranslation
-from .lib.plugins import plugins, load_plugin # This initiates the plugin loading ceremony
-from .lib.configuration import *
-from .lib.udev import udevadm_info
-parser = ArgumentParser()
 
-__version__ = "2.5.4"
+if TYPE_CHECKING:
+	_: Any
+
+
+__version__ = "2.8.0"
 storage['__version__'] = __version__
 
-# add the custome _ as a builtin, it can now be used anywhere in the
+# add the custom _ as a builtin, it can now be used anywhere in the
 # project to mark strings as translatable with _('translate me')
 DeferredTranslation.install()
 
+# Log various information about hardware before starting the installation. This might assist in troubleshooting
+debug(f"Hardware model detected: {SysInfo.sys_vendor()} {SysInfo.product_name()}; UEFI mode: {SysInfo.has_uefi()}")
+debug(f"Processor model detected: {SysInfo.cpu_model()}")
+debug(f"Memory statistics: {SysInfo.mem_available()} available out of {SysInfo.mem_total()} total installed")
+debug(f"Virtualization detected: {SysInfo.virtualization()}; is VM: {SysInfo.is_vm()}")
+debug(f"Graphics devices detected: {SysInfo._graphics_devices().keys()}")
+
+# For support reasons, we'll log the disk layout pre installation to match against post-installation layout
+debug(f"Disk states before installing: {disk.disk_layouts()}")
+
+if 'sphinx' not in sys.modules and os.getuid() != 0:
+	print(_("Archinstall requires root privileges to run. See --help for more."))
+	exit(1)
+
+parser = ArgumentParser()
+
 
 def define_arguments():
 	"""
@@ -66,20 +77,25 @@ def define_arguments():
 	parser.add_argument("-v", "--version", action="version", version="%(prog)s " + __version__)
 	parser.add_argument("--config", nargs="?", help="JSON configuration file or URL")
 	parser.add_argument("--creds", nargs="?", help="JSON credentials configuration file")
-	parser.add_argument("--disk_layouts","--disk_layout","--disk-layouts","--disk-layout",nargs="?",
-					help="JSON disk layout file")
 	parser.add_argument("--silent", action="store_true",
 						help="WARNING: Disables all prompts for input and confirmation. If no configuration is provided, this is ignored")
 	parser.add_argument("--dry-run", "--dry_run", action="store_true",
 						help="Generates a configuration file and then exits instead of performing an installation")
 	parser.add_argument("--script", default="guided", nargs="?", help="Script to run for installation", type=str)
-	parser.add_argument("--mount-point","--mount_point", nargs="?", type=str, help="Define an alternate mount point for installation")
+	parser.add_argument("--mount-point", "--mount_point", nargs="?", type=str,
+						help="Define an alternate mount point for installation")
+	parser.add_argument("--skip-ntp", action="store_true", help="Disables NTP checks during installation", default=False)
 	parser.add_argument("--debug", action="store_true", default=False, help="Adds debug info into the log")
-	parser.add_argument("--offline", action="store_true", default=False, help="Disabled online upstream services such as package search and key-ring auto update.")
-	parser.add_argument("--no-pkg-lookups", action="store_true", default=False, help="Disabled package validation specifically prior to starting installation.")
+	parser.add_argument("--offline", action="store_true", default=False,
+						help="Disabled online upstream services such as package search and key-ring auto update.")
+	parser.add_argument("--no-pkg-lookups", action="store_true", default=False,
+						help="Disabled package validation specifically prior to starting installation.")
 	parser.add_argument("--plugin", nargs="?", type=str)
+	parser.add_argument("--skip-version-check", action="store_true",
+						help="Skip the version check when running archinstall")
+
 
-def parse_unspecified_argument_list(unknowns :list, multiple :bool = False, error :bool = False) -> dict:
+def parse_unspecified_argument_list(unknowns: list, multiple: bool = False, err: bool = False) -> dict:
 	"""We accept arguments not defined to the parser. (arguments "ad hoc").
 	Internally argparse return to us a list of words so we have to parse its contents, manually.
 	We accept following individual syntax for each argument
@@ -95,51 +111,52 @@ def parse_unspecified_argument_list(unknowns :list, multiple :bool = False, erro
 	argument value value ...
 	which isn't am error if multiple is specified
 	"""
-	tmp_list = unknowns[:]   # wastes a few bytes, but avoids any collateral effect of the destructive nature of the pop method()
+	tmp_list = unknowns[:]  # wastes a few bytes, but avoids any collateral effect of the destructive nature of the pop method()
 	config = {}
 	key = None
 	last_key = None
 	while tmp_list:
-		element = tmp_list.pop(0)			  # retrieve an element of the list
-		if element.startswith('--'):		   # is an argument ?
-			if '=' in element:				 # uses the arg=value syntax ?
+		element = tmp_list.pop(0)  # retrieve an element of the list
+		if element.startswith('--'):  # is an argument ?
+			if '=' in element:  # uses the arg=value syntax ?
 				key, value = [x.strip() for x in element[2:].split('=', 1)]
 				config[key] = value
-				last_key = key				 # for multiple handling
-				key = None					 # we have the kwy value pair we need
+				last_key = key  # for multiple handling
+				key = None  # we have the kwy value pair we need
 			else:
 				key = element[2:]
-				config[key] = True   # every argument starts its lifecycle as boolean
+				config[key] = True  # every argument starts its lifecycle as boolean
 		else:
 			if element == '=':
 				continue
 			if key:
 				config[key] = element
-				last_key = key # multiple
+				last_key = key  # multiple
 				key = None
 			else:
 				if multiple and last_key:
-					if isinstance(config[last_key],str):
-						config[last_key] = [config[last_key],element]
+					if isinstance(config[last_key], str):
+						config[last_key] = [config[last_key], element]
 					else:
 						config[last_key].append(element)
-				elif error:
+				elif err:
 					raise ValueError(f"Entry {element} is not related to any argument")
 				else:
 					print(f" We ignore the entry {element} as it isn't related to any argument")
 	return config
 
-def cleanup_empty_args(args :typing.Union[Namespace, dict]) -> dict:
+
+def cleanup_empty_args(args: Union[Namespace, Dict]) -> Dict:
 	"""
 	Takes arguments (dictionary or argparse Namespace) and removes any
 	None values. This ensures clean mergers during dict.update(args)
 	"""
-	if type(args) == Namespace:
+	if type(args) is Namespace:
 		args = vars(args)
 
 	clean_args = {}
 	for key, val in args.items():
-		if type(val) == dict:
+		if isinstance(val, dict):
 			val = cleanup_empty_args(val)
 
 		if val is not None:
@@ -147,6 +164,7 @@ def cleanup_empty_args(args :typing.Union[Namespace, dict]) -> dict:
 
 	return clean_args
 
+
 def get_arguments() -> Dict[str, Any]:
 	""" The handling of parameters from the command line
 	Is done on following steps:
@@ -161,7 +179,7 @@ def get_arguments() -> Dict[str, Any]:
 	3) Amend
 		Change whatever is needed on the configuration dictionary (it could be done in post_process_arguments but  this ougth to be left to changes anywhere else in the code, not in the arguments dictionary
 	"""
-	config = {}
+	config: Dict[str, Any] = {}
 	args, unknowns = parser.parse_known_args()
 	# preprocess the JSON files.
 	# TODO Expand the url access to the other JSON file arguments ?
@@ -174,15 +192,15 @@ def get_arguments() -> Dict[str, Any]:
 			exit(1)
 
 	# load the parameters. first the known, then the unknowns
-	args = cleanup_empty_args(args)
-	config.update(args)
+	clean_args = cleanup_empty_args(args)
+	config.update(clean_args)
 	config.update(parse_unspecified_argument_list(unknowns))
 	# amend the parameters (check internal consistency)
 	# Installation can't be silent if config is not passed
-	if args.get('config') is None:
+	if clean_args.get('config') is None:
 		config["silent"] = False
 	else:
-		config["silent"] = args.get('silent')
+		config["silent"] = clean_args.get('silent')
 
 	# avoiding a compatibility issue
 	if 'dry-run' in config:
@@ -190,60 +208,52 @@ def get_arguments() -> Dict[str, Any]:
 
 	return config
 
+
 def load_config():
 	"""
 	refine and set some arguments. Formerly at the scripts
 	"""
 	from .lib.models import NetworkConfiguration
 
+	arguments['locale_config'] = locale.LocaleConfiguration.parse_arg(arguments)
+
 	if (archinstall_lang := arguments.get('archinstall-language', None)) is not None:
 		arguments['archinstall-language'] = TranslationHandler().get_language_by_name(archinstall_lang)
 
-	if arguments.get('harddrives', None) is not None:
-		if type(arguments['harddrives']) is str:
-			arguments['harddrives'] = arguments['harddrives'].split(',')
-		arguments['harddrives'] = [BlockDevice(BlockDev) for BlockDev in arguments['harddrives']]
-		# Temporarily disabling keep_partitions if config file is loaded
-		# Temporary workaround to make Desktop Environments work
+	if disk_config := arguments.get('disk_config', {}):
+		arguments['disk_config'] = disk.DiskLayoutConfiguration.parse_arg(disk_config)
 
-	if arguments.get('profile', None) is not None:
-		if type(arguments.get('profile', None)) is dict:
-			arguments['profile'] = Profile(None, arguments.get('profile', None)['path'])
-		else:
-			arguments['profile'] = Profile(None, arguments.get('profile', None))
+	if profile_config := arguments.get('profile_config', None):
+		arguments['profile_config'] = profile.ProfileConfiguration.parse_arg(profile_config)
 
-	storage['_desktop_profile'] = arguments.get('desktop-environment', None)
-
-	if arguments.get('mirror-region', None) is not None:
-		if type(arguments.get('mirror-region', None)) is dict:
-			arguments['mirror-region'] = arguments.get('mirror-region', None)
-		else:
-			selected_region = arguments.get('mirror-region', None)
-			arguments['mirror-region'] = {selected_region: list_mirrors()[selected_region]}
-
-	arguments.setdefault('sys-language', 'en_US')
-	arguments.setdefault('sys-encoding', 'utf-8')
-
-	if arguments.get('gfx_driver', None) is not None:
-		storage['gfx_driver_packages'] = AVAILABLE_GFX_DRIVERS.get(arguments.get('gfx_driver', None), None)
+	if mirror_config := arguments.get('mirror_config', None):
+		arguments['mirror_config'] = mirrors.MirrorConfiguration.parse_args(mirror_config)
 
 	if arguments.get('servers', None) is not None:
 		storage['_selected_servers'] = arguments.get('servers', None)
 
-	if arguments.get('nic', None) is not None:
-		handler = NetworkConfigurationHandler()
-		handler.parse_arguments(arguments.get('nic'))
-		arguments['nic'] = handler.configuration
+	if (net_config := arguments.get('network_config', None)) is not None:
+		config = NetworkConfiguration.parse_arg(net_config)
+		arguments['network_config'] = config
 
 	if arguments.get('!users', None) is not None or arguments.get('!superusers', None) is not None:
 		users = arguments.get('!users', None)
 		superusers = arguments.get('!superusers', None)
-		arguments['!users'] = User.parse_arguments(users, superusers)
+		arguments['!users'] = models.User.parse_arguments(users, superusers)
+
+	if arguments.get('bootloader', None) is not None:
+		arguments['bootloader'] = models.Bootloader.from_arg(arguments['bootloader'])
 
-	if arguments.get('disk_encryption', None) is not None and arguments.get('disk_layouts', None) is not None:
+		if arguments.get('uki') and not arguments['bootloader'].has_uki_support():
+			arguments['uki'] = False
+
+	if arguments.get('audio_config', None) is not None:
+		arguments['audio_config'] = models.AudioConfiguration.parse_arg(arguments['audio_config'])
+
+	if arguments.get('disk_encryption', None) is not None and disk_config is not None:
 		password = arguments.get('encryption_password', '')
-		arguments['disk_encryption'] = DiskEncryption.parse_arg(
-			arguments['disk_layouts'],
+		arguments['disk_encryption'] = disk.DiskEncryption.parse_arg(
+			arguments['disk_config'],
 			arguments['disk_encryption'],
 			password
 		)
@@ -251,62 +261,84 @@ def load_config():
 
 def post_process_arguments(arguments):
 	storage['arguments'] = arguments
-	if arguments.get('mount_point'):
-		storage['MOUNT_POINT'] = arguments['mount_point']
+	if mountpoint := arguments.get('mount_point', None):
+		storage['MOUNT_POINT'] = Path(mountpoint)
 
 	if arguments.get('debug', False):
-		log(f"Warning: --debug mode will write certain credentials to {storage['LOG_PATH']}/{storage['LOG_FILE']}!", fg="red", level=logging.WARNING)
+		warn(f"Warning: --debug mode will write certain credentials to {storage['LOG_PATH']}/{storage['LOG_FILE']}!")
 
 	if arguments.get('plugin', None):
-		load_plugin(arguments['plugin'])
-
-	if arguments.get('disk_layouts', None) is not None:
-		layout_storage = {}
-		if not json_stream_to_structure('--disk_layouts',arguments['disk_layouts'],layout_storage):
-			exit(1)
-		else:
-			if arguments.get('harddrives') is None:
-				arguments['harddrives'] = [disk for disk in layout_storage]
-			# backward compatibility. Change partition.format for partition.wipe
-			for disk in layout_storage:
-				for i, partition in enumerate(layout_storage[disk].get('partitions',[])):
-					if 'format' in partition:
-						partition['wipe'] = partition['format']
-						del partition['format']
-					elif 'btrfs' in partition:
-						partition['btrfs']['subvolumes'] = Subvolume.parse_arguments(partition['btrfs']['subvolumes'])
-			arguments['disk_layouts'] = layout_storage
+		path = arguments['plugin']
+		load_plugin(path)
 
 	load_config()
 
 
 define_arguments()
-arguments = get_arguments()
+arguments: Dict[str, Any] = get_arguments()
 post_process_arguments(arguments)
 
+
 # @archinstall.plugin decorator hook to programmatically add
-# plugins in runtime. Useful in profiles and other things.
+# plugins in runtime. Useful in profiles_bck and other things.
 def plugin(f, *args, **kwargs):
 	plugins[f.__name__] = f
 
 
-def run_as_a_module():
+def _check_new_version():
+	info("Checking version...")
+
+	try:
+		Pacman.run("-Sy")
+	except Exception as e:
+		debug(f'Failed to perform version check: {e}')
+		info(f'Arch Linux mirrors are not reachable. Please check your internet connection')
+		exit(1)
+
+	upgrade = None
+
+	try:
+		upgrade = Pacman.run("-Qu archinstall").decode()
+	except Exception as e:
+		debug(f'Failed determine pacman version: {e}')
+
+	if upgrade:
+		text = f'New version available: {upgrade}'
+		info(text)
+		time.sleep(3)
+
+
+def main():
 	"""
-	Since we're running this as a 'python -m archinstall' module OR
-	a nuitka3 compiled version of the project.
-	This function and the file __main__ acts as a entry point.
+	This can either be run as the compiled and installed application: python setup.py install
+	OR straight as a module: python -m archinstall
+	In any case we will be attempting to load the provided script to be run from the scripts/ folder
 	"""
+	if not arguments.get('skip_version_check', False):
+		_check_new_version()
 
-	# Add another path for finding profiles, so that list_profiles() in Script() can find guided.py, unattended.py etc.
-	storage['PROFILE_PATH'].append(os.path.abspath(f'{os.path.dirname(__file__)}/examples'))
-	try:
-		script = Script(arguments.get('script', None))
-	except ProfileNotFound as err:
-		print(f"Couldn't find file: {err}")
-		sys.exit(1)
+	script = arguments.get('script', None)
+
+	if script is None:
+		print('No script to run provided')
 
-	os.chdir(os.path.abspath(os.path.dirname(__file__)))
+	mod_name = f'archinstall.scripts.{script}'
+	# by loading the module we'll automatically run the script
+	importlib.import_module(mod_name)
+
+
+def run_as_a_module():
+	try:
+		main()
+	except Exception as e:
+		err = ''.join(traceback.format_exception(e))
+		error(err)
+
+		text = (
+			'Archinstall experienced the above error. If you think this is a bug, please report it to\n'
+			'https://github.com/archlinux/archinstall and include the log file "/var/log/archinstall/install.log".\n\n'
+			'Hint: To extract the log from a live ISO \ncurl -F\'file=@/var/log/archinstall/install.log\' https://0x0.st\n'
+		)
 
-	# Remove the example directory from the PROFILE_PATH, to avoid guided.py etc shows up in user input questions.
-	storage['PROFILE_PATH'].pop()
-	script.execute()
+		warn(text)
+		exit(1)