blext.ui

blext.ui.download_wheels

Implements a UI that reports the download progress of Python wheels.

CallbacksDownloadWheel pydantic-model

Bases: BaseModel

Callbacks to trigger in the process of downloading wheels.

Notes

• All that a download agent has to ensure, is to allow the user to specify equivalent callbacks to these.
• The callback return values are never used for any purpose.

ATTRIBUTE	DESCRIPTION
cb_start_wheel_download	Called as a wheel is starting to be downloaded. Called with the wheel, and with its download path. Always called before cb_update_wheel_download for a wheel. TYPE: Callable[[PyDepWheel, Path], Any]
cb_update_wheel_download	Called when an actively downloading wheel Called with the wheel, and its download path, and newly downloaded bytes. Always called after cb_start_wheel_download for a wheel. TYPE: Callable[[PyDepWheel, Path, int], Any]
cb_finish_wheel_download	Called when an actively downloading wheel No other callbacks are called for a wheel after this. TYPE: Callable[[PyDepWheel, Path], Any]

See Also

• blext.pydeps.network.download_wheels: Download agent that uses equivalent callbacks.
• blext.ui.ui_download_wheels: Context manager that provides this object.

Fields:

• cb_start_wheel_download (Callable[[PyDepWheel, Path], Any])
• cb_update_wheel_download (Callable[[PyDepWheel, Path, int], Any])
• cb_finish_wheel_download (Callable[[PyDepWheel, Path], Any])

ui_download_wheels

ui_download_wheels(
	wheels_to_download: frozenset[PyDepWheel],
	*,
	console: Console,
	fps: int = 24,
) -> Generator[CallbacksDownloadWheel, None, None]

Context manager creating a terminal UI to communicate wheel downloading.

Notes

Yields callbacks to call during the download progress, in order for the UI to update correctly.

PARAMETER	DESCRIPTION
wheels_to_download	Set of wheels to download. TYPE: frozenset[PyDepWheel]
console	rich console to print the UI to. TYPE: Console
fps	Number of updates to the terminal, per second. TYPE: int DEFAULT: 24

See Also

blext.ui.download_wheels.CallbacksDownloadWheel: For more on when to call each callback.

Source code in blext/ui/download_wheels.py

