third_party/pigweed/repo/pw_package/py/pw_package/package_manager.py

   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
  16 import argparse
  17 import dataclasses
  18 import logging
  19 import os
  20 import pathlib
  21 import shutil
  22 from typing import Dict, List, Sequence, Tuple
  23
  24 _LOG: logging.Logger = logging.getLogger(__name__)
  25
  26
  27 class 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
  69 def 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
  78 class Packages:
  79     all: Tuple[str, ...]
  80     installed: Tuple[str, ...]
  81     available: Tuple[str, ...]
  82
  83
  84 class 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
 126 class 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
 177 def 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
 198 def run(**kwargs):
 199     return PackageManagerCLI().run(**kwargs)