ASInstaller/examples/guided.py

import logging
import os
import shlex
import time
import json

import archinstall
from archinstall import ConfigurationOutput, Menu
from archinstall.lib.general import SysCommand, reboot, SysCallError
from archinstall.lib.models.network_configuration import NetworkConfigurationHandler
from archinstall.lib.disk.lvmoperation import lvm_layout
from typing import Dict, Any
def parted_fix():
    import pexpect
    proc = pexpect.spawn("parted -l")
    proc.expect(['Fix/Ignore',pexpect.EOF])
    proc.sendline("Fix")
    proc.expect(pexpect.EOF)

def partprobe(devname=None):
    if devname:
        SysCommand(f'partprobe {devname}', peak_output=True)
    else:
        SysCommand('partprobe', peak_output=True)

def find_lvm(part):
    if "children" in part:
        for pp in part["children"]:
            if pp["type"] == "lvm":
                yield pp["name"]
            for n in find_lvm(pp):
                yield n

def find_allname(part):
    names = []
    names.append(part["name"])
    if "children" in part:
        for pp in part["children"]:
            names += find_allname(pp)
    return names

def all_pvnames():
    pvlist = []
    output2 = SysCommand(f'pvs --reportformat json').decode()
    if output2:
        data2 = json.loads(output2)
        report = data2["report"][0]
        if "pv" in report and len(report["pv"]) > 0:
            pvlist += report["pv"]
    return pvlist

def remove_lvm(devpath):
    output = SysCommand(f'lsblk --json {devpath}').decode()
    if output:
        data = json.loads(output)
        dev_tree = data["blockdevices"][0]
        allnames = [f'/dev/{x}' for x in find_allname(dev_tree)]
        # print("dbg3// all dev names:", allnames)

        pvs = all_pvnames()

        # print("dbg4// pvs:", pvs)

        for pv in pvs:
            if pv["vg_name"] and pv["pv_name"] in allnames:
                # print("dbg5// remove group:", pv)
                SysCommand(f'vgremove -f {pv["vg_name"]}', peak_output=True)


if archinstall.arguments.get('help'):
    print("See acosail.com for help.")
    exit(0)
if os.getuid() != 0:
    print(_("ASInstall requires root privileges to run. See --help for more."))
    exit(1)

# Log various information about hardware before starting the installation. This might assist in troubleshooting
# {archinstall.product_name()}
archinstall.log(f"Hardware model detected: {archinstall.sys_vendor()} ; UEFI mode: {archinstall.has_uefi()}", level=logging.INFO)
archinstall.log(f"Processor model detected: {archinstall.cpu_model()}", level=logging.DEBUG)
archinstall.log(f"Memory statistics: {archinstall.mem_available()} available out of {archinstall.mem_total()} total installed", level=logging.DEBUG)
archinstall.log(f"Virtualization detected: {archinstall.virtualization()}; is VM: {archinstall.is_vm()}", level=logging.DEBUG)
archinstall.log(f"Graphics devices detected: {archinstall.graphics_devices().keys()}", level=logging.DEBUG)

# For support reasons, we'll log the disk layout pre installation to match against post-installation layout
archinstall.log(f"Disk states before installing: {archinstall.disk_layouts()}", level=logging.DEBUG)


