improved eta

CHANGELOG.md

@@ -49,6 +49,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
   - Changed `Tabs`
   - Changed `TextArea`
   - Changed `Tree`
+- Improved ETA calculation for ProgressBar https://github.com/Textualize/textual/pull/4271
 - BREAKING: `AppFocus` and `AppBlur` are now posted when the terminal window gains or loses focus, if the terminal supports this https://github.com/Textualize/textual/pull/4265
   - When the terminal window loses focus, the currently-focused widget will also lose focus.
   - When the terminal window regains focus, the previously-focused widget will regain focus.

docs/examples/widgets/progress_bar_isolated_.py

@@ -1,4 +1,5 @@
 from textual.app import App, ComposeResult
+from textual.clock import MockClock
 from textual.containers import Center, Middle
 from textual.timer import Timer
 from textual.widgets import Footer, ProgressBar
@@ -11,13 +12,14 @@ class IndeterminateProgressBar(App[None]):
     """Timer to simulate progress happening."""
 
     def compose(self) -> ComposeResult:
+        self.clock = MockClock()
         with Center():
             with Middle():
-                yield ProgressBar()
+                yield ProgressBar(clock=self.clock)
         yield Footer()
 
     def on_mount(self) -> None:
-        """Set up a timer to simulate progess happening."""
+        """Set up a timer to simulate progress happening."""
         self.progress_timer = self.set_interval(1 / 10, self.make_progress, pause=True)
 
     def make_progress(self) -> None:
@@ -31,14 +33,18 @@ def action_start(self) -> None:
 
     def key_f(self) -> None:
         # Freeze time for the indeterminate progress bar.
-        self.query_one(ProgressBar).query_one("#bar")._get_elapsed_time = lambda: 5
+        self.clock.set_time(5)
+        self.refresh()
 
     def key_t(self) -> None:
         # Freeze time to show always the same ETA.
-        self.query_one(ProgressBar).query_one("#eta")._get_elapsed_time = lambda: 3.9
-        self.query_one(ProgressBar).update(total=100, progress=39)
+        self.clock.set_time(0)
+        self.query_one(ProgressBar).update(total=100, progress=0)
+        self.clock.set_time(3.9)
+        self.query_one(ProgressBar).update(progress=39)
 
     def key_u(self) -> None:
+        self.refresh()
         self.query_one(ProgressBar).update(total=100, progress=100)
 
 

docs/examples/widgets/progress_bar_styled.py

@@ -18,7 +18,7 @@ def compose(self) -> ComposeResult:
         yield Footer()
 
     def on_mount(self) -> None:
-        """Set up a timer to simulate progess happening."""
+        """Set up a timer to simulate progress happening."""
         self.progress_timer = self.set_interval(1 / 10, self.make_progress, pause=True)
 
     def make_progress(self) -> None:

docs/examples/widgets/progress_bar_styled_.py

@@ -1,4 +1,5 @@
 from textual.app import App, ComposeResult
+from textual.clock import MockClock
 from textual.containers import Center, Middle
 from textual.timer import Timer
 from textual.widgets import Footer, ProgressBar
@@ -12,13 +13,14 @@ class StyledProgressBar(App[None]):
     """Timer to simulate progress happening."""
 
     def compose(self) -> ComposeResult:
+        self.clock = MockClock()
         with Center():
             with Middle():
-                yield ProgressBar()
+                yield ProgressBar(clock=self.clock)
         yield Footer()
 
     def on_mount(self) -> None:
-        """Set up a timer to simulate progess happening."""
+        """Set up a timer to simulate progress happening."""
         self.progress_timer = self.set_interval(1 / 10, self.make_progress, pause=True)
 
     def make_progress(self) -> None:
@@ -29,15 +31,18 @@ def action_start(self) -> None:
         """Start the progress tracking."""
         self.query_one(ProgressBar).update(total=100)
         self.progress_timer.resume()
+        self.query_one(ProgressBar).refresh()
 
     def key_f(self) -> None:
         # Freeze time for the indeterminate progress bar.
-        self.query_one(ProgressBar).query_one("#bar")._get_elapsed_time = lambda: 5
+        self.clock.set_time(5.0)
 
     def key_t(self) -> None:
         # Freeze time to show always the same ETA.
-        self.query_one(ProgressBar).query_one("#eta")._get_elapsed_time = lambda: 3.9
-        self.query_one(ProgressBar).update(total=100, progress=39)
+        self.clock.set_time(0)
+        self.query_one(ProgressBar).update(total=100, progress=0)
+        self.clock.set_time(3.9)
+        self.query_one(ProgressBar).update(progress=39)
 
     def key_u(self) -> None:
         self.query_one(ProgressBar).update(total=100, progress=100)

src/textual/clock.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from time import monotonic
+from typing import Callable
+
+import rich.repr
+
+
+@rich.repr.auto(angular=True)
+class Clock:
+    """An object to get relative time.
+
+    The `time` attribute of clock will return the time in seconds since the
+    Clock was created or reset.
+
+    """
+
+    def __init__(self, *, get_time: Callable[[], float] = monotonic) -> None:
+        """Create a clock.
+
+        Args:
+            get_time: A callable to get time in seconds.
+            start: Start the clock (time is 0 unless clock has been started).
+        """
+        self._get_time = get_time
+        self._start_time = self._get_time()
+
+    def __rich_repr__(self) -> rich.repr.Result:
+        yield self.time
+
+    def clone(self) -> Clock:
+        """Clone the Clock with an independent time."""
+        return Clock(get_time=self._get_time)
+
+    def reset(self) -> None:
+        """Reset the clock."""
+        self._start_time = self._get_time()
+
+    @property
+    def time(self) -> float:
+        """Time since creation or reset."""
+        return self._get_time() - self._start_time
+
+
+class MockClock(Clock):
+    """A mock clock object where the time may be explicitly set."""
+
+    def __init__(self, time: float = 0.0) -> None:
+        """Construct a mock clock."""
+        self._time = time
+        super().__init__(get_time=lambda: self._time)
+
+    def clone(self) -> MockClock:
+        """Clone the mocked clock (clone will return the same time as original)."""
+        clock = MockClock(self._time)
+        clock._get_time = self._get_time
+        clock._time = self._time
+        return clock
+
+    def reset(self) -> None:
+        """A null-op because it doesn't make sense to reset a mocked clock."""
+
+    def set_time(self, time: float) -> None:
+        """Set the time for the clock.
+
+        Args:
+            time: Time to set.
+        """
+        self._time = time
+
+    @property
+    def time(self) -> float:
+        """Time since creation or reset."""
+        return self._get_time()

