Added smoothing eta to fix #280. The previous algorithm could be really jumpy in some cases and has been replaced with an exponential moving average. Double exponential moving average is also available

Author: wolph

progressbar/algorithms.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+import abc
+from datetime import timedelta
+
+
+class SmoothingAlgorithm(abc.ABC):
+
+    @abc.abstractmethod
+    def __init__(self, **kwargs):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def update(self, new_value: float, elapsed: timedelta) -> float:
+        '''Updates the algorithm with a new value and returns the smoothed
+        value.
+        '''
+        raise NotImplementedError
+
+
+class ExponentialMovingAverage(SmoothingAlgorithm):
+    '''
+    The Exponential Moving Average (EMA) is an exponentially weighted moving
+    average that reduces the lag that's typically associated with a simple
+    moving average. It's more responsive to recent changes in data.
+    '''
+
+    def __init__(self, alpha: float = 0.5) -> None:
+        self.alpha = alpha
+        self.value = 0
+
+    def update(self, new_value: float, elapsed: timedelta) -> float:
+        self.value = self.alpha * new_value + (1 - self.alpha) * self.value
+        return self.value
+
+
+class DoubleExponentialMovingAverage(SmoothingAlgorithm):
+    '''
+    The Double Exponential Moving Average (DEMA) is essentially an EMA of an
+    EMA, which reduces the lag that's typically associated with a simple EMA.
+    It's more responsive to recent changes in data.
+    '''
+
+    def __init__(self, alpha: float=0.5) -> None:
+        self.alpha = alpha
+        self.ema1 = 0
+        self.ema2 = 0
+
+    def update(self, new_value: float, elapsed: timedelta) -> float:
+        self.ema1 = self.alpha * new_value + (1 - self.alpha) * self.ema1
+        self.ema2 = self.alpha * self.ema1 + (1 - self.alpha) * self.ema2
+        return 2 * self.ema1 - self.ema2
+

tests/test_algorithms.py

@@ -0,0 +1,47 @@
+import pytest
+from datetime import timedelta
+
+from progressbar import algorithms
+
+
+def test_ema_initialization():
+    ema = algorithms.ExponentialMovingAverage()
+    assert ema.alpha == 0.5
+    assert ema.value == 0
+
+@pytest.mark.parametrize('alpha, new_value, expected', [
+    (0.5, 10, 5),
+    (0.1, 20, 2),
+    (0.9, 30, 27),
+    (0.3, 15, 4.5),
+    (0.7, 40, 28),
+    (0.5, 0, 0),
+    (0.2, 100, 20),
+    (0.8, 50, 40)
+])
+def test_ema_update(alpha, new_value, expected):
+    ema = algorithms.ExponentialMovingAverage(alpha)
+    result = ema.update(new_value, timedelta(seconds=1))
+    assert result == expected
+
+def test_dema_initialization():
+    dema = algorithms.DoubleExponentialMovingAverage()
+    assert dema.alpha == 0.5
+    assert dema.ema1 == 0
+    assert dema.ema2 == 0
+
+@pytest.mark.parametrize('alpha, new_value, expected', [
+    (0.5, 10, 7.5),
+    (0.1, 20, 3.8),
+    (0.9, 30, 29.7),
+    (0.3, 15, 7.65),
+    (0.5, 0, 0),
+    (0.2, 100, 36.0),
+    (0.8, 50, 48.0)
+])
+def test_dema_update(alpha, new_value, expected):
+    dema = algorithms.DoubleExponentialMovingAverage(alpha)
+    result = dema.update(new_value, timedelta(seconds=1))
+    assert result == expected
+
+# Additional test functions can be added here as needed.

tests/test_monitor_progress.py

@@ -140,20 +140,18 @@ def test_rapid_updates(testdir):
     )
     result.stderr.lines = _non_empty_lines(result.stderr.lines)
     pprint.pprint(result.stderr.lines, width=70)