def ask_user_questions():
    """
            First, we'll ask the user for a bunch of user input.
            Not until we're satisfied with what we want to install
            will we continue with the actual installation steps.
    """

    archinstall.SysCommand('local-install lvm2 aio')
    partprobe()
    time.sleep(0.5)
    # ref: https://github.com/archlinux/archinstall/pull/831
    # we'll set NTP to true by default since this is also
    # the default value specified in the menu options; in
    # case it will be changed by the user we'll also update
    # the system immediately
    global_menu = archinstall.GlobalMenu(data_store=archinstall.arguments)

    # global_menu.enable('archinstall-language')

    # global_menu.enable('keyboard-layout')

    # Set which region to download packages from during the installation
    # global_menu.enable('mirror-region')

    global_menu.enable('sys-language')
    global_menu.enable('sys-encoding')

    # Ask which harddrives/block-devices we will install to
    # and convert them into archinstall.BlockDevice() objects.
    global_menu.enable('harddrives')

    global_menu.enable('disk_layouts')

    global_menu.enable('lvmdrives')

    global_menu.enable('lvm_layouts')

    # Get disk encryption password (or skip if blank)
    # global_menu.enable('!encryption-password')

    if archinstall.arguments.get('advanced', False) or archinstall.arguments.get('HSM', None):
        # Enables the use of HSM
        global_menu.enable('HSM')

    # Ask which boot-loader to use (will only ask if we're in UEFI mode, otherwise will default to GRUB)
    # global_menu.enable('bootloader')

    # global_menu.enable('swap')

    # Get the hostname for the machine
    global_menu.enable('hostname')

    # Ask for a root password (optional, but triggers requirement for super-user if skipped)
    global_menu.enable('!root-password')

    global_menu.enable('!users')

    # Ask for archinstall-specific profiles (such as desktop environments etc)
    # global_menu.enable('profile')

    # Ask about audio server selection if one is not already set
    # global_menu.enable('audio')

    # Ask for preferred kernel:
    # global_menu.enable('kernels')


    # if archinstall.arguments.get('advanced', False):
    # 	# Enable parallel downloads
    # 	global_menu.enable('parallel downloads')

    # Ask or Call the helper function that asks the user to optionally configure a network.
    # global_menu.enable('nic')

    # global_menu.enable('timezone')

    global_menu.enable("base_image")

    global_menu.enable('packages')

    # global_menu.enable('ntp')

    # global_menu.enable('additional-repositories')

    global_menu.enable('__separator__')

    global_menu.enable('save_config')
    global_menu.enable('install')
    global_menu.enable('reboot')
    global_menu.enable('abort')

    global_menu.run()


def perform_filesystem_operations():
    """
            Issue a final warning before we continue with something un-revertable.
            We mention the drive one last time, and count from 5 to 0.
    """

    if archinstall.arguments.get('harddrives', None):
        print(_(f" ! Formatting {archinstall.arguments['harddrives']} in "), end='')
        archinstall.do_countdown()

        """
            Setup the blockdevice, filesystem (and optionally encryption).
            Once that's done, we'll hand over to perform_installation()
        """
        mode = archinstall.GPT
        if archinstall.has_uefi() is False:
            mode = archinstall.MBR

        for drive in archinstall.arguments.get('harddrives', []):
            if archinstall.arguments.get('disk_layouts', {}).get(drive.path):
                with archinstall.Filesystem(drive, mode) as fs:
                    fs.load_layout(archinstall.arguments['disk_layouts'][drive.path])

    if archinstall.arguments.get('lvmdrives', None):
        print(_(f" ! Formatting {archinstall.arguments['lvmdrives']} in "), end='')
        """
            Setup the lvm device, filesystem
            Once that's done, we'll hand over to perform_installation()
        """
        lv2 = lvm_layout(lvmdrives=archinstall.arguments.get('lvmdrives', []))
        # print("lvmdrives:===========>>>>>",lv2)


def install_grub(installation : archinstall.Installer, boot_dev_path: str):
    target = installation.target

    installation.log("Installing grub...", level=logging.INFO)
    if archinstall.has_uefi():
        archinstall.SysCommand('mount -t efivarfs none {}/sys/firmware/efi/efivars/'.format(target))
        archinstall.SysCommand('mount -t efivarfs none /sys/firmware/efi/efivars/')
        archinstall.SysCommand('chroot {} bash -c "grub-install"'.format(target))
    else:
        archinstall.SysCommand('bash -c "cp -a /mnt/pool/other/grub-pc*.deb {}/tmp"'.format(target))
        archinstall.SysCommand('chroot {} bash -c "dpkg -i /tmp/grub-pc*.deb"'.format(target))
        archinstall.SysCommand('chroot {} bash -c "grub-install {}"'.format(target, boot_dev_path))

    archinstall.SysCommand('chroot {} bash -c "grub-mkconfig >> /boot/grub/grub.cfg"'.format(target))

    # if archinstall.has_uefi():
    #     installation.log("Installing finish", level=logging.INFO)

    installation.log("Installing grub finish", level=logging.INFO)

