scripts: west_commands: blobs: support blobs cache

.ruff-excludes.toml

@@ -1284,7 +1284,6 @@ exclude = [
     "./scripts/utils/pinctrl_nrf_migrate.py",
     "./scripts/utils/twister_to_list.py",
     "./scripts/west_commands/bindesc.py",
-    "./scripts/west_commands/blobs.py",
     "./scripts/west_commands/boards.py",
     "./scripts/west_commands/build.py",
     "./scripts/west_commands/build_helpers.py",

doc/develop/west/zephyr-cmds.rst

@@ -220,6 +220,19 @@ the specific blobs that are fetched, by passing a regular expression::
   # For example, only download esp32 blobs, skip the other variants
   west blobs fetch hal_espressif --allow-regex 'lib/esp32/.*'
 
+An auto-cache directory can be provided via the ``--auto-cache`` cli argument
+or via the ``blobs.auto-cache-dir`` config option. When enabled, the auto-cache
+directory is automatically populated whenever a blob is missing and downloaded.
+
+One or more additional cache directories (separated by ``;``) can be provided
+in ``--cache-dirs`` cli argument or ``blobs.cache-dir`` config option.
+
+``west blobs fetch`` searches all configured cache directories (including the
+auto-cache) for a matching blob filename. Cached files may be stored either
+under their original filename or with a SHA-256 suffix (``<filename>.<sha>``).
+If found, the blob is copied from the cache to the blob path; otherwise
+it is downloaded from its url to the blob path.
+
 .. _west-twister:
 
 Twister wrapper: ``west twister``

scripts/west_commands/blobs.py

@@ -5,6 +5,7 @@
 import argparse
 import os
 import re
+import shutil
 import sys
 import textwrap
 from pathlib import Path
@@ -18,7 +19,6 @@
 
 
 class Blobs(WestCommand):
-
     DEFAULT_LIST_FMT = '{module} {status} {path} {type} {abspath}'
 
     def __init__(self):
@@ -27,7 +27,8 @@ def __init__(self):
             # Keep this in sync with the string in west-commands.yml.
             'work with binary blobs',
             'Work with binary blobs',
-            accepts_unknown_args=False)
+            accepts_unknown_args=False,
+        )
 
     def do_add_parser(self, parser_adder):
         parser = parser_adder.add_parser(
@@ -61,22 +62,30 @@ def do_add_parser(self, parser_adder):
             - uri: URI to the remote location of the blob
             - description: blob text description
             - doc-url: URL to the documentation for this blob
-            '''))
+            '''),
+        )
 
         # Remember to update west-completion.bash if you add or remove
         # flags
-        parser.add_argument('subcmd', nargs=1,
-                            choices=['list', 'fetch', 'clean'],
-                            help='sub-command to execute')
+        parser.add_argument(
+            'subcmd', nargs=1, choices=['list', 'fetch', 'clean'], help='sub-command to execute'
+        )
 
-        parser.add_argument('modules', metavar='MODULE', nargs='*',
-                            help='''zephyr modules to operate on;
-                            all modules will be used if not given''')
+        parser.add_argument(
+            'modules',
+            metavar='MODULE',
+            nargs='*',
+            help='''zephyr modules to operate on;
+                            all modules will be used if not given''',
+        )
 
         group = parser.add_argument_group('west blob list options')