-    result.stderr.fnmatch_lines(
-        [
-            '  0% (0 of 10) |      | Elapsed Time: ?:00:00 ETA:  --:--:--',
-            ' 10% (1 of 10) |      | Elapsed Time: ?:00:01 ETA:   ?:00:09',
-            ' 20% (2 of 10) |#     | Elapsed Time: ?:00:02 ETA:   ?:00:08',
-            ' 30% (3 of 10) |#     | Elapsed Time: ?:00:03 ETA:   ?:00:07',
-            ' 40% (4 of 10) |##    | Elapsed Time: ?:00:04 ETA:   ?:00:06',
-            ' 50% (5 of 10) |###   | Elapsed Time: ?:00:05 ETA:   ?:00:05',
-            ' 60% (6 of 10) |###   | Elapsed Time: ?:00:07 ETA:   ?:00:06',
-            ' 70% (7 of 10) |####  | Elapsed Time: ?:00:09 ETA:   ?:00:06',
-            ' 80% (8 of 10) |####  | Elapsed Time: ?:00:11 ETA:   ?:00:04',
-            ' 90% (9 of 10) |##### | Elapsed Time: ?:00:13 ETA:   ?:00:02',
-            '100% (10 of 10) |#####| Elapsed Time: ?:00:15 Time:  ?:00:15',
-        ],
+    result.stderr.fnmatch_lines(['  0% (0 of 10) |      | Elapsed Time: 0:00:00 ETA:  --:--:--',
+         ' 10% (1 of 10) |      | Elapsed Time: 0:00:01 ETA:   0:00:09',
+         ' 20% (2 of 10) |#     | Elapsed Time: 0:00:02 ETA:   0:00:08',
+         ' 30% (3 of 10) |#     | Elapsed Time: 0:00:03 ETA:   0:00:07',
+         ' 40% (4 of 10) |##    | Elapsed Time: 0:00:04 ETA:   0:00:06',
+         ' 50% (5 of 10) |###   | Elapsed Time: 0:00:05 ETA:   0:00:05',
+         ' 60% (6 of 10) |###   | Elapsed Time: 0:00:07 ETA:   0:00:04',
+         ' 70% (7 of 10) |####  | Elapsed Time: 0:00:09 ETA:   0:00:03',
+         ' 80% (8 of 10) |####  | Elapsed Time: 0:00:11 ETA:   0:00:02',
+         ' 90% (9 of 10) |##### | Elapsed Time: 0:00:13 ETA:   0:00:01',
+         '100% (10 of 10) |#####| Elapsed Time: 0:00:15 Time:  0:00:15',
+     ]
     )
 
 

tests/test_windows.py

@@ -0,0 +1,70 @@
+import time
+import sys
+import pytest
+
+if sys.platform.startswith('win'):
+    import win32console  # "pip install pypiwin32" to get this
+else:
+    pytest.skip('skipping windows-only tests', allow_module_level=True)
+
+
+import progressbar
+
+_WIDGETS = [progressbar.Percentage(), ' ',
+            progressbar.Bar(), ' ',
+            progressbar.FileTransferSpeed(), ' ',
+            progressbar.ETA()]
+_MB = 1024 * 1024
+
+
+# ---------------------------------------------------------------------------
+def scrape_console(line_count):
+    pcsb = win32console.GetStdHandle(win32console.STD_OUTPUT_HANDLE)
+    csbi = pcsb.GetConsoleScreenBufferInfo()
+    col_max = csbi['Size'].X
+    row_max = csbi['CursorPosition'].Y
+
+    line_count = min(line_count, row_max)
+    lines = []
+    for row in range(line_count):
+        pct = win32console.PyCOORDType(0, row + row_max - line_count)
+        line = pcsb.ReadConsoleOutputCharacter(col_max, pct)
+        lines.append(line.rstrip())
+    return lines
+
+
+# ---------------------------------------------------------------------------
+def runprogress():
+    print('***BEGIN***')
+    b = progressbar.ProgressBar(widgets=['example.m4v: '] + _WIDGETS,
+                                max_value=10 * _MB)
+    for i in range(10):
+        b.update((i + 1) * _MB)
+        time.sleep(0.25)
+    b.finish()
+    print('***END***')
+    return 0
+
+
+# ---------------------------------------------------------------------------
+def find(L, x):
+    try:
+        return L.index(x)
+    except ValueError:
+        return -sys.maxsize
+
+
+# ---------------------------------------------------------------------------
+def test_windows(argv):
+    runprogress()
+
+    scraped_lines = scrape_console(100)
+    scraped_lines.reverse()  # reverse lines so we find the LAST instances of output.
+    index_begin = find(scraped_lines, '***BEGIN***')
+    index_end = find(scraped_lines, '***END***')
+
+    if index_end + 2 != index_begin:
+        print('ERROR: Unexpected multi-line output from progressbar')
+        print(f'{index_begin=} {index_end=}')
+        return 1
+    return 0