def extract_rootfs(installation):
    target = installation.target

    image_name = archinstall.arguments.get('base_image')

    syspack = ""
    filenames = os.listdir("/mnt/images")
    for fname in filenames:
        if fname.startswith("rootfs") and image_name in fname:
            syspack = fname
            break
    if syspack:
        installation.log("Installing system: {}".format(syspack), level=logging.INFO)
        archinstall.SysCommand('sh -c "xz -d -c {} | tar -x -C {}"'.format("/mnt/images/" + syspack, target))
        installation.log("Installing system finish.", level=logging.INFO)
    else:
        installation.log("Can't find base image: {}".format(image_name), level=logging.ERROR)
        raise ValueError("[Installer] Can't find base image: {}".format(image_name))

def mount_target_paths(installation):
    target = installation.target
    archinstall.SysCommand('mkdir -p {}/run'.format(target))
    archinstall.SysCommand('mkdir -p {}/proc'.format(target))
    archinstall.SysCommand('mkdir -p {}/sys'.format(target))
    archinstall.SysCommand('mkdir -p {}/dev'.format(target))

    archinstall.SysCommand('mkdir -p {}/dev'.format(target))
    archinstall.SysCommand('mkdir -p {}/media'.format(target))

    archinstall.SysCommand('mount -t tmpfs none {}/run'.format(target))
    archinstall.SysCommand('mount -t proc proc {}/proc'.format(target))
    archinstall.SysCommand('mount -t sysfs sys {}/sys'.format(target))
    archinstall.SysCommand('mount -o bind /dev {}/dev'.format(target))

    archinstall.SysCommand('mount -o bind /mnt {}/media'.format(target))

def install_kernel(installation):

    target = installation.target
    try:
        # install lvm2 package
        installation.log("Installing LVM package: {}".format("..."), level=logging.DEBUG)
        archinstall.SysCommand('chroot {} bash -c "dpkg -i /media/pool/main/liba/libaio/*.deb"'.format(target))
        archinstall.SysCommand('chroot {} bash -c "dpkg -i /media/pool/main/l/lvm2/*.deb"'.format(target))
        archinstall.SysCommand('chroot {} bash -c "dpkg -i /media/pool/main/c/cryptsetup/*.deb"'.format(target))

        installation.log("Installing Kenel: {}".format("..."), level=logging.INFO)
        archinstall.SysCommand('mkdir -p {}/tmp'.format(target))
        archinstall.SysCommand('cp -ar /mnt/pool/other/kernel {}/tmp'.format(target))
        archinstall.SysCommand('chroot {} bash -c "dpkg -i /tmp/kernel/linux-*.deb"'.format(target))
        cmdline = ""
        with open("/proc/cmdline", "r") as cmdfs:
            cmdline = cmdfs.read()
        if False and "nomwv" not in cmdline:
            archinstall.SysCommand('chroot {} bash -c "dpkg -i /tmp/kernel/mwv207-*.deb"'.format(target))
        archinstall.SysCommand("chroot {} bash -c 'depmod `ls /lib/modules | tail -n 1`'".format(target))
        archinstall.SysCommand('rm -fr {}/tmp/kernel'.format(target))
    except SysCallError as error:
        installation.log(f"install_kernel error: {error!r}", level=logging.ERROR, fg="red")
        raise error
    installation.log("Installing Kenel finish.", level=logging.INFO)

