diff --git a/openpilot/selfdrive/ui/sunnypilot/layouts/settings/models.py b/openpilot/selfdrive/ui/sunnypilot/layouts/settings/models.py index 1d12f9f5b..22cc818ff 100644 --- a/openpilot/selfdrive/ui/sunnypilot/layouts/settings/models.py +++ b/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 diff --git a/openpilot/selfdrive/ui/sunnypilot/mici/layouts/models.py b/openpilot/selfdrive/ui/sunnypilot/mici/layouts/models.py index 99e0b06e8..c1f1f0915 100644 --- a/openpilot/selfdrive/ui/sunnypilot/mici/layouts/models.py +++ b/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")) diff --git a/openpilot/sunnypilot/models/manager.py b/openpilot/sunnypilot/models/manager.py index bdee8ac94..f1cca741b 100644 --- a/openpilot/sunnypilot/models/manager.py +++ b/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) diff --git a/openpilot/sunnypilot/models/tests/test_manager_download.py b/openpilot/sunnypilot/models/tests/test_manager_download.py index 662a776e9..3446441ee 100644 --- a/openpilot/sunnypilot/models/tests/test_manager_download.py +++ b/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)