`support.progress`¶

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

Progress tracker.

This context manager is used to indicate progress of a slow process within applet code in a generic way, decoupling it from a specific type of user interface. For example, if an applet is used with a CLI frontend, terminal escape sequences could be used to repeatedly display a text description on the last used line, whereas with a GUI frontend, a progress bar widget would be used.

To use it, modify applet code that processes multiple items by wrapping it in with Progress(...):. For example, let’s consider a method that uploads a bitstream via SPIControllerInterface:

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

To add progress indication here, the bitstream will need to be explicitly chunked (while avoiding copies using memoryview) and individual chunks iterated through:

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)

Note

The synchronize() call is added because the write() method does not wait for the operation to complete; without it, the load() method would quickly finish, with the command buffer draining slowly after. In most cases (e.g. Flash programming) there is a requirement to wait for feedback, making explicit synchronization unnecessary.

The particluar case of chunking a sequence (of bytes, or otherwise) is common enough that this class provides a helper method for it, chunks(). Using this helper the load() function becomes:

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()

With this change to the applet code, a CLI frontend can now indicate the progress of the operation, for example, by printing:

programming... 13% (65536/500000 bytes)
programming... 26% (131072/500000 bytes)
(and so on until completion)

classmethod chunks(items: T, chunk_size: int, **kwargs) → Generator¶

Chunked progress tracker.

This helper method exists to handle the case where a sequence must be brought into chunks for processing, but progress must be tracked per-item, not per-chunk. For example, when manipulating large byte sequences, it would be too inefficient to process bytes one by one just to report progress, but processing 64 KiB chunks is usually fine.

All keyword arguments not explicitly declared are passed to the class constructor. Returns a generator yielding the result of indexing slices of items up to chunk_size in length, and advancing progress whenever control is returned.

To split a byte array into chunks of up to chunk_size, such as for a write operation, use:

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

To split an address range into chunks of up to chunk_size, such as for a read operation, use:

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`

Note

Avoid passing bytes or bytearray values as items; wrap them in memoryview so that chunks reference the original bytes instead of copying.

property action: str¶

Action taking progress.

Should have a present participle with no punctuation at the end, e.g. "writing" or "programming flash".

property item: str | None¶

Item being operated on.

Should be a singular noun, e.g. "byte" or "B". If None, then the unit of processing is unspecified: either clear from context, or not feasible to specify exactly.

property scale: Literal[1, 1000, 1024]¶

Scale of item count.

May be one of 1, 1000, or 1024. If not 1, an appropriate SI prefix (K, M, G… or Ki, Mi, Gi…) will be used when counting items.

property total: int | None¶

Total number of items.

If None, impossible to know until the operation finishes.

property done: int¶: Number of completed items.

advance(count: int)¶

Indicate completion of count items.

Advancing progress beyond self.total completed items may result in confusing UI behavior, but will not raise errors. However, count must not be negative.

__enter__()¶

Adds self to the stack of displayed progress indicators.

This dunder method is called implicitly by the with Progress(...): block.

__exit__(exc_type, exc_value, traceback)¶

Unregister the progress indicator.

Removes self from the stack of displayed progress indicators.

This dunder method is called implicitly by the with Progress(...): block.

support.progress¶

`support.progress`¶