def do_config(installation):
    target = installation.target

    ############
    # config apt sources
    ############
    installation.log("Config apt source list")
    with open(f"{target}/etc/apt/sources.list", "w") as fw:
        fw.write("""deb http://mirrors.huaweicloud.com/debian/ bullseye main
deb-src http://mirrors.huaweicloud.com/debian/ bullseye main
""")

    ############
    # config apt sources
    ############
    installation.log("Config machine id")
    try:
        archinstall.SysCommand("chroot {} bash -c 'rm -f /etc/machine-id /var/lib/dbus/machine-id'".format(target))
        archinstall.SysCommand("chroot {} bash -c 'dbus-uuidgen --ensure=/etc/machine-id'".format(target))
        archinstall.SysCommand("chroot {} bash -c 'dbus-uuidgen --ensure'".format(target))
    except SysCallError as error:
        installation.log(f"machine id config error: {error!r}", level=logging.ERROR, fg="red")
        raise error

    
    ############
    # clear tmp
    ############
    try:
        archinstall.SysCommand("bash -c 'rm -f /mnt/tmp/*'".format(target))
    except SysCallError as error:
        installation.log(f"clear tmp failed: {error!r}", level=logging.WARN, fg="yellow")

def install_packages(installation,additional_packages):
    target = installation.target
    if len(additional_packages) == 0:
        return
    try:
        # install additional_packages package
        tmpFolder = '/tmp/acosail'
        installation.log("Installing additional_packages package, {}".format("..."), level=logging.INFO)
        archinstall.SysCommand('chroot {} bash -c "mkdir -p {}"'.format(target,tmpFolder))
        archinstall.SysCommand('chroot {} bash -c "cp -a /media/pool/other/plugins/ {}"'.format(target,tmpFolder))
        for key_name in additional_packages:
            installation.log("Installing {} package...".format(key_name), level=logging.INFO)
            if key_name == 'kde-plasma':
                installation.log("tar -zxvf {} to folder {}".format(key_name,tmpFolder))
                archinstall.SysCommand('chroot {} bash -c "dpkg -i /media/pool/other/plugins/{}/*.deb"'.format(target,'psmisc'))
                archinstall.SysCommand('chroot {} bash -c "tar -zxvf {}/plugins/{}.tgz -C /tmp/acosail/plugins/"'.format(target,tmpFolder,key_name))
                archinstall.SysCommand('chroot {} bash -c "dpkg -i {}/plugins/{}/*.deb"'.format(target,tmpFolder,key_name))
                
            # change default dm
            time.sleep(0.1)
            archinstall.SysCommand('chroot {} bash -c "rm -rf /etc/systemd/system/display-manager.service"'.format(target))
            archinstall.SysCommand('chroot {} bash -c "ln -s /lib/systemd/system/lightdm.service /etc/systemd/system/display-manager.service"'.format(target))
            archinstall.SysCommand('chroot {} bash -c "mv /usr/share/xsessions/lightdm-xsession.desktop /usr/share/xsessions/qlightdm-xsession.desktop"'.format(target))
    except SysCallError as error:
        installation.log(f"additional_packages error: {error!r}", level=logging.ERROR, fg="red")
        raise error

    archinstall.SysCommand('chroot {} bash -c "rm -rf {}"'.format(target,tmpFolder))
    installation.log("Installing additional_packages finish.", level=logging.INFO)
    installation.log("And The default startup session may not be KDE-Plasma DM, and you can modify it on the login page.", level=logging.INFO)


