yt_dlp/downloader/fragment.py

@@ -257,7 +257,6 @@ class FragmentFD(FileDownloader):
             frag_total_bytes = s.get('total_bytes') or 0
             s['fragment_info_dict'] = s.pop('info_dict', {})
 
-            # XXX: Fragment resume is not accounted for here
             if not ctx['live']:
                 estimated_size = (
                     (ctx['complete_frags_downloaded_bytes'] + frag_total_bytes)
@@ -270,11 +269,13 @@ class FragmentFD(FileDownloader):
 
             if s['status'] == 'finished':
                 state['fragment_index'] += 1
-                ctx['fragment_index'] = state['fragment_index']
                 progress.thread_reset()
-
+                ctx['fragment_index'] = state['fragment_index']
                 state['downloaded_bytes'] = ctx['complete_frags_downloaded_bytes'] = progress.downloaded
-            ctx['speed'] = state['speed'] = progress.smooth_speed
+                ctx['speed'] = state['speed'] = progress.speed
+            else:
+                state['downloaded_bytes'] = progress.downloaded
+                ctx['speed'] = state['speed'] = progress.speed
                 state['eta'] = progress.eta
 
             self._hook_progress(state, info_dict)

yt_dlp/utils/progress.py

@@ -4,38 +4,44 @@ import bisect
 import threading
 import time
 
+# FIXME: monotonic has terrible resolution on Windows
+TIME_PROVIDER = time.perf_counter
 
+
+# XXX: Fragment resume is not accounted for here
 class ProgressCalculator:
-    # Time to calculate the speed over (in nanoseconds)
-    SAMPLING_WINDOW = 1_000_000_000
-    # Factor for the exponential moving average (from 0 = prev to 1 = current)
-    SMOOTHING_FACTOR = 0.3
+    # Time to calculate the average over (in seconds)
+    WINDOW_SIZE = 2.0
+    # Minimum time before we add another datapoint (in seconds)
+    # This is *NOT* the same as the time between a progress change
+    UPDATE_TIME = 0.1
 
-    def __init__(self, initial: int):
-        self.downloaded = initial or 0
-
-        self.elapsed: float = 0
-        self.speed: float = 0
-        self.smooth_speed: int = 0
-        self.eta: float | None = None
+    def __init__(self, initial):
+        self.downloaded = initial if initial else 0
 
+        self.elapsed = 0
+        self.speed = None
+        self.eta = None
         self._total = None
-        self._start_time = time.monotonic_ns()
 
         self._lock = threading.Lock()
-        self._thread_sizes: dict[int, int] = {}
+        self._start_time = TIME_PROVIDER()
+        self._last_update = self._start_time
 
-        self._times = [self._start_time]
-        self._downloaded = [self.downloaded]
+        self._times: list[float] = []
+        self._speeds: list[int] = []
+
+        self._thread_updates: dict[int, float] = {}
+        self._thread_sizes: dict[int, int] = {}
 
     @property
     def total(self):
         return self._total
 
     @total.setter
-    def total(self, value: int | None):
+    def total(self, value):
         with self._lock:
-            if not value:
+            if not value or value <= 0.01:
                 value = None
             elif value < self.downloaded:
                 value = self.downloaded
@@ -46,6 +52,7 @@ class ProgressCalculator:
         current_thread = threading.get_ident()
         with self._lock:
             self._thread_sizes[current_thread] = 0
+            self._thread_updates[current_thread] = TIME_PROVIDER()
 
     def update(self, size: int | None):
         if not size:
@@ -54,34 +61,39 @@ class ProgressCalculator:
         current_thread = threading.get_ident()
 
         with self._lock:
-            last_size = self._thread_sizes.get(current_thread, 0)
+            current_time = TIME_PROVIDER()
+            last_size = self._thread_sizes.get(current_thread) or 0
+            last_update = self._thread_updates.get(current_thread) or self._start_time
+            chunk = size - last_size
+            print(f'    [{threading.get_ident()}] {last_update} -> {current_time} ({chunk}B)')
+
+            update_time = self._update(chunk, current_time, last_update)
+            if update_time:
+                self._thread_updates[current_thread] = current_time
             self._thread_sizes[current_thread] = size
-            self._update(size - last_size)
-
-    def _update(self, size: int):
-        current_time = time.monotonic_ns()
 
+    def _update(self, size, current_time, last_update):
         self.downloaded += size
-        self.elapsed = (current_time - self._start_time) / 1_000_000_000
+        self.elapsed = current_time - self._start_time
         if self.total is not None and self.downloaded > self.total:
             self._total = self.downloaded
 
         self._times.append(current_time)
-        self._downloaded.append(self.downloaded)
+        self._speeds.append(size / (current_time - last_update))
 
-        offset = bisect.bisect_left(self._times, current_time - self.SAMPLING_WINDOW)
-        del self._times[:offset]
-        del self._downloaded[:offset]
-
-        download_time = current_time - self._times[0]
-        if not download_time:
+        if current_time - self.UPDATE_TIME <= last_update:
             return
-        downloaded_bytes = self.downloaded - self._downloaded[0]
+        self._last_update = current_time
 
-        self.speed = downloaded_bytes * 1_000_000_000 / download_time
-        self.smooth_speed = int(self.SMOOTHING_FACTOR * self.speed + (1 - self.SMOOTHING_FACTOR) * self.smooth_speed)
+        offset = bisect.bisect_left(self._times, current_time - self.WINDOW_SIZE)
+        del self._times[:offset]
+        del self._speeds[:offset]
+
+        weights = tuple(1 + (point - current_time) / self.WINDOW_SIZE for point in self._times)
+        # Same as `statistics.fmean(self.data_points, weights)`, but weights is >=3.11
+        self.speed = sum(map(float.__mul__, self._speeds, weights)) / sum(weights)
 
         if not self.total:
             self.eta = None
-        elif self.speed:
+        else:
             self.eta = (self.total - self.downloaded) / self.speed