Skip to content

progress

Progress

Progress handler for vectorial operations.

Use as context. Total number of operations must be known beforehand. Call update(op, n=1) to increment tracking.

Source code in linalg\progress.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class Progress:
    """Progress handler for vectorial operations.

    Use as context. Total number of operations must be known beforehand.
    Call `update(op, n=1)` to increment tracking.
    """

    def __init__(self, totals:dict[str,int], descprefix:str=''):
        """Create a new progress handler.

        `totals` should be a dictionary with the tracked operations as keys
        and the total number of operations as values.
        """
        self.pbars:dict[str,tqdm_type] = \
                {o:tqdm(desc=descprefix+o, total=t) for o, t in totals.items()}

    def __enter__(self) -> Self:
        return self

    def __exit__(self, exc_type:  Optional[Type[BaseException]],
                       exc_value: Optional[BaseException],
                       tb:        Optional[TracebackType]) -> bool:
        for pbar in self.pbars.values():
            pbar.close()
        return False

    def update(self, op:str, n:int=1) -> None:
        """Increment the operation `op` progress by `n`.

        If `op` is not tracked, nothing happens.
        """
        pbar = self.pbars.get(op)
        if pbar is not None:
            pbar.update(n)

    def add(self, a:Any, b:Any) -> Any:
        c:Any = a + b
        self.update('+')
        return c

    def mul(self, a:Any, b:Any) -> Any:
        c:Any = a * b
        self.update('*')
        return c

    def truediv(self, a:Any, b:Any) -> Any:
        c:Any = a / b
        self.update('/')
        return c

    def sum_default(self, iterable:Iterable[Any], *, initial:Any=MISSING, default:Any=0) -> Any:
        return reduce_default(self.add, iterable, initial=initial, default=default)

    def prod_default(self, iterable:Iterable[Any], *, initial:Any=MISSING, default:Any=1) -> Any:
        return reduce_default(self.mul, iterable, initial=initial, default=default)

    def sumprod_default(self, a:Iterable[Any], b:Iterable[Any], *, initial:Any=MISSING, default:Any=0) -> Any:
        return self.sum_default(map(self.mul, a, b), initial=initial, default=default)

__init__(totals, descprefix='')

Create a new progress handler.

totals should be a dictionary with the tracked operations as keys and the total number of operations as values.

Source code in linalg\progress.py 
21
22
23
24
25
26
27
28
def __init__(self, totals:dict[str,int], descprefix:str=''):
    """Create a new progress handler.

    `totals` should be a dictionary with the tracked operations as keys
    and the total number of operations as values.
    """
    self.pbars:dict[str,tqdm_type] = \
            {o:tqdm(desc=descprefix+o, total=t) for o, t in totals.items()}

update(op, n=1)

Increment the operation op progress by n.

If op is not tracked, nothing happens.

Source code in linalg\progress.py 
40
41
42
43
44
45
46
47
def update(self, op:str, n:int=1) -> None:
    """Increment the operation `op` progress by `n`.

    If `op` is not tracked, nothing happens.
    """
    pbar = self.pbars.get(op)
    if pbar is not None:
        pbar.update(n)