def perform_installation2(mountpoint):
    with archinstall.Installer(mountpoint, kernels=archinstall.arguments.get('kernels', ['linux'])) as installation:
        disk_layouts = None
        if archinstall.arguments.get('disk_layouts'):
            disk_layouts = archinstall.arguments['disk_layouts']

        if (lvmdrives := archinstall.arguments.get('lvmdrives',None)):
            # print("lvmdrives=====>>>>",lvmdrives)
            for drive in lvmdrives:
                disk_layouts[drive.vg_name] = {"partitions":drive.partitions}
        # print("after:====>>>",disk_layouts)

        installation.mount_ordered_layout(disk_layouts)

        # Placing /boot check during installation because this will catch both re-use and wipe scenarios.

        boot_dev_path = None
        for partition in installation.partitions:
            archinstall.log(f"part mount: {partition.mountpoint}, install boot:{installation.target + '/boot'}", level=logging.DEBUG)
            if str(partition.mountpoint) == installation.target + '/boot':
                boot_dev_path = partition.block_device.path
                if partition.size < 0.19: # ~200 MiB in GiB
                    raise archinstall.DiskError(f"The selected /boot partition in use is not large enough to properly install a boot loader. Please resize it to at least 200MiB and re-run the installation.")

        pass
        archinstall.log(f"boot dev:{boot_dev_path}", level=logging.DEBUG)
        # 0 install dep package
        archinstall.SysCommand('local-install libpam0g bash_5 passwd_4 grub-installer arch-install grub efi di-utils-mapdevfs partman-utils')
        archinstall.SysCommand('local-install dpkg mawk')

        archinstall.SysCommand('depmod')
        
        if archinstall.has_uefi():
            archinstall.SysCommand('modprobe efivarfs')

        # 1 extract root files
        extract_rootfs(installation)

        # 2 update fstab
        installation.genfstab()

        # 3 reinstall kernel
        mount_target_paths(installation)
        install_kernel(installation)

        # 4 grub-install
        install_grub(installation, boot_dev_path)

        # 5 add user/passwd
        if (root_pw := archinstall.arguments.get('!root-password', None)) and len(root_pw):
            installation.user_set_pw('root', root_pw)
        
        # 6 do config
        do_config(installation)
        # 7 install adational packages
        if archinstall.arguments.get('base_image_develop',False):
            addition_packages = archinstall.arguments.get('packages',[])
            install_packages(installation,addition_packages)

        input(str(_('Install finished, user:[acosail] password is [ ].\nPress Enter to reboot.')))
        reboot()