@contextlib.contextmanager
def ui_download_wheels(
	wheels_to_download: frozenset[pydeps.PyDepWheel],
	*,
	console: rich.console.Console,
	fps: int = 24,
) -> collections.abc.Generator[CallbacksDownloadWheel, None, None]:
	"""Context manager creating a terminal UI to communicate wheel downloading.

	Notes:
		Yields callbacks to call during the download progress, in order for the UI to update correctly.

	Parameters:
		wheels_to_download: Set of wheels to download.
		console: `rich` console to print the UI to.
		fps: Number of updates to the terminal, per second.


	See Also:
		`blext.ui.download_wheels.CallbacksDownloadWheel`: For more on when to call each callback.
	"""
	bytes_to_download = sum(int(wheel.size) for wheel in wheels_to_download)

	####################
	# - Progress: Overall Download
	####################
	progress_download_all = rich.progress.Progress(
		rich.progress.SpinnerColumn(),
		'{task.description}',
		rich.progress.BarColumn(
			bar_width=None,
		),
		rich.progress.DownloadColumn(),
		expand=True,
		console=console,
	)
	task_download_all = progress_download_all.add_task(
		'Downloading',
		total=bytes_to_download,
		start=True,
	)

	####################
	# - Progress: Individual Wheels
	####################
	progress_download_wheels = {
		wheel: rich.progress.Progress(
			rich.progress.BarColumn(
				bar_width=None,
			),
			rich.progress.DownloadColumn(),
			expand=True,
			console=console,
		)
		for wheel in wheels_to_download
	}
	task_download_wheels: dict[pydeps.PyDepWheel, None | rich.progress.TaskID] = (
		dict.fromkeys(wheels_to_download)
	)

	####################
	# - Callbacks
	####################
	def cb_start_wheel_download(
		wheel: pydeps.PyDepWheel,
		_: Path,
	) -> None:
		task_download_wheels.update(
			{
				wheel: progress_download_wheels[wheel].add_task(
					wheel.filename,
					total=int(wheel.size),
				)
			}
		)

	def cb_update_wheel_download(
		wheel: pydeps.PyDepWheel,
		_: Path,
		advance: int,
	) -> None:
		task_download_wheel = task_download_wheels[wheel]
		if task_download_wheel is not None:
			progress_download_all.advance(task_download_all, advance=advance)
			progress_download_wheels[wheel].advance(
				task_download_wheel,
				advance=advance,
			)
		else:
			msg = f'`rich.progress.Task` was not initialized for `{wheel.filename}` - something is very wrong!'
			raise RuntimeError(msg)

	def cb_finish_wheel_download(
		wheel: pydeps.PyDepWheel,
		_: Path,
	) -> None:
		del task_download_wheels[wheel]
		del progress_download_wheels[wheel]

	####################
	# - Layout: Static UI
	####################
	max_wheel_project_length = max(
		[
			*[len(wheel.project) for wheel in wheels_to_download],
			len('Name'),
		]
	)

	def layout() -> rich.console.Group:
		"""Generate a static UI, valid at one moment in time."""
		# Progress (Overall)
		table_overall_progress = rich.table.Table(box=None)
		table_overall_progress.add_row(progress_download_all)
		table_overall_progress.add_row()

		# Progress (by Wheel)
		table_wheel_progress = rich.table.Table(expand=True, box=None)
		table_wheel_progress.add_column(
			'Name',
			header_style='bold',
			justify='left',
			no_wrap=True,
			min_width=max_wheel_project_length - 6,
		)
		table_wheel_progress.add_column(
			'Platforms',
			header_style='bold',
			style='italic',
			justify='left',
			no_wrap=False,
			ratio=2,
		)
		table_wheel_progress.add_column(
			'Progress',
			header_style='bold',
			justify='left',
			no_wrap=True,
			ratio=6,  ## Make this column greedy
		)
		for wheel in sorted(progress_download_wheels, key=lambda wheel: wheel.filename):
			if task_download_wheels[wheel] is not None:
				table_wheel_progress.add_row(
					wheel.project,
					wheel.pretty_bl_platforms,
					progress_download_wheels[wheel],
				)

		return rich.console.Group(
			table_overall_progress,
			table_wheel_progress,
		)

	####################
	# - Live: Dynamic UI
	####################
	with rich.live.Live(
		get_renderable=layout,
		console=console,
		transient=True,
		refresh_per_second=fps,
	):
		yield CallbacksDownloadWheel(
			cb_start_wheel_download=functools.partial(
				cb_start_wheel_download,
			),
			cb_update_wheel_download=functools.partial(
				cb_update_wheel_download,
			),
			cb_finish_wheel_download=functools.partial(
				cb_finish_wheel_download,
			),
		)

	progress_download_all.stop_task(task_download_all)

---

blext.ui.prepack_extension

Implements a UI that reports the progress of extension pre-packing.

CallbacksPrepackExtension pydantic-model

Bases: BaseModel

Callbacks to trigger in the process of pre-packing an extension.

Notes

• All callbacks additionally take a live= keyword argument
• All that a pre-pack agent has to ensure, is to allow the user to specify equivalent callbacks to these.
• The callback return values are never used for any purpose.

ATTRIBUTE	DESCRIPTION
cb_pre_file_write	Called as a file is about to be pre-packed. Called with the host path, and the zipfile path. Always called before cb_post_file_write for a file. TYPE: PrepackCallback
cb_post_file_write	Called after a file has been pre-packed. Called with the host path, and the zipfile path. Always called before cb_post_file_write for a file. TYPE: PrepackCallback

See Also

• blext.pack.prepack_extension: Prepack agent that uses equivalent callbacks.
• blext.ui.ui_prepack_extension: Context manager that provides this object.

Fields:

• cb_pre_file_write (PrepackCallback)
• cb_post_file_write (PrepackCallback)

PrepackCallback

Bases: Protocol

Callback for use in CallbacksPrepackExtension.

__call__

__call__(
	path: Path, zipfile_path: Path, *, live: Live
) -> Any

Signature of the callback.

Source code in blext/ui/prepack_extension.py

def __call__(  # pyright: ignore[reportAny]
	self,
	path: Path,
	zipfile_path: Path,
	*,
	live: rich.live.Live,
) -> typ.Any:
	"""Signature of the callback."""