-        group.add_argument('-f', '--format',
-                            help='''format string to use to list each blob;
-                                    see FORMAT STRINGS below''')
+        group.add_argument(
+            '-f',
+            '--format',
+            help='''format string to use to list each blob;
+                                    see FORMAT STRINGS below''',
+        )
 
         group = parser.add_argument_group('west blobs fetch options')
         group.add_argument(
@@ -86,8 +95,24 @@ def do_add_parser(self, parser_adder):
                     Only local paths matching this regex will be fetched.
                     Note that local paths are relative to the module directory''',
         )
-        group.add_argument('-a', '--auto-accept', action='store_true',
-                            help='''auto accept license if the fetching needs click-through''')
+        group.add_argument(
+            '-a',
+            '--auto-accept',
+            action='store_true',
+            help='''auto accept license if the fetching needs click-through''',
+        )
+        group.add_argument(
+            '--cache-dirs',
+            help='''Semicolon-separated list of directories to search for cached
+                                    blobs before downloading. Cache files may use the original
+                                    filename or be suffixed with `.<sha256>`.''',
+        )
+        group.add_argument(
+            '--auto-cache',
+            help='''Path to a directory that is automatically populated when a blob
+                                    is downloaded. Cached blobs are stored using the original
+                                    filename suffixed with `.<sha256>`.''',
+        )
 
         return parser
 
@@ -121,26 +146,87 @@ def list(self, args):
     def ensure_folder(self, path):
         path.parent.mkdir(parents=True, exist_ok=True)
 
-    def fetch_blob(self, url, path):
+    def get_cached_blob(self, blob, cache_dirs) -> Path | None:
+        """
+        Look for a cached blob in the provided cache directories.
+        A blob may be stored using either its original name or suffixed with
+        its SHA256 hash (i.e., "<name>.<sha256>").
+        Return the first matching path, or None if not found.
+        """
+        blob_name = Path(blob["path"]).name
+        sha256 = blob["sha256"]
+        candidate_names = [
+            f"{blob_name}.{sha256}",  # suffixed version
+            blob_name,  # plain version
+        ]
+
+        for cache_dir in cache_dirs:
+            if not cache_dir.exists():
+                continue
+            for name in candidate_names:
+                candidate_path = cache_dir / name
+                if (
+                    zephyr_module.get_blob_status(candidate_path, sha256)
+                    == zephyr_module.BLOB_PRESENT
+                ):
+                    return candidate_path
+        return None
+
+    def download_blob(self, url, path):
         scheme = urlparse(url).scheme
-        self.dbg(f'Fetching {path} with {scheme}')
+        self.dbg(f'Fetching blob from url {url} with {scheme} to path: {path}')
         import fetchers
-        fetcher = fetchers.get_fetcher_cls(scheme)
 
+        fetcher = fetchers.get_fetcher_cls(scheme)
         self.dbg(f'Found fetcher: {fetcher}')
         inst = fetcher()
         self.ensure_folder(path)
         inst.fetch(url, path)
 
+    def fetch_blob(self, args, blob):
+        url = blob['url']
+        path = Path(blob['path'])
+
+        # collect existing cache dirs specified as args, otherwise from west config
+        cache_dirs = args.cache_dirs
+        auto_cache_dir = args.auto_cache
+        if self.has_config:
+            if not cache_dirs:
+                cache_dirs = self.config.get('blobs.cache-dirs', default='')
+            if not auto_cache_dir:
+                auto_cache_dir = self.config.get('blobs.auto-cache')
+        cache_dirs = [Path(p) for p in (cache_dirs or '').split(';') if p]
+        auto_cache_dir = Path(auto_cache_dir)
+
+        # search for cached blob in the cache-dirs
+        cached_blob = self.get_cached_blob(blob, cache_dirs)
+
+        # If not cached: Search in the auto-cache and download blob to auto-cache
+        if not cached_blob and auto_cache_dir:
+            cached_blob = self.get_cached_blob(blob, [auto_cache_dir])
+            if not cached_blob:
+                name = path.name
+                sha256 = blob['sha256']
+                self.download_blob(url, auto_cache_dir / f'{name}.{sha256}')
+                cached_blob = self.get_cached_blob(blob, [auto_cache_dir])
+
+        # copy the cached blob or download it
+        if cached_blob:
+            self.dbg(f'Copy cached blob: {cached_blob}')
+            shutil.copy(cached_blob, path)
+        else:
+            self.download_blob(url, path)
+
     # Compare the checksum of a file we've just downloaded
     # to the digest in blob metadata, warn user if they differ.
     def verify_blob(self, blob) -> bool:
         self.dbg(f"Verifying blob {blob['module']}: {blob['abspath']}")
 
         status = zephyr_module.get_blob_status(blob['abspath'], blob['sha256'])
         if status == zephyr_module.BLOB_OUTDATED:
-            self.err(textwrap.dedent(
-                f'''\
+            self.err(
+                textwrap.dedent(
+                    f'''\
                 The checksum of the downloaded file does not match that
                 in the blob metadata:
                 - if it is not certain that the download was successful,
@@ -153,7 +239,9 @@ def verify_blob(self, blob) -> bool:
                 Module: {blob['module']}
                 Blob:   {blob['path']}
                 URL:    {blob['url']}
-                Info:   {blob['description']}'''))
+                Info:   {blob['description']}'''
+                )
+            )
             return False
         return True
 
@@ -176,9 +264,11 @@ def fetch(self, args):
 
             if blob['click-through'] and not args.auto_accept:
                 while True:
-                    user_input = input("For this blob, need to read and accept "
-                                       "license to continue. Read it?\n"
-                                       "Please type 'y' or 'n' and press enter to confirm: ")
+                    user_input = input(
+                        "For this blob, need to read and accept "
+                        "license to continue. Read it?\n"
+                        "Please type 'y' or 'n' and press enter to confirm: "
+                    )
                     if user_input.upper() == "Y" or user_input.upper() == "N":
                         break
 
@@ -191,16 +281,18 @@ def fetch(self, args):
                     print(license_content)
 
                 while True:
-                    user_input = input("Accept license to continue?\n"
-                                       "Please type 'y' or 'n' and press enter to confirm: ")
+                    user_input = input(
+                        "Accept license to continue?\n"
+                        "Please type 'y' or 'n' and press enter to confirm: "
+                    )
                     if user_input.upper() == "Y" or user_input.upper() == "N":
                         break
 
                 if user_input.upper() != "Y":
                     self.wrn('Skip fetching this blob.')
                     continue
 
-            self.fetch_blob(blob['url'], blob['abspath'])
+            self.fetch_blob(args, blob)
             if not self.verify_blob(blob):
                 bad_checksum_count += 1