def perform_installation(mountpoint):
    """
    Performs the installation steps on a block device.
    Only requirement is that the block devices are
    formatted and setup prior to entering this function.
    """

    with archinstall.Installer(mountpoint, kernels=archinstall.arguments.get('kernels', ['linux'])) as installation:
        # Mount all the drives to the desired mountpoint
        # This *can* be done outside of the installation, but the installer can deal with it.
        if archinstall.arguments.get('disk_layouts'):
            installation.mount_ordered_layout(archinstall.arguments['disk_layouts'])

        # Placing /boot check during installation because this will catch both re-use and wipe scenarios.
        for partition in installation.partitions:
            if partition.mountpoint == installation.target + '/boot':
                if partition.size < 0.19: # ~200 MiB in GiB
                    raise archinstall.DiskError(f"The selected /boot partition in use is not large enough to properly install a boot loader. Please resize it to at least 200MiB and re-run the installation.")

        # if len(mirrors):
        # Certain services might be running that affects the system during installation.
        # Currently, only one such service is "reflector.service" which updates /etc/pacman.d/mirrorlist
        # We need to wait for it before we continue since we opted in to use a custom mirror/region.
        installation.log('Waiting for automatic mirror selection (reflector) to complete.', level=logging.INFO)
        while archinstall.service_state('reflector') not in ('dead', 'failed'):
            time.sleep(1)

        # If we've activated NTP, make sure it's active in the ISO too and
        # make sure at least one time-sync finishes before we continue with the installation
        if archinstall.arguments.get('ntp', False):
            # Activate NTP in the ISO
            archinstall.SysCommand('timedatectl set-ntp true')

            # TODO: This block might be redundant, but this service is not activated unless
            # `timedatectl set-ntp true` is executed.
            logged = False
            while archinstall.service_state('dbus-org.freedesktop.timesync1.service') not in ('running'):
                if not logged:
                    installation.log(f"Waiting for dbus-org.freedesktop.timesync1.service to enter running state", level=logging.INFO)
                    logged = True
                time.sleep(1)

            logged = False
            while 'Server: n/a' in archinstall.SysCommand('timedatectl timesync-status --no-pager --property=Server --value'):
                if not logged:
                    installation.log(f"Waiting for timedatectl timesync-status to report a timesync against a server", level=logging.INFO)
                    logged = True
                time.sleep(1)

        # Set mirrors used by pacstrap (outside of installation)
        if archinstall.arguments.get('mirror-region', None):
            archinstall.use_mirrors(archinstall.arguments['mirror-region'])  # Set the mirrors for the live medium

        # Retrieve list of additional repositories and set boolean values appropriately
        if archinstall.arguments.get('additional-repositories', None) is not None:
            enable_testing = 'testing' in archinstall.arguments.get('additional-repositories', None)
            enable_multilib = 'multilib' in archinstall.arguments.get('additional-repositories', None)
        else:
            enable_testing = False
            enable_multilib = False

        if installation.minimal_installation(
                        testing=enable_testing, multilib=enable_multilib, hostname=archinstall.arguments['hostname'],
                        locales=[f"{archinstall.arguments['sys-language']} {archinstall.arguments['sys-encoding'].upper()}"]):
            if archinstall.arguments.get('mirror-region') is not None:
                if archinstall.arguments.get("mirrors", None) is not None:
                    installation.set_mirrors(archinstall.arguments['mirror-region'])  # Set the mirrors in the installation medium
            if archinstall.arguments.get('swap'):
                installation.setup_swap('zram')
            if archinstall.arguments.get("bootloader") == "grub-install" and archinstall.has_uefi():
                installation.add_additional_packages("grub")
            installation.add_bootloader(archinstall.arguments["bootloader"])

            # If user selected to copy the current ISO network configuration
            # Perform a copy of the config
            network_config = archinstall.arguments.get('nic', None)

            if network_config:
                handler = NetworkConfigurationHandler(network_config)
                handler.config_installer(installation)

            if archinstall.arguments.get('audio', None) is not None:
                installation.log(f"This audio server will be used: {archinstall.arguments.get('audio', None)}", level=logging.INFO)
                if archinstall.arguments.get('audio', None) == 'pipewire':
                    archinstall.Application(installation, 'pipewire').install()
                elif archinstall.arguments.get('audio', None) == 'pulseaudio':
                    print('Installing pulseaudio ...')
                    installation.add_additional_packages("pulseaudio")
            else:
                installation.log("No audio server will be installed.", level=logging.INFO)

            if archinstall.arguments.get('packages', None) and archinstall.arguments.get('packages', None)[0] != '':
                installation.add_additional_packages(archinstall.arguments.get('packages', None))

            if archinstall.arguments.get('profile', None):
                installation.install_profile(archinstall.arguments.get('profile', None))

            if users := archinstall.arguments.get('!users', None):
                installation.create_users(users)

            if timezone := archinstall.arguments.get('timezone', None):
                installation.set_timezone(timezone)

            if archinstall.arguments.get('ntp', False):
                installation.activate_time_syncronization()

            if archinstall.accessibility_tools_in_use():
                installation.enable_espeakup()

            if (root_pw := archinstall.arguments.get('!root-password', None)) and len(root_pw):
                installation.user_set_pw('root', root_pw)

            # This step must be after profile installs to allow profiles to install language pre-requisits.
            # After which, this step will set the language both for console and x11 if x11 was installed for instance.
            installation.set_keyboard_language(archinstall.arguments['keyboard-layout'])

            if archinstall.arguments['profile'] and archinstall.arguments['profile'].has_post_install():
                with archinstall.arguments['profile'].load_instructions(namespace=f"{archinstall.arguments['profile'].namespace}.py") as imported:
                    if not imported._post_install():
                        archinstall.log(' * Profile\'s post configuration requirements was not fulfilled.', fg='red')
                        exit(1)

        # If the user provided a list of services to be enabled, pass the list to the enable_service function.
        # Note that while it's called enable_service, it can actually take a list of services and iterate it.
        if archinstall.arguments.get('services', None):
            installation.enable_service(*archinstall.arguments['services'])

        # If the user provided custom commands to be run post-installation, execute them now.
        if archinstall.arguments.get('custom-commands', None):
            archinstall.run_custom_user_commands(archinstall.arguments['custom-commands'], installation)

        installation.genfstab()

        installation.log("For post-installation tips, see https://wiki.archlinux.org/index.php/Installation_guide#Post-installation", fg="yellow")
        if not archinstall.arguments.get('silent'):
            prompt = str(_('Would you like to chroot into the newly created installation and perform post-installation configuration?'))
            choice = Menu(prompt, Menu.yes_no(), default_option=Menu.yes()).run()
            if choice.value == Menu.yes():
                try:
                    installation.drop_to_shell()
                except:
                    pass

    # For support reasons, we'll log the disk layout post installation (crash or no crash)
    archinstall.log(f"Disk states after installing: {archinstall.disk_layouts()}", level=logging.DEBUG)