ui_prepack_extension

ui_prepack_extension(
	files_to_prepack: frozendict[Path, Path]
	| dict[Path, Path],
	*,
	console: Console,
) -> Generator[CallbacksPrepackExtension, None, None]

Context manager creating a terminal UI to communicate extension prepacking progress.

Notes

Yields callbacks to call during the download progress, in order for the UI to update correctly.

PARAMETER	DESCRIPTION
files_to_prepack	Files to be prepack. Maps an absolute host filesystem path, to a relative zipfile path. TYPE: frozendict[Path, Path] | dict[Path, Path]
console	rich console to print the UI to. TYPE: Console

See Also

blext.ui.download_wheels.CallbacksDownloadWheel: For more on when to call each callback.

Source code in blext/ui/prepack_extension.py

@contextlib.contextmanager
def ui_prepack_extension(
	files_to_prepack: frozendict[Path, Path] | dict[Path, Path],
	*,
	console: rich.console.Console,
) -> collections.abc.Generator[CallbacksPrepackExtension, None, None]:
	"""Context manager creating a terminal UI to communicate extension prepacking progress.

	Notes:
		Yields callbacks to call during the download progress, in order for the UI to update correctly.

	Parameters:
		files_to_prepack: Files to be prepack.
			Maps an absolute host filesystem path, to a relative zipfile path.
		console: `rich` console to print the UI to.


	See Also:
		`blext.ui.download_wheels.CallbacksDownloadWheel`: For more on when to call each callback.
	"""
	file_sizes = {path: path.stat().st_size for path in files_to_prepack}
	bytes_to_prepack = sum(file_size for file_size in file_sizes.values())

	####################
	# - Progress: Overall Download
	####################
	remaining_files_to_prepack = set(files_to_prepack.keys())
	progress_prepack = rich.progress.Progress(
		rich.progress.SpinnerColumn(),
		'{task.description}',
		rich.progress.BarColumn(
			bar_width=None,
		),
		rich.progress.DownloadColumn(),
		expand=True,
		console=console,
	)
	task_prepack = progress_prepack.add_task(
		'Pre-Packing',
		total=bytes_to_prepack,
		start=True,
	)

	####################
	# - Callbacks
	####################
	def cb_pre_file_write(*_, **__) -> None:  # pyright: ignore[reportUnusedParameter, reportMissingParameterType, reportUnknownParameterType]
		pass

	def cb_post_file_write(
		path: Path,
		_: Path,
		*,
		live: rich.live.Live,
	) -> None:
		progress_prepack.advance(
			task_prepack,
			advance=file_sizes[path],
		)
		remaining_files_to_prepack.remove(path)
		live.refresh()

	####################
	# - Layout: Static UI
	####################
	def layout() -> rich.console.Group:
		"""Generate a static UI, valid at one moment in time."""
		# Progress (Overall)
		table_overall_progress = rich.table.Table(box=None)
		table_overall_progress.add_row(progress_prepack)
		table_overall_progress.add_row()

		# Progress (by Wheel)
		table_wheel_progress = rich.table.Table(expand=True, box=None)
		table_wheel_progress.add_column(
			'Size',
			header_style='bold',
			justify='left',
			no_wrap=True,
		)
		table_wheel_progress.add_column(
			'Packed Path',
			header_style='bold',
			justify='right',
			no_wrap=False,
		)
		for i, path in enumerate(
			sorted(remaining_files_to_prepack, key=lambda path: file_sizes[path])
		):
			style = {'style': 'green'} if i == 0 else {}
			table_wheel_progress.add_row(
				pyd.ByteSize(file_sizes[path]).human_readable(decimal=True),
				str(Path('wheels') / path.name),
				**style,  # pyright: ignore[reportArgumentType]
			)

		return rich.console.Group(
			table_overall_progress,
			table_wheel_progress,
		)

	####################
	# - Live: Dynamic UI
	####################
	with rich.live.Live(
		get_renderable=layout,
		console=console,
		transient=True,
		auto_refresh=False,
	) as live:
		yield CallbacksPrepackExtension(
			cb_pre_file_write=functools.partial(cb_pre_file_write, live=live),
			cb_post_file_write=functools.partial(cb_post_file_write, live=live),
		)

		progress_prepack.stop_task(task_prepack)