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