From 486ad7dd6d195dd435c5da58d241e14742f60485 Mon Sep 17 00:00:00 2001 From: Anton Hvornum Date: Sat, 28 May 2022 15:40:36 +0200 Subject: Removes btrfs subvolume warnings on incorrect subvolume locations (#1267) * Adding debug information * Adding debug information * Adding debug information * Removed a 'already-a-subvolume' check as it requires more information. * Adding debug information * Adding debug information * Made sure Partition().subvolumes() only attempts to retrieve btrfs subvolume information if fstype==btrfs. * Removed debug information --- archinstall/lib/disk/partition.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 73c88597..151775b1 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -310,8 +310,9 @@ class Partition: def iterate_children_recursively(information): for child in information.get('children', []): if target := child.get('target'): - if subvolume := subvolume_info_from_path(pathlib.Path(target)): - yield subvolume + if child.get('fstype') == 'btrfs': + if subvolume := subvolume_info_from_path(pathlib.Path(target)): + yield subvolume if child.get('children'): for subchild in iterate_children_recursively(child): @@ -320,8 +321,9 @@ class Partition: for mountpoint in self.mount_information: if result := findmnt(pathlib.Path(mountpoint['target'])): for filesystem in result.get('filesystems', []): - if subvolume := subvolume_info_from_path(pathlib.Path(mountpoint['target'])): - yield subvolume + if mountpoint.get('fstype') == 'btrfs': + if subvolume := subvolume_info_from_path(pathlib.Path(mountpoint['target'])): + yield subvolume for child in iterate_children_recursively(filesystem): yield child -- cgit v1.2.3-70-g09d2 From 7dbea73514b35cbaa18c156895bf6f416b2345ca Mon Sep 17 00:00:00 2001 From: Anton Hvornum Date: Sun, 29 May 2022 11:25:28 +0200 Subject: Cleanup and version changes in prep for release --- PKGBUILD | 2 +- archinstall/lib/disk/partition.py | 10 ---------- 2 files changed, 1 insertion(+), 11 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/PKGBUILD b/PKGBUILD index 5821bee8..d8e89ae2 100644 --- a/PKGBUILD +++ b/PKGBUILD @@ -4,7 +4,7 @@ # Contributor: demostanis worlds pkgname=archinstall -pkgver=2.4.3rc1 +pkgver=2.5.0 #pkgver=$(git describe --long | sed 's/\([^-]*-g\)/r\1/;s/-/./g') pkgrel=1 pkgdesc="Just another guided/automated Arch Linux installer with a twist" diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 151775b1..2c9f50c2 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -160,16 +160,6 @@ class Partition: def boot(self) -> bool: output = json.loads(SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')) - # Get the bootable flag from the sfdisk output: - # { - # "partitiontable": { - # "device":"/dev/loop0", - # "partitions": [ - # {"node":"/dev/loop0p1", "start":2048, "size":10483712, "type":"83", "bootable":true} - # ] - # } - # } - for partition in output.get('partitiontable', {}).get('partitions', []): if partition['node'] == self.path: # first condition is for MBR disks, second for GPT disks -- cgit v1.2.3-70-g09d2 From a7ca037a26de53fd242f89bc6a90fd53337b4d13 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Tue, 7 Jun 2022 01:28:46 +1000 Subject: Update the subvolume menu - fix for #1278 (#1297) * Update subvolume * Add mypy compliance Co-authored-by: Daniel Girtler Co-authored-by: Anton Hvornum --- .github/workflows/mypy.yaml | 2 +- archinstall/__init__.py | 6 +- archinstall/lib/disk/btrfs/__init__.py | 130 +------------ archinstall/lib/disk/btrfs/btrfs_helpers.py | 94 ++++------ archinstall/lib/disk/btrfs/btrfspartition.py | 6 +- archinstall/lib/disk/btrfs/btrfssubvolume.py | 191 -------------------- archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py | 192 ++++++++++++++++++++ archinstall/lib/disk/helpers.py | 9 +- archinstall/lib/disk/mapperdev.py | 10 +- archinstall/lib/disk/partition.py | 10 +- archinstall/lib/disk/user_guides.py | 19 +- archinstall/lib/installer.py | 55 ++---- archinstall/lib/menu/global_menu.py | 33 ++-- archinstall/lib/menu/list_manager.py | 39 ++-- archinstall/lib/models/subvolume.py | 68 +++++++ archinstall/lib/models/users.py | 6 +- .../lib/user_interaction/partitioning_conf.py | 12 +- .../lib/user_interaction/subvolume_config.py | 201 +++++++-------------- 18 files changed, 462 insertions(+), 621 deletions(-) delete mode 100644 archinstall/lib/disk/btrfs/btrfssubvolume.py create mode 100644 archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py create mode 100644 archinstall/lib/models/subvolume.py (limited to 'archinstall/lib/disk/partition.py') diff --git a/.github/workflows/mypy.yaml b/.github/workflows/mypy.yaml index 8463afda..b0901b38 100644 --- a/.github/workflows/mypy.yaml +++ b/.github/workflows/mypy.yaml @@ -15,4 +15,4 @@ jobs: # one day this will be enabled # run: mypy --strict --module archinstall || exit 0 - name: run mypy - run: mypy --follow-imports=silent archinstall/lib/menu/selection_menu.py archinstall/lib/menu/global_menu.py archinstall/lib/models/network_configuration.py archinstall/lib/menu/list_manager.py archinstall/lib/user_interaction/network_conf.py archinstall/lib/models/users.py archinstall/lib/translation.py + run: mypy --follow-imports=silent archinstall/lib/menu/selection_menu.py archinstall/lib/menu/global_menu.py archinstall/lib/models/network_configuration.py archinstall/lib/menu/list_manager.py archinstall/lib/user_interaction/network_conf.py archinstall/lib/models/users.py archinstall/lib/user_interaction/subvolume_config.py archinstall/lib/disk/btrfs/btrfs_helpers.py archinstall/lib/translation.py diff --git a/archinstall/__init__.py b/archinstall/__init__.py index abcad3ba..1a360c67 100644 --- a/archinstall/__init__.py +++ b/archinstall/__init__.py @@ -226,8 +226,6 @@ def post_process_arguments(arguments): load_plugin(arguments['plugin']) if arguments.get('disk_layouts', None) is not None: - # if 'disk_layouts' not in storage: - # storage['disk_layouts'] = {} layout_storage = {} if not json_stream_to_structure('--disk_layouts',arguments['disk_layouts'],layout_storage): exit(1) @@ -236,10 +234,12 @@ def post_process_arguments(arguments): 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',[])): + 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 load_config() diff --git a/archinstall/lib/disk/btrfs/__init__.py b/archinstall/lib/disk/btrfs/__init__.py index 90c58145..3c183112 100644 --- a/archinstall/lib/disk/btrfs/__init__.py +++ b/archinstall/lib/disk/btrfs/__init__.py @@ -2,8 +2,7 @@ from __future__ import annotations import pathlib import glob import logging -import re -from typing import Union, Dict, TYPE_CHECKING, Any, Iterator +from typing import Union, Dict, TYPE_CHECKING # https://stackoverflow.com/a/39757388/929999 if TYPE_CHECKING: @@ -15,30 +14,15 @@ from .btrfs_helpers import ( setup_subvolumes as setup_subvolumes, mount_subvolume as mount_subvolume ) -from .btrfssubvolume import BtrfsSubvolume as BtrfsSubvolume +from .btrfssubvolumeinfo import BtrfsSubvolumeInfo as BtrfsSubvolume from .btrfspartition import BTRFSPartition as BTRFSPartition -from ..helpers import get_mount_info from ...exceptions import DiskError, Deprecated from ...general import SysCommand from ...output import log -from ...exceptions import SysCallError -def get_subvolume_info(path :pathlib.Path) -> Dict[str, Any]: - try: - output = SysCommand(f"btrfs subvol show {path}").decode() - except SysCallError as error: - print('Error:', error) - result = {} - for line in output.replace('\r\n', '\n').split('\n'): - if ':' in line: - key, val = line.replace('\t', '').split(':', 1) - result[key.strip().lower().replace(' ', '_')] = val.strip() - - return result - -def create_subvolume(installation :Installer, subvolume_location :Union[pathlib.Path, str]) -> bool: +def create_subvolume(installation: Installer, subvolume_location :Union[pathlib.Path, str]) -> bool: """ This function uses btrfs to create a subvolume. @@ -71,112 +55,6 @@ def create_subvolume(installation :Installer, subvolume_location :Union[pathlib. if (cmd := SysCommand(f"btrfs subvolume create {target}")).exit_code != 0: raise DiskError(f"Could not create a subvolume at {target}: {cmd}") -def _has_option(option :str,options :list) -> bool: - """ auxiliary routine to check if an option is present in a list. - we check if the string appears in one of the options, 'cause it can appear in several forms (option, option=val,...) - """ - if not options: - return False - - for item in options: - if option in item: - return True - - return False - -def manage_btrfs_subvolumes(installation :Installer, - partition :Dict[str, str],) -> list: +def manage_btrfs_subvolumes(installation :Installer, partition :Dict[str, str]) -> list: raise Deprecated("Use setup_subvolumes() instead.") - - from copy import deepcopy - """ we do the magic with subvolumes in a centralized place - parameters: - * the installation object - * the partition dictionary entry which represents the physical partition - returns - * mountpoinst, the list which contains all the "new" partititon to be mounted - - We expect the partition has been mounted as / , and it to be unmounted after the processing - Then we create all the subvolumes inside btrfs as demand - We clone then, both the partition dictionary and the object inside it and adapt it to the subvolume needs - Then we return a list of "new" partitions to be processed as "normal" partitions - # TODO For encrypted devices we need some special processing prior to it - """ - # We process each of the pairs - # th mount info dict has an entry for the path of the mountpoint (named 'mountpoint') and 'options' which is a list - # of mount options (or similar used by brtfs) - mountpoints = [] - subvolumes = partition['btrfs']['subvolumes'] - for name, right_hand in subvolumes.items(): - try: - # we normalize the subvolume name (getting rid of slash at the start if exists. In our implementation has no semantic load - every subvolume is created from the top of the hierarchy- and simplifies its further use - if name.startswith('/'): - name = name[1:] - # renormalize the right hand. - location = None - subvol_options = [] - # no contents, so it is not to be mounted - if not right_hand: - location = None - # just a string. per backward compatibility the mount point - elif isinstance(right_hand,str): - location = right_hand - # a dict. two elements 'mountpoint' (obvious) and and a mount options list ¿? - elif isinstance(right_hand,dict): - location = right_hand.get('mountpoint',None) - subvol_options = right_hand.get('options',[]) - # we create the subvolume - create_subvolume(installation,name) - # Make the nodatacow processing now - # It will be the main cause of creation of subvolumes which are not to be mounted - # it is not an options which can be established by subvolume (but for whole file systems), and can be - # set up via a simple attribute change in a directory (if empty). And here the directories are brand new - if 'nodatacow' in subvol_options: - if (cmd := SysCommand(f"chattr +C {installation.target}/{name}")).exit_code != 0: - raise DiskError(f"Could not set nodatacow attribute at {installation.target}/{name}: {cmd}") - # entry is deleted so nodatacow doesn't propagate to the mount options - del subvol_options[subvol_options.index('nodatacow')] - # Make the compress processing now - # it is not an options which can be established by subvolume (but for whole file systems), and can be - # set up via a simple attribute change in a directory (if empty). And here the directories are brand new - # in this way only zstd compression is activaded - # TODO WARNING it is not clear if it should be a standard feature, so it might need to be deactivated - if 'compress' in subvol_options: - if not _has_option('compress',partition.get('filesystem',{}).get('mount_options',[])): - if (cmd := SysCommand(f"chattr +c {installation.target}/{name}")).exit_code != 0: - raise DiskError(f"Could not set compress attribute at {installation.target}/{name}: {cmd}") - # entry is deleted so compress doesn't propagate to the mount options - del subvol_options[subvol_options.index('compress')] - # END compress processing. - # we do not mount if THE basic partition will be mounted or if we exclude explicitly this subvolume - if not partition['mountpoint'] and location is not None: - # we begin to create a fake partition entry. First we copy the original -the one that corresponds to - # the primary partition. We make a deepcopy to avoid altering the original content in any case - fake_partition = deepcopy(partition) - # we start to modify entries in the "fake partition" to match the needs of the subvolumes - # to avoid any chance of entering in a loop (not expected) we delete the list of subvolumes in the copy - del fake_partition['btrfs'] - fake_partition['encrypted'] = False - fake_partition['generate-encryption-key-file'] = False - # Mount destination. As of now the right hand part - fake_partition['mountpoint'] = location - # we load the name in an attribute called subvolume, but i think it is not needed anymore, 'cause the mount logic uses a different path. - fake_partition['subvolume'] = name - # here we add the special mount options for the subvolume, if any. - # if the original partition['options'] is not a list might give trouble - if fake_partition.get('filesystem',{}).get('mount_options',[]): - fake_partition['filesystem']['mount_options'].extend(subvol_options) - else: - fake_partition['filesystem']['mount_options'] = subvol_options - # Here comes the most exotic part. The dictionary attribute 'device_instance' contains an instance of Partition. This instance will be queried along the mount process at the installer. - # As the rest will query there the path of the "partition" to be mounted, we feed it with the bind name needed to mount subvolumes - # As we made a deepcopy we have a fresh instance of this object we can manipulate problemless - fake_partition['device_instance'].path = f"{partition['device_instance'].path}[/{name}]" - - # Well, now that this "fake partition" is ready, we add it to the list of the ones which are to be mounted, - # as "normal" ones - mountpoints.append(fake_partition) - except Exception as e: - raise e - return mountpoints diff --git a/archinstall/lib/disk/btrfs/btrfs_helpers.py b/archinstall/lib/disk/btrfs/btrfs_helpers.py index 5fa94314..ab528388 100644 --- a/archinstall/lib/disk/btrfs/btrfs_helpers.py +++ b/archinstall/lib/disk/btrfs/btrfs_helpers.py @@ -1,72 +1,42 @@ -import pathlib import logging -from typing import Optional +from pathlib import Path +from typing import Optional, Dict, Any, TYPE_CHECKING +from ...models.subvolume import Subvolume from ...exceptions import SysCallError, DiskError from ...general import SysCommand from ...output import log from ..helpers import get_mount_info -from .btrfssubvolume import BtrfsSubvolume +from .btrfssubvolumeinfo import BtrfsSubvolumeInfo +if TYPE_CHECKING: + from .btrfspartition import BTRFSPartition + from ...installer import Installer -def mount_subvolume(installation, device, name, subvolume_information): - # we normalize the subvolume name (getting rid of slash at the start if exists. In our implementation has no semantic load. - # Every subvolume is created from the top of the hierarchy- and simplifies its further use - name = name.lstrip('/') - - # renormalize the right hand. - mountpoint = subvolume_information.get('mountpoint', None) - if not mountpoint: - return None - - if type(mountpoint) == str: - mountpoint = pathlib.Path(mountpoint) - installation_target = installation.target - if type(installation_target) == str: - installation_target = pathlib.Path(installation_target) +def mount_subvolume(installation: 'Installer', device: 'BTRFSPartition', subvolume: Subvolume): + # we normalize the subvolume name (getting rid of slash at the start if exists. + # In our implementation has no semantic load. + # Every subvolume is created from the top of the hierarchy- and simplifies its further use + name = subvolume.name.lstrip('/') + mountpoint = Path(subvolume.mountpoint) + installation_target = Path(installation.target) mountpoint = installation_target / mountpoint.relative_to(mountpoint.anchor) mountpoint.mkdir(parents=True, exist_ok=True) - - mount_options = subvolume_information.get('options', []) - if not any('subvol=' in x for x in mount_options): - mount_options += [f'subvol={name}'] + mount_options = subvolume.options + [f'subvol={name}'] log(f"Mounting subvolume {name} on {device} to {mountpoint}", level=logging.INFO, fg="gray") SysCommand(f"mount {device.path} {mountpoint} -o {','.join(mount_options)}") -def setup_subvolumes(installation, partition_dict): - """ - Taken from: ..user_guides.py - - partition['btrfs'] = { - "subvolumes" : { - "@": "/", - "@home": "/home", - "@log": "/var/log", - "@pkg": "/var/cache/pacman/pkg", - "@.snapshots": "/.snapshots" - } - } - """ +def setup_subvolumes(installation: 'Installer', partition_dict: Dict[str, Any]): log(f"Setting up subvolumes: {partition_dict['btrfs']['subvolumes']}", level=logging.INFO, fg="gray") - for name, right_hand in partition_dict['btrfs']['subvolumes'].items(): + + for subvolume in partition_dict['btrfs']['subvolumes']: # we normalize the subvolume name (getting rid of slash at the start if exists. In our implementation has no semantic load. # Every subvolume is created from the top of the hierarchy- and simplifies its further use - name = name.lstrip('/') - - # renormalize the right hand. - # mountpoint = None - subvol_options = [] - - match right_hand: - # case str(): # backwards-compatability - # mountpoint = right_hand - case dict(): - # mountpoint = right_hand.get('mountpoint', None) - subvol_options = right_hand.get('options', []) + name = subvolume.name.lstrip('/') # We create the subvolume using the BTRFSPartition instance. # That way we ensure not only easy access, but also accurate mount locations etc. @@ -76,27 +46,25 @@ def setup_subvolumes(installation, partition_dict): # It will be the main cause of creation of subvolumes which are not to be mounted # it is not an options which can be established by subvolume (but for whole file systems), and can be # set up via a simple attribute change in a directory (if empty). And here the directories are brand new - if 'nodatacow' in subvol_options: + if subvolume.nodatacow: if (cmd := SysCommand(f"chattr +C {installation.target}/{name}")).exit_code != 0: raise DiskError(f"Could not set nodatacow attribute at {installation.target}/{name}: {cmd}") - # entry is deleted so nodatacow doesn't propagate to the mount options - del subvol_options[subvol_options.index('nodatacow')] + # Make the compress processing now # it is not an options which can be established by subvolume (but for whole file systems), and can be # set up via a simple attribute change in a directory (if empty). And here the directories are brand new # in this way only zstd compression is activaded # TODO WARNING it is not clear if it should be a standard feature, so it might need to be deactivated - if 'compress' in subvol_options: + if subvolume.compress: if not any(['compress' in filesystem_option for filesystem_option in partition_dict.get('filesystem', {}).get('mount_options', [])]): if (cmd := SysCommand(f"chattr +c {installation.target}/{name}")).exit_code != 0: raise DiskError(f"Could not set compress attribute at {installation.target}/{name}: {cmd}") - # entry is deleted so compress doesn't propagate to the mount options - del subvol_options[subvol_options.index('compress')] -def subvolume_info_from_path(path :pathlib.Path) -> Optional[BtrfsSubvolume]: + +def subvolume_info_from_path(path: Path) -> Optional[BtrfsSubvolumeInfo]: try: - subvolume_name = None + subvolume_name = '' result = {} for index, line in enumerate(SysCommand(f"btrfs subvolume show {path}")): if index == 0: @@ -110,14 +78,14 @@ def subvolume_info_from_path(path :pathlib.Path) -> Optional[BtrfsSubvolume]: # allows for hooking in a pre-processor to do this we have to do it here: result[key.lower().replace(' ', '_').replace('(s)', 's')] = value.strip() - return BtrfsSubvolume(**{'full_path' : path, 'name' : subvolume_name, **result}) - + return BtrfsSubvolumeInfo(**{'full_path' : path, 'name' : subvolume_name, **result}) # type: ignore except SysCallError as error: log(f"Could not retrieve subvolume information from {path}: {error}", level=logging.WARNING, fg="orange") return None -def find_parent_subvolume(path :pathlib.Path, filters=[]): + +def find_parent_subvolume(path: Path, filters=[]) -> Optional[BtrfsSubvolumeInfo]: # A root path cannot have a parent if str(path) == '/': return None @@ -127,6 +95,8 @@ def find_parent_subvolume(path :pathlib.Path, filters=[]): if found_mount['target'] == '/': return None - return find_parent_subvolume(path.parent, traverse=True, filters=[*filters, found_mount['target']]) + return find_parent_subvolume(path.parent, filters=[*filters, found_mount['target']]) + + return subvolume - return subvolume \ No newline at end of file + return None diff --git a/archinstall/lib/disk/btrfs/btrfspartition.py b/archinstall/lib/disk/btrfs/btrfspartition.py index 6f7487e4..a05f1527 100644 --- a/archinstall/lib/disk/btrfs/btrfspartition.py +++ b/archinstall/lib/disk/btrfs/btrfspartition.py @@ -15,7 +15,7 @@ from .btrfs_helpers import ( if TYPE_CHECKING: from ...installer import Installer - from .btrfssubvolume import BtrfsSubvolume + from .btrfssubvolumeinfo import BtrfsSubvolumeInfo class BTRFSPartition(Partition): def __init__(self, *args, **kwargs): @@ -50,7 +50,7 @@ class BTRFSPartition(Partition): for child in iterate_children(filesystem): yield child - def create_subvolume(self, subvolume :pathlib.Path, installation :Optional['Installer'] = None) -> 'BtrfsSubvolume': + def create_subvolume(self, subvolume :pathlib.Path, installation :Optional['Installer'] = None) -> 'BtrfsSubvolumeInfo': """ Subvolumes have to be created within a mountpoint. This means we need to get the current installation target. @@ -117,4 +117,4 @@ class BTRFSPartition(Partition): # And deal with it here: SysCommand(f"btrfs subvolume create {subvolume}") - return subvolume_info_from_path(subvolume) \ No newline at end of file + return subvolume_info_from_path(subvolume) diff --git a/archinstall/lib/disk/btrfs/btrfssubvolume.py b/archinstall/lib/disk/btrfs/btrfssubvolume.py deleted file mode 100644 index bc7db612..00000000 --- a/archinstall/lib/disk/btrfs/btrfssubvolume.py +++ /dev/null @@ -1,191 +0,0 @@ -import pathlib -import datetime -import logging -import string -import random -import shutil -from dataclasses import dataclass -from typing import Optional, List# , TYPE_CHECKING -from functools import cached_property - -# if TYPE_CHECKING: -# from ..blockdevice import BlockDevice - -from ...exceptions import DiskError -from ...general import SysCommand -from ...output import log -from ...storage import storage - -@dataclass -class BtrfsSubvolume: - full_path :pathlib.Path - name :str - uuid :str - parent_uuid :str - creation_time :datetime.datetime - subvolume_id :int - generation :int - gen_at_creation :int - parent_id :int - top_level_id :int - send_transid :int - send_time :datetime.datetime - receive_transid :int - received_uuid :Optional[str] = None - flags :Optional[str] = None - receive_time :Optional[datetime.datetime] = None - snapshots :Optional[List] = None - - def __post_init__(self): - self.full_path = pathlib.Path(self.full_path) - - # Convert "-" entries to `None` - if self.parent_uuid == "-": - self.parent_uuid = None - if self.received_uuid == "-": - self.received_uuid = None - if self.flags == "-": - self.flags = None - if self.receive_time == "-": - self.receive_time = None - if self.snapshots == "": - self.snapshots = [] - - # Convert timestamps into datetime workable objects (and preserve timezone by using ISO formats) - self.creation_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.creation_time)) - self.send_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.send_time)) - if self.receive_time: - self.receive_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.receive_time)) - - @property - def parent_subvolume(self): - from .btrfs_helpers import find_parent_subvolume - - return find_parent_subvolume(self.full_path) - - @property - def root(self) -> bool: - from .btrfs_helpers import subvolume_info_from_path - - # TODO: Make this function traverse storage['MOUNT_POINT'] and find the first - # occurrence of a mountpoint that is a btrfs volume instead of lazy assume / is a subvolume. - # It would also be nice if it could use findmnt(self.full_path) and traverse backwards - # finding the last occurrence of a subvolume which 'self' belongs to. - if volume := subvolume_info_from_path(storage['MOUNT_POINT']): - return self.full_path == volume.full_path - - return False - - @cached_property - def partition(self): - from ..helpers import findmnt, get_parent_of_partition, all_blockdevices - from ..partition import Partition - from ..blockdevice import BlockDevice - from ..mapperdev import MapperDev - from .btrfspartition import BTRFSPartition - from .btrfs_helpers import subvolume_info_from_path - - try: - # If the subvolume is mounted, it's pretty trivial to lookup the partition (parent) device. - if filesystem := findmnt(self.full_path).get('filesystems', []): - if source := filesystem[0].get('source', None): - # Strip away subvolume definitions from findmnt - if '[' in source: - source = source[:source.find('[')] - - if filesystem[0].get('fstype', '') == 'btrfs': - return BTRFSPartition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) - elif filesystem[0].get('source', '').startswith('/dev/mapper'): - return MapperDev(source) - else: - return Partition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) - except DiskError: - # Subvolume has never been mounted, we have no reliable way of finding where it is. - # But we have the UUID of the partition, and can begin looking for it by mounting - # all blockdevices that we can reliably support.. This is taxing tho and won't cover all devices. - - log(f"Looking up {self}, this might take time.", fg="orange", level=logging.WARNING) - for blockdevice, instance in all_blockdevices(mappers=True, partitions=True, error=True).items(): - if type(instance) in (Partition, MapperDev): - we_mounted_it = False - detection_mountpoint = instance.mountpoint - if not detection_mountpoint: - if type(instance) == Partition and instance.encrypted: - # TODO: Perhaps support unlocking encrypted volumes? - # This will cause a lot of potential user interactions tho. - log(f"Ignoring {blockdevice} because it's encrypted.", fg="gray", level=logging.DEBUG) - continue - - detection_mountpoint = pathlib.Path(f"/tmp/{''.join([random.choice(string.ascii_letters) for x in range(20)])}") - detection_mountpoint.mkdir(parents=True, exist_ok=True) - - instance.mount(str(detection_mountpoint)) - we_mounted_it = True - - if (filesystem := findmnt(detection_mountpoint)) and (filesystem := filesystem.get('filesystems', [])): - if subvolume := subvolume_info_from_path(filesystem[0]['target']): - if subvolume.uuid == self.uuid: - # The top level subvolume matched of ourselves, - # which means the instance we're iterating has the subvol we're looking for. - log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") - return instance - - def iterate_children(struct): - for child in struct.get('children', []): - if '[' in child.get('source', ''): - yield subvolume_info_from_path(child['target']) - - for sub_child in iterate_children(child): - yield sub_child - - for child in iterate_children(filesystem[0]): - if child.uuid == self.uuid: - # We found a child within the instance that has the subvol we're looking for. - log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") - return instance - - if we_mounted_it: - instance.unmount() - shutil.rmtree(detection_mountpoint) - - @cached_property - def mount_options(self) -> Optional[List[str]]: - from ..helpers import findmnt - - if filesystem := findmnt(self.full_path).get('filesystems', []): - return filesystem[0].get('options').split(',') - - def convert_to_ISO_format(self, time_string): - time_string_almost_done = time_string.replace(' ', 'T', 1).replace(' ', '') - iso_string = f"{time_string_almost_done[:-2]}:{time_string_almost_done[-2:]}" - return iso_string - - def mount(self, mountpoint :pathlib.Path, options=None, include_previously_known_options=True): - from ..helpers import findmnt - - try: - if mnt_info := findmnt(pathlib.Path(mountpoint), traverse=False): - log(f"Unmounting {mountpoint} as it was already mounted using {mnt_info}") - SysCommand(f"umount {mountpoint}") - except DiskError: - # No previously mounted device at the mountpoint - pass - - if not options: - options = [] - - try: - if include_previously_known_options and (cached_options := self.mount_options): - options += cached_options - except DiskError: - pass - - if not any('subvol=' in x for x in options): - options += f'subvol={self.name}' - - SysCommand(f"mount {self.partition.path} {mountpoint} -o {','.join(options)}") - log(f"{self} has successfully been mounted to {mountpoint}", level=logging.INFO, fg="gray") - - def unmount(self, recurse :bool = True): - SysCommand(f"umount {'-R' if recurse else ''} {self.full_path}") - log(f"Successfully unmounted {self}", level=logging.INFO, fg="gray") \ No newline at end of file diff --git a/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py b/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py new file mode 100644 index 00000000..5f5bdea6 --- /dev/null +++ b/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py @@ -0,0 +1,192 @@ +import pathlib +import datetime +import logging +import string +import random +import shutil +from dataclasses import dataclass +from typing import Optional, List# , TYPE_CHECKING +from functools import cached_property + +# if TYPE_CHECKING: +# from ..blockdevice import BlockDevice + +from ...exceptions import DiskError +from ...general import SysCommand +from ...output import log +from ...storage import storage + + +@dataclass +class BtrfsSubvolumeInfo: + full_path :pathlib.Path + name :str + uuid :str + parent_uuid :str + creation_time :datetime.datetime + subvolume_id :int + generation :int + gen_at_creation :int + parent_id :int + top_level_id :int + send_transid :int + send_time :datetime.datetime + receive_transid :int + received_uuid :Optional[str] = None + flags :Optional[str] = None + receive_time :Optional[datetime.datetime] = None + snapshots :Optional[List] = None + + def __post_init__(self): + self.full_path = pathlib.Path(self.full_path) + + # Convert "-" entries to `None` + if self.parent_uuid == "-": + self.parent_uuid = None + if self.received_uuid == "-": + self.received_uuid = None + if self.flags == "-": + self.flags = None + if self.receive_time == "-": + self.receive_time = None + if self.snapshots == "": + self.snapshots = [] + + # Convert timestamps into datetime workable objects (and preserve timezone by using ISO formats) + self.creation_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.creation_time)) + self.send_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.send_time)) + if self.receive_time: + self.receive_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.receive_time)) + + @property + def parent_subvolume(self): + from .btrfs_helpers import find_parent_subvolume + + return find_parent_subvolume(self.full_path) + + @property + def root(self) -> bool: + from .btrfs_helpers import subvolume_info_from_path + + # TODO: Make this function traverse storage['MOUNT_POINT'] and find the first + # occurrence of a mountpoint that is a btrfs volume instead of lazy assume / is a subvolume. + # It would also be nice if it could use findmnt(self.full_path) and traverse backwards + # finding the last occurrence of a subvolume which 'self' belongs to. + if volume := subvolume_info_from_path(storage['MOUNT_POINT']): + return self.full_path == volume.full_path + + return False + + @cached_property + def partition(self): + from ..helpers import findmnt, get_parent_of_partition, all_blockdevices + from ..partition import Partition + from ..blockdevice import BlockDevice + from ..mapperdev import MapperDev + from .btrfspartition import BTRFSPartition + from .btrfs_helpers import subvolume_info_from_path + + try: + # If the subvolume is mounted, it's pretty trivial to lookup the partition (parent) device. + if filesystem := findmnt(self.full_path).get('filesystems', []): + if source := filesystem[0].get('source', None): + # Strip away subvolume definitions from findmnt + if '[' in source: + source = source[:source.find('[')] + + if filesystem[0].get('fstype', '') == 'btrfs': + return BTRFSPartition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) + elif filesystem[0].get('source', '').startswith('/dev/mapper'): + return MapperDev(source) + else: + return Partition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) + except DiskError: + # Subvolume has never been mounted, we have no reliable way of finding where it is. + # But we have the UUID of the partition, and can begin looking for it by mounting + # all blockdevices that we can reliably support.. This is taxing tho and won't cover all devices. + + log(f"Looking up {self}, this might take time.", fg="orange", level=logging.WARNING) + for blockdevice, instance in all_blockdevices(mappers=True, partitions=True, error=True).items(): + if type(instance) in (Partition, MapperDev): + we_mounted_it = False + detection_mountpoint = instance.mountpoint + if not detection_mountpoint: + if type(instance) == Partition and instance.encrypted: + # TODO: Perhaps support unlocking encrypted volumes? + # This will cause a lot of potential user interactions tho. + log(f"Ignoring {blockdevice} because it's encrypted.", fg="gray", level=logging.DEBUG) + continue + + detection_mountpoint = pathlib.Path(f"/tmp/{''.join([random.choice(string.ascii_letters) for x in range(20)])}") + detection_mountpoint.mkdir(parents=True, exist_ok=True) + + instance.mount(str(detection_mountpoint)) + we_mounted_it = True + + if (filesystem := findmnt(detection_mountpoint)) and (filesystem := filesystem.get('filesystems', [])): + if subvolume := subvolume_info_from_path(filesystem[0]['target']): + if subvolume.uuid == self.uuid: + # The top level subvolume matched of ourselves, + # which means the instance we're iterating has the subvol we're looking for. + log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") + return instance + + def iterate_children(struct): + for child in struct.get('children', []): + if '[' in child.get('source', ''): + yield subvolume_info_from_path(child['target']) + + for sub_child in iterate_children(child): + yield sub_child + + for child in iterate_children(filesystem[0]): + if child.uuid == self.uuid: + # We found a child within the instance that has the subvol we're looking for. + log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") + return instance + + if we_mounted_it: + instance.unmount() + shutil.rmtree(detection_mountpoint) + + @cached_property + def mount_options(self) -> Optional[List[str]]: + from ..helpers import findmnt + + if filesystem := findmnt(self.full_path).get('filesystems', []): + return filesystem[0].get('options').split(',') + + def convert_to_ISO_format(self, time_string): + time_string_almost_done = time_string.replace(' ', 'T', 1).replace(' ', '') + iso_string = f"{time_string_almost_done[:-2]}:{time_string_almost_done[-2:]}" + return iso_string + + def mount(self, mountpoint :pathlib.Path, options=None, include_previously_known_options=True): + from ..helpers import findmnt + + try: + if mnt_info := findmnt(pathlib.Path(mountpoint), traverse=False): + log(f"Unmounting {mountpoint} as it was already mounted using {mnt_info}") + SysCommand(f"umount {mountpoint}") + except DiskError: + # No previously mounted device at the mountpoint + pass + + if not options: + options = [] + + try: + if include_previously_known_options and (cached_options := self.mount_options): + options += cached_options + except DiskError: + pass + + if not any('subvol=' in x for x in options): + options += f'subvol={self.name}' + + SysCommand(f"mount {self.partition.path} {mountpoint} -o {','.join(options)}") + log(f"{self} has successfully been mounted to {mountpoint}", level=logging.INFO, fg="gray") + + def unmount(self, recurse :bool = True): + SysCommand(f"umount {'-R' if recurse else ''} {self.full_path}") + log(f"Successfully unmounted {self}", level=logging.INFO, fg="gray") diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py index 85c0390f..660594ed 100644 --- a/archinstall/lib/disk/helpers.py +++ b/archinstall/lib/disk/helpers.py @@ -8,6 +8,8 @@ import time import glob from typing import Union, List, Iterator, Dict, Optional, Any, TYPE_CHECKING # https://stackoverflow.com/a/39757388/929999 +from ..models.subvolume import Subvolume + if TYPE_CHECKING: from .partition import Partition @@ -469,6 +471,7 @@ def convert_device_to_uuid(path :str) -> str: raise DiskError(f"Could not retrieve the UUID of {path} within a timely manner.") + def has_mountpoint(partition: Union[dict,Partition,MapperDev], target: str, strict: bool = True) -> bool: """ Determine if a certain partition is mounted (or has a mountpoint) as specific target (path) Coded for clarity rather than performance @@ -485,10 +488,12 @@ def has_mountpoint(partition: Union[dict,Partition,MapperDev], target: str, stri """ # we create the mountpoint list if isinstance(partition,dict): - subvols = partition.get('btrfs',{}).get('subvolumes',{}) - mountpoints = [partition.get('mountpoint'),] + [subvols[subvol] if isinstance(subvols[subvol],str) or not subvols[subvol] else subvols[subvol].get('mountpoint') for subvol in subvols] + subvolumes: List[Subvolume] = partition.get('btrfs',{}).get('subvolumes', []) + mountpoints = [partition.get('mountpoint')] + mountpoints += [volume.mountpoint for volume in subvolumes] else: mountpoints = [partition.mountpoint,] + [subvol.target for subvol in partition.subvolumes] + # we check if strict or target == '/': if target in mountpoints: diff --git a/archinstall/lib/disk/mapperdev.py b/archinstall/lib/disk/mapperdev.py index 913dbc13..49137ae9 100644 --- a/archinstall/lib/disk/mapperdev.py +++ b/archinstall/lib/disk/mapperdev.py @@ -10,7 +10,7 @@ from ..general import SysCommand from ..output import log if TYPE_CHECKING: - from .btrfs import BtrfsSubvolume + from .btrfs import BtrfsSubvolumeInfo @dataclass class MapperDev: @@ -37,12 +37,12 @@ class MapperDev: for slave in glob.glob(f"/sys/class/block/{dm_device.name}/slaves/*"): partition_belonging_to_dmcrypt_device = pathlib.Path(slave).name - + try: uevent_data = SysCommand(f"blkid -o export /dev/{partition_belonging_to_dmcrypt_device}").decode() except SysCallError as error: log(f"Could not get information on device /dev/{partition_belonging_to_dmcrypt_device}: {error}", level=logging.ERROR, fg="red") - + information = uevent(uevent_data) block_device = BlockDevice(get_parent_of_partition('/dev/' / pathlib.Path(information['DEVNAME']))) @@ -75,10 +75,10 @@ class MapperDev: return get_filesystem_type(self.path) @property - def subvolumes(self) -> Iterator['BtrfsSubvolume']: + def subvolumes(self) -> Iterator['BtrfsSubvolumeInfo']: from .btrfs import subvolume_info_from_path for mountpoint in self.mount_information: if target := mountpoint.get('target'): if subvolume := subvolume_info_from_path(pathlib.Path(target)): - yield subvolume \ No newline at end of file + yield subvolume diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 2c9f50c2..6f25a5f7 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -14,7 +14,7 @@ from ..exceptions import DiskError, SysCallError, UnknownFilesystemFormat from ..output import log from ..general import SysCommand from .btrfs.btrfs_helpers import subvolume_info_from_path -from .btrfs.btrfssubvolume import BtrfsSubvolume +from .btrfs.btrfssubvolumeinfo import BtrfsSubvolumeInfo class Partition: def __init__(self, @@ -185,7 +185,7 @@ class Partition: for i in range(storage['DISK_RETRY_ATTEMPTS']): if not self.partprobe(): raise DiskError(f"Could not perform partprobe on {self.device_path}") - + time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * i)) partuuid = self._safe_part_uuid @@ -294,9 +294,9 @@ class Partition: return bind_name @property - def subvolumes(self) -> Iterator[BtrfsSubvolume]: + def subvolumes(self) -> Iterator[BtrfsSubvolumeInfo]: from .helpers import findmnt - + def iterate_children_recursively(information): for child in information.get('children', []): if target := child.get('target'): @@ -452,7 +452,7 @@ class Partition: if retry is True: log(f"Retrying in {storage.get('DISK_TIMEOUTS', 1)} seconds.", level=logging.WARNING, fg="orange") time.sleep(storage.get('DISK_TIMEOUTS', 1)) - + return self.format(filesystem, path, log_formatting, options, retry=False) if get_filesystem_type(path) == 'crypto_LUKS' or get_filesystem_type(self.real_device) == 'crypto_LUKS': diff --git a/archinstall/lib/disk/user_guides.py b/archinstall/lib/disk/user_guides.py index 5fa6bfdc..5809c073 100644 --- a/archinstall/lib/disk/user_guides.py +++ b/archinstall/lib/disk/user_guides.py @@ -3,6 +3,8 @@ import logging from typing import Optional, Dict, Any, List, TYPE_CHECKING # https://stackoverflow.com/a/39757388/929999 +from ..models.subvolume import Subvolume + if TYPE_CHECKING: from .blockdevice import BlockDevice _: Any @@ -107,17 +109,14 @@ def suggest_single_disk_layout(block_device :BlockDevice, # https://unix.stackexchange.com/questions/246976/btrfs-subvolume-uuid-clash # https://github.com/classy-giraffe/easy-arch/blob/main/easy-arch.sh layout[block_device.path]['partitions'][1]['btrfs'] = { - "subvolumes" : { - "@":"/", - "@home": "/home", - "@log": "/var/log", - "@pkg": "/var/cache/pacman/pkg", - "@.snapshots": "/.snapshots" - } + 'subvolumes': [ + Subvolume('@', '/'), + Subvolume('@home', '/home'), + Subvolume('@log', '/var/log'), + Subvolume('@pkg', '/var/cache/pacman/pkg'), + Subvolume('@.snapshots', '/.snapshots') + ] } - # else: - # pass # ... implement a guided setup - elif using_home_partition: # If we don't want to use subvolumes, # But we want to be able to re-use data between re-installs.. diff --git a/archinstall/lib/installer.py b/archinstall/lib/installer.py index bf296c2e..97c2492d 100644 --- a/archinstall/lib/installer.py +++ b/archinstall/lib/installer.py @@ -24,6 +24,7 @@ from .disk.partition import get_mount_fs_type from .exceptions import DiskError, ServiceException, RequirementError, HardwareIncompatibilityError, SysCallError from .hsm import fido2_enroll from .models.users import User +from .models.subvolume import Subvolume if TYPE_CHECKING: _: Any @@ -263,47 +264,25 @@ class Installer: hsm_device_path = storage['arguments']['HSM'] fido2_enroll(hsm_device_path, partition['device_instance'], password) - # we manage the btrfs partitions - if any(btrfs_subvolumes := [entry for entry in list_part if entry.get('btrfs', {}).get('subvolumes', {})]): - for partition in btrfs_subvolumes: - if mount_options := ','.join(partition.get('filesystem',{}).get('mount_options',[])): - self.mount(partition['device_instance'], "/", options=mount_options) - else: - self.mount(partition['device_instance'], "/") - - setup_subvolumes( - installation=self, - partition_dict=partition - ) + btrfs_subvolumes = [entry for entry in list_part if entry.get('btrfs', {}).get('subvolumes', [])] - partition['device_instance'].unmount() + for partition in btrfs_subvolumes: + device_instance = partition['device_instance'] + mount_options = partition.get('filesystem', {}).get('mount_options', []) + self.mount(device_instance, "/", options=','.join(mount_options)) + setup_subvolumes(installation=self, partition_dict=partition) + device_instance.unmount() # We then handle any special cases, such as btrfs - if any(btrfs_subvolumes := [entry for entry in list_part if entry.get('btrfs', {}).get('subvolumes', {})]): - for partition_information in btrfs_subvolumes: - for name, mountpoint in sorted(partition_information['btrfs']['subvolumes'].items(), key=lambda item: item[1]): - btrfs_subvolume_information = {} - - match mountpoint: - case str(): # backwards-compatability - btrfs_subvolume_information['mountpoint'] = mountpoint - btrfs_subvolume_information['options'] = [] - case dict(): - btrfs_subvolume_information['mountpoint'] = mountpoint.get('mountpoint', None) - btrfs_subvolume_information['options'] = mountpoint.get('options', []) - case _: - continue - - if mountpoint_parsed := btrfs_subvolume_information.get('mountpoint'): - # We cache the mount call for later - mount_queue[mountpoint_parsed] = lambda device=partition_information['device_instance'], \ - name=name, \ - subvolume_information=btrfs_subvolume_information: mount_subvolume( - installation=self, - device=device, - name=name, - subvolume_information=subvolume_information - ) + for partition in btrfs_subvolumes: + subvolumes: List[Subvolume] = partition['btrfs']['subvolumes'] + for subvolume in sorted(subvolumes, key=lambda item: item.mountpoint): + # We cache the mount call for later + mount_queue[subvolume.mountpoint] = lambda sub_vol=subvolume, device=partition['device_instance']: mount_subvolume( + installation=self, + device=device, + subvolume=sub_vol + ) # We mount ordinary partitions, and we sort them by the mountpoint for partition in sorted([entry for entry in list_part if entry.get('mountpoint', False)], key=lambda part: part['mountpoint']): diff --git a/archinstall/lib/menu/global_menu.py b/archinstall/lib/menu/global_menu.py index cb61168d..49083517 100644 --- a/archinstall/lib/menu/global_menu.py +++ b/archinstall/lib/menu/global_menu.py @@ -325,22 +325,23 @@ class GlobalMenu(GeneralMenu): def _select_harddrives(self, old_harddrives : list) -> List: harddrives = select_harddrives(old_harddrives) - if len(harddrives) == 0: - 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, Menu.yes_no(), default_option=Menu.yes(), skip=False).run() - - if choice.value == Menu.no(): - return self._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['disk_layouts'].set_current_selection(None) - storage['arguments']['disk_layouts'] = {} + if harddrives: + if len(harddrives) == 0: + 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, Menu.yes_no(), default_option=Menu.yes(), skip=False).run() + + if choice.value == Menu.no(): + return self._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['disk_layouts'].set_current_selection(None) + storage['arguments']['disk_layouts'] = {} return harddrives diff --git a/archinstall/lib/menu/list_manager.py b/archinstall/lib/menu/list_manager.py index 7e051528..40d01ce3 100644 --- a/archinstall/lib/menu/list_manager.py +++ b/archinstall/lib/menu/list_manager.py @@ -137,34 +137,35 @@ class ListManager: else: self._default_action = [str(default_action),] - self.header = header if header else None - self.cancel_action = str(_('Cancel')) - self.confirm_action = str(_('Confirm and exit')) - self.separator = '' - self.bottom_list = [self.confirm_action,self.cancel_action] - self.bottom_item = [self.cancel_action] - self.base_actions = base_actions if base_actions else [str(_('Add')),str(_('Copy')),str(_('Edit')),str(_('Delete'))] + self._header = header if header else None + self._cancel_action = str(_('Cancel')) + self._confirm_action = str(_('Confirm and exit')) + self._separator = '' + self._bottom_list = [self._confirm_action, self._cancel_action] + self._bottom_item = [self._cancel_action] + self._base_actions = base_actions if base_actions else [str(_('Add')), str(_('Copy')), str(_('Edit')), str(_('Delete'))] self._original_data = copy.deepcopy(base_list) self._data = copy.deepcopy(base_list) # as refs, changes are immediate + # default values for the null case self.target: Optional[Any] = None self.action = self._null_action - if len(self._data) == 0 and self._null_action: - self._data = self.exec_action(self._data) - def run(self): while True: # this will return a dictionary with the key as the menu entry to be displayed # and the value is the original value from the self._data container + data_formatted = self.reformat(self._data) options = list(data_formatted.keys()) - options.append(self.separator) + + if len(options) > 0: + options.append(self._separator) if self._default_action: options += self._default_action - options += self.bottom_list + options += self._bottom_list system('clear') @@ -174,12 +175,12 @@ class ListManager: sort=False, clear_screen=False, clear_menu_on_exit=False, - header=self.header, + header=self._header, skip_empty_entries=True, skip=False ).run() - if not target.value or target.value in self.bottom_list: + if not target.value or target.value in self._bottom_list: self.action = target break @@ -201,13 +202,13 @@ class ListManager: # Possible enhancement. If run_actions returns false a message line indicating the failure self.run_actions(target.value) - if target.value == self.cancel_action: # TODO dubious + if target.value == self._cancel_action: # TODO dubious return self._original_data # return the original list else: return self._data def run_actions(self,prompt_data=None): - options = self.action_list() + self.bottom_item + options = self.action_list() + self._bottom_item prompt = _("Select an action for < {} >").format(prompt_data if prompt_data else self.target) choice = Menu( prompt, @@ -215,13 +216,13 @@ class ListManager: sort=False, clear_screen=False, clear_menu_on_exit=False, - preset_values=self.bottom_item, + preset_values=self._bottom_item, show_search_hint=False ).run() self.action = choice.value - if self.action and self.action != self.cancel_action: + if self.action and self.action != self._cancel_action: self._data = self.exec_action(self._data) """ @@ -243,7 +244,7 @@ class ListManager: can define alternate action list or customize the list for each item. Executed after any item is selected, contained in self.target """ - return self.base_actions + return self._base_actions def exec_action(self, data: Any): """ diff --git a/archinstall/lib/models/subvolume.py b/archinstall/lib/models/subvolume.py new file mode 100644 index 00000000..34a09227 --- /dev/null +++ b/archinstall/lib/models/subvolume.py @@ -0,0 +1,68 @@ +from dataclasses import dataclass +from typing import List, Any, Dict + + +@dataclass +class Subvolume: + name: str + mountpoint: str + compress: bool = False + nodatacow: bool = False + + def display(self) -> str: + options_str = ','.join(self.options) + return f'{_("Subvolume")}: {self.name:15} {_("Mountpoint")}: {self.mountpoint:20} {_("Options")}: {options_str}' + + @property + def options(self) -> List[str]: + options = [ + 'compress' if self.compress else '', + 'nodatacow' if self.nodatacow else '' + ] + return [o for o in options if len(o)] + + def json(self) -> Dict[str, Any]: + return { + 'name': self.name, + 'mountpoint': self.mountpoint, + 'compress': self.compress, + 'nodatacow': self.nodatacow + } + + @classmethod + def _parse(cls, config_subvolumes: List[Dict[str, Any]]) -> List['Subvolume']: + subvolumes = [] + for entry in config_subvolumes: + if not entry.get('name', None) or not entry.get('mountpoint', None): + continue + + subvolumes.append( + Subvolume( + entry['name'], + entry['mountpoint'], + entry.get('compress', False), + entry.get('nodatacow', False) + ) + ) + + return subvolumes + + @classmethod + def _parse_backwards_compatible(cls, config_subvolumes) -> List['Subvolume']: + subvolumes = [] + for name, mountpoint in config_subvolumes.items(): + if not name or not mountpoint: + continue + + subvolumes.append(Subvolume(name, mountpoint)) + + return subvolumes + + @classmethod + def parse_arguments(cls, config_subvolumes: Any) -> List['Subvolume']: + if isinstance(config_subvolumes, list): + return cls._parse(config_subvolumes) + elif isinstance(config_subvolumes, dict): + return cls._parse_backwards_compatible(config_subvolumes) + + raise ValueError('Unknown disk layout btrfs subvolume format') diff --git a/archinstall/lib/models/users.py b/archinstall/lib/models/users.py index f72cabde..a8feb9ef 100644 --- a/archinstall/lib/models/users.py +++ b/archinstall/lib/models/users.py @@ -27,8 +27,10 @@ class User: } def display(self) -> str: - strength = PasswordStrength.strength(self.password) - password = '*' * len(self.password) + f' ({strength.value})' + password = '*' * (len(self.password) if self.password else 0) + if password: + strength = PasswordStrength.strength(self.password) + password += f' ({strength.value})' return f'{_("Username")}: {self.username:16} {_("Password")}: {password:20} sudo: {str(self.sudo)}' @classmethod diff --git a/archinstall/lib/user_interaction/partitioning_conf.py b/archinstall/lib/user_interaction/partitioning_conf.py index caf5f5df..63b7c7df 100644 --- a/archinstall/lib/user_interaction/partitioning_conf.py +++ b/archinstall/lib/user_interaction/partitioning_conf.py @@ -351,18 +351,16 @@ def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, 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'] = {} + 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'] + result = SubvolumeList(_("Manage btrfs subvolumes for current partition"), prev).run() + block_device_struct["partitions"][partition]['btrfs']['subvolumes'] = result return block_device_struct + def select_encrypted_partitions( title :str, partitions :List[Partition], diff --git a/archinstall/lib/user_interaction/subvolume_config.py b/archinstall/lib/user_interaction/subvolume_config.py index 94e6f5d7..a54ec891 100644 --- a/archinstall/lib/user_interaction/subvolume_config.py +++ b/archinstall/lib/user_interaction/subvolume_config.py @@ -1,155 +1,94 @@ -from typing import Dict, List +from typing import Dict, List, Optional, Any, TYPE_CHECKING from ..menu.list_manager import ListManager from ..menu.menu import MenuSelectionType -from ..menu.selection_menu import Selector, GeneralMenu from ..menu.text_input import TextInput from ..menu import Menu +from ..models.subvolume import Subvolume + +if TYPE_CHECKING: + _: Any -""" -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, data: Dict) -> Dict: - def presentation(key :str, value :Dict): - 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 - - formatted = {presentation(k, v): k for k, v in data.items()} - return {k: v for k, v in sorted(formatted.items(), key=lambda e: e[0])} + def __init__(self, prompt: str, current_volumes: List[Subvolume]): + self._actions = [ + str(_('Add subvolume')), + str(_('Edit subvolume')), + str(_('Delete subvolume')) + ] + super().__init__(prompt, current_volumes, self._actions, self._actions[0]) - def action_list(self): - return super().action_list() + def reformat(self, data: List[Subvolume]) -> Dict[str, Subvolume]: + return {e.display(): e for e in data} - def exec_action(self, data: Dict): - if self.target: - origkey, origval = list(self.target.items())[0] - else: - origkey = None + def action_list(self): + active_user = self.target if self.target else None - if self.action == str(_('Delete')): - del data[origkey] + if active_user is None: + return [self._actions[0]] else: - 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 elem in ['name','mountpoint','options']: - add_menu.exec_option(elem) - else: - SubvolumeMenu(self.target,self.action).run() + return self._actions[1:] - data.update(self.target) + def _prompt_options(self, editing: Optional[Subvolume] = None) -> List[str]: + preset_options = [] + if editing: + preset_options = editing.options - return data - - -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): - 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) -> List[str]: - # def __init__(self, title, p_options, skip=True, multi=False, default_option=None, sort=True): choice = Menu( str(_("Select the desired subvolume options ")), ['nodatacow','compress'], skip=True, - preset_values=value, + preset_values=preset_options, multi=True ).run() if choice.type_ == MenuSelectionType.Selection: - return choice.value + return choice.value # type: ignore return [] + + def _add_subvolume(self, editing: Optional[Subvolume] = None) -> Optional[Subvolume]: + name = TextInput(f'\n\n{_("Subvolume name")}: ', editing.name if editing else '').run() + + if not name: + return None + + mountpoint = TextInput(f'\n{_("Subvolume mountpoint")}: ', editing.mountpoint if editing else '').run() + + if not mountpoint: + return None + + options = self._prompt_options(editing) + + subvolume = Subvolume(name, mountpoint) + subvolume.compress = 'compress' in options + subvolume.nodatacow = 'nodatacow' in options + + return subvolume + + def exec_action(self, data: List[Subvolume]) -> List[Subvolume]: + if self.target: + active_subvolume = self.target + else: + active_subvolume = None + + if self.action == self._actions[0]: # add + new_subvolume = self._add_subvolume() + + if new_subvolume is not None: + # in case a user with the same username as an existing user + # was created we'll replace the existing one + data = [d for d in data if d.name != new_subvolume.name] + data += [new_subvolume] + elif self.action == self._actions[1]: # edit subvolume + new_subvolume = self._add_subvolume(active_subvolume) + + if new_subvolume is not None: + # we'll remove the original subvolume and add the modified version + data = [d for d in data if d.name != active_subvolume.name and d.name != new_subvolume.name] + data += [new_subvolume] + elif self.action == self._actions[2]: # delete + data = [d for d in data if d != active_subvolume] + + return data -- cgit v1.2.3-70-g09d2 From fd131c8fe956858fef802f8e53a530daa0b5df47 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Tue, 7 Jun 2022 19:00:48 +1000 Subject: Update blockdevice (#1289) * Update blockdevice class Co-authored-by: Daniel Girtler --- .github/workflows/mypy.yaml | 2 +- archinstall/lib/disk/blockdevice.py | 344 +++++++++++++++++------------------- archinstall/lib/disk/filesystem.py | 2 +- archinstall/lib/disk/partition.py | 14 +- archinstall/lib/menu/global_menu.py | 2 +- archinstall/lib/output.py | 4 +- 6 files changed, 177 insertions(+), 191 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/.github/workflows/mypy.yaml b/.github/workflows/mypy.yaml index b0901b38..01d4741f 100644 --- a/.github/workflows/mypy.yaml +++ b/.github/workflows/mypy.yaml @@ -15,4 +15,4 @@ jobs: # one day this will be enabled # run: mypy --strict --module archinstall || exit 0 - name: run mypy - run: mypy --follow-imports=silent archinstall/lib/menu/selection_menu.py archinstall/lib/menu/global_menu.py archinstall/lib/models/network_configuration.py archinstall/lib/menu/list_manager.py archinstall/lib/user_interaction/network_conf.py archinstall/lib/models/users.py archinstall/lib/user_interaction/subvolume_config.py archinstall/lib/disk/btrfs/btrfs_helpers.py archinstall/lib/translation.py + run: mypy --follow-imports=silent archinstall/lib/menu/selection_menu.py archinstall/lib/menu/global_menu.py archinstall/lib/models/network_configuration.py archinstall/lib/menu/list_manager.py archinstall/lib/user_interaction/network_conf.py archinstall/lib/models/users.py archinstall/lib/disk/blockdevice.py archinstall/lib/user_interaction/subvolume_config.py archinstall/lib/disk/btrfs/btrfs_helpers.py archinstall/lib/translation.py diff --git a/archinstall/lib/disk/blockdevice.py b/archinstall/lib/disk/blockdevice.py index c7b69205..4e207bf4 100644 --- a/archinstall/lib/disk/blockdevice.py +++ b/archinstall/lib/disk/blockdevice.py @@ -1,13 +1,11 @@ from __future__ import annotations -import os import json import logging import time -from functools import cached_property -from typing import Optional, Dict, Any, Iterator, Tuple, List, TYPE_CHECKING -# https://stackoverflow.com/a/39757388/929999 -if TYPE_CHECKING: - from .partition import Partition + +from collections import OrderedDict +from dataclasses import dataclass +from typing import Optional, Dict, Any, Iterator, List, TYPE_CHECKING from ..exceptions import DiskError, SysCallError from ..output import log @@ -15,18 +13,44 @@ from ..general import SysCommand from ..storage import storage +if TYPE_CHECKING: + from .partition import Partition + _: Any + + +@dataclass +class BlockSizeInfo: + start: str + end: str + size: str + + +@dataclass +class BlockInfo: + pttype: str + ptuuid: str + size: int + tran: Optional[str] + rota: bool + free_space: Optional[List[BlockSizeInfo]] + + class BlockDevice: def __init__(self, path :str, info :Optional[Dict[str, Any]] = None): if not info: from .helpers import all_blockdevices # If we don't give any information, we need to auto-fill it. # Otherwise any subsequent usage will break. - info = all_blockdevices(partitions=False)[path].info + self.info = all_blockdevices(partitions=False)[path].info + else: + self.info = info - self.path = path - self.info = info + self._path = path self.keep_partitions = True - self.part_cache = {} + self._block_info = self._fetch_information() + self._partitions: Dict[str, 'Partition'] = {} + + self._load_partitions() # TODO: Currently disk encryption is a BIT misleading. # It's actually partition-encryption, but for future-proofing this @@ -35,70 +59,114 @@ class BlockDevice: def __repr__(self, *args :str, **kwargs :str) -> str: return self._str_repr - @cached_property + @property + def path(self) -> str: + return self._path + + @property def _str_repr(self) -> str: - return f"BlockDevice({self.device_or_backfile}, size={self._safe_size}GB, free_space={self._safe_free_space}, bus_type={self.bus_type})" - - @cached_property - def display_info(self) -> str: - columns = { - str(_('Device')): self.device_or_backfile, - str(_('Size')): f'{self._safe_size}GB', - str(_('Free space')): f'{self._safe_free_space}', + return f"BlockDevice({self._device_or_backfile}, size={self.size}GB, free_space={self._safe_free_space()}, bus_type={self.bus_type})" + + def as_json(self) -> Dict[str, Any]: + return { + str(_('Device')): self._device_or_backfile, + str(_('Size')): f'{self.size}GB', + str(_('Free space')): f'{self._safe_free_space()}', str(_('Bus-type')): f'{self.bus_type}' } - padding = max([len(k) for k in columns.keys()]) - - pretty = '' - for k, v in columns.items(): - k = k.ljust(padding, ' ') - pretty += f'{k} = {v}\n' - - return pretty.rstrip() - - def __iter__(self) -> Iterator[Partition]: + def __iter__(self) -> Iterator['Partition']: for partition in self.partitions: yield self.partitions[partition] def __getitem__(self, key :str, *args :str, **kwargs :str) -> Any: if hasattr(self, key): return getattr(self, key) - elif key not in self.info: - raise KeyError(f'{self} does not contain information: "{key}"') - return self.info[key] + + if self.info and key in self.info: + return self.info[key] + + raise KeyError(f'{self.info} does not contain information: "{key}"') def __len__(self) -> int: return len(self.partitions) def __lt__(self, left_comparitor :'BlockDevice') -> bool: - return self.path < left_comparitor.path + return self._path < left_comparitor.path def json(self) -> str: """ json() has precedence over __dump__, so this is a way to give less/partial information for user readability. """ - return self.path + return self._path def __dump__(self) -> Dict[str, Dict[str, Any]]: return { - self.path : { - 'partuuid' : self.uuid, - 'wipe' : self.info.get('wipe', None), - 'partitions' : [part.__dump__() for part in self.partitions.values()] + self._path: { + 'partuuid': self.uuid, + 'wipe': self.info.get('wipe', None), + 'partitions': [part.__dump__() for part in self.partitions.values()] } } - @property - def partition_type(self) -> str: - output = json.loads(SysCommand(f"lsblk --json -o+PTTYPE {self.path}").decode('UTF-8')) + def _call_lsblk(self, path: str) -> Dict[str, Any]: + output = SysCommand(f'lsblk --json -b -o+SIZE,PTTYPE,ROTA,TRAN,PTUUID {self._path}').decode('UTF-8') + if output: + lsblk_info = json.loads(output) + return lsblk_info + + raise DiskError(f'Failed to read disk "{self.path}" with lsblk') + + def _load_partitions(self): + from .partition import Partition - for device in output['blockdevices']: - return device['pttype'] + lsblk_info = self._call_lsblk(self._path) + device = lsblk_info['blockdevices'][0] + self._partitions.clear() - @cached_property - def device_or_backfile(self) -> str: + if children := device.get('children', None): + root = f'/dev/{device["name"]}' + for child in children: + part_id = child['name'].removeprefix(device['name']) + self._partitions[part_id] = Partition(root + part_id, block_device=self, part_id=part_id) + + def _get_free_space(self) -> Optional[List[BlockSizeInfo]]: + # NOTE: parted -s will default to `cancel` on prompt, skipping any partition + # that is "outside" the disk. in /dev/sr0 this is usually the case with Archiso, + # so the free will ignore the ESP partition and just give the "free" space. + # Doesn't harm us, but worth noting in case something weird happens. + try: + output = SysCommand(f"parted -s --machine {self._path} print free").decode('utf-8') + if output: + free_lines = [line for line in output.split('\n') if 'free' in line] + sizes = [] + for free_space in free_lines: + _, start, end, size, *_ = free_space.strip('\r\n;').split(':') + sizes.append(BlockSizeInfo(start, end, size)) + + return sizes + except SysCallError as error: + log(f"Could not get free space on {self._path}: {error}", level=logging.DEBUG) + + return None + + def _fetch_information(self) -> BlockInfo: + lsblk_info = self._call_lsblk(self._path) + device = lsblk_info['blockdevices'][0] + free_space = self._get_free_space() + + return BlockInfo( + pttype=device['pttype'], + ptuuid=device['ptuuid'], + size=device['size'], + tran=device['tran'], + rota=device['rota'], + free_space=free_space + ) + + @property + def _device_or_backfile(self) -> Optional[str]: """ Returns the actual device-endpoint of the BlockDevice. If it's a loop-back-device it returns the back-file, @@ -118,7 +186,7 @@ class BlockDevice: return None @property - def device(self) -> str: + def device(self) -> Optional[str]: """ Returns the device file of the BlockDevice. If it's a loop-back-device it returns the /dev/X device, @@ -126,168 +194,84 @@ class BlockDevice: And if it's a crypto-device it returns the parent device """ if "DEVTYPE" not in self.info: - raise DiskError(f'Could not locate backplane info for "{self.path}"') + raise DiskError(f'Could not locate backplane info for "{self._path}"') if self.info['DEVTYPE'] in ['disk','loop']: - return self.path + return self._path elif self.info['DEVTYPE'][:4] == 'raid': # This should catch /dev/md## raid devices - return self.path + return self._path elif self.info['DEVTYPE'] == 'crypt': if 'pkname' not in self.info: - raise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.') + raise DiskError(f'A crypt device ({self._path}) without a parent kernel device name.') return f"/dev/{self.info['pkname']}" else: - log(f"Unknown blockdevice type for {self.path}: {self.info['DEVTYPE']}", level=logging.DEBUG) + log(f"Unknown blockdevice type for {self._path}: {self.info['DEVTYPE']}", level=logging.DEBUG) - # if not stat.S_ISBLK(os.stat(full_path).st_mode): - # raise DiskError(f'Selected disk "{full_path}" is not a block device.') - - @property - def partitions(self) -> Dict[str, Partition]: - from .filesystem import Partition - - self.partprobe() - result = SysCommand(['/usr/bin/lsblk', '-J', self.path]) - - if b'not a block device' in result: - raise DiskError(f'Can not read partitions off something that isn\'t a block device: {self.path}') - - if not result[:1] == b'{': - raise DiskError('Error getting JSON output from:', f'/usr/bin/lsblk -J {self.path}') - - r = json.loads(result.decode('UTF-8')) - if len(r['blockdevices']) and 'children' in r['blockdevices'][0]: - root_path = f"/dev/{r['blockdevices'][0]['name']}" - for part in r['blockdevices'][0]['children']: - part_id = part['name'][len(os.path.basename(self.path)):] - if part_id not in self.part_cache: - # TODO: Force over-write even if in cache? - if part_id not in self.part_cache or self.part_cache[part_id].size != part['size']: - self.part_cache[part_id] = Partition(root_path + part_id, block_device=self, part_id=part_id) - - return {k: self.part_cache[k] for k in sorted(self.part_cache)} + return None @property - def partition(self) -> Partition: - all_partitions = self.partitions - return [all_partitions[k] for k in all_partitions] + def partition_type(self) -> str: + return self._block_info.pttype @property - def partition_table_type(self) -> int: - # TODO: Don't hardcode :) - # Remove if we don't use this function anywhere - from .filesystem import GPT - return GPT - - @cached_property def uuid(self) -> str: - log('BlockDevice().uuid is untested!', level=logging.WARNING, fg='yellow') - """ - Returns the disk UUID as returned by lsblk. - This is more reliable than relying on /dev/disk/by-partuuid as - it doesn't seam to be able to detect md raid partitions. - """ - return SysCommand(f'blkid -s PTUUID -o value {self.path}').decode('UTF-8') - - @cached_property - def _safe_size(self) -> float: - from .helpers import convert_size_to_gb - - try: - output = json.loads(SysCommand(f"lsblk --json -b -o+SIZE {self.path}").decode('UTF-8')) - except SysCallError: - return -1.0 + return self._block_info.ptuuid - for device in output['blockdevices']: - return convert_size_to_gb(device['size']) - - @cached_property + @property def size(self) -> float: from .helpers import convert_size_to_gb + return convert_size_to_gb(self._block_info.size) - output = json.loads(SysCommand(f"lsblk --json -b -o+SIZE {self.path}").decode('UTF-8')) - - for device in output['blockdevices']: - return convert_size_to_gb(device['size']) - - @cached_property - def bus_type(self) -> str: - output = json.loads(SysCommand(f"lsblk --json -o+ROTA,TRAN {self.path}").decode('UTF-8')) - - for device in output['blockdevices']: - return device['tran'] + @property + def bus_type(self) -> Optional[str]: + return self._block_info.tran - @cached_property + @property def spinning(self) -> bool: - output = json.loads(SysCommand(f"lsblk --json -o+ROTA,TRAN {self.path}").decode('UTF-8')) + return self._block_info.rota - for device in output['blockdevices']: - return device['rota'] is True + @property + def partitions(self) -> Dict[str, 'Partition']: + self._partprobe() + self._load_partitions() + return OrderedDict(sorted(self._partitions.items())) - @cached_property - def _safe_free_space(self) -> Tuple[str, ...]: - try: - return '+'.join(part[2] for part in self.free_space) - except SysCallError: - return '?' + @property + def partition(self) -> List['Partition']: + return list(self.partitions.values()) - @cached_property - def free_space(self) -> Tuple[str, ...]: - # NOTE: parted -s will default to `cancel` on prompt, skipping any partition - # that is "outside" the disk. in /dev/sr0 this is usually the case with Archiso, - # so the free will ignore the ESP partition and just give the "free" space. - # Doesn't harm us, but worth noting in case something weird happens. - try: - for line in SysCommand(f"parted -s --machine {self.path} print free"): - if 'free' in (free_space := line.decode('UTF-8')): - _, start, end, size, *_ = free_space.strip('\r\n;').split(':') - yield (start, end, size) - except SysCallError as error: - log(f"Could not get free space on {self.path}: {error}", level=logging.DEBUG) - - @cached_property - def largest_free_space(self) -> List[str]: - info = [] - for space_info in self.free_space: - if not info: - info = space_info - else: - # [-1] = size - if space_info[-1] > info[-1]: - info = space_info - return info - - @cached_property + @property def first_free_sector(self) -> str: - if info := self.largest_free_space: - start = info[0] + if block_size := self._largest_free_space(): + return block_size.start else: - start = '512MB' - return start + return '512MB' - @cached_property + @property def first_end_sector(self) -> str: - if info := self.largest_free_space: - end = info[1] + if block_size := self._largest_free_space(): + return block_size.end else: - end = f"{self.size}GB" - return end - - def partprobe(self) -> bool: - return SysCommand(['partprobe', self.path]).exit_code == 0 + return f"{self.size}GB" + + def _safe_free_space(self) -> str: + if self._block_info.free_space: + sizes = [free_space.size for free_space in self._block_info.free_space] + return '+'.join(sizes) + return '?' + + def _largest_free_space(self) -> Optional[BlockSizeInfo]: + if self._block_info.free_space: + sorted_sizes = sorted(self._block_info.free_space, key=lambda x: x.size, reverse=True) + return sorted_sizes[0] + return None - def has_partitions(self) -> int: - return len(self.partitions) - - def has_mount_point(self, mountpoint :str) -> bool: - for partition in self.partitions: - if self.partitions[partition].mountpoint == mountpoint: - return True - return False + def _partprobe(self) -> bool: + return SysCommand(['partprobe', self._path]).exit_code == 0 def flush_cache(self) -> None: - self.part_cache = {} + self._load_partitions() def get_partition(self, uuid :Optional[str] = None, partuuid :Optional[str] = None) -> Partition: if not uuid and not partuuid: @@ -296,7 +280,7 @@ class BlockDevice: for count in range(storage.get('DISK_RETRY_ATTEMPTS', 5)): for partition_index, partition in self.partitions.items(): try: - if uuid and partition.uuid.lower() == uuid.lower(): + if uuid and partition.uuid and partition.uuid.lower() == uuid.lower(): return partition elif partuuid and partition.part_uuid.lower() == partuuid.lower(): return partition @@ -308,8 +292,8 @@ class BlockDevice: log(f"uuid {uuid} or {partuuid} not found. Waiting {storage.get('DISK_TIMEOUTS', 1) * count}s for next attempt",level=logging.DEBUG) time.sleep(storage.get('DISK_TIMEOUTS', 1) * count) - + log(f"Could not find {uuid}/{partuuid} in disk after 5 retries", level=logging.INFO) - log(f"Cache: {self.part_cache}") + log(f"Cache: {self._partitions}") log(f"Partitions: {self.partitions.items()}") raise DiskError(f"Partition {uuid}/{partuuid} was never found on {self} despite several attempts.") diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py index cc29a491..1c7a801b 100644 --- a/archinstall/lib/disk/filesystem.py +++ b/archinstall/lib/disk/filesystem.py @@ -100,7 +100,7 @@ class Filesystem: partition['device_instance'] = self.blockdevice.get_partition(uuid=partition_uuid) except DiskError: partition['device_instance'] = self.blockdevice.get_partition(partuuid=partition_uuid) - + log(_("Re-using partition instance: {}").format(partition['device_instance']), level=logging.DEBUG, fg="gray") else: log(f"{self}.load_layout() doesn't know how to work without 'wipe' being set or UUID ({partition.get('PARTUUID')}) was given and found.", fg="yellow", level=logging.WARNING) diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 6f25a5f7..062c79ab 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -175,7 +175,7 @@ class Partition: return device['pttype'] @property - def part_uuid(self) -> Optional[str]: + def part_uuid(self) -> str: """ Returns the PARTUUID as returned by lsblk. This is more reliable than relying on /dev/disk/by-partuuid as @@ -222,7 +222,7 @@ class Partition: For instance when you want to get a __repr__ of the class. """ if not self.partprobe(): - if self.block_device.info.get('TYPE') == 'iso9660': + if self.block_device.partition_type == 'iso9660': return None log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) @@ -230,7 +230,7 @@ class Partition: try: return SysCommand(f'blkid -s UUID -o value {self.device_path}').decode('UTF-8').strip() except SysCallError as error: - if self.block_device.info.get('TYPE') == 'iso9660': + if self.block_device.partition_type == 'iso9660': # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) return None @@ -244,15 +244,15 @@ class Partition: For instance when you want to get a __repr__ of the class. """ if not self.partprobe(): - if self.block_device.info.get('TYPE') == 'iso9660': + if self.block_device.partition_type == 'iso9660': return None log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) try: - return SysCommand(f'blkid -s PARTUUID -o value {self.device_path}').decode('UTF-8').strip() + return self.block_device.uuid except SysCallError as error: - if self.block_device.info.get('TYPE') == 'iso9660': + if self.block_device.partition_type == 'iso9660': # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) return None @@ -399,7 +399,7 @@ class Partition: elif filesystem == 'vfat': options = ['-F32'] + options - + log(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}") if (handle := SysCommand(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") self.filesystem = filesystem diff --git a/archinstall/lib/menu/global_menu.py b/archinstall/lib/menu/global_menu.py index 49083517..abd652a9 100644 --- a/archinstall/lib/menu/global_menu.py +++ b/archinstall/lib/menu/global_menu.py @@ -240,7 +240,7 @@ class GlobalMenu(GeneralMenu): selector = self._menu_options['harddrives'] if selector.has_selection(): drives = selector.current_selection - return '\n\n'.join([d.display_info for d in drives]) + return FormattedOutput.as_table(drives) return None def _prev_disk_layouts(self) -> Optional[str]: diff --git a/archinstall/lib/output.py b/archinstall/lib/output.py index 29b73bc4..e2b38ce6 100644 --- a/archinstall/lib/output.py +++ b/archinstall/lib/output.py @@ -11,7 +11,9 @@ class FormattedOutput: @classmethod def values(cls, o: Any) -> Dict[str, Any]: - if hasattr(o, 'json'): + if hasattr(o, 'as_json'): + return o.as_json() + elif hasattr(o, 'json'): return o.json() else: return o.__dict__ -- cgit v1.2.3-70-g09d2 From 5c3c1312a49e1c110d4c5825fbb8242868544900 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Tue, 14 Jun 2022 22:38:39 +1000 Subject: Update list manager (#1331) * Rework partition management * Update * Update list manager * Update * Update Co-authored-by: Daniel Girtler --- archinstall/lib/disk/partition.py | 35 ++- archinstall/lib/menu/list_manager.py | 262 +++------------------ .../lib/user_interaction/manage_users_conf.py | 22 +- archinstall/lib/user_interaction/network_conf.py | 20 +- .../lib/user_interaction/partitioning_conf.py | 14 +- .../lib/user_interaction/subvolume_config.py | 32 ++- 6 files changed, 109 insertions(+), 276 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 062c79ab..17c24d57 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -17,14 +17,16 @@ from .btrfs.btrfs_helpers import subvolume_info_from_path from .btrfs.btrfssubvolumeinfo import BtrfsSubvolumeInfo class Partition: - def __init__(self, + def __init__( + self, path: str, block_device: BlockDevice, part_id :Optional[str] = None, filesystem :Optional[str] = None, mountpoint :Optional[str] = None, encrypted :bool = False, - autodetect_filesystem :bool = True): + autodetect_filesystem :bool = True + ): if not part_id: part_id = os.path.basename(path) @@ -76,7 +78,30 @@ class Partition: else: return f'Partition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, fs={self.filesystem}{mount_repr})' + def as_json(self) -> Dict[str, Any]: + """ + this is used for the table representation of the partition (see FormattedOutput) + """ + partition_info = { + 'type': 'primary', + 'PARTUUID': self._safe_uuid, + 'wipe': self.allow_formatting, + 'boot': self.boot, + 'ESP': self.boot, + 'mountpoint': self.target_mountpoint, + 'encrypted': self._encrypted, + 'start': self.start, + 'size': self.end, + 'filesystem': self.filesystem_type + } + + return partition_info + def __dump__(self) -> Dict[str, Any]: + # TODO remove this in favour of as_json + + log(get_filesystem_type(self.path)) + return { 'type': 'primary', 'PARTUUID': self._safe_uuid, @@ -88,10 +113,14 @@ class Partition: 'start': self.start, 'size': self.end, 'filesystem': { - 'format': get_filesystem_type(self.path) + 'format': self.filesystem_type } } + @property + def filesystem_type(self) -> Optional[str]: + return get_filesystem_type(self.path) + @property def mountpoint(self) -> Optional[str]: try: diff --git a/archinstall/lib/menu/list_manager.py b/archinstall/lib/menu/list_manager.py index fe491caa..e7a9c2ac 100644 --- a/archinstall/lib/menu/list_manager.py +++ b/archinstall/lib/menu/list_manager.py @@ -1,94 +1,7 @@ -#!/usr/bin/python -""" -# Purpose -ListManager is a widget based on `menu` which allows the handling of repetitive operations in a list. -Imagine you have a list and want to add/copy/edit/delete their elements. With this widget you will be shown the list -``` -Vamos alla - -Use ESC to skip - - -> uno : 1 -dos : 2 -tres : 3 -cuatro : 4 -==> -Confirm and exit -Cancel -(Press "/" to search) -``` -Once you select one of the elements of the list, you will be promted with the action to be done to the selected element -``` - -uno : 1 -dos : 2 -> tres : 3 -cuatro : 4 -==> -Confirm and exit -Cancel -(Press "/" to search) - -Select an action for < {'tres': 3} > - - -Add -Copy -Edit -Delete -> Cancel -``` -You execute the action for this element (which might or not involve user interaction) and return to the list main page -till you call one of the options `confirm and exit` which returns the modified list or `cancel` which returns the original list unchanged. -If the list is empty one action can be defined as default (usually Add). We call it **null_action** -YOu can also define a **default_action** which will appear below the separator, not tied to any element of the list. Barring explicit definition, default_action will be the null_action -``` -==> -Add -Confirm and exit -Cancel -(Press "/" to search) -``` -The default implementation can handle simple lists and a key:value dictionary. The default actions are the shown above. -A sample of basic usage is included at the end of the source. - -More sophisticaded uses can be achieved by -* changing the action list and the null_action during initialization -``` - opciones = ListManager('Vamos alla',opciones,[str(_('Add')),str(_('Delete'))],_('Add')).run() -``` -* And using following methods to overwrite/define user actions and other details: -* * `reformat`. To change the appearance of the list elements -* * `action_list`. To modify the content of the action list once an element is defined. F.i. to avoid Delete to appear for certain elements, or to add/modify action based in the value of the element. -* * `exec_action` which contains the actual code to be executed when an action is selected - -The contents in the base class of this methods serve for a very basic usage, and are to be taken as samples. Thus the best use of this class would be to subclass in your code - -``` - class ObjectList(archinstall.ListManager): - def __init__(prompt,list): - self.ObjectAction = [... list of actions ...] - self.ObjectNullAction = one ObjectAction - super().__init__(prompt,list,ObjectActions,ObjectNullAction) - def reformat(self): - ... beautfy the output of the list - def action_list(self): - ... if you need some changes to the action list based on self.target - def exec_action(self): - if self.action == self.ObjectAction[0]: - performFirstAction(self.target, ...) - - ... - resultList = ObjectList(prompt,originallist).run() -``` - -""" import copy from os import system -from typing import Union, Any, TYPE_CHECKING, Dict, Optional, Tuple, List +from typing import Any, TYPE_CHECKING, Dict, Optional, Tuple, List -from .text_input import TextInput from .menu import Menu if TYPE_CHECKING: @@ -98,58 +11,38 @@ if TYPE_CHECKING: class ListManager: def __init__( self, - prompt :str, - base_list :Union[list,dict] , - base_actions :list = None, - null_action :str = None, - default_action :Union[str,list] = None, - header :Union[str,list] = None): + prompt: str, + entries: List[Any], + base_actions: List[str], + sub_menu_actions: List[str] + ): """ - param :prompt Text which will appear at the header + :param prompt: Text which will appear at the header type param: string | DeferredTranslation - param :base:_list list/dict of option to be shown / mainpulated - type param: list | dict - - param base_actions an alternate list of actions to the items of the object + :param entries: list/dict of option to be shown / manipulated type param: list - param: null_action action which will be taken (if any) when base_list is empty - type param: string - - param: default_action action which will be presented at the bottom of the list. Shouldn't need a target. If not present, null_action is set there. - Both Null and Default actions can be defined outside the base_actions list, as long as they are launched in exec_action - type param: string or list + :param base_actions: list of actions that is displayed in the main list manager, + usually global actions such as 'Add...' + type param: list - param: header one or more header lines for the list - type param: string or list + :param sub_menu_actions: list of actions available for a chosen entry + type param: list """ + self._original_data = copy.deepcopy(entries) + self._data = copy.deepcopy(entries) explainer = str(_('\n Choose an object from the list, and select one of the available actions for it to execute')) self._prompt = prompt + explainer if prompt else explainer - self._null_action = str(null_action) if null_action else None - - if not default_action: - self._default_action = [self._null_action] - elif isinstance(default_action,(list,tuple)): - self._default_action = default_action - else: - self._default_action = [str(default_action)] - - self._header = header if header else '' - self._cancel_action = str(_('Cancel')) - self._confirm_action = str(_('Confirm and exit')) self._separator = '' - self._bottom_list = [self._confirm_action, self._cancel_action] - self._bottom_item = [self._cancel_action] - self._base_actions = base_actions if base_actions else [str(_('Add')), str(_('Copy')), str(_('Edit')), str(_('Delete'))] - self._original_data = copy.deepcopy(base_list) - self._data = copy.deepcopy(base_list) # as refs, changes are immediate + self._confirm_action = str(_('Confirm and exit')) + self._cancel_action = str(_('Cancel')) - # default values for the null case - self.target: Optional[Any] = None - self.action = self._null_action + self._terminate_actions = [self._confirm_action, self._cancel_action] + self._base_actions = base_actions + self._sub_menu_actions = sub_menu_actions def run(self): while True: @@ -158,11 +51,6 @@ class ListManager: data_formatted = self.reformat(self._data) options, header = self._prepare_selection(data_formatted) - menu_header = self._header - - if header: - menu_header += header - system('clear') choice = Menu( @@ -177,27 +65,13 @@ class ListManager: show_search_hint=False ).run() - if not choice.value or choice.value in self._bottom_list: - self.action = choice + if choice.value in self._base_actions: + self._data = self.handle_action(choice.value, None, self._data) + elif choice.value in self._terminate_actions: break - - if choice.value and choice.value in self._default_action: - self.action = choice.value - self.target = None - self._data = self.exec_action(self._data) - continue - - if isinstance(self._data, dict): - data_key = data_formatted[choice.value] - key = self._data[data_key] - self.target = {data_key: key} - elif isinstance(self._data, list): - self.target = [d for d in self._data if d == data_formatted[choice.value]][0] - else: - self.target = self._data[data_formatted[choice.value]] - - # Possible enhancement. If run_actions returns false a message line indicating the failure - self.run_actions(choice.value) + else: # an entry of the existing selection was choosen + selected_entry = data_formatted[choice.value] + self._run_actions_on_entry(selected_entry) if choice.value == self._cancel_action: return self._original_data # return the original list @@ -217,16 +91,14 @@ class ListManager: if len(options) > 0: options.append(self._separator) - if self._default_action: - # done only for mypy -> todo fix the self._default_action declaration - options += [action for action in self._default_action if action] + options += self._base_actions + options += self._terminate_actions - options += self._bottom_list return options, header - def run_actions(self,prompt_data=''): - options = self.action_list() + self._bottom_item - display_value = self.selected_action_display(self.target) if self.target else prompt_data + def _run_actions_on_entry(self, entry: Any): + options = self._sub_menu_actions + [self._cancel_action] + display_value = self.selected_action_display(entry) prompt = _("Select an action for '{}'").format(display_value) @@ -236,14 +108,11 @@ class ListManager: sort=False, clear_screen=False, clear_menu_on_exit=False, - preset_values=self._bottom_item, show_search_hint=False ).run() - self.action = choice.value - - if self.action and self.action != self._cancel_action: - self._data = self.exec_action(self._data) + if choice.value and choice.value != self._cancel_action: + self._data = self.handle_action(choice.value, entry, self._data) def selected_action_display(self, selection: Any) -> str: # this will return the value to be displayed in the @@ -256,64 +125,7 @@ class ListManager: # in the header value (useful when displaying tables) raise NotImplementedError('Please implement me in the child class') - def action_list(self): - """ - can define alternate action list or customize the list for each item. - Executed after any item is selected, contained in self.target - """ - active_entry = self.target if self.target else None - - if active_entry is None: - return [self._base_actions[0]] - else: - return self._base_actions[1:] - - def exec_action(self, data: Any): - """ - what's executed one an item (self.target) and one action (self.action) is selected. - Should be overwritten by the user - The result is expected to update self._data in this routine, else it is ignored - The basic code is useful for simple lists and dictionaries (key:value pairs, both strings) - """ - # TODO guarantee unicity - if isinstance(self._data,list): - if self.action == str(_('Add')): - self.target = TextInput(_('Add: '),None).run() - self._data.append(self.target) - if self.action == str(_('Copy')): - while True: - target = TextInput(_('Copy to: '),self.target).run() - if target != self.target: - self._data.append(self.target) - break - elif self.action == str(_('Edit')): - tgt = self.target - idx = self._data.index(self.target) - result = TextInput(_('Edit: '),tgt).run() - self._data[idx] = result - elif self.action == str(_('Delete')): - del self._data[self._data.index(self.target)] - elif isinstance(self._data,dict): - # allows overwrites - if self.target: - origkey,origval = list(self.target.items())[0] - else: - origkey = None - origval = None - if self.action == str(_('Add')): - key = TextInput(_('Key: '),None).run() - value = TextInput(_('Value: '),None).run() - self._data[key] = value - if self.action == str(_('Copy')): - while True: - key = TextInput(_('Copy to new key:'),origkey).run() - if key != origkey: - self._data[key] = origval - break - elif self.action == str(_('Edit')): - value = TextInput(_('Edit {}: ').format(origkey), origval).run() - self._data[origkey] = value - elif self.action == str(_('Delete')): - del self._data[origkey] - - return self._data + def handle_action(self, action: Any, entry: Optional[Any], data: List[Any]) -> List[Any]: + # this function is called when a base action or + # a specific action for an entry is triggered + raise NotImplementedError('Please implement me in the child class') diff --git a/archinstall/lib/user_interaction/manage_users_conf.py b/archinstall/lib/user_interaction/manage_users_conf.py index 33c31342..a97328c2 100644 --- a/archinstall/lib/user_interaction/manage_users_conf.py +++ b/archinstall/lib/user_interaction/manage_users_conf.py @@ -25,7 +25,7 @@ class UserList(ListManager): str(_('Promote/Demote user')), str(_('Delete User')) ] - super().__init__(prompt, lusers, self._actions, self._actions[0]) + super().__init__(prompt, lusers, [self._actions[0]], self._actions[1:]) def reformat(self, data: List[User]) -> Dict[str, User]: table = FormattedOutput.as_table(data) @@ -45,27 +45,25 @@ class UserList(ListManager): def selected_action_display(self, user: User) -> str: return user.username - def exec_action(self, data: List[User]) -> List[User]: - active_user = self.target if self.target else None - - if self.action == self._actions[0]: # add + def handle_action(self, action: str, entry: Optional[User], data: List[User]) -> List[User]: + if action == self._actions[0]: # add new_user = self._add_user() if new_user is not None: # in case a user with the same username as an existing user # was created we'll replace the existing one data = [d for d in data if d.username != new_user.username] data += [new_user] - elif self.action == self._actions[1]: # change password - prompt = str(_('Password for user "{}": ').format(active_user.username)) + elif action == self._actions[1]: # change password + prompt = str(_('Password for user "{}": ').format(entry.username)) new_password = get_password(prompt=prompt) if new_password: - user = next(filter(lambda x: x == active_user, data), 1) + user = next(filter(lambda x: x == entry, data), 1) user.password = new_password - elif self.action == self._actions[2]: # promote/demote - user = next(filter(lambda x: x == active_user, data), 1) + elif action == self._actions[2]: # promote/demote + user = next(filter(lambda x: x == entry, data), 1) user.sudo = False if user.sudo else True - elif self.action == self._actions[3]: # delete - data = [d for d in data if d != active_user] + elif action == self._actions[3]: # delete + data = [d for d in data if d != entry] return data diff --git a/archinstall/lib/user_interaction/network_conf.py b/archinstall/lib/user_interaction/network_conf.py index 4f22b790..1908603e 100644 --- a/archinstall/lib/user_interaction/network_conf.py +++ b/archinstall/lib/user_interaction/network_conf.py @@ -29,7 +29,7 @@ class ManualNetworkConfig(ListManager): str(_('Delete interface')) ] - super().__init__(prompt, ifaces, self._actions, self._actions[0]) + super().__init__(prompt, ifaces, [self._actions[0]], self._actions[1:]) def reformat(self, data: List[NetworkConfiguration]) -> Dict[str, Optional[NetworkConfiguration]]: table = FormattedOutput.as_table(data) @@ -49,21 +49,19 @@ class ManualNetworkConfig(ListManager): def selected_action_display(self, iface: NetworkConfiguration) -> str: return iface.iface if iface.iface else '' - def exec_action(self, data: List[NetworkConfiguration]): - active_iface: Optional[NetworkConfiguration] = self.target if self.target else None - - if self.action == self._actions[0]: # add + def handle_action(self, action: str, entry: Optional[NetworkConfiguration], data: List[NetworkConfiguration]): + if action == self._actions[0]: # add iface_name = self._select_iface(data) if iface_name: iface = NetworkConfiguration(NicType.MANUAL, iface=iface_name) iface = self._edit_iface(iface) data += [iface] - elif active_iface: - if self.action == self._actions[1]: # edit interface - data = [d for d in data if d.iface != active_iface.iface] - data.append(self._edit_iface(active_iface)) - elif self.action == self._actions[2]: # delete - data = [d for d in data if d != active_iface] + elif entry: + if action == self._actions[1]: # edit interface + data = [d for d in data if d.iface != entry.iface] + data.append(self._edit_iface(entry)) + elif action == self._actions[2]: # delete + data = [d for d in data if d != entry] return data diff --git a/archinstall/lib/user_interaction/partitioning_conf.py b/archinstall/lib/user_interaction/partitioning_conf.py index 0f4784b6..8bf76121 100644 --- a/archinstall/lib/user_interaction/partitioning_conf.py +++ b/archinstall/lib/user_interaction/partitioning_conf.py @@ -154,9 +154,6 @@ def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, block_device_struct = {"partitions": [partition.__dump__() for partition in block_device.partitions.values()]} original_layout = copy.deepcopy(block_device_struct) - # Test code: [part.__dump__() for part in block_device.partitions.values()] - # TODO: Squeeze in BTRFS subvolumes here - new_partition = str(_('Create a new partition')) suggest_partition_layout = str(_('Suggest partition layout')) delete_partition = str(_('Delete a partition')) @@ -209,6 +206,7 @@ def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, return original_layout elif task == save_and_exit: break + if task == new_partition: from ..disk import valid_parted_position @@ -222,8 +220,9 @@ def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, if fs_choice.type_ == MenuSelectionType.Esc: continue - prompt = _('Enter the start sector (percentage or block number, default: {}): ').format( - block_device.first_free_sector) + prompt = str(_('Enter the start sector (percentage or block number, default: {}): ')).format( + block_device.first_free_sector + ) start = input(prompt).strip() if not start.strip(): @@ -232,8 +231,9 @@ def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, else: end_suggested = '100%' - prompt = _('Enter the end sector of the partition (percentage or block number, ex: {}): ').format( - end_suggested) + prompt = str(_('Enter the end sector of the partition (percentage or block number, ex: {}): ')).format( + end_suggested + ) end = input(prompt).strip() if not end.strip(): diff --git a/archinstall/lib/user_interaction/subvolume_config.py b/archinstall/lib/user_interaction/subvolume_config.py index e2797bba..94150dee 100644 --- a/archinstall/lib/user_interaction/subvolume_config.py +++ b/archinstall/lib/user_interaction/subvolume_config.py @@ -12,13 +12,13 @@ if TYPE_CHECKING: class SubvolumeList(ListManager): - def __init__(self, prompt: str, current_volumes: List[Subvolume]): + def __init__(self, prompt: str, subvolumes: List[Subvolume]): self._actions = [ str(_('Add subvolume')), str(_('Edit subvolume')), str(_('Delete subvolume')) ] - super().__init__(prompt, current_volumes, self._actions, self._actions[0]) + super().__init__(prompt, subvolumes, [self._actions[0]], self._actions[1:]) def reformat(self, data: List[Subvolume]) -> Dict[str, Optional[Subvolume]]: table = FormattedOutput.as_table(data) @@ -75,13 +75,8 @@ class SubvolumeList(ListManager): return subvolume - def exec_action(self, data: List[Subvolume]) -> List[Subvolume]: - if self.target: - active_subvolume = self.target - else: - active_subvolume = None - - if self.action == self._actions[0]: # add + def handle_action(self, action: str, entry: Optional[Subvolume], data: List[Subvolume]) -> List[Subvolume]: + if action == self._actions[0]: # add new_subvolume = self._add_subvolume() if new_subvolume is not None: @@ -89,14 +84,15 @@ class SubvolumeList(ListManager): # was created we'll replace the existing one data = [d for d in data if d.name != new_subvolume.name] data += [new_subvolume] - elif self.action == self._actions[1]: # edit subvolume - new_subvolume = self._add_subvolume(active_subvolume) - - if new_subvolume is not None: - # we'll remove the original subvolume and add the modified version - data = [d for d in data if d.name != active_subvolume.name and d.name != new_subvolume.name] - data += [new_subvolume] - elif self.action == self._actions[2]: # delete - data = [d for d in data if d != active_subvolume] + elif entry is not None: + if action == self._actions[1]: # edit subvolume + new_subvolume = self._add_subvolume(entry) + + if new_subvolume is not None: + # we'll remove the original subvolume and add the modified version + data = [d for d in data if d.name != entry.name and d.name != new_subvolume.name] + data += [new_subvolume] + elif action == self._actions[2]: # delete + data = [d for d in data if d != entry] return data -- cgit v1.2.3-70-g09d2 From 9194f6d85965f435f8d0ae44ba20e73cc761eb44 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Tue, 26 Jul 2022 18:46:50 +1000 Subject: Cleanup partition (#1333) * Cleanup partition * Update * Remove unused method * Update partitioning * Update * Update * Fix mypy Co-authored-by: Daniel Girtler --- archinstall/lib/disk/blockdevice.py | 10 +- archinstall/lib/disk/btrfs/btrfspartition.py | 19 +- archinstall/lib/disk/filesystem.py | 26 +- archinstall/lib/disk/helpers.py | 10 +- archinstall/lib/disk/partition.py | 429 +++++++++------------ archinstall/lib/luks.py | 6 +- .../lib/user_interaction/partitioning_conf.py | 10 - 7 files changed, 227 insertions(+), 283 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/archinstall/lib/disk/blockdevice.py b/archinstall/lib/disk/blockdevice.py index 4e207bf4..736bacbc 100644 --- a/archinstall/lib/disk/blockdevice.py +++ b/archinstall/lib/disk/blockdevice.py @@ -88,9 +88,6 @@ class BlockDevice: raise KeyError(f'{self.info} does not contain information: "{key}"') - def __len__(self) -> int: - return len(self.partitions) - def __lt__(self, left_comparitor :'BlockDevice') -> bool: return self._path < left_comparitor.path @@ -121,6 +118,8 @@ class BlockDevice: def _load_partitions(self): from .partition import Partition + self._partitions.clear() + lsblk_info = self._call_lsblk(self._path) device = lsblk_info['blockdevices'][0] self._partitions.clear() @@ -233,8 +232,6 @@ class BlockDevice: @property def partitions(self) -> Dict[str, 'Partition']: - self._partprobe() - self._load_partitions() return OrderedDict(sorted(self._partitions.items())) @property @@ -282,7 +279,7 @@ class BlockDevice: try: if uuid and partition.uuid and partition.uuid.lower() == uuid.lower(): return partition - elif partuuid and partition.part_uuid.lower() == partuuid.lower(): + elif partuuid and partition.part_uuid and partition.part_uuid.lower() == partuuid.lower(): return partition except DiskError as error: # Most likely a blockdevice that doesn't support or use UUID's @@ -291,6 +288,7 @@ class BlockDevice: pass log(f"uuid {uuid} or {partuuid} not found. Waiting {storage.get('DISK_TIMEOUTS', 1) * count}s for next attempt",level=logging.DEBUG) + self.flush_cache() time.sleep(storage.get('DISK_TIMEOUTS', 1) * count) log(f"Could not find {uuid}/{partuuid} in disk after 5 retries", level=logging.INFO) diff --git a/archinstall/lib/disk/btrfs/btrfspartition.py b/archinstall/lib/disk/btrfs/btrfspartition.py index a05f1527..d04c9b98 100644 --- a/archinstall/lib/disk/btrfs/btrfspartition.py +++ b/archinstall/lib/disk/btrfs/btrfspartition.py @@ -17,22 +17,11 @@ if TYPE_CHECKING: from ...installer import Installer from .btrfssubvolumeinfo import BtrfsSubvolumeInfo + class BTRFSPartition(Partition): def __init__(self, *args, **kwargs): Partition.__init__(self, *args, **kwargs) - def __repr__(self, *args :str, **kwargs :str) -> str: - mount_repr = '' - if self.mountpoint: - mount_repr = f", mounted={self.mountpoint}" - elif self.target_mountpoint: - mount_repr = f", rel_mountpoint={self.target_mountpoint}" - - if self._encrypted: - return f'BTRFSPartition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, parent={self.real_device}, fs={self.filesystem}{mount_repr})' - else: - return f'BTRFSPartition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, fs={self.filesystem}{mount_repr})' - @property def subvolumes(self): for filesystem in findmnt(pathlib.Path(self.path), recurse=True).get('filesystems', []): @@ -40,11 +29,11 @@ class BTRFSPartition(Partition): yield subvolume_info_from_path(filesystem['target']) def iterate_children(struct): - for child in struct.get('children', []): + for c in struct.get('children', []): if '[' in child.get('source', ''): - yield subvolume_info_from_path(child['target']) + yield subvolume_info_from_path(c['target']) - for sub_child in iterate_children(child): + for sub_child in iterate_children(c): yield sub_child for child in iterate_children(filesystem): diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py index 1c7a801b..90656308 100644 --- a/archinstall/lib/disk/filesystem.py +++ b/archinstall/lib/disk/filesystem.py @@ -210,7 +210,14 @@ class Filesystem: # TODO: Implement this with declarative profiles instead. raise ValueError("Installation().use_entire_disk() has to be re-worked.") - def add_partition(self, partition_type :str, start :str, end :str, partition_format :Optional[str] = None, skip_mklabel :bool = False) -> Partition: + def add_partition( + self, + partition_type :str, + start :str, + end :str, + partition_format :Optional[str] = None, + skip_mklabel :bool = False + ) -> Partition: log(f'Adding partition to {self.blockdevice}, {start}->{end}', level=logging.INFO) if len(self.blockdevice.partitions) == 0 and skip_mklabel is False: @@ -232,6 +239,7 @@ class Filesystem: except DiskError: pass + # TODO this check should probably run in the setup process rather than during the installation if self.mode == MBR: if len(self.blockdevice.partitions) > 3: DiskError("Too many partitions on disk, MBR disks can only have 3 primary partitions") @@ -246,14 +254,9 @@ class Filesystem: if self.parted(parted_string): for count in range(storage.get('DISK_RETRY_ATTEMPTS', 3)): self.partprobe() + self.blockdevice.flush_cache() - new_partition_uuids = [] - for partition in self.blockdevice.partitions.values(): - try: - new_partition_uuids.append(partition.part_uuid) - except DiskError: - pass - + new_partition_uuids = [partition.part_uuid for partition in self.blockdevice.partitions.values()] new_partuuid_set = (set(previous_partuuids) ^ set(new_partition_uuids)) if len(new_partuuid_set) and (new_partuuid := new_partuuid_set.pop()): @@ -263,17 +266,20 @@ class Filesystem: log(f'Blockdevice: {self.blockdevice}', level=logging.ERROR, fg="red") log(f'Partitions: {self.blockdevice.partitions}', level=logging.ERROR, fg="red") log(f'Partition set: {new_partuuid_set}', level=logging.ERROR, fg="red") - log(f'New UUID: {[new_partuuid]}', level=logging.ERROR, fg="red") + log(f'New PARTUUID: {[new_partuuid]}', level=logging.ERROR, fg="red") log(f'get_partition(): {self.blockdevice.get_partition}', level=logging.ERROR, fg="red") raise err else: log(f"Could not get UUID for partition. Waiting {storage.get('DISK_TIMEOUTS', 1) * count}s before retrying.",level=logging.DEBUG) time.sleep(storage.get('DISK_TIMEOUTS', 1) * count) + total_partitions = set([partition.part_uuid for partition in self.blockdevice.partitions.values()]) + total_partitions.update(previous_partuuids) + # TODO: This should never be able to happen log(f"Could not find the new PARTUUID after adding the partition.", level=logging.ERROR, fg="red") log(f"Previous partitions: {previous_partuuids}", level=logging.ERROR, fg="red") - log(f"New partitions: {(previous_partuuids ^ {partition.part_uuid for partition in self.blockdevice.partitions.values()})}", level=logging.ERROR, fg="red") + log(f"New partitions: {total_partitions}", level=logging.ERROR, fg="red") raise DiskError(f"Could not add partition using: {parted_string}") def set_name(self, partition: int, name: str) -> bool: diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py index 660594ed..c8ac564e 100644 --- a/archinstall/lib/disk/helpers.py +++ b/archinstall/lib/disk/helpers.py @@ -370,7 +370,7 @@ def get_all_targets(data :Dict[str, Any], filters :Dict[str, None] = {}) -> Dict return filters -def get_partitions_in_use(mountpoint :str) -> List[Partition]: +def get_partitions_in_use(mountpoint :str) -> Dict[str, Any]: from .partition import Partition try: @@ -393,8 +393,12 @@ def get_partitions_in_use(mountpoint :str) -> List[Partition]: if not type(blockdev) in (Partition, MapperDev): continue - for blockdev_mountpoint in blockdev.mount_information: - block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev + if isinstance(blockdev, Partition): + for blockdev_mountpoint in blockdev.mountpoints: + block_devices_mountpoints[blockdev_mountpoint] = blockdev + else: + for blockdev_mountpoint in blockdev.mount_information: + block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev log(f'Filtering available mounts {block_devices_mountpoints} to those under {mountpoint}', level=logging.DEBUG) diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 17c24d57..4028f114 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -1,14 +1,16 @@ import glob -import pathlib import time import logging import json import os import hashlib +import typing +from dataclasses import dataclass +from pathlib import Path from typing import Optional, Dict, Any, List, Union, Iterator from .blockdevice import BlockDevice -from .helpers import find_mountpoint, get_filesystem_type, convert_size_to_gb, split_bind_name +from .helpers import get_filesystem_type, convert_size_to_gb, split_bind_name from ..storage import storage from ..exceptions import DiskError, SysCallError, UnknownFilesystemFormat from ..output import log @@ -16,6 +18,26 @@ from ..general import SysCommand from .btrfs.btrfs_helpers import subvolume_info_from_path from .btrfs.btrfssubvolumeinfo import BtrfsSubvolumeInfo + +@dataclass +class PartitionInfo: + pttype: str + partuuid: str + uuid: str + start: Optional[int] + end: Optional[int] + bootable: bool + size: float + sector_size: int + filesystem_type: str + mountpoints: List[Path] + + def get_first_mountpoint(self) -> Optional[Path]: + if len(self.mountpoints) > 0: + return self.mountpoints[0] + return None + + class Partition: def __init__( self, @@ -25,38 +47,37 @@ class Partition: filesystem :Optional[str] = None, mountpoint :Optional[str] = None, encrypted :bool = False, - autodetect_filesystem :bool = True + autodetect_filesystem :bool = True, ): - if not part_id: part_id = os.path.basename(path) - self.block_device = block_device - if type(self.block_device) is str: + if type(block_device) is str: raise ValueError(f"Partition()'s 'block_device' parameter has to be a archinstall.BlockDevice() instance!") - self.path = path - self.part_id = part_id - self.target_mountpoint = mountpoint - self.filesystem = filesystem + self.block_device = block_device + self._path = path + self._part_id = part_id + self._target_mountpoint = mountpoint self._encrypted = None - self.encrypted = encrypted - self.allow_formatting = False + self._encrypted = encrypted + self._wipe = False + self._type = 'primary' if mountpoint: self.mount(mountpoint) - try: - self.mount_information = list(find_mountpoint(self.path)) - except DiskError: - self.mount_information = [{}] + self._partition_info = self._fetch_information() - if not self.filesystem and autodetect_filesystem: - self.filesystem = get_filesystem_type(path) + if not autodetect_filesystem and filesystem: + self._partition_info.filesystem_type = filesystem - if self.filesystem == 'crypto_LUKS': - self.encrypted = True + if self._partition_info.filesystem_type == 'crypto_LUKS': + self._encrypted = True + # I hate doint this but I'm currently unsure where this + # is acutally used to be able to fix the typing issues properly + @typing.no_type_check def __lt__(self, left_comparitor :BlockDevice) -> bool: if type(left_comparitor) == Partition: left_comparitor = left_comparitor.path @@ -64,254 +85,191 @@ class Partition: left_comparitor = str(left_comparitor) # The goal is to check if /dev/nvme0n1p1 comes before /dev/nvme0n1p5 - return self.path < left_comparitor + return self._path < left_comparitor def __repr__(self, *args :str, **kwargs :str) -> str: mount_repr = '' - if self.mountpoint: - mount_repr = f", mounted={self.mountpoint}" - elif self.target_mountpoint: - mount_repr = f", rel_mountpoint={self.target_mountpoint}" + if mountpoint := self._partition_info.get_first_mountpoint(): + mount_repr = f", mounted={mountpoint}" + elif self._target_mountpoint: + mount_repr = f", rel_mountpoint={self._target_mountpoint}" + + classname = self.__class__.__name__ if self._encrypted: - return f'Partition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, parent={self.real_device}, fs={self.filesystem}{mount_repr})' + return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, parent={self.real_device}, fs={self._partition_info.filesystem_type}{mount_repr})' else: - return f'Partition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, fs={self.filesystem}{mount_repr})' + return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, fs={self._partition_info.filesystem_type}{mount_repr})' def as_json(self) -> Dict[str, Any]: """ this is used for the table representation of the partition (see FormattedOutput) """ partition_info = { - 'type': 'primary', - 'PARTUUID': self._safe_uuid, - 'wipe': self.allow_formatting, + 'type': self._type, + 'PARTUUID': self.part_uuid, + 'wipe': self._wipe, 'boot': self.boot, 'ESP': self.boot, - 'mountpoint': self.target_mountpoint, + 'mountpoint': self._target_mountpoint, 'encrypted': self._encrypted, 'start': self.start, 'size': self.end, - 'filesystem': self.filesystem_type + 'filesystem': self._partition_info.filesystem_type } return partition_info def __dump__(self) -> Dict[str, Any]: # TODO remove this in favour of as_json - - log(get_filesystem_type(self.path)) - return { - 'type': 'primary', - 'PARTUUID': self._safe_uuid, - 'wipe': self.allow_formatting, + 'type': self._type, + 'PARTUUID': self.part_uuid, + 'wipe': self._wipe, 'boot': self.boot, 'ESP': self.boot, - 'mountpoint': self.target_mountpoint, + 'mountpoint': self._target_mountpoint, 'encrypted': self._encrypted, 'start': self.start, 'size': self.end, 'filesystem': { - 'format': self.filesystem_type + 'format': self._partition_info.filesystem_type } } - @property - def filesystem_type(self) -> Optional[str]: - return get_filesystem_type(self.path) + def _call_lsblk(self) -> Dict[str, Any]: + self.partprobe() + output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8') - @property - def mountpoint(self) -> Optional[str]: - try: - data = json.loads(SysCommand(f"findmnt --json -R {self.path}").decode()) - for filesystem in data['filesystems']: - return pathlib.Path(filesystem.get('target')) + if output: + lsblk_info = json.loads(output) + return lsblk_info - except SysCallError as error: - # Not mounted anywhere most likely - log(f"Could not locate mount information for {self.path}: {error}", level=logging.DEBUG, fg="grey") - pass - - return None + raise DiskError(f'Failed to read disk "{self.device_path}" with lsblk') - @property - def sector_size(self) -> Optional[int]: - output = json.loads(SysCommand(f"lsblk --json -o+LOG-SEC {self.device_path}").decode('UTF-8')) + def _call_sfdisk(self) -> Dict[str, Any]: + output = SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8') - for device in output['blockdevices']: - return device.get('log-sec', None) + if output: + sfdisk_info = json.loads(output) + partitions = sfdisk_info.get('partitiontable', {}).get('partitions', []) + node = list(filter(lambda x: x['node'] == self._path, partitions)) - @property - def start(self) -> Optional[str]: - output = json.loads(SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')) + if len(node) > 0: + return node[0] - for partition in output.get('partitiontable', {}).get('partitions', []): - if partition['node'] == self.path: - return partition['start'] # * self.sector_size + return {} - @property - def end(self) -> Optional[str]: - # TODO: actually this is size in sectors unit - # TODO: Verify that the logic holds up, that 'size' is the size without 'start' added to it. - output = json.loads(SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')) + raise DiskError(f'Failed to read disk "{self.block_device.path}" with sfdisk') - for partition in output.get('partitiontable', {}).get('partitions', []): - if partition['node'] == self.path: - return partition['size'] # * self.sector_size + def _fetch_information(self) -> PartitionInfo: + lsblk_info = self._call_lsblk() + sfdisk_info = self._call_sfdisk() + device = lsblk_info['blockdevices'][0] - @property - def end_sectors(self) -> Optional[str]: - output = json.loads(SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')) + mountpoints = [Path(mountpoint) for mountpoint in device['mountpoints'] if mountpoint] + bootable = sfdisk_info.get('bootable', False) or sfdisk_info.get('type', '') == 'C12A7328-F81F-11D2-BA4B-00A0C93EC93B' - for partition in output.get('partitiontable', {}).get('partitions', []): - if partition['node'] == self.path: - return partition['start'] + partition['size'] + return PartitionInfo( + pttype=device['pttype'], + partuuid=device['partuuid'], + uuid=device['uuid'], + sector_size=device['log-sec'], + size=convert_size_to_gb(device['size']), + start=sfdisk_info.get('start', None), + end=sfdisk_info.get('size', None), + bootable=bootable, + filesystem_type=device['fstype'], + mountpoints=mountpoints + ) @property - def size(self) -> Optional[float]: - for i in range(storage['DISK_RETRY_ATTEMPTS']): - self.partprobe() - time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * i)) - - try: - lsblk = json.loads(SysCommand(f"lsblk --json -b -o+SIZE {self.device_path}").decode()) - - for device in lsblk['blockdevices']: - return convert_size_to_gb(device['size']) - except SysCallError as error: - if error.exit_code == 8192: - return None - else: - raise error + def target_mountpoint(self) -> Optional[str]: + return self._target_mountpoint @property - def boot(self) -> bool: - output = json.loads(SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8')) - - for partition in output.get('partitiontable', {}).get('partitions', []): - if partition['node'] == self.path: - # first condition is for MBR disks, second for GPT disks - return partition.get('bootable', False) or partition.get('type','') == 'C12A7328-F81F-11D2-BA4B-00A0C93EC93B' - - return False + def path(self) -> str: + return self._path @property - def partition_type(self) -> Optional[str]: - lsblk = json.loads(SysCommand(f"lsblk --json -o+PTTYPE {self.device_path}").decode('UTF-8')) - - for device in lsblk['blockdevices']: - return device['pttype'] + def filesystem(self) -> str: + return self._partition_info.filesystem_type @property - def part_uuid(self) -> str: - """ - Returns the PARTUUID as returned by lsblk. - This is more reliable than relying on /dev/disk/by-partuuid as - it doesn't seam to be able to detect md raid partitions. - For bind mounts all the subvolumes share the same uuid - """ - for i in range(storage['DISK_RETRY_ATTEMPTS']): - if not self.partprobe(): - raise DiskError(f"Could not perform partprobe on {self.device_path}") - - time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * i)) - - partuuid = self._safe_part_uuid - if partuuid: - return partuuid - - raise DiskError(f"Could not get PARTUUID for {self.path} using 'blkid -s PARTUUID -o value {self.path}'") + def mountpoint(self) -> Optional[Path]: + if len(self.mountpoints) > 0: + return self.mountpoints[0] + return None @property - def uuid(self) -> Optional[str]: - """ - Returns the UUID as returned by lsblk for the **partition**. - This is more reliable than relying on /dev/disk/by-uuid as - it doesn't seam to be able to detect md raid partitions. - For bind mounts all the subvolumes share the same uuid - """ - for i in range(storage['DISK_RETRY_ATTEMPTS']): - if not self.partprobe(): - raise DiskError(f"Could not perform partprobe on {self.device_path}") + def mountpoints(self) -> List[Path]: + return self._partition_info.mountpoints - time.sleep(storage.get('DISK_TIMEOUTS', 1) * i) - - partuuid = self._safe_uuid - if partuuid: - return partuuid - - raise DiskError(f"Could not get PARTUUID for {self.path} using 'blkid -s PARTUUID -o value {self.path}'") + @property + def sector_size(self) -> int: + return self._partition_info.sector_size @property - def _safe_uuid(self) -> Optional[str]: - """ - A near copy of self.uuid but without any delays. - This function should only be used where uuid is not crucial. - For instance when you want to get a __repr__ of the class. - """ - if not self.partprobe(): - if self.block_device.partition_type == 'iso9660': - return None + def start(self) -> Optional[int]: + return self._partition_info.start - log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) + @property + def end(self) -> Optional[int]: + return self._partition_info.end - try: - return SysCommand(f'blkid -s UUID -o value {self.device_path}').decode('UTF-8').strip() - except SysCallError as error: - if self.block_device.partition_type == 'iso9660': - # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) - return None + @property + def end_sectors(self) -> Optional[int]: + start = self._partition_info.start + end = self._partition_info.end + if start and end: + return start + end + return None - log(f"Could not get PARTUUID of partition using 'blkid -s UUID -o value {self.device_path}': {error}") + @property + def size(self) -> Optional[float]: + return self._partition_info.size @property - def _safe_part_uuid(self) -> Optional[str]: - """ - A near copy of self.uuid but without any delays. - This function should only be used where uuid is not crucial. - For instance when you want to get a __repr__ of the class. - """ - if not self.partprobe(): - if self.block_device.partition_type == 'iso9660': - return None + def boot(self) -> bool: + return self._partition_info.bootable - log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) + @property + def partition_type(self) -> Optional[str]: + return self._partition_info.pttype - try: - return self.block_device.uuid - except SysCallError as error: - if self.block_device.partition_type == 'iso9660': - # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) - return None + @property + def part_uuid(self) -> str: + return self._partition_info.partuuid - log(f"Could not get PARTUUID of partition using 'blkid -s PARTUUID -o value {self.device_path}': {error}") + @property + def uuid(self) -> Optional[str]: + return self._partition_info.uuid @property def encrypted(self) -> Union[bool, None]: return self._encrypted - @encrypted.setter - def encrypted(self, value: bool) -> None: - self._encrypted = value - @property def parent(self) -> str: return self.real_device @property def real_device(self) -> str: - for blockdevice in json.loads(SysCommand('lsblk -J').decode('UTF-8'))['blockdevices']: - if parent := self.find_parent_of(blockdevice, os.path.basename(self.device_path)): - return f"/dev/{parent}" - # raise DiskError(f'Could not find appropriate parent for encrypted partition {self}') - return self.path + output = SysCommand('lsblk -J').decode('UTF-8') + + if output: + for blockdevice in json.loads(output)['blockdevices']: + if parent := self.find_parent_of(blockdevice, os.path.basename(self.device_path)): + return f"/dev/{parent}" + return self._path + + raise DiskError('Unable to get disk information for command "lsblk -J"') @property def device_path(self) -> str: - """ for bind mounts returns the phisical path of the partition + """ for bind mounts returns the physical path of the partition """ - device_path, bind_name = split_bind_name(self.path) + device_path, bind_name = split_bind_name(self._path) return device_path @property @@ -319,7 +277,7 @@ class Partition: """ for bind mounts returns the bind name (subvolume path). Returns none if this property does not exist """ - device_path, bind_name = split_bind_name(self.path) + device_path, bind_name = split_bind_name(self._path) return bind_name @property @@ -330,29 +288,29 @@ class Partition: for child in information.get('children', []): if target := child.get('target'): if child.get('fstype') == 'btrfs': - if subvolume := subvolume_info_from_path(pathlib.Path(target)): + if subvolume := subvolume_info_from_path(Path(target)): yield subvolume if child.get('children'): for subchild in iterate_children_recursively(child): yield subchild - for mountpoint in self.mount_information: - if result := findmnt(pathlib.Path(mountpoint['target'])): - for filesystem in result.get('filesystems', []): - if mountpoint.get('fstype') == 'btrfs': - if subvolume := subvolume_info_from_path(pathlib.Path(mountpoint['target'])): + if self._partition_info.filesystem_type == 'btrfs': + for mountpoint in self._partition_info.mountpoints: + if result := findmnt(mountpoint): + for filesystem in result.get('filesystems', []): + if subvolume := subvolume_info_from_path(mountpoint): yield subvolume - for child in iterate_children_recursively(filesystem): - yield child + for child in iterate_children_recursively(filesystem): + yield child def partprobe(self) -> bool: try: if self.block_device: return 0 == SysCommand(f'partprobe {self.block_device.device}').exit_code except SysCallError as error: - log(f"Unreliable results might be given for {self.path} due to partprobe error: {error}", level=logging.DEBUG) + log(f"Unreliable results might be given for {self._path} due to partprobe error: {error}", level=logging.DEBUG) return False @@ -364,19 +322,20 @@ class Partition: with luks2(self, storage.get('ENC_IDENTIFIER', 'ai') + 'loop', password, auto_unmount=True) as unlocked_device: return unlocked_device.filesystem except SysCallError: - return None + pass + return None def has_content(self) -> bool: - fs_type = get_filesystem_type(self.path) + fs_type = self._partition_info.filesystem_type if not fs_type or "swap" in fs_type: return False temporary_mountpoint = '/tmp/' + hashlib.md5(bytes(f"{time.time()}", 'UTF-8') + os.urandom(12)).hexdigest() - temporary_path = pathlib.Path(temporary_mountpoint) + temporary_path = Path(temporary_mountpoint) temporary_path.mkdir(parents=True, exist_ok=True) - if (handle := SysCommand(f'/usr/bin/mount {self.path} {temporary_mountpoint}')).exit_code != 0: - raise DiskError(f'Could not mount and check for content on {self.path} because: {b"".join(handle)}') + if (handle := SysCommand(f'/usr/bin/mount {self._path} {temporary_mountpoint}')).exit_code != 0: + raise DiskError(f'Could not mount and check for content on {self._path} because: {handle}') files = len(glob.glob(f"{temporary_mountpoint}/*")) iterations = 0 @@ -387,14 +346,14 @@ class Partition: return True if files > 0 else False - def encrypt(self, *args :str, **kwargs :str) -> str: + def encrypt(self, password: Optional[str] = None) -> str: """ A wrapper function for luks2() instances and the .encrypt() method of that instance. """ from ..luks import luks2 handle = luks2(self, None, None) - return handle.encrypt(self, *args, **kwargs) + return handle.encrypt(self, password=password) def format(self, filesystem :Optional[str] = None, path :Optional[str] = None, log_formatting :bool = True, options :List[str] = [], retry :bool = True) -> bool: """ @@ -402,17 +361,17 @@ class Partition: the formatting functionality and in essence the support for the given filesystem. """ if filesystem is None: - filesystem = self.filesystem + filesystem = self._partition_info.filesystem_type if path is None: - path = self.path + path = self._path # This converts from fat32 -> vfat to unify filesystem names filesystem = get_mount_fs_type(filesystem) # To avoid "unable to open /dev/x: No such file or directory" start_wait = time.time() - while pathlib.Path(path).exists() is False and time.time() - start_wait < 10: + while Path(path).exists() is False and time.time() - start_wait < 10: time.sleep(0.025) if log_formatting: @@ -422,57 +381,57 @@ class Partition: if filesystem == 'btrfs': options = ['-f'] + options - if 'UUID:' not in (mkfs := SysCommand(f"/usr/bin/mkfs.btrfs {' '.join(options)} {path}").decode('UTF-8')): + mkfs = SysCommand(f"/usr/bin/mkfs.btrfs {' '.join(options)} {path}").decode('UTF-8') + if mkfs and 'UUID:' not in mkfs: raise DiskError(f'Could not format {path} with {filesystem} because: {mkfs}') - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'vfat': options = ['-F32'] + options log(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}") if (handle := SysCommand(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'ext4': options = ['-F'] + options if (handle := SysCommand(f"/usr/bin/mkfs.ext4 {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'ext2': options = ['-F'] + options if (handle := SysCommand(f"/usr/bin/mkfs.ext2 {' '.join(options)} {path}")).exit_code != 0: - raise DiskError(f'Could not format {path} with {filesystem} because: {b"".join(handle)}') - self.filesystem = 'ext2' - + raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") + self._partition_info.filesystem_type = 'ext2' elif filesystem == 'xfs': options = ['-f'] + options if (handle := SysCommand(f"/usr/bin/mkfs.xfs {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'f2fs': options = ['-f'] + options if (handle := SysCommand(f"/usr/bin/mkfs.f2fs {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'ntfs3': options = ['-f'] + options if (handle := SysCommand(f"/usr/bin/mkfs.ntfs -Q {' '.join(options)} {path}")).exit_code != 0: raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}") - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem elif filesystem == 'crypto_LUKS': # from ..luks import luks2 # encrypted_partition = luks2(self, None, None) # encrypted_partition.format(path) - self.filesystem = filesystem + self._partition_info.filesystem_type = filesystem else: raise UnknownFilesystemFormat(f"Fileformat '{filesystem}' is not yet implemented.") @@ -485,9 +444,9 @@ class Partition: return self.format(filesystem, path, log_formatting, options, retry=False) if get_filesystem_type(path) == 'crypto_LUKS' or get_filesystem_type(self.real_device) == 'crypto_LUKS': - self.encrypted = True + self._encrypted = True else: - self.encrypted = False + self._encrypted = False return True @@ -499,18 +458,18 @@ class Partition: if parent := self.find_parent_of(child, name, parent=data['name']): return parent + return None + def mount(self, target :str, fs :Optional[str] = None, options :str = '') -> bool: - if not self.mountpoint: + if not self._partition_info.get_first_mountpoint(): log(f'Mounting {self} to {target}', level=logging.INFO) if not fs: - if not self.filesystem: - raise DiskError(f'Need to format (or define) the filesystem on {self} before mounting.') - fs = self.filesystem + fs = self._partition_info.filesystem_type fs_type = get_mount_fs_type(fs) - pathlib.Path(target).mkdir(parents=True, exist_ok=True) + Path(target).mkdir(parents=True, exist_ok=True) if self.bind_name: device_path = self.device_path @@ -520,7 +479,7 @@ class Partition: else: options = f"subvol={self.bind_name}" else: - device_path = self.path + device_path = self._path try: if options: mnt_handle = SysCommand(f"/usr/bin/mount -t {fs_type} -o {options} {device_path} {target}") @@ -529,7 +488,7 @@ class Partition: # TODO: Should be redundant to check for exit_code if mnt_handle.exit_code != 0: - raise DiskError(f"Could not mount {self.path} to {target} using options {options}") + raise DiskError(f"Could not mount {self._path} to {target} using options {options}") except SysCallError as err: raise err @@ -538,19 +497,17 @@ class Partition: return False def unmount(self) -> bool: - worker = SysCommand(f"/usr/bin/umount {self.path}") + worker = SysCommand(f"/usr/bin/umount {self._path}") + exit_code = worker.exit_code # Without to much research, it seams that low error codes are errors. # And above 8k is indicators such as "/dev/x not mounted.". # So anything in between 0 and 8k are errors (?). - if 0 < worker.exit_code < 8000: - raise SysCallError(f"Could not unmount {self.path} properly: {worker}", exit_code=worker.exit_code) + if exit_code and 0 < exit_code < 8000: + raise SysCallError(f"Could not unmount {self._path} properly: {worker}", exit_code=exit_code) return True - def umount(self) -> bool: - return self.unmount() - def filesystem_supported(self) -> bool: """ The support for a filesystem (this partition) is tested by calling @@ -559,7 +516,7 @@ class Partition: 2. UnknownFilesystemFormat that indicates that we don't support the given filesystem type """ try: - self.format(self.filesystem, '/dev/null', log_formatting=False, allow_formatting=True) + self.format(self._partition_info.filesystem_type, '/dev/null', log_formatting=False) except (SysCallError, DiskError): pass # We supported it, but /dev/null is not formattable as expected so the mkfs call exited with an error code except UnknownFilesystemFormat as err: diff --git a/archinstall/lib/luks.py b/archinstall/lib/luks.py index ac480b11..7e4534d8 100644 --- a/archinstall/lib/luks.py +++ b/archinstall/lib/luks.py @@ -22,9 +22,9 @@ from .disk.btrfs import BTRFSPartition class luks2: def __init__(self, - partition :Partition, - mountpoint :str, - password :str, + partition: Partition, + mountpoint: Optional[str], + password: Optional[str], key_file :Optional[str] = None, auto_unmount :bool = False, *args :str, diff --git a/archinstall/lib/user_interaction/partitioning_conf.py b/archinstall/lib/user_interaction/partitioning_conf.py index 8bf76121..f2e6b881 100644 --- a/archinstall/lib/user_interaction/partitioning_conf.py +++ b/archinstall/lib/user_interaction/partitioning_conf.py @@ -140,16 +140,6 @@ def get_default_partition_layout( return suggest_multi_disk_layout(block_devices, advanced_options=advanced_options) -def select_individual_blockdevice_usage(block_devices: list) -> Dict[str, Any]: - result = {} - - for device in block_devices: - layout = manage_new_and_existing_partitions(device) - result[device.path] = layout - - return result - - def manage_new_and_existing_partitions(block_device: 'BlockDevice') -> Dict[str, Any]: # noqa: max-complexity: 50 block_device_struct = {"partitions": [partition.__dump__() for partition in block_device.partitions.values()]} original_layout = copy.deepcopy(block_device_struct) -- cgit v1.2.3-70-g09d2 From 3da03a192e3dc47c0e0c08302d28e9f3a62bcd0f Mon Sep 17 00:00:00 2001 From: Werner Llácer Date: Mon, 1 Aug 2022 10:26:51 +0200 Subject: Solves issue 1343. Could not locate partition after creation (#1355) * Solves issue 1343. Could not locate partition after creation * Added some flake fixes. Co-authored-by: Anton Hvornum --- archinstall/lib/disk/partition.py | 62 +++++++++++++++++++++++++++++++++++++++ archinstall/lib/general.py | 2 +- archinstall/lib/plugins.py | 2 +- 3 files changed, 64 insertions(+), 2 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 4028f114..f70bf907 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -243,6 +243,68 @@ class Partition: @property def uuid(self) -> Optional[str]: + """ + Returns the UUID as returned by lsblk for the **partition**. + This is more reliable than relying on /dev/disk/by-uuid as + it doesn't seam to be able to detect md raid partitions. + For bind mounts all the subvolumes share the same uuid + """ + for i in range(storage['DISK_RETRY_ATTEMPTS']): + if not self.partprobe(): + raise DiskError(f"Could not perform partprobe on {self.device_path}") + + time.sleep(storage.get('DISK_TIMEOUTS', 1) * i) + + partuuid = self._safe_uuid + if partuuid: + return partuuid + + raise DiskError(f"Could not get PARTUUID for {self.path} using 'blkid -s PARTUUID -o value {self.path}'") + + @property + def _safe_uuid(self) -> Optional[str]: + """ + A near copy of self.uuid but without any delays. + This function should only be used where uuid is not crucial. + For instance when you want to get a __repr__ of the class. + """ + if not self.partprobe(): + if self.block_device.partition_type == 'iso9660': + return None + + log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) + + try: + return SysCommand(f'blkid -s UUID -o value {self.device_path}').decode('UTF-8').strip() + except SysCallError as error: + if self.block_device.partition_type == 'iso9660': + # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) + return None + + log(f"Could not get PARTUUID of partition using 'blkid -s UUID -o value {self.device_path}': {error}") + + @property + def _safe_part_uuid(self) -> Optional[str]: + """ + A near copy of self.uuid but without any delays. + This function should only be used where uuid is not crucial. + For instance when you want to get a __repr__ of the class. + """ + if not self.partprobe(): + if self.block_device.partition_type == 'iso9660': + return None + + log(f"Could not reliably refresh PARTUUID of partition {self.device_path} due to partprobe error.", level=logging.DEBUG) + + try: + return SysCommand(f'blkid -s PARTUUID -o value {self.device_path}').decode('UTF-8').strip() + except SysCallError as error: + if self.block_device.partition_type == 'iso9660': + # Parent device is a Optical Disk (.iso dd'ed onto a device for instance) + return None + + log(f"Could not get PARTUUID of partition using 'blkid -s PARTUUID -o value {self.device_path}': {error}") + return self._partition_info.uuid @property diff --git a/archinstall/lib/general.py b/archinstall/lib/general.py index 3ec1d685..27f444e8 100644 --- a/archinstall/lib/general.py +++ b/archinstall/lib/general.py @@ -37,7 +37,7 @@ else: def unregister(self, fileno :int, *args :List[Any], **kwargs :Dict[str, Any]) -> None: try: - del(self.monitoring[fileno]) + del(self.monitoring[fileno]) # noqa: E275 except: pass diff --git a/archinstall/lib/plugins.py b/archinstall/lib/plugins.py index 99e3811c..f771aacb 100644 --- a/archinstall/lib/plugins.py +++ b/archinstall/lib/plugins.py @@ -60,7 +60,7 @@ def import_via_path(path :str, namespace :Optional[str] = None) -> ModuleType: log(f"The above error was detected when loading the plugin: {path}", fg="red", level=logging.ERROR) try: - del(sys.modules[namespace]) + del(sys.modules[namespace]) # noqa: E275 except: pass -- cgit v1.2.3-70-g09d2 From 0f5b91c7d733e94ffcad2fd8dd01774631b0c15a Mon Sep 17 00:00:00 2001 From: Anton Hvornum Date: Tue, 30 Aug 2022 22:59:23 +0200 Subject: Fixing issue where blkid causes SysCallException (#1445) * Moving a partprobe() call to better allow for cache updates * Trying to improve Partition()._fetch_information() * Removed a sleep() for debugging purposes * Tweaked a sleep --- archinstall/lib/disk/filesystem.py | 6 ++++-- archinstall/lib/disk/helpers.py | 5 +++++ archinstall/lib/disk/partition.py | 18 ++++++++++++++++-- archinstall/lib/exceptions.py | 8 ++++++-- archinstall/lib/general.py | 2 +- 5 files changed, 32 insertions(+), 7 deletions(-) (limited to 'archinstall/lib/disk/partition.py') diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py index 90656308..5d5952a0 100644 --- a/archinstall/lib/disk/filesystem.py +++ b/archinstall/lib/disk/filesystem.py @@ -253,7 +253,6 @@ class Filesystem: if self.parted(parted_string): for count in range(storage.get('DISK_RETRY_ATTEMPTS', 3)): - self.partprobe() self.blockdevice.flush_cache() new_partition_uuids = [partition.part_uuid for partition in self.blockdevice.partitions.values()] @@ -271,7 +270,10 @@ class Filesystem: raise err else: log(f"Could not get UUID for partition. Waiting {storage.get('DISK_TIMEOUTS', 1) * count}s before retrying.",level=logging.DEBUG) - time.sleep(storage.get('DISK_TIMEOUTS', 1) * count) + self.partprobe() + time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1))) + else: + print("Parted did not return True during partition creation") total_partitions = set([partition.part_uuid for partition in self.blockdevice.partitions.values()]) total_partitions.update(previous_partuuids) diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py index 60efe724..a148a5db 100644 --- a/archinstall/lib/disk/helpers.py +++ b/archinstall/lib/disk/helpers.py @@ -229,12 +229,17 @@ def all_blockdevices(mappers=False, partitions=False, error=False) -> Dict[str, try: information = get_loop_info(device_path) if not information: + print("Exit code for blkid -p -o export was:", ex.exit_code) raise SysCallError("Could not get loop information", exit_code=1) except SysCallError: + print("Not a loop device, trying uevent rules.") information = get_blockdevice_uevent(pathlib.Path(block_device).readlink().name) else: + # We could not reliably get any information, perhaps the disk is clean of information? + print("Raising ex because:", ex.exit_code) raise ex + # return instances information = enrich_blockdevice_information(information) diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index f70bf907..56a7d436 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -139,7 +139,19 @@ class Partition: def _call_lsblk(self) -> Dict[str, Any]: self.partprobe() - output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8') + # This sleep might be overkill, but lsblk is known to + # work against a chaotic cache that can change during call + # causing no information to be returned (blkid is better) + # time.sleep(1) + + # TODO: Maybe incorporate a re-try system here based on time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1))) + + try: + output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8') + except SysCallError as error: + # It appears as if lsblk can return exit codes like 8192 to indicate something. + # But it does return output so we'll try to catch it. + output = error.worker.decode('UTF-8') if output: lsblk_info = json.loads(output) @@ -165,7 +177,9 @@ class Partition: def _fetch_information(self) -> PartitionInfo: lsblk_info = self._call_lsblk() sfdisk_info = self._call_sfdisk() - device = lsblk_info['blockdevices'][0] + + if not (device := lsblk_info.get('blockdevices', [None])[0]): + raise DiskError(f'Failed to retrieve information for "{self.device_path}" with lsblk') mountpoints = [Path(mountpoint) for mountpoint in device['mountpoints'] if mountpoint] bootable = sfdisk_info.get('bootable', False) or sfdisk_info.get('type', '') == 'C12A7328-F81F-11D2-BA4B-00A0C93EC93B' diff --git a/archinstall/lib/exceptions.py b/archinstall/lib/exceptions.py index a16faa3f..a66e4e04 100644 --- a/archinstall/lib/exceptions.py +++ b/archinstall/lib/exceptions.py @@ -1,4 +1,7 @@ -from typing import Optional +from typing import Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from .general import SysCommandWorker class RequirementError(BaseException): pass @@ -17,10 +20,11 @@ class ProfileError(BaseException): class SysCallError(BaseException): - def __init__(self, message :str, exit_code :Optional[int] = None) -> None: + def __init__(self, message :str, exit_code :Optional[int] = None, worker :Optional['SysCommandWorker'] = None) -> None: super(SysCallError, self).__init__(message) self.message = message self.exit_code = exit_code + self.worker = worker class PermissionError(BaseException): diff --git a/archinstall/lib/general.py b/archinstall/lib/general.py index 61c4358e..d76b7036 100644 --- a/archinstall/lib/general.py +++ b/archinstall/lib/general.py @@ -272,7 +272,7 @@ class SysCommandWorker: log(args[1], level=logging.DEBUG, fg='red') if self.exit_code != 0: - raise SysCallError(f"{self.cmd} exited with abnormal exit code [{self.exit_code}]: {self._trace_log[-500:]}", self.exit_code) + raise SysCallError(f"{self.cmd} exited with abnormal exit code [{self.exit_code}]: {self._trace_log[-500:]}", self.exit_code, worker=self) def is_alive(self) -> bool: self.poll() -- cgit v1.2.3-70-g09d2