We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[master] Lag in menu after selecting a harddrive
Running `viztracer` generates a 2.6G trace and crashes most machines trying to analyze it.
Here's the only evidence of the issue thus far:

https://user-images.githubusercontent.com/861439/167908067-694654d5-56a3-4baa-8bcb-9378472c8677.mov


https://user-images.githubusercontent.com/861439/167908074-be9c081f-2180-42ea-84bc-7fe0fed73e20.mov

I haven't been able to track it down.
Reported by `@rez` on discord.

Doesn't always happen, I tried a "speedrum" earlier today of `master` and it worked, with a slight hint of the lag: https://cdn.discordapp.com/attachments/726808924401172483/973994546126024714/archinstall_speedrun.gif

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `archinstall/lib/disk/blockdevice.py`
Content:
```
1 from __future__ import annotations
2 import os
3 import json
4 import logging
5 import time
6 from typing import Optional, Dict, Any, Iterator, Tuple, List, TYPE_CHECKING
7 # https://stackoverflow.com/a/39757388/929999
8 if TYPE_CHECKING:
9 	from .partition import Partition
10 
11 from ..exceptions import DiskError, SysCallError
12 from ..output import log
13 from ..general import SysCommand
14 from ..storage import storage
15 
16 
17 class BlockDevice:
18 	def __init__(self, path :str, info :Optional[Dict[str, Any]] = None):
19 		if not info:
20 			from .helpers import all_blockdevices
21 			# If we don't give any information, we need to auto-fill it.
22 			# Otherwise any subsequent usage will break.
23 			info = all_blockdevices(partitions=False)[path].info
24 
25 		self.path = path
26 		self.info = info
27 		self.keep_partitions = True
28 		self.part_cache = {}
29 
30 		# TODO: Currently disk encryption is a BIT misleading.
31 		#       It's actually partition-encryption, but for future-proofing this
32 		#       I'm placing the encryption password on a BlockDevice level.
33 
34 	def __repr__(self, *args :str, **kwargs :str) -> str:
35 		return f"BlockDevice({self.device_or_backfile}, size={self._safe_size}GB, free_space={self._safe_free_space}, bus_type={self.bus_type})"
36 
37 	def __iter__(self) -> Iterator[Partition]:
38 		for partition in self.partitions:
39 			yield self.partitions[partition]
40 
41 	def __getitem__(self, key :str, *args :str, **kwargs :str) -> Any:
42 		if hasattr(self, key):
43 			return getattr(self, key)
44 		elif key not in self.info:
45 			raise KeyError(f'{self} does not contain information: "{key}"')
46 		return self.info[key]
47 
48 	def __len__(self) -> int:
49 		return len(self.partitions)
50 
51 	def __lt__(self, left_comparitor :'BlockDevice') -> bool:
52 		return self.path < left_comparitor.path
53 
54 	def json(self) -> str:
55 		"""
56 		json() has precedence over __dump__, so this is a way
57 		to give less/partial information for user readability.
58 		"""
59 		return self.path
60 
61 	def __dump__(self) -> Dict[str, Dict[str, Any]]:
62 		return {
63 			self.path : {
64 				'partuuid' : self.uuid,
65 				'wipe' : self.info.get('wipe', None),
66 				'partitions' : [part.__dump__() for part in self.partitions.values()]
67 			}
68 		}
69 
70 	@property
71 	def partition_type(self) -> str:
72 		output = json.loads(SysCommand(f"lsblk --json -o+PTTYPE {self.path}").decode('UTF-8'))
73 
74 		for device in output['blockdevices']:
75 			return device['pttype']
76 
77 	@property
78 	def device_or_backfile(self) -> str:
79 		"""
80 		Returns the actual device-endpoint of the BlockDevice.
81 		If it's a loop-back-device it returns the back-file,
82 		For other types it return self.device
83 		"""
84 		if self.info.get('type') == 'loop':
85 			return self.info['back-file']
86 		else:
87 			return self.device
88 
89 	@property
90 	def mountpoint(self) -> None:
91 		"""
92 		A dummy function to enable transparent comparisons of mountpoints.
93 		As blockdevices can't be mounted directly, this will always be None
94 		"""
95 		return None
96 
97 	@property
98 	def device(self) -> str:
99 		"""
100 		Returns the device file of the BlockDevice.
101 		If it's a loop-back-device it returns the /dev/X device,
102 		If it's a ATA-drive it returns the /dev/X device
103 		And if it's a crypto-device it returns the parent device
104 		"""
105 		if "DEVTYPE" not in self.info:
106 			raise DiskError(f'Could not locate backplane info for "{self.path}"')
107 
108 		if self.info['DEVTYPE'] in ['disk','loop']:
109 			return self.path
110 		elif self.info['DEVTYPE'][:4] == 'raid':
111 			# This should catch /dev/md## raid devices
112 			return self.path
113 		elif self.info['DEVTYPE'] == 'crypt':
114 			if 'pkname' not in self.info:
115 				raise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.')
116 			return f"/dev/{self.info['pkname']}"
117 		else:
118 			log(f"Unknown blockdevice type for {self.path}: {self.info['DEVTYPE']}", level=logging.DEBUG)
119 
120 	# 	if not stat.S_ISBLK(os.stat(full_path).st_mode):
121 	# 		raise DiskError(f'Selected disk "{full_path}" is not a block device.')
122 
123 	@property
124 	def partitions(self) -> Dict[str, Partition]:
125 		from .filesystem import Partition
126 
127 		self.partprobe()
128 		result = SysCommand(['/usr/bin/lsblk', '-J', self.path])
129 
130 		if b'not a block device' in result:
131 			raise DiskError(f'Can not read partitions off something that isn\'t a block device: {self.path}')
132 
133 		if not result[:1] == b'{':
134 			raise DiskError('Error getting JSON output from:', f'/usr/bin/lsblk -J {self.path}')
135 
136 		r = json.loads(result.decode('UTF-8'))
137 		if len(r['blockdevices']) and 'children' in r['blockdevices'][0]:
138 			root_path = f"/dev/{r['blockdevices'][0]['name']}"
139 			for part in r['blockdevices'][0]['children']:
140 				part_id = part['name'][len(os.path.basename(self.path)):]
141 				if part_id not in self.part_cache:
142 					# TODO: Force over-write even if in cache?
143 					if part_id not in self.part_cache or self.part_cache[part_id].size != part['size']:
144 						self.part_cache[part_id] = Partition(root_path + part_id, block_device=self, part_id=part_id)
145 
146 		return {k: self.part_cache[k] for k in sorted(self.part_cache)}
147 
148 	@property
149 	def partition(self) -> Partition:
150 		all_partitions = self.partitions
151 		return [all_partitions[k] for k in all_partitions]
152 
153 	@property
154 	def partition_table_type(self) -> int:
155 		# TODO: Don't hardcode :)
156 		# Remove if we don't use this function anywhere
157 		from .filesystem import GPT
158 		return GPT
159 
160 	@property
161 	def uuid(self) -> str:
162 		log('BlockDevice().uuid is untested!', level=logging.WARNING, fg='yellow')
163 		"""
164 		Returns the disk UUID as returned by lsblk.
165 		This is more reliable than relying on /dev/disk/by-partuuid as
166 		it doesn't seam to be able to detect md raid partitions.
167 		"""
168 		return SysCommand(f'blkid -s PTUUID -o value {self.path}').decode('UTF-8')
169 
170 	@property
171 	def _safe_size(self) -> float:
172 		from .helpers import convert_size_to_gb
173 
174 		try:
175 			output = json.loads(SysCommand(f"lsblk --json -b -o+SIZE {self.path}").decode('UTF-8'))
176 		except SysCallError:
177 			return -1.0
178 
179 		for device in output['blockdevices']:
180 			return convert_size_to_gb(device['size'])
181 
182 	@property
183 	def size(self) -> float:
184 		from .helpers import convert_size_to_gb
185 
186 		output = json.loads(SysCommand(f"lsblk --json -b -o+SIZE {self.path}").decode('UTF-8'))
187 
188 		for device in output['blockdevices']:
189 			return convert_size_to_gb(device['size'])
190 
191 	@property
192 	def bus_type(self) -> str:
193 		output = json.loads(SysCommand(f"lsblk --json -o+ROTA,TRAN {self.path}").decode('UTF-8'))
194 
195 		for device in output['blockdevices']:
196 			return device['tran']
197 
198 	@property
199 	def spinning(self) -> bool:
200 		output = json.loads(SysCommand(f"lsblk --json -o+ROTA,TRAN {self.path}").decode('UTF-8'))
201 
202 		for device in output['blockdevices']:
203 			return device['rota'] is True
204 
205 	@property
206 	def _safe_free_space(self) -> Tuple[str, ...]:
207 		try:
208 			return '+'.join(part[2] for part in self.free_space)
209 		except SysCallError:
210 			return '?'
211 
212 	@property
213 	def free_space(self) -> Tuple[str, ...]:
214 		# NOTE: parted -s will default to `cancel` on prompt, skipping any partition
215 		# that is "outside" the disk. in /dev/sr0 this is usually the case with Archiso,
216 		# so the free will ignore the ESP partition and just give the "free" space.
217 		# Doesn't harm us, but worth noting in case something weird happens.
218 		try:
219 			for line in SysCommand(f"parted -s --machine {self.path} print free"):
220 				if 'free' in (free_space := line.decode('UTF-8')):
221 					_, start, end, size, *_ = free_space.strip('\r\n;').split(':')
222 					yield (start, end, size)
223 		except SysCallError as error:
224 			log(f"Could not get free space on {self.path}: {error}", level=logging.DEBUG)
225 
226 	@property
227 	def largest_free_space(self) -> List[str]:
228 		info = []
229 		for space_info in self.free_space:
230 			if not info:
231 				info = space_info
232 			else:
233 				# [-1] = size
234 				if space_info[-1] > info[-1]:
235 					info = space_info
236 		return info
237 
238 	@property
239 	def first_free_sector(self) -> str:
240 		if info := self.largest_free_space:
241 			start = info[0]
242 		else:
243 			start = '512MB'
244 		return start
245 
246 	@property
247 	def first_end_sector(self) -> str:
248 		if info := self.largest_free_space:
249 			end = info[1]
250 		else:
251 			end = f"{self.size}GB"
252 		return end
253 
254 	def partprobe(self) -> bool:
255 		return SysCommand(['partprobe', self.path]).exit_code == 0
256 
257 	def has_partitions(self) -> int:
258 		return len(self.partitions)
259 
260 	def has_mount_point(self, mountpoint :str) -> bool:
261 		for partition in self.partitions:
262 			if self.partitions[partition].mountpoint == mountpoint:
263 				return True
264 		return False
265 
266 	def flush_cache(self) -> None:
267 		self.part_cache = {}
268 
269 	def get_partition(self, uuid :str) -> Partition:
270 		count = 0
271 		while count < 5:
272 			for partition_uuid, partition in self.partitions.items():
273 				if partition.uuid.lower() == uuid.lower():
274 					return partition
275 			else:
276 				log(f"uuid {uuid} not found. Waiting for {count +1} time",level=logging.DEBUG)
277 				time.sleep(float(storage['arguments'].get('disk-sleep', 0.2)))
278 				count += 1
279 		else:
280 			log(f"Could not find {uuid} in disk after 5 retries",level=logging.INFO)
281 			print(f"Cache: {self.part_cache}")
282 			print(f"Partitions: {self.partitions.items()}")
283 			print(f"UUID: {[uuid]}")
284 			raise DiskError(f"New partition {uuid} never showed up after adding new partition on {self}")
285
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/archinstall/lib/disk/blockdevice.py b/archinstall/lib/disk/blockdevice.py
--- a/archinstall/lib/disk/blockdevice.py
+++ b/archinstall/lib/disk/blockdevice.py
@@ -3,6 +3,7 @@
 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:
@@ -32,6 +33,10 @@
 		#       I'm placing the encryption password on a BlockDevice level.
 
 	def __repr__(self, *args :str, **kwargs :str) -> str:
+		return self._str_repr
+	
+	@cached_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})"
 
 	def __iter__(self) -> Iterator[Partition]:

{"golden_diff": "diff --git a/archinstall/lib/disk/blockdevice.py b/archinstall/lib/disk/blockdevice.py\n--- a/archinstall/lib/disk/blockdevice.py\n+++ b/archinstall/lib/disk/blockdevice.py\n@@ -3,6 +3,7 @@\n import json\n import logging\n import time\n+from functools import cached_property\n from typing import Optional, Dict, Any, Iterator, Tuple, List, TYPE_CHECKING\n # https://stackoverflow.com/a/39757388/929999\n if TYPE_CHECKING:\n@@ -32,6 +33,10 @@\n \t\t#       I'm placing the encryption password on a BlockDevice level.\n \n \tdef __repr__(self, *args :str, **kwargs :str) -> str:\n+\t\treturn self._str_repr\n+\t\n+\t@cached_property\n+\tdef _str_repr(self) -> str:\n \t\treturn f\"BlockDevice({self.device_or_backfile}, size={self._safe_size}GB, free_space={self._safe_free_space}, bus_type={self.bus_type})\"\n \n \tdef __iter__(self) -> Iterator[Partition]:\n", "issue": "[master] Lag in menu after selecting a harddrive\nRunning `viztracer` generates a 2.6G trace and crashes most machines trying to analyze it.\r\nHere's the only evidence of the issue thus far:\r\n\r\nhttps://user-images.githubusercontent.com/861439/167908067-694654d5-56a3-4baa-8bcb-9378472c8677.mov\r\n\r\n\r\nhttps://user-images.githubusercontent.com/861439/167908074-be9c081f-2180-42ea-84bc-7fe0fed73e20.mov\r\n\r\nI haven't been able to track it down.\r\nReported by `@rez` on discord.\r\n\r\nDoesn't always happen, I tried a \"speedrum\" earlier today of `master` and it worked, with a slight hint of the lag: https://cdn.discordapp.com/attachments/726808924401172483/973994546126024714/archinstall_speedrun.gif\n", "before_files": [{"content": "from __future__ import annotations\nimport os\nimport json\nimport logging\nimport time\nfrom typing import Optional, Dict, Any, Iterator, Tuple, List, TYPE_CHECKING\n# https://stackoverflow.com/a/39757388/929999\nif TYPE_CHECKING:\n\tfrom .partition import Partition\n\nfrom ..exceptions import DiskError, SysCallError\nfrom ..output import log\nfrom ..general import SysCommand\nfrom ..storage import storage\n\n\nclass BlockDevice:\n\tdef __init__(self, path :str, info :Optional[Dict[str, Any]] = None):\n\t\tif not info:\n\t\t\tfrom .helpers import all_blockdevices\n\t\t\t# If we don't give any information, we need to auto-fill it.\n\t\t\t# Otherwise any subsequent usage will break.\n\t\t\tinfo = all_blockdevices(partitions=False)[path].info\n\n\t\tself.path = path\n\t\tself.info = info\n\t\tself.keep_partitions = True\n\t\tself.part_cache = {}\n\n\t\t# TODO: Currently disk encryption is a BIT misleading.\n\t\t#       It's actually partition-encryption, but for future-proofing this\n\t\t#       I'm placing the encryption password on a BlockDevice level.\n\n\tdef __repr__(self, *args :str, **kwargs :str) -> str:\n\t\treturn f\"BlockDevice({self.device_or_backfile}, size={self._safe_size}GB, free_space={self._safe_free_space}, bus_type={self.bus_type})\"\n\n\tdef __iter__(self) -> Iterator[Partition]:\n\t\tfor partition in self.partitions:\n\t\t\tyield self.partitions[partition]\n\n\tdef __getitem__(self, key :str, *args :str, **kwargs :str) -> Any:\n\t\tif hasattr(self, key):\n\t\t\treturn getattr(self, key)\n\t\telif key not in self.info:\n\t\t\traise KeyError(f'{self} does not contain information: \"{key}\"')\n\t\treturn self.info[key]\n\n\tdef __len__(self) -> int:\n\t\treturn len(self.partitions)\n\n\tdef __lt__(self, left_comparitor :'BlockDevice') -> bool:\n\t\treturn self.path < left_comparitor.path\n\n\tdef json(self) -> str:\n\t\t\"\"\"\n\t\tjson() has precedence over __dump__, so this is a way\n\t\tto give less/partial information for user readability.\n\t\t\"\"\"\n\t\treturn self.path\n\n\tdef __dump__(self) -> Dict[str, Dict[str, Any]]:\n\t\treturn {\n\t\t\tself.path : {\n\t\t\t\t'partuuid' : self.uuid,\n\t\t\t\t'wipe' : self.info.get('wipe', None),\n\t\t\t\t'partitions' : [part.__dump__() for part in self.partitions.values()]\n\t\t\t}\n\t\t}\n\n\t@property\n\tdef partition_type(self) -> str:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+PTTYPE {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['pttype']\n\n\t@property\n\tdef device_or_backfile(self) -> str:\n\t\t\"\"\"\n\t\tReturns the actual device-endpoint of the BlockDevice.\n\t\tIf it's a loop-back-device it returns the back-file,\n\t\tFor other types it return self.device\n\t\t\"\"\"\n\t\tif self.info.get('type') == 'loop':\n\t\t\treturn self.info['back-file']\n\t\telse:\n\t\t\treturn self.device\n\n\t@property\n\tdef mountpoint(self) -> None:\n\t\t\"\"\"\n\t\tA dummy function to enable transparent comparisons of mountpoints.\n\t\tAs blockdevices can't be mounted directly, this will always be None\n\t\t\"\"\"\n\t\treturn None\n\n\t@property\n\tdef device(self) -> str:\n\t\t\"\"\"\n\t\tReturns the device file of the BlockDevice.\n\t\tIf it's a loop-back-device it returns the /dev/X device,\n\t\tIf it's a ATA-drive it returns the /dev/X device\n\t\tAnd if it's a crypto-device it returns the parent device\n\t\t\"\"\"\n\t\tif \"DEVTYPE\" not in self.info:\n\t\t\traise DiskError(f'Could not locate backplane info for \"{self.path}\"')\n\n\t\tif self.info['DEVTYPE'] in ['disk','loop']:\n\t\t\treturn self.path\n\t\telif self.info['DEVTYPE'][:4] == 'raid':\n\t\t\t# This should catch /dev/md## raid devices\n\t\t\treturn self.path\n\t\telif self.info['DEVTYPE'] == 'crypt':\n\t\t\tif 'pkname' not in self.info:\n\t\t\t\traise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.')\n\t\t\treturn f\"/dev/{self.info['pkname']}\"\n\t\telse:\n\t\t\tlog(f\"Unknown blockdevice type for {self.path}: {self.info['DEVTYPE']}\", level=logging.DEBUG)\n\n\t# \tif not stat.S_ISBLK(os.stat(full_path).st_mode):\n\t# \t\traise DiskError(f'Selected disk \"{full_path}\" is not a block device.')\n\n\t@property\n\tdef partitions(self) -> Dict[str, Partition]:\n\t\tfrom .filesystem import Partition\n\n\t\tself.partprobe()\n\t\tresult = SysCommand(['/usr/bin/lsblk', '-J', self.path])\n\n\t\tif b'not a block device' in result:\n\t\t\traise DiskError(f'Can not read partitions off something that isn\\'t a block device: {self.path}')\n\n\t\tif not result[:1] == b'{':\n\t\t\traise DiskError('Error getting JSON output from:', f'/usr/bin/lsblk -J {self.path}')\n\n\t\tr = json.loads(result.decode('UTF-8'))\n\t\tif len(r['blockdevices']) and 'children' in r['blockdevices'][0]:\n\t\t\troot_path = f\"/dev/{r['blockdevices'][0]['name']}\"\n\t\t\tfor part in r['blockdevices'][0]['children']:\n\t\t\t\tpart_id = part['name'][len(os.path.basename(self.path)):]\n\t\t\t\tif part_id not in self.part_cache:\n\t\t\t\t\t# TODO: Force over-write even if in cache?\n\t\t\t\t\tif part_id not in self.part_cache or self.part_cache[part_id].size != part['size']:\n\t\t\t\t\t\tself.part_cache[part_id] = Partition(root_path + part_id, block_device=self, part_id=part_id)\n\n\t\treturn {k: self.part_cache[k] for k in sorted(self.part_cache)}\n\n\t@property\n\tdef partition(self) -> Partition:\n\t\tall_partitions = self.partitions\n\t\treturn [all_partitions[k] for k in all_partitions]\n\n\t@property\n\tdef partition_table_type(self) -> int:\n\t\t# TODO: Don't hardcode :)\n\t\t# Remove if we don't use this function anywhere\n\t\tfrom .filesystem import GPT\n\t\treturn GPT\n\n\t@property\n\tdef uuid(self) -> str:\n\t\tlog('BlockDevice().uuid is untested!', level=logging.WARNING, fg='yellow')\n\t\t\"\"\"\n\t\tReturns the disk UUID as returned by lsblk.\n\t\tThis is more reliable than relying on /dev/disk/by-partuuid as\n\t\tit doesn't seam to be able to detect md raid partitions.\n\t\t\"\"\"\n\t\treturn SysCommand(f'blkid -s PTUUID -o value {self.path}').decode('UTF-8')\n\n\t@property\n\tdef _safe_size(self) -> float:\n\t\tfrom .helpers import convert_size_to_gb\n\n\t\ttry:\n\t\t\toutput = json.loads(SysCommand(f\"lsblk --json -b -o+SIZE {self.path}\").decode('UTF-8'))\n\t\texcept SysCallError:\n\t\t\treturn -1.0\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn convert_size_to_gb(device['size'])\n\n\t@property\n\tdef size(self) -> float:\n\t\tfrom .helpers import convert_size_to_gb\n\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -b -o+SIZE {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn convert_size_to_gb(device['size'])\n\n\t@property\n\tdef bus_type(self) -> str:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+ROTA,TRAN {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['tran']\n\n\t@property\n\tdef spinning(self) -> bool:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+ROTA,TRAN {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['rota'] is True\n\n\t@property\n\tdef _safe_free_space(self) -> Tuple[str, ...]:\n\t\ttry:\n\t\t\treturn '+'.join(part[2] for part in self.free_space)\n\t\texcept SysCallError:\n\t\t\treturn '?'\n\n\t@property\n\tdef free_space(self) -> Tuple[str, ...]:\n\t\t# NOTE: parted -s will default to `cancel` on prompt, skipping any partition\n\t\t# that is \"outside\" the disk. in /dev/sr0 this is usually the case with Archiso,\n\t\t# so the free will ignore the ESP partition and just give the \"free\" space.\n\t\t# Doesn't harm us, but worth noting in case something weird happens.\n\t\ttry:\n\t\t\tfor line in SysCommand(f\"parted -s --machine {self.path} print free\"):\n\t\t\t\tif 'free' in (free_space := line.decode('UTF-8')):\n\t\t\t\t\t_, start, end, size, *_ = free_space.strip('\\r\\n;').split(':')\n\t\t\t\t\tyield (start, end, size)\n\t\texcept SysCallError as error:\n\t\t\tlog(f\"Could not get free space on {self.path}: {error}\", level=logging.DEBUG)\n\n\t@property\n\tdef largest_free_space(self) -> List[str]:\n\t\tinfo = []\n\t\tfor space_info in self.free_space:\n\t\t\tif not info:\n\t\t\t\tinfo = space_info\n\t\t\telse:\n\t\t\t\t# [-1] = size\n\t\t\t\tif space_info[-1] > info[-1]:\n\t\t\t\t\tinfo = space_info\n\t\treturn info\n\n\t@property\n\tdef first_free_sector(self) -> str:\n\t\tif info := self.largest_free_space:\n\t\t\tstart = info[0]\n\t\telse:\n\t\t\tstart = '512MB'\n\t\treturn start\n\n\t@property\n\tdef first_end_sector(self) -> str:\n\t\tif info := self.largest_free_space:\n\t\t\tend = info[1]\n\t\telse:\n\t\t\tend = f\"{self.size}GB\"\n\t\treturn end\n\n\tdef partprobe(self) -> bool:\n\t\treturn SysCommand(['partprobe', self.path]).exit_code == 0\n\n\tdef has_partitions(self) -> int:\n\t\treturn len(self.partitions)\n\n\tdef has_mount_point(self, mountpoint :str) -> bool:\n\t\tfor partition in self.partitions:\n\t\t\tif self.partitions[partition].mountpoint == mountpoint:\n\t\t\t\treturn True\n\t\treturn False\n\n\tdef flush_cache(self) -> None:\n\t\tself.part_cache = {}\n\n\tdef get_partition(self, uuid :str) -> Partition:\n\t\tcount = 0\n\t\twhile count < 5:\n\t\t\tfor partition_uuid, partition in self.partitions.items():\n\t\t\t\tif partition.uuid.lower() == uuid.lower():\n\t\t\t\t\treturn partition\n\t\t\telse:\n\t\t\t\tlog(f\"uuid {uuid} not found. Waiting for {count +1} time\",level=logging.DEBUG)\n\t\t\t\ttime.sleep(float(storage['arguments'].get('disk-sleep', 0.2)))\n\t\t\t\tcount += 1\n\t\telse:\n\t\t\tlog(f\"Could not find {uuid} in disk after 5 retries\",level=logging.INFO)\n\t\t\tprint(f\"Cache: {self.part_cache}\")\n\t\t\tprint(f\"Partitions: {self.partitions.items()}\")\n\t\t\tprint(f\"UUID: {[uuid]}\")\n\t\t\traise DiskError(f\"New partition {uuid} never showed up after adding new partition on {self}\")\n", "path": "archinstall/lib/disk/blockdevice.py"}], "after_files": [{"content": "from __future__ import annotations\nimport os\nimport json\nimport logging\nimport time\nfrom functools import cached_property\nfrom typing import Optional, Dict, Any, Iterator, Tuple, List, TYPE_CHECKING\n# https://stackoverflow.com/a/39757388/929999\nif TYPE_CHECKING:\n\tfrom .partition import Partition\n\nfrom ..exceptions import DiskError, SysCallError\nfrom ..output import log\nfrom ..general import SysCommand\nfrom ..storage import storage\n\n\nclass BlockDevice:\n\tdef __init__(self, path :str, info :Optional[Dict[str, Any]] = None):\n\t\tif not info:\n\t\t\tfrom .helpers import all_blockdevices\n\t\t\t# If we don't give any information, we need to auto-fill it.\n\t\t\t# Otherwise any subsequent usage will break.\n\t\t\tinfo = all_blockdevices(partitions=False)[path].info\n\n\t\tself.path = path\n\t\tself.info = info\n\t\tself.keep_partitions = True\n\t\tself.part_cache = {}\n\n\t\t# TODO: Currently disk encryption is a BIT misleading.\n\t\t#       It's actually partition-encryption, but for future-proofing this\n\t\t#       I'm placing the encryption password on a BlockDevice level.\n\n\tdef __repr__(self, *args :str, **kwargs :str) -> str:\n\t\treturn self._str_repr\n\t\n\t@cached_property\n\tdef _str_repr(self) -> str:\n\t\treturn f\"BlockDevice({self.device_or_backfile}, size={self._safe_size}GB, free_space={self._safe_free_space}, bus_type={self.bus_type})\"\n\n\tdef __iter__(self) -> Iterator[Partition]:\n\t\tfor partition in self.partitions:\n\t\t\tyield self.partitions[partition]\n\n\tdef __getitem__(self, key :str, *args :str, **kwargs :str) -> Any:\n\t\tif hasattr(self, key):\n\t\t\treturn getattr(self, key)\n\t\telif key not in self.info:\n\t\t\traise KeyError(f'{self} does not contain information: \"{key}\"')\n\t\treturn self.info[key]\n\n\tdef __len__(self) -> int:\n\t\treturn len(self.partitions)\n\n\tdef __lt__(self, left_comparitor :'BlockDevice') -> bool:\n\t\treturn self.path < left_comparitor.path\n\n\tdef json(self) -> str:\n\t\t\"\"\"\n\t\tjson() has precedence over __dump__, so this is a way\n\t\tto give less/partial information for user readability.\n\t\t\"\"\"\n\t\treturn self.path\n\n\tdef __dump__(self) -> Dict[str, Dict[str, Any]]:\n\t\treturn {\n\t\t\tself.path : {\n\t\t\t\t'partuuid' : self.uuid,\n\t\t\t\t'wipe' : self.info.get('wipe', None),\n\t\t\t\t'partitions' : [part.__dump__() for part in self.partitions.values()]\n\t\t\t}\n\t\t}\n\n\t@property\n\tdef partition_type(self) -> str:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+PTTYPE {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['pttype']\n\n\t@property\n\tdef device_or_backfile(self) -> str:\n\t\t\"\"\"\n\t\tReturns the actual device-endpoint of the BlockDevice.\n\t\tIf it's a loop-back-device it returns the back-file,\n\t\tFor other types it return self.device\n\t\t\"\"\"\n\t\tif self.info.get('type') == 'loop':\n\t\t\treturn self.info['back-file']\n\t\telse:\n\t\t\treturn self.device\n\n\t@property\n\tdef mountpoint(self) -> None:\n\t\t\"\"\"\n\t\tA dummy function to enable transparent comparisons of mountpoints.\n\t\tAs blockdevices can't be mounted directly, this will always be None\n\t\t\"\"\"\n\t\treturn None\n\n\t@property\n\tdef device(self) -> str:\n\t\t\"\"\"\n\t\tReturns the device file of the BlockDevice.\n\t\tIf it's a loop-back-device it returns the /dev/X device,\n\t\tIf it's a ATA-drive it returns the /dev/X device\n\t\tAnd if it's a crypto-device it returns the parent device\n\t\t\"\"\"\n\t\tif \"DEVTYPE\" not in self.info:\n\t\t\traise DiskError(f'Could not locate backplane info for \"{self.path}\"')\n\n\t\tif self.info['DEVTYPE'] in ['disk','loop']:\n\t\t\treturn self.path\n\t\telif self.info['DEVTYPE'][:4] == 'raid':\n\t\t\t# This should catch /dev/md## raid devices\n\t\t\treturn self.path\n\t\telif self.info['DEVTYPE'] == 'crypt':\n\t\t\tif 'pkname' not in self.info:\n\t\t\t\traise DiskError(f'A crypt device ({self.path}) without a parent kernel device name.')\n\t\t\treturn f\"/dev/{self.info['pkname']}\"\n\t\telse:\n\t\t\tlog(f\"Unknown blockdevice type for {self.path}: {self.info['DEVTYPE']}\", level=logging.DEBUG)\n\n\t# \tif not stat.S_ISBLK(os.stat(full_path).st_mode):\n\t# \t\traise DiskError(f'Selected disk \"{full_path}\" is not a block device.')\n\n\t@property\n\tdef partitions(self) -> Dict[str, Partition]:\n\t\tfrom .filesystem import Partition\n\n\t\tself.partprobe()\n\t\tresult = SysCommand(['/usr/bin/lsblk', '-J', self.path])\n\n\t\tif b'not a block device' in result:\n\t\t\traise DiskError(f'Can not read partitions off something that isn\\'t a block device: {self.path}')\n\n\t\tif not result[:1] == b'{':\n\t\t\traise DiskError('Error getting JSON output from:', f'/usr/bin/lsblk -J {self.path}')\n\n\t\tr = json.loads(result.decode('UTF-8'))\n\t\tif len(r['blockdevices']) and 'children' in r['blockdevices'][0]:\n\t\t\troot_path = f\"/dev/{r['blockdevices'][0]['name']}\"\n\t\t\tfor part in r['blockdevices'][0]['children']:\n\t\t\t\tpart_id = part['name'][len(os.path.basename(self.path)):]\n\t\t\t\tif part_id not in self.part_cache:\n\t\t\t\t\t# TODO: Force over-write even if in cache?\n\t\t\t\t\tif part_id not in self.part_cache or self.part_cache[part_id].size != part['size']:\n\t\t\t\t\t\tself.part_cache[part_id] = Partition(root_path + part_id, block_device=self, part_id=part_id)\n\n\t\treturn {k: self.part_cache[k] for k in sorted(self.part_cache)}\n\n\t@property\n\tdef partition(self) -> Partition:\n\t\tall_partitions = self.partitions\n\t\treturn [all_partitions[k] for k in all_partitions]\n\n\t@property\n\tdef partition_table_type(self) -> int:\n\t\t# TODO: Don't hardcode :)\n\t\t# Remove if we don't use this function anywhere\n\t\tfrom .filesystem import GPT\n\t\treturn GPT\n\n\t@property\n\tdef uuid(self) -> str:\n\t\tlog('BlockDevice().uuid is untested!', level=logging.WARNING, fg='yellow')\n\t\t\"\"\"\n\t\tReturns the disk UUID as returned by lsblk.\n\t\tThis is more reliable than relying on /dev/disk/by-partuuid as\n\t\tit doesn't seam to be able to detect md raid partitions.\n\t\t\"\"\"\n\t\treturn SysCommand(f'blkid -s PTUUID -o value {self.path}').decode('UTF-8')\n\n\t@property\n\tdef _safe_size(self) -> float:\n\t\tfrom .helpers import convert_size_to_gb\n\n\t\ttry:\n\t\t\toutput = json.loads(SysCommand(f\"lsblk --json -b -o+SIZE {self.path}\").decode('UTF-8'))\n\t\texcept SysCallError:\n\t\t\treturn -1.0\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn convert_size_to_gb(device['size'])\n\n\t@property\n\tdef size(self) -> float:\n\t\tfrom .helpers import convert_size_to_gb\n\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -b -o+SIZE {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn convert_size_to_gb(device['size'])\n\n\t@property\n\tdef bus_type(self) -> str:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+ROTA,TRAN {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['tran']\n\n\t@property\n\tdef spinning(self) -> bool:\n\t\toutput = json.loads(SysCommand(f\"lsblk --json -o+ROTA,TRAN {self.path}\").decode('UTF-8'))\n\n\t\tfor device in output['blockdevices']:\n\t\t\treturn device['rota'] is True\n\n\t@property\n\tdef _safe_free_space(self) -> Tuple[str, ...]:\n\t\ttry:\n\t\t\treturn '+'.join(part[2] for part in self.free_space)\n\t\texcept SysCallError:\n\t\t\treturn '?'\n\n\t@property\n\tdef free_space(self) -> Tuple[str, ...]:\n\t\t# NOTE: parted -s will default to `cancel` on prompt, skipping any partition\n\t\t# that is \"outside\" the disk. in /dev/sr0 this is usually the case with Archiso,\n\t\t# so the free will ignore the ESP partition and just give the \"free\" space.\n\t\t# Doesn't harm us, but worth noting in case something weird happens.\n\t\ttry:\n\t\t\tfor line in SysCommand(f\"parted -s --machine {self.path} print free\"):\n\t\t\t\tif 'free' in (free_space := line.decode('UTF-8')):\n\t\t\t\t\t_, start, end, size, *_ = free_space.strip('\\r\\n;').split(':')\n\t\t\t\t\tyield (start, end, size)\n\t\texcept SysCallError as error:\n\t\t\tlog(f\"Could not get free space on {self.path}: {error}\", level=logging.DEBUG)\n\n\t@property\n\tdef largest_free_space(self) -> List[str]:\n\t\tinfo = []\n\t\tfor space_info in self.free_space:\n\t\t\tif not info:\n\t\t\t\tinfo = space_info\n\t\t\telse:\n\t\t\t\t# [-1] = size\n\t\t\t\tif space_info[-1] > info[-1]:\n\t\t\t\t\tinfo = space_info\n\t\treturn info\n\n\t@property\n\tdef first_free_sector(self) -> str:\n\t\tif info := self.largest_free_space:\n\t\t\tstart = info[0]\n\t\telse:\n\t\t\tstart = '512MB'\n\t\treturn start\n\n\t@property\n\tdef first_end_sector(self) -> str:\n\t\tif info := self.largest_free_space:\n\t\t\tend = info[1]\n\t\telse:\n\t\t\tend = f\"{self.size}GB\"\n\t\treturn end\n\n\tdef partprobe(self) -> bool:\n\t\treturn SysCommand(['partprobe', self.path]).exit_code == 0\n\n\tdef has_partitions(self) -> int:\n\t\treturn len(self.partitions)\n\n\tdef has_mount_point(self, mountpoint :str) -> bool:\n\t\tfor partition in self.partitions:\n\t\t\tif self.partitions[partition].mountpoint == mountpoint:\n\t\t\t\treturn True\n\t\treturn False\n\n\tdef flush_cache(self) -> None:\n\t\tself.part_cache = {}\n\n\tdef get_partition(self, uuid :str) -> Partition:\n\t\tcount = 0\n\t\twhile count < 5:\n\t\t\tfor partition_uuid, partition in self.partitions.items():\n\t\t\t\tif partition.uuid.lower() == uuid.lower():\n\t\t\t\t\treturn partition\n\t\t\telse:\n\t\t\t\tlog(f\"uuid {uuid} not found. Waiting for {count +1} time\",level=logging.DEBUG)\n\t\t\t\ttime.sleep(float(storage['arguments'].get('disk-sleep', 0.2)))\n\t\t\t\tcount += 1\n\t\telse:\n\t\t\tlog(f\"Could not find {uuid} in disk after 5 retries\",level=logging.INFO)\n\t\t\tprint(f\"Cache: {self.part_cache}\")\n\t\t\tprint(f\"Partitions: {self.partitions.items()}\")\n\t\t\tprint(f\"UUID: {[uuid]}\")\n\t\t\traise DiskError(f\"New partition {uuid} never showed up after adding new partition on {self}\")\n", "path": "archinstall/lib/disk/blockdevice.py"}]}