Class SysCommand

archinstall/lib/command.py:227–331

class SysCommand:
	def __init__(
		self,
		cmd: str | list[str],
		peek_output: bool | None = False,
		environment_vars: dict[str, str] | None = None,
		working_directory: str = './',
		remove_vt100_escape_codes_from_lines: bool = True,
	):
		self.cmd = cmd
		self.peek_output = peek_output
		self.environment_vars = environment_vars
		self.working_directory = working_directory
		self.remove_vt100_escape_codes_from_lines = remove_vt100_escape_codes_from_lines

		self.session: SysCommandWorker | None = None
		self.create_session()

	def __enter__(self) -> SysCommandWorker | None:
		return self.session

	def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
		# b''.join(sys_command('sync')) # No need to, since the underlying fs() object will call sync.
		# TODO: https://stackoverflow.com/questions/28157929/how-to-safely-handle-an-exception-inside-a-context-manager

		if exc_type is not None:
			error(str(exc_value))

	def __iter__(self, *args: list[Any], **kwargs: dict[str, Any]) -> Iterator[bytes]:
		if self.session:
			yield from self.session

	def __getitem__(self, key: slice) -> bytes:
		if not self.session:
			raise KeyError('SysCommand() does not have an active session.')
		elif type(key) is slice:
			start = key.start or 0
			end = key.stop or len(self.session._trace_log)

			return self.session._trace_log[start:end]
		else:
			raise ValueError("SysCommand() doesn't have key & value pairs, only slices, SysCommand('ls')[:10] as an example.")

	@override
	def __repr__(self, *args: list[Any], **kwargs: dict[str, Any]) -> str:
		return self.decode('UTF-8', errors='backslashreplace') or ''

	def create_session(self) -> bool:
		"""
		Initiates a :ref:`SysCommandWorker` session in this class ``.session``.
		It then proceeds to poll the process until it ends, after which it also
		clears any printed output if ``.peek_output=True``.
		"""
		if self.session:
			return True

		with SysCommandWorker(
			self.cmd,