models: publish the download speed and derive the eta from it (#2052)

* models: publish the download speed and derive the eta from it

Fills in DownloadProgress.speed (bytes per second, declared with the parallel
downloader). The reporter measures the rate between its 0.5 s reports, smooths
it, and derives the eta from the bytes remaining at that rate instead of
extrapolating elapsed time. Speed is zeroed when a download ends, fails or is
cancelled and for a cached artifact, and copied across artifacts that share a
file like the other progress fields.

The big UI's Models page shows "12.3 MB/s" next to the progress bar and the
mici models panel shows it beside the percentage. Both take the max across a
bundle's artifacts, not the sum, since artifacts sharing a file mirror the same
transfer and only one file is ever in flight.

A stall decays the smoothed speed toward 0 without ever reaching it, so below
0.05 MB/s, which would only display as "0.0 MB/s", it snaps to 0. Both UIs
then show "--.- MB/s" while downloading with no rate (before the first reading
or during a stall) instead of a stuck "0.0 MB/s" or nothing, and the speed
recovers straight to the real rate once data flows again. The snap also keeps
the eta in range: without it a stall of about 25 s drove remaining / speed past
the capnp UInt32 eta, and the capnp error failed the download outright instead
of leaving the stall to the read timeout and retries. A test holds a piece
stalled until the published speed reaches exactly 0, and errors on that
overflow without the snap.

Assisted-by: Claude Code (Claude Opus 4.8)
Assisted-by: Claude Code (Claude Opus 5.5)

* remove unneeded comments

Assisted-by: Claude Code (Claude Opus 5.5)

* models: take the speed baseline before the workers start

The reporter took its starting byte count after the piece workers were
submitted, and starting each pool thread lets earlier workers run, so
bytes landed in that gap never counted toward a speed sample. Take the
baseline in _download_file before any worker exists and pass it in.

The test runs every piece inline in submit, so the whole file lands
before the reporter starts; with the late baseline the speed stays 0.

Assisted-by: Claude Code (Claude Opus 5.5)

openpilot/selfdrive/ui/sunnypilot/layouts/settings/models.py

@@ -243,7 +243,9 @@ class ModelsLayout(Widget):
     if ds.verifying in statuses:
       return {"name": name, "downloading": True, "progress": progress, "status_text": tr("verifying")}
     if ds.downloading in statuses:
-      return {"name": name, "downloading": True, "progress": progress}
+      speed = max((p.speed for p in progresses), default=0.0)
+      status_text = f"{speed / 1024 ** 2:.1f} MB/s" if speed > 0 else "--.- MB/s"
+      return {"name": name, "downloading": True, "progress": progress, "status_text": status_text}
     if statuses <= {ds.downloaded, ds.cached}:
       return {"name": name, "text_color": ON_COLOR, "icon": "icons/checkmark.png"}
     # circled_slash is authored grey; tinting it again only darkens it

openpilot/selfdrive/ui/sunnypilot/mici/layouts/models.py

@@ -257,6 +257,7 @@ class ModelsLayoutMici(NavScroller):
       count = 0
       verifying = False
       canceling = False
+      speed = 0.0
       for model in manager.selectedBundle.models:
         count += 1
         p = model.artifact.downloadProgress
@@ -265,6 +266,7 @@ class ModelsLayoutMici(NavScroller):
           progress += p.progress
           verifying = verifying or p.status == custom.ModelManagerSP.DownloadStatus.verifying
           canceling = canceling or p.canceling
+          speed = max(speed, p.speed)  # only one file downloads at a time, so the non-zero speed is the live one
         elif p.status in (custom.ModelManagerSP.DownloadStatus.downloaded,
                           custom.ModelManagerSP.DownloadStatus.cached):
           progress += 100.0
@@ -279,7 +281,10 @@ class ModelsLayoutMici(NavScroller):
       self.current_model_info.current_model_text.set_text(name_text)
       self.current_model_info.info_header.set_text(tr("progress") + self._download_progress)
       self.current_model_info.info_header._shimmer = True
-      self.current_model_info.info_text.set_text(f"{progress/count:.2f}%")
+      progress_text = f"{progress/count:.2f}%"
+      if not verifying and not canceling:
+        progress_text += f"  |  {speed / 1024 ** 2:.1f} MB/s" if speed > 0 else "  |  --.- MB/s"
+      self.current_model_info.info_text.set_text(progress_text)
 
     elif manager.selectedBundle and manager.selectedBundle.status == custom.ModelManagerSP.DownloadStatus.downloaded:
       self.current_model_info.info_header.set_text(tr("downloaded"))

openpilot/sunnypilot/models/manager.py

@@ -34,6 +34,8 @@ TRANSIENT_HTTP_STATUS = frozenset({408, 429, 500, 502, 503, 504})
 REPORT_INTERVAL = 0.5  # seconds between progress publications
 # byte ranges count encoded bytes but iter_content yields decoded ones, so ask for an unencoded body
 NO_COMPRESSION = {"Accept-Encoding": "identity"}
+SPEED_SMOOTHING = 0.7  # weight of the previous speed sample in the published speed
+SPEED_FLOOR = 0.05 * 1024 ** 2  # bytes/s; anything slower would only display as 0.0 MB/s, so it reads as no rate
 
 
 class DownloadCancelled(Exception):
@@ -154,6 +156,7 @@ class ModelManagerSP:
         artifact.downloadProgress.status = source_artifact.downloadProgress.status
         artifact.downloadProgress.progress = source_artifact.downloadProgress.progress
         artifact.downloadProgress.eta = source_artifact.downloadProgress.eta
+        artifact.downloadProgress.speed = source_artifact.downloadProgress.speed
 
   def _calculate_eta(self, filename: str, progress: float) -> int:
     if filename not in self._download_start_times or progress <= 0:
@@ -167,28 +170,43 @@ class ModelManagerSP:
     eta = total_estimated_time - elapsed_time
     return max(1, int(eta))  # Return at least 1 second if download is ongoing
 
-  def _set_progress(self, artifact, status, progress: float, eta: int = 0) -> None:
+  def _set_progress(self, artifact, status, progress: float, eta: int = 0, speed: float = 0.0) -> None:
     artifact.downloadProgress.status = status
     artifact.downloadProgress.progress = progress
     artifact.downloadProgress.eta = eta
+    artifact.downloadProgress.speed = speed
     self._sync_artifact_progress(artifact)
     self._report_status()
 
-  def _publish_progress(self, artifact, done_bytes: int, total: int) -> None:
+  def _publish_progress(self, artifact, done_bytes: int, total: int, speed: float) -> None:
     # held at 99 until the file passes its hash check
     progress = min(99.0, done_bytes / total * 100) if total > 0 else 0.0
-    eta = self._calculate_eta(artifact.fileName, progress)
-    self._set_progress(artifact, custom.ModelManagerSP.DownloadStatus.downloading, progress, eta)
+    if speed > 0 and total > 0:
+      eta = max(1, int((total - done_bytes) / speed))
+    else:
+      eta = self._calculate_eta(artifact.fileName, progress)
+    self._set_progress(artifact, custom.ModelManagerSP.DownloadStatus.downloading, progress, eta, speed)
 
-  async def _report_until_done(self, tasks: list[asyncio.Future], artifact, progress: _Progress, total: int) -> None:
+  async def _report_until_done(self, tasks: list[asyncio.Future], artifact, progress: _Progress, total: int,
+                               started: tuple[int, float]) -> None:
     pending = set(tasks)
+    last_bytes, last_time = started
+    speed = 0.0
     while pending:
       done, pending = await asyncio.wait(pending, timeout=REPORT_INTERVAL)
       for task in done:
         task.result()  # re-raises a worker failure
       if self._download_interrupted():
         raise DownloadCancelled("Download cancelled")
-      self._publish_progress(artifact, progress.snapshot(), total)
+      now = time.monotonic()
+      done_bytes = progress.snapshot()
+      if (dt := now - last_time) > 0:
+        instant = max(0.0, (done_bytes - last_bytes) / dt)  # bytes/s this tick; clamped since retried pieces can lower the byte count
+        speed = instant if speed == 0 else SPEED_SMOOTHING * speed + (1 - SPEED_SMOOTHING) * instant
+        if speed < SPEED_FLOOR:  # a stall decays toward 0 without reaching it, and a near-0 speed overflows the eta
+          speed = 0.0
+      last_time, last_bytes = now, done_bytes
+      self._publish_progress(artifact, done_bytes, total, speed)
 
   async def _download_file(self, url: str, path: str, artifact) -> None:
     self._download_start_times[artifact.fileName] = time.monotonic()
@@ -199,12 +217,13 @@ class ModelManagerSP:
     await loop.run_in_executor(None, _prepare_target, path, total)
     progress = _Progress()
     cancel = threading.Event()
+    started = progress.snapshot(), time.monotonic()  # before any worker can land bytes
 
     with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_PIECES) as pool:
       tasks = [loop.run_in_executor(pool, _fetch_piece, url, path, i, start, end, progress, cancel, self._block_size)
                for i, (start, end) in enumerate(pieces)]
       try:
