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, 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 def __init__(self, name): 33 self._name = name 34 35 @property 36 def name(self): 37 return self._name 38 39 def install(self, path: pathlib.Path) -> None: # pylint: disable=no-self-use 40 """Install the package at path. 41 42 Install the package in path. Cannot assume this directory is empty—it 43 may need to be deleted or updated. 44 """ 45 46 def remove(self, path: pathlib.Path) -> None: # pylint: disable=no-self-use 47 """Remove the package from path. 48 49 Removes the directory containing the package. For most packages this 50 should be sufficient to remove the package, and subclasses should not 51 need to override this package. 52 """ 53 if os.path.exists(path): 54 shutil.rmtree(path) 55 56 def status(self, path: pathlib.Path) -> bool: # pylint: disable=no-self-use 57 """Returns if package is installed at path and current. 58 59 This method will be skipped if the directory does not exist. 60 """ 61 62 def info(self, path: pathlib.Path) -> Sequence[str]: # pylint: disable=no-self-use 63 """Returns a short string explaining how to enable the package.""" 64 65 66_PACKAGES: Dict[str, Package] = {} 67 68 69def register(package_class: type, name: str = None) -> None: 70 if name: 71 obj = package_class(name) 72 else: 73 obj = package_class() 74 _PACKAGES[obj.name] = obj 75 76 77@dataclasses.dataclass 78class Packages: 79 all: Tuple[str, ...] 80 installed: Tuple[str, ...] 81 available: Tuple[str, ...] 82 83 84class PackageManager: 85 """Install and remove optional packages.""" 86 def __init__(self, root: pathlib.Path): 87 self._pkg_root = root 88 os.makedirs(root, exist_ok=True) 89 90 def install(self, package: str, force: bool = False) -> None: 91 pkg = _PACKAGES[package] 92 if force: 93 self.remove(package) 94 pkg.install(self._pkg_root / pkg.name) 95 96 def remove(self, package: str) -> None: 97 pkg = _PACKAGES[package] 98 pkg.remove(self._pkg_root / pkg.name) 99 100 def status(self, package: str) -> bool: 101 pkg = _PACKAGES[package] 102 path = self._pkg_root / pkg.name 103 return os.path.isdir(path) and pkg.status(path) 104 105 def list(self) -> Packages: 106 installed = [] 107 available = [] 108 for package in sorted(_PACKAGES.keys()): 109 pkg = _PACKAGES[package] 110 if pkg.status(self._pkg_root / pkg.name): 111 installed.append(pkg.name) 112 else: 113 available.append(pkg.name) 114 115 return Packages( 116 all=tuple(_PACKAGES.keys()), 117 installed=tuple(installed), 118 available=tuple(available), 119 ) 120 121 def info(self, package: str) -> Sequence[str]: 122 pkg = _PACKAGES[package] 123 return pkg.info(self._pkg_root / pkg.name) 124 125 126class PackageManagerCLI: 127 """Command-line interface to PackageManager.""" 128 def __init__(self): 129 self._mgr: PackageManager = None 130 131 def install(self, package: str, force: bool = False) -> int: 132 _LOG.info('Installing %s...', package) 133 self._mgr.install(package, force) 134 _LOG.info('Installing %s...done.', package) 135 for line in self._mgr.info(package): 136 _LOG.info('%s', line) 137 return 0 138 139 def remove(self, package: str) -> int: 140 _LOG.info('Removing %s...', package) 141 self._mgr.remove(package) 142 _LOG.info('Removing %s...done.', package) 143 return 0 144 145 def status(self, package: str) -> int: 146 if self._mgr.status(package): 147 _LOG.info('%s is installed.', package) 148 for line in self._mgr.info(package): 149 _LOG.info('%s', line) 150 return 0 151 152 _LOG.info('%s is not installed.', package) 153 return -1 154 155 def list(self) -> int: 156 packages = self._mgr.list() 157 158 _LOG.info('Installed packages:') 159 for package in packages.installed: 160 _LOG.info(' %s', package) 161 for line in self._mgr.info(package): 162 _LOG.info(' %s', line) 163 _LOG.info('') 164 165 _LOG.info('Available packages:') 166 for package in packages.available: 167 _LOG.info(' %s', package) 168 _LOG.info('') 169 170 return 0 171 172 def run(self, command: str, pkg_root: pathlib.Path, **kwargs) -> int: 173 self._mgr = PackageManager(pkg_root.resolve()) 174 return getattr(self, command)(**kwargs) 175 176 177def parse_args(argv: List[str] = None) -> argparse.Namespace: 178 parser = argparse.ArgumentParser("Manage packages.") 179 parser.add_argument( 180 '--package-root', 181 '-e', 182 dest='pkg_root', 183 type=pathlib.Path, 184 default=(pathlib.Path(os.environ['_PW_ACTUAL_ENVIRONMENT_ROOT']) / 185 'packages'), 186 ) 187 subparsers = parser.add_subparsers(dest='command', required=True) 188 install = subparsers.add_parser('install') 189 install.add_argument('--force', '-f', action='store_true') 190 remove = subparsers.add_parser('remove') 191 status = subparsers.add_parser('status') 192 for cmd in (install, remove, status): 193 cmd.add_argument('package', choices=_PACKAGES.keys()) 194 _ = subparsers.add_parser('list') 195 return parser.parse_args(argv) 196 197 198def run(**kwargs): 199 return PackageManagerCLI().run(**kwargs) 200