# if archinstall.arguments.get('skip-mirror-check', True) is False and archinstall.check_mirror_reachable() is False:
# 	log_file = os.path.join(archinstall.storage.get('LOG_PATH', None), archinstall.storage.get('LOG_FILE', None))
# 	archinstall.log(f"Arch Linux mirrors are not reachable. Please check your internet connection and the log file '{log_file}'.", level=logging.INFO, fg="red")
# 	exit(1)

# if not archinstall.arguments.get('offline', True):
# 	latest_version_archlinux_keyring = max([k.pkg_version for k in archinstall.find_package('archlinux-keyring')])

# 	# If we want to check for keyring updates
# 	# and the installed package version is lower than the upstream version
# 	if archinstall.arguments.get('skip-keyring-update', False) is False and \
# 		archinstall.installed_package('archlinux-keyring').version < latest_version_archlinux_keyring:

# 		# Then we update the keyring in the ISO environment
# 		if not archinstall.update_keyring():
# 			log_file = os.path.join(archinstall.storage.get('LOG_PATH', None), archinstall.storage.get('LOG_FILE', None))
# 			archinstall.log(f"Failed to update the keyring. Please check your internet connection and the log file '{log_file}'.", level=logging.INFO, fg="red")
# 			exit(1)

def user_set_pw(target, user, password):
    combo = f'{user}:{password}'
    echo = shlex.join(['echo', combo])
    sh = shlex.join(['sh', '-c', echo])

    result = SysCommand(f"chroot {target} " + sh[:-1] + " | chpasswd'", peak_output=True)

