• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 Dict, List, Optional, Sequence, Tuple
23
24_LOG: logging.Logger = logging.getLogger(__name__)
25
26
27class Package:
28    """Package to be installed.
29
30    Subclass this to implement installation of a specific package.
31    """
32
33    def __init__(self, name):
34        self._name = name
35        self._allow_use_in_downstream = True
36
37    @property
38    def name(self):
39        return self._name
40
41    @property
42    def allow_use_in_downstream(self):
43        return self._allow_use_in_downstream
44
45    def install(
46        self, path: pathlib.Path
47    ) -> None:  # pylint: disable=no-self-use
48        """Install the package at path.
49
50        Install the package in path. Cannot assume this directory is empty—it
51        may need to be deleted or updated.
52        """
53
54    def remove(self, path: pathlib.Path) -> None:  # pylint: disable=no-self-use
55        """Remove the package from path.
56
57        Removes the directory containing the package. For most packages this
58        should be sufficient to remove the package, and subclasses should not
59        need to override this package.
60        """
61        if os.path.exists(path):
62            shutil.rmtree(path)
63
64    def status(  # pylint: disable=no-self-use
65        self,
66        path: pathlib.Path,  # pylint: disable=unused-argument
67    ) -> bool:
68        """Returns if package is installed at path and current.
69
70        This method will be skipped if the directory does not exist.
71        """
72        return False
73
74    def info(  # pylint: disable=no-self-use
75        self,
76        path: pathlib.Path,  # pylint: disable=unused-argument
77    ) -> Sequence[str]:
78        """Returns a short string explaining how to enable the package."""
79        return []
80
81
82_PACKAGES: Dict[str, Package] = {}
83
84
85def register(package_class: type, *args, **kwargs) -> None:
86    obj = package_class(*args, **kwargs)
87    _PACKAGES[obj.name] = obj
88
89
90@dataclasses.dataclass
91class Packages:
92    all: Tuple[str, ...]
93    installed: Tuple[str, ...]
94    available: Tuple[str, ...]
95
96
97class UpstreamOnlyPackageError(Exception):
98    def __init__(self, pkg_name):
99        super().__init__(
100            f'Package {pkg_name} is an upstream-only package--it should be '
101            'imported as a submodule and not a package'
102        )
103
104
105class PackageManager:
106    """Install and remove optional packages."""
107
108    def __init__(self, root: pathlib.Path):
109        self._pkg_root = root
110        os.makedirs(root, exist_ok=True)
111
112    def install(self, package: str, force: bool = False) -> None:
113        """Install the named package.
114
115        Args:
116            package: The name of the package to install.
117            force: Install the package regardless of whether it's already
118                installed or if it's not "allowed" to be installed on this
119                project.
120        """
121
122        pkg = _PACKAGES[package]
123        if not pkg.allow_use_in_downstream:
124            if os.environ.get('PW_ROOT') != os.environ.get('PW_PROJECT_ROOT'):
125                if force:
126                    _LOG.warning(str(UpstreamOnlyPackageError(pkg.name)))
127                else:
128                    raise UpstreamOnlyPackageError(pkg.name)
129
130        if force:
131            self.remove(package)
132        pkg.install(self._pkg_root / pkg.name)
133
134    def remove(self, package: str) -> None:
135        pkg = _PACKAGES[package]
136        pkg.remove(self._pkg_root / pkg.name)
137
138    def status(self, package: str) -> bool:
139        pkg = _PACKAGES[package]
140        path = self._pkg_root / pkg.name
141        return os.path.isdir(path) and pkg.status(path)
142
143    def list(self) -> Packages:
144        installed = []
145        available = []
146        for package in sorted(_PACKAGES.keys()):
147            pkg = _PACKAGES[package]
148            if pkg.status(self._pkg_root / pkg.name):
149                installed.append(pkg.name)
150            else:
151                available.append(pkg.name)
152
153        return Packages(
154            all=tuple(_PACKAGES.keys()),
155            installed=tuple(installed),
156            available=tuple(available),
157        )
158
159    def info(self, package: str) -> Sequence[str]:
160        pkg = _PACKAGES[package]
161        return pkg.info(self._pkg_root / pkg.name)
162
163
164class PackageManagerCLI:
165    """Command-line interface to PackageManager."""
166
167    def __init__(self):
168        self._mgr: PackageManager = None
169
170    def install(self, package: str, force: bool = False) -> int:
171        _LOG.info('Installing %s...', package)
172        self._mgr.install(package, force)
173        _LOG.info('Installing %s...done.', package)
174        for line in self._mgr.info(package):
175            _LOG.info('%s', line)
176        return 0
177
178    def remove(self, package: str) -> int:
179        _LOG.info('Removing %s...', package)
180        self._mgr.remove(package)
181        _LOG.info('Removing %s...done.', package)
182        return 0
183
184    def status(self, package: str) -> int:
185        if self._mgr.status(package):
186            _LOG.info('%s is installed.', package)
187            for line in self._mgr.info(package):
188                _LOG.info('%s', line)
189            return 0
190
191        _LOG.info('%s is not installed.', package)
192        return -1
193
194    def list(self) -> int:
195        packages = self._mgr.list()
196
197        _LOG.info('Installed packages:')
198        for package in packages.installed:
199            _LOG.info('  %s', package)
200            for line in self._mgr.info(package):
201                _LOG.info('    %s', line)
202        _LOG.info('')
203
204        _LOG.info('Available packages:')
205        for package in packages.available:
206            _LOG.info('  %s', package)
207        _LOG.info('')
208
209        return 0
210
211    def run(self, command: str, pkg_root: pathlib.Path, **kwargs) -> int:
212        self._mgr = PackageManager(pkg_root.resolve())
213        return getattr(self, command)(**kwargs)
214
215
216def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace:
217    parser = argparse.ArgumentParser("Manage packages.")
218    parser.add_argument(
219        '--package-root',
220        '-e',
221        dest='pkg_root',
222        type=pathlib.Path,
223        default=pathlib.Path(os.environ['PW_PACKAGE_ROOT']),
224    )
225    subparsers = parser.add_subparsers(dest='command', required=True)
226    install = subparsers.add_parser('install')
227    install.add_argument('--force', '-f', action='store_true')
228    remove = subparsers.add_parser('remove')
229    status = subparsers.add_parser('status')
230    for cmd in (install, remove, status):
231        cmd.add_argument('package', choices=_PACKAGES.keys())
232    _ = subparsers.add_parser('list')
233    return parser.parse_args(argv)
234
235
236def run(**kwargs):
237    return PackageManagerCLI().run(**kwargs)
238