qubes-installer-qubes-os/anaconda/pyanaconda/ui/tui/spokes/storage.py

# Text storage configuration spoke classes
#
# Copyright (C) 2012  Red Hat, Inc.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# the GNU General Public License v.2, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY expressed or implied, including the implied warranties of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
# Public License for more details.  You should have received a copy of the
# GNU General Public License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
# source code or documentation are not subject to the GNU General Public
# License and may only be used or replicated with the express permission of
# Red Hat, Inc.
#
# Red Hat Author(s): Jesse Keating <jkeating@redhat.com>
#
# Some of the code here is copied from pyanaconda/ui/gui/spokes/storage.py
# which has the same license and authored by David Lehman <dlehman@redhat.com>
#

from pyanaconda.ui.lib.disks import getDisks, size_str
from pyanaconda.ui.tui.spokes import NormalTUISpoke
from pyanaconda.ui.tui.simpleline import TextWidget, CheckboxWidget

from pykickstart.constants import AUTOPART_TYPE_LVM, AUTOPART_TYPE_BTRFS, AUTOPART_TYPE_PLAIN
from blivet.size import Size
from blivet.errors import StorageError
from pyanaconda.flags import flags
from pyanaconda.kickstart import doKickstartStorage
from pyanaconda.threads import threadMgr, AnacondaThread
from pyanaconda.constants import THREAD_STORAGE, THREAD_STORAGE_WATCHER
from pyanaconda.constants_text import INPUT_PROCESSED
from pyanaconda.i18n import _, P_
from pyanaconda.bootloader import BootLoaderError

from pykickstart.constants import CLEARPART_TYPE_ALL, CLEARPART_TYPE_LINUX, CLEARPART_TYPE_NONE
from pykickstart.errors import KickstartValueError

from collections import OrderedDict

import logging
log = logging.getLogger("anaconda")

__all__ = ["StorageSpoke", "AutoPartSpoke"]

CLEARALL = _("Use All Space")
CLEARLINUX = _("Replace Existing Linux system(s)")
CLEARNONE = _("Use Free Space")

PARTTYPES = {CLEARALL: CLEARPART_TYPE_ALL, CLEARLINUX: CLEARPART_TYPE_LINUX,
             CLEARNONE: CLEARPART_TYPE_NONE}

