python - 分离进度跟踪和循环逻辑

Question

假设我想使用进度条打印机 ProgressMeter 跟踪循环的进度(如 recipe 中所述)。

def bigIteration(collection):
    for element in collection:
        doWork(element)

我希望能够打开和关闭进度条。出于性能原因，我还想每 x 步更新一次。我天真的方法是

def bigIteration(collection, progressbar=True):
    if progressBar:
        pm = progress.ProgressMeter(total=len(collection))
        pc = 0
    for element in collection:
        if progressBar:
            pc += 1
            if pc % 100 = 0:
                pm.update(pc)
        doWork(element)

但是，我并不满意。从“美学”的角度来看，循环的功能代码现在被通用进度跟踪代码“污染”了。

你能想出一种方法来清楚地分离进度跟踪代码和功能代码吗？ (可以有一个进度跟踪装饰器什么的吗？)

Answer 1

看起来这段代码会受益于 null object pattern .

# a progress bar that uses ProgressMeter
class RealProgressBar:
     pm = Nothing
     def setMaximum(self, max):
         pm = progress.ProgressMeter(total=max)
         pc = 0
     def progress(self):
        pc += 1
        if pc % 100 = 0:
            pm.update(pc)

# a fake progress bar that does nothing
class NoProgressBar:
    def setMaximum(self, max):
         pass 
    def progress(self):
         pass

# Iterate with a given progress bar
def bigIteration(collection, progressBar=NoProgressBar()):
    progressBar.setMaximum(len(collection))
    for element in collection:
        progressBar.progress()
        doWork(element)

bigIteration(collection, RealProgressBar())

(请原谅我的法语，呃，Python，它不是我的母语；)不过希望你能理解。)

这让您可以将进度更新逻辑从循环中移出，但您仍然有一些与进度相关的调用。

如果您从集合中创建一个在迭代时自动跟踪进度的生成器，则可以删除此部分。

 # turn a collection into one that shows progress when iterated
 def withProgress(collection, progressBar=NoProgressBar()):
      progressBar.setMaximum(len(collection))
      for element in collection:
           progressBar.progress();
           yield element

 # simple iteration function
 def bigIteration(collection):
    for element in collection:
        doWork(element)

 # let's iterate with progress reports
 bigIteration(withProgress(collection, RealProgressBar()))

这种方法使您的 bigIteration 函数保持原样并且具有高度可组合性。例如，假设您还想在您的这个大迭代中添加取消。只需创建另一个恰好可取消的生成器即可。

# highly simplified cancellation token
# probably needs synchronization
class CancellationToken:
     cancelled = False
     def isCancelled(self):
         return cancelled
     def cancel(self):
         cancelled = True

# iterates a collection with cancellation support
def withCancellation(collection, cancelToken):
     for element in collection:
         if cancelToken.isCancelled():
             break
         yield element

progressCollection = withProgress(collection, RealProgressBar())
cancellableCollection = withCancellation(progressCollection, cancelToken)
bigIteration(cancellableCollection)

# meanwhile, on another thread...
cancelToken.cancel()