firmware download: reuse a file already in the cache unless --force

cli/src/mpftp/cli.py

@@ -3144,7 +3144,8 @@ def build_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="Prefer .uf2 (default: .bin for esp32 serial, .uf2 for rp2/samd)",
     )
-    fwdd.add_argument("--force", action="store_true", help="Refresh catalog cache")
+    fwdd.add_argument("--force", action="store_true",
+                      help="Refresh the catalog cache and download the file again")
     fwdd.set_defaults(func=cmd_firmware)
 
     fwd = fwsub.add_parser("detect", parents=[fw_sel, device_opts],

cli/src/mpftp/firmware.py

@@ -2703,6 +2703,8 @@ def progress(done: int, total: int) -> None:
             progress=progress,
             on_url=lambda url: emit_log(f"[mpftp] downloading {url}"),
         )
+        if out.get("cached"):
+            emit_log(f"[mpftp] already downloaded: {out['artifact']} (--force fetches it again)")
         out.pop("ready", None)
         out.setdefault("flashOffset", None)
         emit_result(True, ready=True, **out)
@@ -2848,7 +2850,8 @@ def add_target(sp: argparse.ArgumentParser) -> None:
         action="store_true",
         help="the .uf2 even on an ESP board (default: .bin for esp32, .uf2 elsewhere)",
     )
-    dld.add_argument("--force", action="store_true", help="refresh catalog cache")
+    dld.add_argument("--force", action="store_true",
+                     help="refresh the catalog cache and download the file again")
     dld.set_defaults(func=do_download)
 
     fl = sub.add_parser("flashers")

cli/src/mpftp/firmware_download.py

@@ -608,25 +608,36 @@ def resolve_remote_flash_offset(
 # Download
 # --------------------------------------------------------------------------- #
 
+def cached_file(url: str, dest_dir: Optional[Path] = None) -> Path:
+    """Where ``download_file`` keeps ``url`` in the firmware cache."""
+    dest_dir = dest_dir or (cache_root() / "files")
+    name = unquote(urlparse(url).path.rsplit("/", 1)[-1]) or "firmware.bin"
+    return dest_dir / name
+
+
 def download_file(
     url: str,
     *,
     dest_dir: Optional[Path] = None,
     progress: Optional[Callable[[int, int], None]] = None,
+    force: bool = False,
 ) -> Path:
-    """Download ``url`` into the firmware cache; a same-sized copy is reused."""
-    dest_dir = dest_dir or (cache_root() / "files")
-    dest_dir.mkdir(parents=True, exist_ok=True)
-    name = unquote(urlparse(url).path.rsplit("/", 1)[-1]) or "firmware.bin"
-    dest = dest_dir / name
+    """Download ``url`` into the firmware cache, or reuse the copy already there.
+
+    A firmware file's name carries its build date and version, so a file of
+    that name in the cache is the same build. Only a finished download is
+    given its name (it arrives as ``.partial``), so one that was cut off is
+    never reused. ``force`` downloads it again.
+    """
+    dest = cached_file(url, dest_dir)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    name = dest.name
+    if not force and dest.is_file() and dest.stat().st_size:
+        return dest
 
     req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
     with urllib.request.urlopen(req, timeout=120) as resp:
         total = int(resp.headers.get("Content-Length") or 0)
-        if dest.is_file() and total and dest.stat().st_size == total:
-            if progress:
-                progress(total, total)
-            return dest
         tmp = dest.with_suffix(dest.suffix + ".partial")
         written = 0
         with open(tmp, "wb") as f:
@@ -665,16 +676,18 @@ def download_board(
     chosen = pick_download(
         entry, version=version, preview=preview, mp_variant=mp_variant, fetch=fetch, uf2=uf2
     )
-    if on_url:
+    cached = not force and cached_file(chosen["url"]).is_file()
+    if on_url and not cached:
         on_url(chosen["url"])
-    path = download_file(chosen["url"], progress=progress)
+    path = download_file(chosen["url"], progress=progress, force=force)
     st = path.stat()
     out: dict[str, Any] = {
         "ready": True,
         "artifact": str(path),
         "size": st.st_size,
         "mtime": st.st_mtime,
         "source": "download",
+        "cached": cached,
         "catalogSource": entry["source"],
         "board": entry["board"],
         "variant": mp_variant or "",

cli/tests/test_firmware_download.py

@@ -226,5 +226,62 @@ def test_board_json_offset(self):
         self.assertEqual(off, "0x2000")
 
 
+class FakeResponse:
+    def __init__(self, body: bytes, length: bool = True):
+        self.body = body
+        self.headers = {"Content-Length": str(len(body))} if length else {}
+
+    def read(self, n: int) -> bytes:
+        chunk, self.body = self.body[:n], self.body[n:]
+        return chunk
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+class CachedDownloadTests(CacheIsolated):
+    """A firmware file already in the cache is reused (mpftp#210)."""
+
+    def download(self, body: bytes, **kw):
+        opened = []
+
+        def urlopen(req, timeout=0):
+            opened.append(req.full_url)
+            # micropython.org sends no Content-Length, so size can't decide.
+            return FakeResponse(body, length=False)
+
+        with mock.patch.object(fd.urllib.request, "urlopen", side_effect=urlopen):
+            out = fd.download_board("ESP32_GENERIC_S3", catalog=self.catalog, fetch=fake_fetch, **kw)
+        return out, opened
+
+    def test_the_second_download_reuses_the_file(self):
+        first, opened = self.download(b"image")
+        self.assertEqual(len(opened), 1)
+        self.assertFalse(first["cached"])
+        second, opened = self.download(b"other")
+        self.assertEqual(opened, [])
+        self.assertTrue(second["cached"])
+        self.assertEqual(Path(second["artifact"]).read_bytes(), b"image")
+
+    def test_force_downloads_it_again(self):
+        self.download(b"image")
+        out, opened = self.download(b"newer", force=True)
+        self.assertEqual(len(opened), 1)
+        self.assertFalse(out["cached"])
+        self.assertEqual(Path(out["artifact"]).read_bytes(), b"newer")
+
+    def test_a_cut_off_download_is_not_reused(self):
+        url = fd.pick_download(self.entry("ESP32_GENERIC_S3"))["url"]
+        partial = fd.cached_file(url)
+        partial.parent.mkdir(parents=True, exist_ok=True)
+        partial.with_suffix(partial.suffix + ".partial").write_bytes(b"ima")
+        out, opened = self.download(b"image")
+        self.assertEqual(len(opened), 1)
+        self.assertEqual(Path(out["artifact"]).read_bytes(), b"image")
+
+
 if __name__ == "__main__":
     unittest.main()

docs/guides/download-and-flash-firmware.md

@@ -30,7 +30,8 @@ whose images are only `.dfu` or `.hex` (stm32, most nrf) isn't listed; use its
 page on micropython.org. In the Firmware panel, choose **Download**, type part
 of the board's name (Pico, ESP32-S3, XIAO), then its variant and version. Choose
 **File** to flash a `.bin` or `.uf2` you already have. Images are kept in
-`~/.mpftp/firmware-cache/`.
+`~/.mpftp/firmware-cache/`, and downloading one that's already there reuses it
+without going online; `--force` fetches it again.
 
 ## Flash an ESP board