src/textual/dom.py

@@ -474,13 +474,11 @@ def __init_subclass__(
         cls._merged_bindings = cls._merge_bindings()
         cls._css_type_names = frozenset(css_type_names)
         cls._computes = frozenset(
-            dict.fromkeys(
-                [
-                    name.lstrip("_")[8:]
-                    for name in dir(cls)
-                    if name.startswith(("_compute_", "compute_"))
-                ]
-            ).keys()
+            [
+                name.lstrip("_")[8:]
+                for name in dir(cls)
+                if name.startswith(("_compute_", "compute_"))
+            ]
         )
 
     def get_component_styles(self, name: str) -> RenderStyles:

src/textual/eta.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import bisect
+from math import ceil
+from time import monotonic
+
+import rich.repr
+
+
+@rich.repr.auto(angular=True)
+class ETA:
+    """Calculate speed and estimate time to arrival."""
+
+    def __init__(
+        self, estimation_period: float = 60, extrapolate_period: float = 30
+    ) -> None:
+        """Create an ETA.
+
+        Args:
+            estimation_period: Period in seconds, used to calculate speed.
+            extrapolate_period: Maximum number of seconds used to estimate progress after last sample.
+        """
+        self.estimation_period = estimation_period
+        self.max_extrapolate = extrapolate_period
+        self._samples: list[tuple[float, float]] = [(0.0, 0.0)]
+        self._add_count = 0
+
+    def __rich_repr__(self) -> rich.repr.Result:
+        yield "speed", self.speed
+        yield "eta", self.get_eta(monotonic())
+
+    @property
+    def first_sample(self) -> tuple[float, float]:
+        """First sample."""
+        assert self._samples, "Assumes samples not empty"
+        return self._samples[0]
+
+    @property
+    def last_sample(self) -> tuple[float, float]:
+        """Last sample."""
+        assert self._samples, "Assumes samples not empty"
+        return self._samples[-1]
+
+    def reset(self) -> None:
+        """Start ETA calculations from current time."""
+        del self._samples[:]
+
+    def add_sample(self, time: float, progress: float) -> None:
+        """Add a new sample.
+
+        Args:
+            time: Time when sample occurred.
+            progress: Progress ratio (0 is start, 1 is complete).
+        """
+        if self._samples and self.last_sample[1] > progress:
+            # If progress goes backwards, we need to reset calculations
+            self.reset()
+        self._samples.append((time, progress))
+        self._add_count += 1
+        if self._add_count % 100 == 0:
+            # Prune periodically so we don't accumulate vast amounts of samples
+            self._prune()
+
+    def _prune(self) -> None:
+        """Prune old samples."""
+        if len(self._samples) <= 10:
+            # Keep at least 10 samples
+            return
+        prune_time = self._samples[-1][0] - self.estimation_period
+        index = bisect.bisect_left(self._samples, (prune_time, 0))
+        del self._samples[:index]
+
+    def _get_progress_at(self, time: float) -> tuple[float, float]:
+        """Get the progress at a specific time."""
+
+        index = bisect.bisect_left(self._samples, (time, 0))
+        if index >= len(self._samples):
+            return self.last_sample
+        if index == 0:
+            return self.first_sample
+        # Linearly interpolate progress between two samples
+        time1, progress1 = self._samples[index - 1]
+        time2, progress2 = self._samples[index]
+        factor = (time - time1) / (time2 - time1)
+        intermediate_progress = progress1 + (progress2 - progress1) * factor
+        return time, intermediate_progress
+
+    @property
+    def speed(self) -> float | None:
+        """The current speed, or `None` if it couldn't be calculated."""
+
+        if len(self._samples) < 2:
+            # Need at least 2 samples to calculate speed
+            return None
+
+        recent_sample_time, progress2 = self.last_sample
+        progress_start_time, progress1 = self._get_progress_at(
+            recent_sample_time - self.estimation_period
+        )
+        time_delta = recent_sample_time - progress_start_time
+        distance = progress2 - progress1
+        speed = distance / time_delta if time_delta else 0
+        return speed
+
+    def get_eta(self, time: float) -> int | None:
+        """Estimated seconds until completion, or `None` if no estimate can be made.
+
+        Args:
+            time: Current time.
+        """
+        speed = self.speed
+        if not speed:
+            # Not enough samples to guess
+            return None
+        recent_time, recent_progress = self.last_sample
+        remaining = 1.0 - recent_progress
+        if remaining <= 0:
+            # Complete
+            return 0
+        # The bar is not complete, so we will extrapolate progress
+        # This will give us a countdown, even with no samples
+        time_since_sample = min(self.max_extrapolate, time - recent_time)
+        extrapolate_progress = speed * time_since_sample
+        # We don't want to extrapolate all the way to 0, as that would erroneously suggest it is finished
+        eta = max(1.0, (remaining - extrapolate_progress) / speed)
+        return ceil(eta)