class StorageSpoke(NormalTUISpoke):
    """
    Storage spoke where users proceed to customize storage features such
    as disk selection, partitioning, and fs type.
    """
    title = _("Installation Destination")
    category = "system"

    def __init__(self, app, data, storage, payload, instclass):
        NormalTUISpoke.__init__(self, app, data, storage, payload, instclass)

        self._ready = False
        self.selected_disks = self.data.ignoredisk.onlyuse[:]

        self.autopart = None
        self.clearPartType = None

        # This list gets set up once in initialize and should not be modified
        # except perhaps to add advanced devices. It will remain the full list
        # of disks that can be included in the install.
        self.disks = []
        self.errors = []
        self.warnings = []

        if not flags.automatedInstall:
            # default to using autopart for interactive installs
            self.data.autopart.autopart = True

    @property
    def completed(self):
        retval = bool(self.storage.rootDevice and not self.errors)

        return retval

    @property
    def ready(self):
        # By default, the storage spoke is not ready.  We have to wait until
        # storageInitialize is done.
        return self._ready and not threadMgr.get(THREAD_STORAGE_WATCHER)

    @property
    def mandatory(self):
        return True

    @property
    def showable(self):
        return not flags.dirInstall

    @property
    def status(self):
        """ A short string describing the current status of storage setup. """
        msg = _("No disks selected")

        if flags.automatedInstall and not self.storage.rootDevice:
            return msg
        elif self.data.ignoredisk.onlyuse:
            msg = P_(("%d disk selected"),
                     ("%d disks selected"),
                     len(self.data.ignoredisk.onlyuse)) % len(self.data.ignoredisk.onlyuse)

            if self.errors:
                msg = _("Error checking storage configuration")
            elif self.warnings:
                msg = _("Warning checking storage configuration")
            # Maybe show what type of clearpart and which disks selected?
            elif self.data.autopart.autopart:
                msg = _("Automatic partitioning selected")
            else:
                msg = _("Custom partitioning selected")

        return msg

    def _update_disk_list(self, disk):
        """ Update self.selected_disks based on the selection."""

        name = disk.name

        # if the disk isn't already selected, select it.
        if name not in self.selected_disks:
            self.selected_disks.append(name)
        # If the disk is already selected, deselect it.
        elif name in self.selected_disks:
            self.selected_disks.remove(name)

    def _update_summary(self):
        """ Update the summary based on the UI. """
        count = 0
        capacity = 0
        free = Size(bytes=0)

        # pass in our disk list so hidden disks' free space is available
        free_space = self.storage.getFreeSpace(disks=self.disks)
        selected = [d for d in self.disks if d.name in self.selected_disks]

        for disk in selected:
            capacity += disk.size
            free += free_space[disk.name][0]
            count += 1

        summary = (P_(("%d disk selected; %s capacity; %s free ..."),
                      ("%d disks selected; %s capacity; %s free ..."),
                      count) % (count, str(Size(en_spec="%f MB" % capacity)), free))

        if len(self.disks) == 0:
            summary = _("No disks detected.  Please shut down the computer, connect at least one disk, and restart to complete installation.")
        elif count == 0:
            summary = (_("No disks selected; please select at least one disk to install to."))

        # Append storage errors to the summary
        if self.errors:
            summary = summary + "\n" + "\n".join(self.errors)
        elif self.warnings:
            summary = summary + "\n" + "\n".join(self.warnings)

        return summary

    def refresh(self, args = None):
        NormalTUISpoke.refresh(self, args)

        # Join the initialization thread to block on it
        # This print is foul.  Need a better message display
        print(_("Probing storage..."))
        threadMgr.wait(THREAD_STORAGE_WATCHER)

        # synchronize our local data store with the global ksdata
        # Commment out because there is no way to select a disk right
        # now without putting it in ksdata.  Seems wrong?
        #self.selected_disks = self.data.ignoredisk.onlyuse[:]
        self.autopart = self.data.autopart.autopart

        message = self._update_summary()

        # loop through the disks and present them.
        for disk in self.disks:
            size = size_str(disk.size)
            c = CheckboxWidget(title="%i) %s: %s (%s)" % (self.disks.index(disk) + 1,
                                                 disk.model, size, disk.name),
                               completed=(disk.name in self.selected_disks))
            self._window += [c, ""]

        self._window += [TextWidget(message), ""]

        return True

    def input(self, args, key):
        """Grab the disk choice and update things"""

        try:
            keyid = int(key) - 1
            self._update_disk_list(self.disks[keyid])
            return INPUT_PROCESSED
        except (ValueError, IndexError):
            if key.lower() == "c":
                if self.selected_disks:
                    newspoke = AutoPartSpoke(self.app, self.data, self.storage,
                                             self.payload, self.instclass)
                    self.app.switch_screen_modal(newspoke)
                    self.apply()
                    self.execute()
                    self.close()
                return INPUT_PROCESSED
            else:
                return key

    def apply(self):
        self.autopart = self.data.autopart.autopart
        self.data.ignoredisk.onlyuse = self.selected_disks[:]
        self.data.clearpart.drives = self.selected_disks[:]

        if self.data.autopart.type is None:
            self.data.autopart.type = AUTOPART_TYPE_LVM

        if self.autopart:
            self.clearPartType = CLEARPART_TYPE_ALL
        else:
            self.clearPartType = CLEARPART_TYPE_NONE

        for disk in self.disks:
            if disk.name not in self.selected_disks and \
               disk in self.storage.devices:
                self.storage.devicetree.hide(disk)
            elif disk.name in self.selected_disks and \
                 disk not in self.storage.devices:
                self.storage.devicetree.unhide(disk)

        self.data.bootloader.location = "mbr"

        if self.data.bootloader.bootDrive and \
           self.data.bootloader.bootDrive not in self.selected_disks:
            self.data.bootloader.bootDrive = ""
            self.storage.bootloader.reset()

        self.storage.config.update(self.data)

        # If autopart is selected we want to remove whatever has been
        # created/scheduled to make room for autopart.
        # If custom is selected, we want to leave alone any storage layout the
        # user may have set up before now.
        self.storage.config.clearNonExistent = self.data.autopart.autopart

    def execute(self):
        print(_("Generating updated storage configuration"))
        try:
            doKickstartStorage(self.storage, self.data, self.instclass)
        except (StorageError, KickstartValueError) as e:
            log.error("storage configuration failed: %s", e)
            print _("storage configuration failed: %s") % e
            self.errors = [str(e)]
            self.data.bootloader.bootDrive = ""
            self.data.clearpart.type = CLEARPART_TYPE_ALL
            self.data.clearpart.initAll = False
            self.storage.config.update(self.data)
            self.storage.autoPartType = self.data.clearpart.type
            self.storage.reset()
            self._ready = True
        except BootLoaderError as e:
            log.error("BootLoader setup failed: %s", e)
            print _("storage configuration failed: %s") % e
            self.errors = [str(e)]
            self.data.bootloader.bootDrive = ""
            self._ready = True
        else:
            print(_("Checking storage configuration..."))
            (self.errors, self.warnings) = self.storage.sanityCheck()
            self._ready = True
            for e in self.errors:
                log.error(e)
                print e
            for w in self.warnings:
                log.warn(w)
                print w

    def initialize(self):
        NormalTUISpoke.initialize(self)

        threadMgr.add(AnacondaThread(name=THREAD_STORAGE_WATCHER,
                                     target=self._initialize))

        self.selected_disks = self.data.ignoredisk.onlyuse[:]
        # Probably need something here to track which disks are selected?

    def _initialize(self):
        """
        Secondary initialize so wait for the storage thread to complete before
        populating our disk list
        """

        threadMgr.wait(THREAD_STORAGE)

        self.disks = sorted(getDisks(self.storage.devicetree),
                            key=lambda d: d.name)
        # if only one disk is available, go ahead and mark it as selected
        if len(self.disks) == 1:
            self._update_disk_list(self.disks[0])

        self._update_summary()
        self._ready = True

