1# Copyright (c) 2022 Nordic Semiconductor ASA
2#
3# SPDX-License-Identifier: Apache-2.0
4
5import argparse
6import os
7from pathlib import Path
8import sys
9import textwrap
10from urllib.parse import urlparse
11
12from west import log
13from west.commands import WestCommand
14
15from zephyr_ext_common import ZEPHYR_BASE
16
17sys.path.append(os.fspath(Path(__file__).parent.parent))
18import zephyr_module
19
20class Blobs(WestCommand):
21
22    DEFAULT_LIST_FMT = '{module} {status} {path} {type} {abspath}'
23
24    def __init__(self):
25        super().__init__(
26            'blobs',
27            # Keep this in sync with the string in west-commands.yml.
28            'work with binary blobs',
29            'Work with binary blobs',
30            accepts_unknown_args=False)
31
32    def do_add_parser(self, parser_adder):
33        parser = parser_adder.add_parser(
34            self.name,
35            help=self.help,
36            formatter_class=argparse.RawDescriptionHelpFormatter,
37            description=self.description,
38            epilog=textwrap.dedent(f'''\
39            FORMAT STRINGS
40            --------------
41
42            Blobs are listed using a Python 3 format string. Arguments
43            to the format string are accessed by name.
44
45            The default format string is:
46
47            "{self.DEFAULT_LIST_FMT}"
48
49            The following arguments are available:
50
51            - module: name of the module that contains this blob
52            - abspath: blob absolute path
53            - status: short status (A: present, M: hash failure, D: not present)
54            - path: blob local path from <module>/zephyr/blobs/
55            - sha256: blob SHA256 hash in hex
56            - type: type of blob
57            - version: version string
58            - license_path: path to the license file for the blob
59            - uri: URI to the remote location of the blob
60            - description: blob text description
61            - doc-url: URL to the documentation for this blob
62            '''))
63
64        # Remember to update west-completion.bash if you add or remove
65        # flags
66        parser.add_argument('subcmd', nargs=1,
67                            choices=['list', 'fetch', 'clean'],
68                            help='sub-command to execute')
69
70        parser.add_argument('modules', metavar='MODULE', nargs='*',
71                            help='''zephyr modules to operate on;
72                            all modules will be used if not given''')
73
74        group = parser.add_argument_group('west blob list options')
75        group.add_argument('-f', '--format',
76                            help='''format string to use to list each blob;
77                                    see FORMAT STRINGS below''')
78
79        return parser
80
81    def get_blobs(self, args):
82        blobs = []
83        modules = args.modules
84        for module in zephyr_module.parse_modules(ZEPHYR_BASE, self.manifest):
85            # Filter by module
86            module_name = module.meta.get('name', None)
87            if len(modules) and module_name not in modules:
88                continue
89
90            blobs += zephyr_module.process_blobs(module.project, module.meta)
91
92        return blobs
93
94    def list(self, args):
95        blobs = self.get_blobs(args)
96        fmt = args.format or self.DEFAULT_LIST_FMT
97        for blob in blobs:
98            log.inf(fmt.format(**blob))
99
100    def ensure_folder(self, path):
101        path.parent.mkdir(parents=True, exist_ok=True)
102
103    def fetch_blob(self, url, path):
104        scheme = urlparse(url).scheme
105        log.dbg(f'Fetching {path} with {scheme}')
106        import fetchers
107        fetcher = fetchers.get_fetcher_cls(scheme)
108
109        log.dbg(f'Found fetcher: {fetcher}')
110        inst = fetcher()
111        self.ensure_folder(path)
112        inst.fetch(url, path)
113
114    def fetch(self, args):
115        blobs = self.get_blobs(args)
116        for blob in blobs:
117            if blob['status'] == 'A':
118                log.dbg('Blob {module}: {abspath} is up to date'.format(**blob))
119                continue
120            log.inf('Fetching blob {module}: {abspath}'.format(**blob))
121            self.fetch_blob(blob['url'], blob['abspath'])
122
123    def clean(self, args):
124        blobs = self.get_blobs(args)
125        for blob in blobs:
126            if blob['status'] == 'D':
127                log.dbg('Blob {module}: {abspath} not in filesystem'.format(**blob))
128                continue
129            log.inf('Deleting blob {module}: {status} {abspath}'.format(**blob))
130            blob['abspath'].unlink()
131
132    def do_run(self, args, _):
133        log.dbg(f'subcmd: \'{args.subcmd[0]}\' modules: {args.modules}')
134
135        subcmd = getattr(self, args.subcmd[0])
136
137        if args.subcmd[0] != 'list' and args.format is not None:
138            log.die(f'unexpected --format argument; this is a "west blobs list" option')
139
140        subcmd(args)
141