Update progress with results

[tlogemann]

Hello,

I would like to update a progress with the results of each stage. Therefore, I am now using this custom update method, see:
main...tlogemann:mr4mp:main

I assume, that progress is not only an Iterable, but a class (like StageProgress here), on that update is callable, see e.g.:

from typing import Optional, Generator

from tqdm import tqdm

from src.models.graph_structures.TreeMergingInput import TreeMergingInput


class StageProgress:
    def __init__(self):
        self.pbar: Optional[tqdm] = None
        self.stage_iterable: Optional[Generator] = None

    def __iter__(self):
        return self

    def __call__(self, stage_iterable):
        if self.pbar is None:
            self.pbar = tqdm()
        if self.stage_iterable is None:
            self.stage_iterable = stage_iterable
        return self

    def __next__(self):
        try:
            result = next(self.stage_iterable)
        except IndexError:
            raise StopIteration
        self.pbar.update(1)
        return result

    def close(self):
        self.pbar.close()

    def update(self, current_result: TreeMergingInput, stage_update: TreeMergingInput):
        # Source: https://stackoverflow.com/a/63693165
        num_nodes = current_result.tree.node_counter
        self.pbar.set_postfix({'num_nodes': num_nodes})

In my case the mapreduce calculation is very resource and time intensive and it is not known how many nodes are generated, thus it is of interest to follow the generation live.

Maybe this idea of updates of the current results after reduction in staged mode is of interest for you.