def perform_installation_diskimg():
    drive = archinstall.arguments.get('harddrives', [])

    # check img localtion
    sysimgpath = ""
    sysimgpathlist = [
            "/root/system_images/sys.img",
            "/run/install/repo/system_images/sys.img"
    ]
    for imgpath in sysimgpathlist:
        if os.path.exists(imgpath):
            sysimgpath = imgpath
            break

    archinstall.log(f"remove lvm from disk: {drive[0]}")
    remove_lvm(drive[0].device)

    archinstall.log(f"restore [{sysimgpath}] system into disk: {drive[0]}")
    cmd = SysCommand(f'dd if="{sysimgpath}" of="{drive[0].device}" status=noxfer bs=5M', peak_output=True)

    archinstall.log(f"File copy is finished.")
    parted_fix()
    partprobe(f"{drive[0].device}")

    # mount rootfs
    # mountpoint = ""
    # mountcheck = SysCommand(f'lsblk --json {drive[0].device}').decode()

    # devpath = []
    # if mountcheck:
    # 	data = json.loads(mountcheck)
    # 	if len(data["blockdevices"]) > 0:
    # 		mnts = []
    # 		deinfo = data["blockdevices"][0]
    # 		# print("dbg7// ", deinfo)
    # 		for part in deinfo["children"]:
    # 			devpath.append(f"/dev/{part['name']}")
    # 			mnts.append(part['mountpoint'])
    # 		mountpoint = mnts[2]

    # if not mountpoint:
    # 	cmd = SysCommand(f'mount {devpath[2]} /run/media/')
    # 	mountpoint = "/run/media"

    # 	SysCommand(f'mount {devpath[1]} /run/media/boot')
    # 	SysCommand(f'mount {devpath[0]} /run/media/boot/efi')

    # for d in ["/dev","/dev/pts","/proc","/sys","/run"]:
    # 	SysCommand(f'mount -B {d} {mountpoint}{d}')

    # with open(f"{mountpoint}/etc/fstab", "w") as fw:
    # 	fw.write(f"{devpath[2]} / ext4 defaults 0 0 \n")
    # 	fw.write(f"{devpath[1]} /boot ext4 defaults 0 0 \n")
    # 	fw.write(f"{devpath[0]} /boot/efi vfat umask=0077,shortname=winnt 0 2 \n")

    # archinstall.log(f"Updating fstab.")
    # SysCommand(f'cat {mountpoint}/etc/fstab', peak_output=True)

    # archinstall.log(f"Updating grub.")

    # try:
    # 	SysCommand(f"grub2-install --recheck {drive[0].device}", peak_output=True)
    # except SysCallError as ex:
    # 	if ex.exit_code in (256):
    # 		pass

    # try:
    # 	SysCommand(f"chroot {mountpoint} grub2-mkconfig -o /boot/efi/EFI/acosail/grub.cfg", peak_output=True)
    # except SysCallError as ex:
    # 	if ex.exit_code in (256,):
    # 		pass

    # archinstall.log(f"Generate initramfs.")
    # SysCommand(f"chroot {mountpoint} dracut -f /boot/initramfs-4.18.0-372.13.1.rt7.170.el8_6.2.aarch64.img -k /lib/modules/4.18.0-372.13.1.rt7.170.el8_6.2.aarch64", peak_output=True)

# 	with open(f"{mountpoint}/boot/efi/EFI/acosail/grubenv", "w") as fw:
# 		fw.write('''# GRUB Environment Block
# saved_entry=233e37fb44c141b68a73de4702e0d54a-4.18.0-372.13.1.rt7.170.el8_6.2.aarch64
# kernelopts=root=UUID=858796f8-e37a-440a-ad32-43682e5a9cb9 ro  rhgb
# boot_success=0
# #######################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################################
# ''')

    input(str(_('Install finished, user:[acosail] password is [ ].\nPress Enter to reboot.')))

    reboot()

    # archinstall.log(f"Updating password.")
    # if (root_pw := archinstall.arguments.get('!root-password', None)) and len(root_pw):
    # 	user_set_pw(mountpoint, 'root', root_pw)


parted_fix()
if not archinstall.arguments.get('silent'):
    ask_user_questions()

config_output = ConfigurationOutput(archinstall.arguments)
# print("config_output===========>>>",archinstall.arguments)
# time.sleep(15)
if not archinstall.arguments.get('silent'):
    config_output.show()
config_output.save()

if archinstall.arguments.get('dry_run'):
    exit(0)

if not archinstall.arguments.get('silent'):
    input(str(_('Press Enter to continue.')))

archinstall.configuration_sanity_check()
perform_filesystem_operations()
perform_installation2('/target')
# perform_installation_diskimg()