support.progress

class glasgow.support.progress.Progress(*, total: int | None = None, action: str, item: str | None = None, scale: Literal[1, 1000, 1024] = 1)

进度跟踪组件。

该上下文管理器用于通用地、与具体界面类型解耦地显示一个小程序代码中进展较慢的流程的进度。例如,如果一个小程序和 CLI 前端配合使用,则它可以使用终端转义序列,在输出所在的最后一行反复更新显示文字描述,而如果是和 GUI 前端配合,则它可以使用一个进度条组件。

如需使用,请修改小程序代码中处理多项元素的部分,将其包在 with Progress(...): 中。例如,假设有一个用 SPIControllerInterface 来上传位流的方法:

async def load(bitstream: bytes):
    async with self._spi.select():
        await self._spi.write(bitstream)

如需在这里加入进度显示,则位流需要被显式地分段(用 memoryview 以避免复制),然后迭代每段分别处理:

async def load(bitstream: bytes):
    bitstream  = memoryview(bitstream)
    chunk_size = 0x10000
    async with self._spi.select():
        with Progress(total=len(bitstream),
                action="programming", item="B", scale=1024) as progress:
            for start in range(0, len(bitstream), chunk_size):
                await self._spi.write(bitstream[start:start + chunk_size])
                await self._spi.synchronize()
                progress.advance(chunk_size)

备注

这里加入了一行调用 synchronize() 的原因是 write() 方法并不等待操作完成。如果没有这一行,则 load() 会快速完成,而之后命令缓冲区会逐渐慢慢被消耗。在大部分情况下(如闪存编程时)等待完成反馈是必须的,此时显式同步不是必要的。

这个将一个序列(元素是字节或其它内容)分段的情况很常见,所以这个类提供一个辅助方法 chunks()。使用此辅助方法,load() 函数可写为:

async def load(bitstream: bytes):
    async with self._spi.select():
        for chunk in Progress.chunks(memoryview(bitstream), 0x10000,
                action="programming", item="B", scale=1024):
            await self._spi.write(chunk)
            await self._spi.synchronize()

经过这样修改小程序代码后,CLI 前端可以显示操作的进度,例如输出如下:

programming... 13% (65536/500000 bytes)
programming... 26% (131072/500000 bytes)
(以此类推,直到完成)
classmethod chunks(items: T, chunk_size: int, **kwargs) → Generator

分段进度跟踪组件。

该辅助方法用于处理一个序列需要分段处理,但是需要按元素数而不是按段数跟踪进度的情况。例如,在处理较大的字节序列的时候,为了汇报进度而一个字节一个字节处理太过低效,而按每段 64 KiB 分段处理一般没有问题。

所有没有显式指定的关键字参数都传给类构造器。返回一个生成器,产生 items 中抽取的长度不超过 chunk_size 的切片,并在每次控制流返回时记录进度前进。如果 len(items) <= chunk_size,则完全不显示进度。

如需将一个字节数组切分为长度不超过 chunk_size 的段(例如为写入操作所用),用法如下:

for chunk in Progress.chunks(memoryview(data), chunk_size,
        action="programming", item="B", scale=1024):
    write_data(chunk) # here `chunk` is a `memoryview`

如需将一个地址范围切分为长度不超过 chunk_size 的段(例如为读取操作所用),用法如下:

for chunk in Progress.chunks(range(start, start + length), chunk_size,
        action="reading", item="B", scale=1024):
    data += read_data(chunk.start, len(chunk)) # here `chunk` is a `range`

备注

避免将 bytes 或 bytearray 值作为 items 传入,而应将其包装在 memoryview 中,这样每个段可以直接引用原来的字节,而不需要复制一份。

property action: str

正在进行的操作。

应包含动词的现在分词形式(如为英文),且后面不带标点符号,如 "writing" 或 "programming flash"。

property item: str | None

所操作的元素。

应为一个单数名词(如为英文),如 "byte" 或 "B"。如为 None,则处理的单位未指定:要么从上下文中可以明确,要么无法准确给出。

property scale: Literal[1, 1000, 1024]

元素数量的基数。

可以为 1、1000、1024 之一。如果不是 1,则在对元素计数时会使用相应的 SI 前缀(K、M、G……或 Ki、Mi、Gi……)。

property total: int | None

元素总数。

如果为 None,则表示在操作完成前无法预知。

property done: int

完成的元素数量。

advance(count: int)

表示新有 count 个元素完成。

将进度前进到超过 self.total 个完成的元素可能会导致 UI 显示令用户困惑,但是不会引发异常。此外,count 不得为负。

__enter__()

注册进度指示组件。

将 self 加入当前显示的进度指示组件的栈中。

该 dunder 方法由 with Progress(...): 块隐式调用。

__exit__(exc_type, exc_value, traceback)

注销进度指示组件。

将 self 从当前显示的进度指示组件的栈中移除。

该 dunder 方法由 with Progress(...): 块隐式调用。

class glasgow.support.progress.TqdmProgressImpl

基于 تقدّم 的终端进度显示实现。

该实现自动由 glasgow.main.run_async() 注册,但是如有需要可以显式使用。

class glasgow.support.progress.QtWidgetsProgressImpl(progressBar: QProgressBar | None, statusLabel: QLabel | None)

为 Qt 应用所用的基于 QtWidgets 的进度显示实现。

该实现需要手动注册。它需要在构造时提供一个 QProgressBar 和/或一个 QLabel;这些部件会被用于显示进度。用法如下:

class MainWindow(QMainWindow, Ui_MainWindow):
    def __init__(self):
        super().__init__()

        self.setupUi(self)
        QtWidgetsProgressImpl(self.progressBar, self.statusLabel).register()

    ...

备注

该实现有一个限制,它无法显示多个进度条。进度信息在更新时总是使用的是最后一个打开的进度指示。