xref: /aosp_15_r20/external/pigweed/pw_package/py/pw_package/package_manager.py (revision 61c4878ac05f98d0ceed94b57d316916de578985) 
1# Copyright 2020 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Install and remove optional packages."""
15
16import argparse
17import dataclasses
18import logging
19import os
20import pathlib
21import shutil
22from typing import Sequence
23
24from pw_env_setup import config_file
25
26_LOG: logging.Logger = logging.getLogger(__name__)
27
28
29class Package:
30    """Package to be installed.
31
32    Subclass this to implement installation of a specific package.
33    """
34
35    def __init__(self, name):
36        self._name = name
37        self._allow_use_in_downstream = True
38
39    @property
40    def name(self):
41        return self._name
42
43    @property
44    def allow_use_in_downstream(self):
45        return self._allow_use_in_downstream
46
47    def install(
48        self, path: pathlib.Path
49    ) -> None:  # pylint: disable=no-self-use
50        """Install the package at path.
51
52        Install the package in path. Cannot assume this directory is empty—it
53        may need to be deleted or updated.
54        """
55
56    def remove(self, path: pathlib.Path) -> None:  # pylint: disable=no-self-use
57        """Remove the package from path.
58
59        Removes the directory containing the package. For most packages this
60        should be sufficient to remove the package, and subclasses should not
61        need to override this package.
62        """
63        if os.path.exists(path):
64            shutil.rmtree(path)
65
66    def status(  # pylint: disable=no-self-use
67        self,
68        path: pathlib.Path,  # pylint: disable=unused-argument
69    ) -> bool:
70        """Returns if package is installed at path and current.
71
72        This method will be skipped if the directory does not exist.
73        """
74        return False
75
76    def info(  # pylint: disable=no-self-use
77        self,
78        path: pathlib.Path,  # pylint: disable=unused-argument
79    ) -> Sequence[str]:
80        """Returns a short string explaining how to enable the package."""
81        return []
82
83
84_PACKAGES: dict[str, Package] = {}
85
86
87def register(package_class: type, *args, **kwargs) -> None:
88    obj = package_class(*args, **kwargs)
89    _PACKAGES[obj.name] = obj
90
91
92@dataclasses.dataclass
93class Packages:
94    all: tuple[str, ...]
95    installed: tuple[str, ...]
96    available: tuple[str, ...]
97
98
99class MiddlewareOnlyPackageError(Exception):
100    def __init__(self, pkg_name):
101        super().__init__(
102            f'Package {pkg_name} is a middleware-only package--it should be '
103            'imported as a submodule and not a package'
104        )
105
106
107class PackageManager:
108    """Install and remove optional packages."""
109
110    def __init__(self, root: pathlib.Path):
111        self._pkg_root = root
112        os.makedirs(root, exist_ok=True)
113
114        config = config_file.load().get('pw', {}).get('pw_package', {})
115        self._allow_middleware_only_packages = config.get(
116            'allow_middleware_only_packages',
117            False,
118        )
119
120    def install(self, package: str, force: bool = False) -> None:
121        """Install the named package.
122
123        Args:
124            package: The name of the package to install.
125            force: Install the package regardless of whether it's already
126                installed or if it's not "allowed" to be installed on this
127                project.
128        """
129
130        pkg = _PACKAGES[package]
131        if not pkg.allow_use_in_downstream:
132            if not self._allow_middleware_only_packages:
133                if force:
134                    _LOG.warning(str(MiddlewareOnlyPackageError(pkg.name)))
135                else:
136                    raise MiddlewareOnlyPackageError(pkg.name)
137
138        if force:
139            self.remove(package)
140        pkg.install(self._pkg_root / pkg.name)
141
142    def remove(self, package: str) -> None:
143        pkg = _PACKAGES[package]
144        pkg.remove(self._pkg_root / pkg.name)
145
146    def status(self, package: str) -> bool:
147        pkg = _PACKAGES[package]
148        path = self._pkg_root / pkg.name
149        return os.path.isdir(path) and pkg.status(path)
150
151    def list(self) -> Packages:
152        installed = []
153        available = []
154        for package in sorted(_PACKAGES.keys()):
155            pkg = _PACKAGES[package]
156            if pkg.status(self._pkg_root / pkg.name):
157                installed.append(pkg.name)
158            else:
159                available.append(pkg.name)
160
161        return Packages(
162            all=tuple(_PACKAGES.keys()),
163            installed=tuple(installed),
164            available=tuple(available),
165        )
166
167    def info(self, package: str) -> Sequence[str]:
168        pkg = _PACKAGES[package]
169        return pkg.info(self._pkg_root / pkg.name)
170
171
172class PackageManagerCLI:
173    """Command-line interface to PackageManager."""
174
175    def __init__(self):
176        self._mgr: PackageManager = None
177
178    def install(self, package: str, force: bool = False) -> int:
179        _LOG.info('Installing %s...', package)
180        self._mgr.install(package, force)
181        _LOG.info('Installing %s...done.', package)
182        for line in self._mgr.info(package):
183            _LOG.info('%s', line)
184        return 0
185
186    def remove(self, package: str) -> int:
187        _LOG.info('Removing %s...', package)
188        self._mgr.remove(package)
189        _LOG.info('Removing %s...done.', package)
190        return 0
191
192    def status(self, package: str) -> int:
193        if self._mgr.status(package):
194            _LOG.info('%s is installed.', package)
195            for line in self._mgr.info(package):
196                _LOG.info('%s', line)
197            return 0
198
199        _LOG.info('%s is not installed.', package)
200        return -1
201
202    def list(self) -> int:
203        packages = self._mgr.list()
204
205        _LOG.info('Installed packages:')
206        for package in packages.installed:
207            _LOG.info('  %s', package)
208            for line in self._mgr.info(package):
209                _LOG.info('    %s', line)
210        _LOG.info('')
211
212        _LOG.info('Available packages:')
213        for package in packages.available:
214            _LOG.info('  %s', package)
215        _LOG.info('')
216
217        return 0
218
219    def run(self, command: str, pkg_root: pathlib.Path, **kwargs) -> int:
220        self._mgr = PackageManager(pkg_root.resolve())
221        return getattr(self, command)(**kwargs)
222
223
224def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
225    parser = argparse.ArgumentParser("Manage packages.")
226    parser.add_argument(
227        '--package-root',
228        '-e',
229        dest='pkg_root',
230        type=pathlib.Path,
231        default=pathlib.Path(os.environ['PW_PACKAGE_ROOT']),
232    )
233    subparsers = parser.add_subparsers(dest='command', required=True)
234    install = subparsers.add_parser('install')
235    install.add_argument('--force', '-f', action='store_true')
236    remove = subparsers.add_parser('remove')
237    status = subparsers.add_parser('status')
238    for cmd in (install, remove, status):
239        cmd.add_argument('package', choices=_PACKAGES.keys())
240    _ = subparsers.add_parser('list')
241    return parser.parse_args(argv)
242
243
244def run(**kwargs):
245    return PackageManagerCLI().run(**kwargs)
246