-        await self._report_until_done(tasks, artifact, progress, total)
+        await self._report_until_done(tasks, artifact, progress, total, started)
       except BaseException as e:
         cancel.set()
         if isinstance(e, DownloadCancelled):
@@ -247,6 +266,7 @@ class ModelManagerSP:
       self._download_start_times.pop(artifact.fileName, None)
       artifact.downloadProgress.status = status.failed
       artifact.downloadProgress.eta = 0
+      artifact.downloadProgress.speed = 0
       self._sync_artifact_progress(artifact)
       if self.selected_bundle:
         self.selected_bundle.status = status.failed
@@ -259,6 +279,7 @@ class ModelManagerSP:
         os.remove(full_path)
       artifact.downloadProgress.status = status.failed
       artifact.downloadProgress.eta = 0
+      artifact.downloadProgress.speed = 0
       self._sync_artifact_progress(artifact)
       if self.selected_bundle:
         self.selected_bundle.status = status.failed
@@ -301,6 +322,7 @@ class ModelManagerSP:
           artifact.downloadProgress.status = custom.ModelManagerSP.DownloadStatus.cached
           artifact.downloadProgress.progress = 100
           artifact.downloadProgress.eta = 0
+          artifact.downloadProgress.speed = 0
         else:
           seen_artifacts.add(artifact.fileName)
           await self._process_artifact(artifact, destination_path)