class AutoPartSpoke(NormalTUISpoke):
    """ Autopartitioning options are presented here. """
    title = _("Autopartitioning Options")
    category = "system"

    def __init__(self, app, data, storage, payload, instclass):
        NormalTUISpoke.__init__(self, app, data, storage, payload, instclass)
        self.clearPartType = self.data.clearpart.type
        self.parttypelist = sorted(PARTTYPES.keys())

    @property
    def indirect(self):
        return True

    def refresh(self, args = None):
        NormalTUISpoke.refresh(self, args)
        # synchronize our local data store with the global ksdata
        self.clearPartType = self.data.clearpart.type
        # I dislike "is None", but bool(0) returns false :(
        if self.clearPartType is None:
            # Default to clearing everything.
            self.clearPartType = CLEARPART_TYPE_ALL

        for parttype in self.parttypelist:
            c = CheckboxWidget(title="%i) %s" % (self.parttypelist.index(parttype) + 1,
                                                 parttype),
                               completed=(PARTTYPES[parttype] == self.clearPartType))
            self._window += [c, ""]

        message = _("Installation requires partitioning of your hard drive. Select what space to use for the install target.")

        self._window += [TextWidget(message), ""]

        return True

    def apply(self):
        # kind of a hack, but if we're actually getting to this spoke, there
        # is no doubt that we are doing autopartitioning, so set autopart to
        # True. In the case of ks installs which may not have defined any
        # partition options, autopart was never set to True, causing some
        # issues. (rhbz#1001061)
        self.data.autopart.autopart = True
        self.data.clearpart.type = self.clearPartType
        self.data.clearpart.initAll = True

    def input(self, args, key):
        """Grab the choice and update things"""

        try:
            keyid = int(key) - 1
        except ValueError:
            if key.lower() == "c":
                newspoke = PartitionSchemeSpoke(self.app, self.data, self.storage,
                                                self.payload, self.instclass)
                self.app.switch_screen_modal(newspoke)
                self.apply()
                self.close()
                return INPUT_PROCESSED
            else:
                return key

        if 0 <= keyid < len(self.parttypelist):
            self.clearPartType = PARTTYPES[self.parttypelist[keyid]]
            self.apply()
        return INPUT_PROCESSED

class PartitionSchemeSpoke(NormalTUISpoke):
    """ Spoke to select what partitioning scheme to use on disk(s). """
    title = _("Partition Scheme Options")
    category = "system"

    # set default FS to LVM, for consistency with graphical behavior
    _selection = 1

    def __init__(self, app, data, storage, payload, instclass):
        NormalTUISpoke.__init__(self, app, data, storage, payload, instclass)
        self.partschemes = OrderedDict([("Standard Partition", AUTOPART_TYPE_PLAIN),
                        ("LVM", AUTOPART_TYPE_LVM), ("BTRFS", AUTOPART_TYPE_BTRFS)])

    @property
    def indirect(self):
        return True

    def refresh(self, args=None):
        NormalTUISpoke.refresh(self, args)

        schemelist = self.partschemes.keys()
        for sch in schemelist:
            box = CheckboxWidget(title="%i) %s" %(schemelist.index(sch) \
                                 + 1, sch), completed=(schemelist.index(sch) \
                                 == self._selection))
            self._window += [box, ""]

        message = _("Select a partition scheme configuration.")
        self._window += [TextWidget(message), ""]
        return True

    def input(self, args, key):
        """ Grab the choice and update things. """

        try:
            keyid = int(key) - 1
        except ValueError:
            if key.lower() == "c":
                self.apply()
                self.close()
                return INPUT_PROCESSED
            else:
                return key

        if 0 <= keyid < len(self.partschemes):
            self._selection = keyid
        return INPUT_PROCESSED

    def apply(self):
        """ Apply our selections. """

        schemelist = self.partschemes.values()
        try:
            self.data.autopart.type = schemelist[self._selection]
        except IndexError:
            # we shouldn't ever see this, but just in case, don't crash.
            # when autopart.type is detected as None in AutoPartSpoke.apply(),
            # it'll automatically just be set to LVM
            pass