Send patches - preferably formatted by git format-patch - to patches at archlinux32 dot org.
summaryrefslogtreecommitdiff
path: root/archinstall/lib
diff options
context:
space:
mode:
Diffstat (limited to 'archinstall/lib')
-rw-r--r--archinstall/lib/disk.py25
-rw-r--r--archinstall/lib/general.py31
-rw-r--r--archinstall/lib/luks.py8
-rw-r--r--archinstall/lib/mirrors.py8
-rw-r--r--archinstall/lib/networking.py4
-rw-r--r--archinstall/lib/packages.py2
-rw-r--r--archinstall/lib/profiles.py2
-rw-r--r--archinstall/lib/services.py5
-rw-r--r--archinstall/lib/tts.py0
9 files changed, 46 insertions, 39 deletions
diff --git a/archinstall/lib/disk.py b/archinstall/lib/disk.py
index b11f2318..0d0285d2 100644
--- a/archinstall/lib/disk.py
+++ b/archinstall/lib/disk.py
@@ -1,6 +1,6 @@
import glob, re, os, json
from collections import OrderedDict
-from .exceptions import *
+from .exceptions import DiskError
from .general import *
ROOT_DIR_PATTERN = re.compile('^.*?/devices')
@@ -21,7 +21,7 @@ class BlockDevice():
return f"BlockDevice({self.device})"
def __getitem__(self, key, *args, **kwargs):
- if not key in self.info:
+ if key not in self.info:
raise KeyError(f'{self} does not contain information: "{key}"')
return self.info[key]
@@ -37,9 +37,9 @@ class BlockDevice():
def __dump__(self):
return {
- 'path' : self.path,
- 'info' : self.info,
- 'partition_cache' : self.part_cache
+ 'path': self.path,
+ 'info': self.info,
+ 'partition_cache': self.part_cache
}
@property
@@ -50,7 +50,8 @@ class BlockDevice():
If it's a ATA-drive it returns the /dev/X device
And if it's a crypto-device it returns the parent device
"""
- if not 'type' in self.info: raise DiskError(f'Could not locate backplane info for "{self.path}"')
+ if "type" not in self.info:
+ raise DiskError(f'Could not locate backplane info for "{self.path}"')
if self.info['type'] == 'loop':
for drive in json.loads(b''.join(sys_command(f'losetup --json', hide_from_log=True)).decode('UTF_8'))['loopdevices']:
@@ -60,7 +61,8 @@ class BlockDevice():
elif self.info['type'] == 'disk':
return self.path
elif self.info['type'] == 'crypt':
- if not 'pkname' in self.info: raise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.')
+ if 'pkname' not in self.info:
+ raise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.')
return f"/dev/{self.info['pkname']}"
# if not stat.S_ISBLK(os.stat(full_path).st_mode):
@@ -97,7 +99,8 @@ class BlockDevice():
class Partition():
def __init__(self, path, part_id=None, size=-1, filesystem=None, mountpoint=None, encrypted=False):
- if not part_id: part_id = os.path.basename(path)
+ if not part_id:
+ part_id = os.path.basename(path)
self.path = path
self.part_id = part_id
self.mountpoint = mountpoint
@@ -115,7 +118,7 @@ class Partition():
log(f'Formatting {self} -> {filesystem}')
if filesystem == 'btrfs':
o = b''.join(sys_command(f'/usr/bin/mkfs.btrfs -f {self.path}'))
- if not b'UUID' in o:
+ if b'UUID' not in o:
raise DiskError(f'Could not format {self.path} with {filesystem} because: {o}')
self.filesystem = 'btrfs'
elif filesystem == 'fat32':
@@ -244,11 +247,11 @@ def device_state(name, *args, **kwargs):
# lsblk --json -l -n -o path
def all_disks(*args, **kwargs):
- if not 'partitions' in kwargs: kwargs['partitions'] = False
+ kwargs.setdefault("partitions", False)
drives = OrderedDict()
#for drive in json.loads(sys_command(f'losetup --json', *args, **lkwargs, hide_from_log=True)).decode('UTF_8')['loopdevices']:
for drive in json.loads(b''.join(sys_command(f'lsblk --json -l -n -o path,size,type,mountpoint,label,pkname', *args, **kwargs, hide_from_log=True)).decode('UTF_8'))['blockdevices']:
if not kwargs['partitions'] and drive['type'] == 'part': continue
drives[drive['path']] = BlockDevice(drive['path'], drive)
- return drives \ No newline at end of file
+ return drives
diff --git a/archinstall/lib/general.py b/archinstall/lib/general.py
index d9b0c147..abcf25f0 100644
--- a/archinstall/lib/general.py
+++ b/archinstall/lib/general.py
@@ -73,9 +73,10 @@ class sys_command():#Thread):
Stolen from archinstall_gui
"""
def __init__(self, cmd, callback=None, start_callback=None, *args, **kwargs):
- if not 'worker_id' in kwargs: kwargs['worker_id'] = gen_uid()
- if not 'emulate' in kwargs: kwargs['emulate'] = False
- if not 'suppress_errors' in kwargs: kwargs['suppress_errors'] = False
+ kwargs.setdefault("worker_id", gen_uid())
+ kwargs.setdefault("emulate", False)
+ kwargs.setdefault("suppress_errors", False)
+
if kwargs['emulate']:
log(f"Starting command '{cmd}' in emulation mode.")
self.raw_cmd = cmd
@@ -85,7 +86,8 @@ class sys_command():#Thread):
raise ValueError(f'Incorrect string to split: {cmd}\n{e}')
self.args = args
self.kwargs = kwargs
- if not 'worker' in self.kwargs: self.kwargs['worker'] = None
+
+ self.kwargs.setdefault("worker", None)
self.callback = callback
self.pid = None
self.exit_code = None
@@ -110,7 +112,8 @@ class sys_command():#Thread):
if not os.path.isdir(self.exec_dir):
os.makedirs(self.exec_dir)
- if start_callback: start_callback(self, *args, **kwargs)
+ if start_callback:
+ start_callback(self, *args, **kwargs)
self.run()
def __iter__(self, *args, **kwargs):
@@ -125,14 +128,14 @@ class sys_command():#Thread):
def dump(self):
return {
- 'status' : self.status,
- 'worker_id' : self.worker_id,
- 'worker_result' : self.trace_log.decode('UTF-8'),
- 'started' : self.started,
- 'ended' : self.ended,
- 'started_pprint' : '{}-{}-{} {}:{}:{}'.format(*time.localtime(self.started)),
- 'ended_pprint' : '{}-{}-{} {}:{}:{}'.format(*time.localtime(self.ended)) if self.ended else None,
- 'exit_code' : self.exit_code
+ 'status': self.status,
+ 'worker_id': self.worker_id,
+ 'worker_result': self.trace_log.decode('UTF-8'),
+ 'started': self.started,
+ 'ended': self.ended,
+ 'started_pprint': '{}-{}-{} {}:{}:{}'.format(*time.localtime(self.started)),
+ 'ended_pprint': '{}-{}-{} {}:{}:{}'.format(*time.localtime(self.ended)) if self.ended else None,
+ 'exit_code': self.exit_code
}
def run(self):
@@ -255,4 +258,4 @@ def prerequisite_check():
return True
def reboot():
- o = b''.join(sys_command(("/usr/bin/reboot"))) \ No newline at end of file
+ o = b''.join(sys_command("/usr/bin/reboot"))
diff --git a/archinstall/lib/luks.py b/archinstall/lib/luks.py
index e56f3bd2..d4ee6632 100644
--- a/archinstall/lib/luks.py
+++ b/archinstall/lib/luks.py
@@ -23,14 +23,15 @@ class luks2():
def encrypt(self, partition, password, key_size=512, hash_type='sha512', iter_time=10000, key_file=None):
log(f'Encrypting {partition}')
- if not key_file: key_file = f'/tmp/{os.path.basename(self.partition.path)}.disk_pw' #TODO: Make disk-pw-file randomly unique?
+ if not key_file:
+ key_file = f"/tmp/{os.path.basename(self.partition.path)}.disk_pw" # TODO: Make disk-pw-file randomly unique?
if type(password) != bytes: password = bytes(password, 'UTF-8')
with open(key_file, 'wb') as fh:
fh.write(password)
o = b''.join(sys_command(f'/usr/bin/cryptsetup -q -v --type luks2 --pbkdf argon2i --hash {hash_type} --key-size {key_size} --iter-time {iter_time} --key-file {os.path.abspath(key_file)} --use-urandom luksFormat {partition.path}'))
- if not b'Command successful.' in o:
+ if b'Command successful.' not in o:
raise DiskError(f'Could not encrypt volume "{partition.path}": {o}')
return key_file
@@ -43,7 +44,8 @@ class luks2():
:param mountpoint: The name without absolute path, for instance "luksdev" will point to /dev/mapper/luksdev
:type mountpoint: str
"""
- if '/' in mountpoint: os.path.basename(mountpoint) # TODO: Raise exception instead?
+ if '/' in mountpoint:
+ os.path.basename(mountpoint) # TODO: Raise exception instead?
sys_command(f'/usr/bin/cryptsetup open {partition.path} {mountpoint} --key-file {os.path.abspath(key_file)} --type luks2')
if os.path.islink(f'/dev/mapper/{mountpoint}'):
return Partition(f'/dev/mapper/{mountpoint}', encrypted=True)
diff --git a/archinstall/lib/mirrors.py b/archinstall/lib/mirrors.py
index a4aade9c..1cd53ed0 100644
--- a/archinstall/lib/mirrors.py
+++ b/archinstall/lib/mirrors.py
@@ -78,16 +78,16 @@ def list_mirrors():
region = 'Unknown region'
for line in response.readlines():
- if len(line.strip()) == 0: continue
+ if len(line.strip()) == 0:
+ continue
line = line.decode('UTF-8').strip('\n').strip('\r')
if line[:3] == '## ':
region = line[3:]
elif line[:10] == '#Server = ':
- if not region in regions:
- regions[region] = {}
+ regions.setdefault(region, {})
- url = line[1:].lstrip('Server = ')
+ url = line.lstrip('#Server = ')
regions[region][url] = True
return regions \ No newline at end of file
diff --git a/archinstall/lib/networking.py b/archinstall/lib/networking.py
index ae4126a9..d24bb911 100644
--- a/archinstall/lib/networking.py
+++ b/archinstall/lib/networking.py
@@ -3,7 +3,6 @@ import socket
import struct
from collections import OrderedDict
-from .exceptions import *
def getHwAddr(ifname):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -13,7 +12,8 @@ def getHwAddr(ifname):
def list_interfaces(skip_loopback=True):
interfaces = OrderedDict()
for index, iface in socket.if_nameindex():
- if skip_loopback and iface == 'lo': continue
+ if skip_loopback and iface == "lo":
+ continue
mac = getHwAddr(iface).replace(':', '-')
interfaces[mac] = iface
diff --git a/archinstall/lib/packages.py b/archinstall/lib/packages.py
index 2f5ebe94..ddf11f7f 100644
--- a/archinstall/lib/packages.py
+++ b/archinstall/lib/packages.py
@@ -11,7 +11,7 @@ def find_package(name):
"""
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
- ssl_context.verify_mode=ssl.CERT_NONE
+ ssl_context.verify_mode = ssl.CERT_NONE
response = urllib.request.urlopen(BASE_URL.format(package=name), context=ssl_context)
data = response.read().decode('UTF-8')
return json.loads(data)
diff --git a/archinstall/lib/profiles.py b/archinstall/lib/profiles.py
index e018f753..322436c0 100644
--- a/archinstall/lib/profiles.py
+++ b/archinstall/lib/profiles.py
@@ -106,7 +106,7 @@ class Profile():
# To avoid profiles importing the wrong 'archinstall',
# we need to ensure that this current archinstall is in sys.path
archinstall_path = os.path.abspath(f'{os.path.dirname(__file__)}/../../')
- if not archinstall_path in sys.path:
+ if archinstall_path not in sys.path:
sys.path.insert(0, archinstall_path)
instructions = self.load_instructions()
diff --git a/archinstall/lib/services.py b/archinstall/lib/services.py
index 43051f0f..8fcdd296 100644
--- a/archinstall/lib/services.py
+++ b/archinstall/lib/services.py
@@ -3,11 +3,10 @@ import os
from .exceptions import *
from .general import *
-def service_state(service_name :str):
+def service_state(service_name: str):
if os.path.splitext(service_name)[1] != '.service':
- service_name += '.service' # Just to be safe
+ service_name += '.service' # Just to be safe
state = b''.join(sys_command(f'systemctl show -p SubState --value {service_name}'))
return state.strip().decode('UTF-8')
- \ No newline at end of file
diff --git a/archinstall/lib/tts.py b/archinstall/lib/tts.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/archinstall/lib/tts.py

Copyright © 2002-2022 Judd Vinet and Aaron Griffin. Copyright © 2022 Erich Eckner.
The Arch Linux name and logo are recognized trademarks. Some rights reserved. The registered trademark Linux® is used pursuant to a sublicense from LMI, the exclusive licensee of Linus Torvalds, owner of the mark on a world-wide basis.