openpilot/sunnypilot/models/tests/test_manager_download.py

@@ -6,6 +6,7 @@ See the LICENSE.md file in the root directory for more details.
 """
 
 import asyncio
+import concurrent.futures
 import contextlib
 import dataclasses
 import hashlib
@@ -129,6 +130,13 @@ def threaded_server(handler):
     thread.join()
 
 
+class InlineExecutor(concurrent.futures.Executor):
+  def submit(self, fn, /, *args, **kwargs):
+    future: concurrent.futures.Future = concurrent.futures.Future()
+    future.set_result(fn(*args, **kwargs))  # the piece lands before the reporter starts
+    return future
+
+
 class ManagerDownloadTestBase(OpenpilotTestCase):
   def setUp(self):
     super().setUp()
@@ -139,7 +147,7 @@ class ManagerDownloadTestBase(OpenpilotTestCase):
     self.addCleanup(self._tmp.cleanup)
     self.dest = self._tmp.name
 
-    for name, value in (('PIECE_SIZE', PIECE_SIZE), ('REPORT_INTERVAL', 0.05), ('PIECE_BACKOFF', 0.01)):
+    for name, value in (('PIECE_SIZE', PIECE_SIZE), ('REPORT_INTERVAL', 0.05), ('PIECE_BACKOFF', 0.01), ('SPEED_FLOOR', 100)):
       self.enterContext(mock.patch.object(manager_module, name, value))
 
     self.reported: list[float] = []
@@ -229,6 +237,31 @@ class TestManagerDownload(ManagerDownloadTestBase):
     assert self.reported and self.reported == sorted(self.reported), "progress only rises"
     assert max(self.reported) <= 99.0, "progress holds below 100 until the hash check"
     assert artifact.fileName not in self.manager._download_start_times
+    assert artifact.downloadProgress.speed > 0, "a running download shows its speed"
+
+  def test_speed_counts_bytes_landed_before_the_first_report(self):
+    self.enterContext(mock.patch.object(manager_module, 'ThreadPoolExecutor', lambda max_workers: InlineExecutor()))
+    artifact = self.download_file()
+    assert artifact.downloadProgress.speed > 0
+
+  def test_eta_comes_from_speed(self):
+    artifact = self.make_artifact()
+    self.manager._publish_progress(artifact, done_bytes=400, total=1000, speed=100)
+    assert artifact.downloadProgress.eta == 6
+
+  def test_stalled_speed_drops_to_zero(self):
+    stall_event = self.stall(0)
+    seen_speed = threading.Event()
+
+    def release_once_zero(*args):
+      if self.artifact.downloadProgress.speed > 0:
+        seen_speed.set()
+      elif seen_speed.is_set():
+        stall_event.set()
+    self.pm.send.side_effect = release_once_zero
+    self.download_file()
+    assert self.server.stall_released_by_event is True, "the stalled speed never reached exactly 0"
+    assert self.artifact.downloadProgress.speed > 0, "the speed must recover once data flows again"
 
   def test_pieces_download_in_parallel(self):
     stall_event = self.stall(0)
@@ -303,7 +336,8 @@ class TestProcessArtifact(ManagerDownloadTestBase):
         self.download_bundle(source)
         assert ACTIVE_BUNDLE_KEYS[source] in store and ACTIVE_BUNDLE_KEYS[other] not in store
         assert self.manager.selected_bundle.status == ds.downloaded
-        assert self.artifact.downloadProgress.progress == 100 and self.artifact.downloadProgress.eta == 0
+        assert self.artifact.downloadProgress.progress == 100
+        assert self.artifact.downloadProgress.eta == 0 and self.artifact.downloadProgress.speed == 0
         assert self.read_path() == WHOLE_BODY
         os.remove(self.path)