Source code for rosdep2.installers

# Copyright (c) 2009, Willow Garage, Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
#     * Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above copyright
#       notice, this list of conditions and the following disclaimer in the
#       documentation and/or other materials provided with the distribution.
#     * Neither the name of the Willow Garage, Inc. nor the names of its
#       contributors may be used to endorse or promote products derived from
#       this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

# Author Tully Foote/tfoote@willowgarage.com, Ken Conley/kwc@willowgarage.com

from __future__ import print_function

import os
import subprocess
import traceback

from rospkg.os_detect import OsDetect

from .core import rd_debug, RosdepInternalError, InstallFailed, print_bold, InvalidData

# kwc: InstallerContext is basically just a bunch of dictionaries with
# defined lookup methods.  It really encompasses two facets of a
# rosdep configuration: the pluggable nature of installers and
# platforms, as well as the resolution of the operating system for a
# specific machine.  It is possible to decouple those two notions,
# though there are some touch points over how this interfaces with the
# rospkg.os_detect library, i.e. how platforms can tweak these
# detectors and how the higher-level APIs can override them.


[docs]class InstallerContext(object): """ :class:`InstallerContext` manages the context of execution for rosdep as it relates to the installers, OS detectors, and other extensible APIs. """ def __init__(self, os_detect=None): """ :param os_detect: (optional) :class:`rospkg.os_detect.OsDetect` instance to use for detecting platforms. If `None`, default instance will be used. """ # platform configuration self.installers = {} self.os_installers = {} self.default_os_installer = {} # stores configuration of which value to use for the OS version key (version number or codename) self.os_version_type = {} # OS detection and override if os_detect is None: os_detect = OsDetect() self.os_detect = os_detect self.os_override = None self.verbose = False def set_verbose(self, verbose): self.verbose = verbose
[docs] def set_os_override(self, os_name, os_version): """ Override the OS detector with *os_name* and *os_version*. See :meth:`InstallerContext.detect_os`. :param os_name: OS name value to use, ``str`` :param os_version: OS version value to use, ``str`` """ if self.verbose: print('overriding OS to [%s:%s]' % (os_name, os_version)) self.os_override = os_name, os_version
def get_os_version_type(self, os_name): return self.os_version_type.get(os_name, OsDetect.get_version) def set_os_version_type(self, os_name, version_type): if not hasattr(version_type, '__call__'): raise ValueError('version type should be a method') self.os_version_type[os_name] = version_type
[docs] def get_os_name_and_version(self): """ Get the OS name and version key to use for resolution and installation. This will be the detected OS name/version unless :meth:`InstallerContext.set_os_override()` has been called. :returns: (os_name, os_version), ``(str, str)`` """ if self.os_override: return self.os_override else: os_name = self.os_detect.get_name() os_key = self.get_os_version_type(os_name) os_version = os_key(self.os_detect) return os_name, os_version
[docs] def get_os_detect(self): """ :returns os_detect: :class:`OsDetect` instance used for detecting platforms. """ return self.os_detect
[docs] def set_installer(self, installer_key, installer): """ Set the installer to use for *installer_key*. This will replace any existing installer associated with the key. *installer_key* should be the same key used for the ``rosdep.yaml`` package manager key. If *installer* is ``None``, this will delete any existing associated installer from this context. :param installer_key: key/name to associate with installer, ``str`` :param installer: :class:`Installer` implementation, ``class``. :raises: :exc:`TypeError` if *installer* is not a subclass of :class:`Installer` """ if installer is None: del self.installers[installer_key] return if not isinstance(installer, Installer): raise TypeError('installer must be a instance of Installer') if self.verbose: print('registering installer [%s]' % (installer_key)) self.installers[installer_key] = installer
[docs] def get_installer(self, installer_key): """ :returns: :class:`Installer` class associated with *installer_key*. :raises: :exc:`KeyError` If not associated installer :raises: :exc:`InstallFailed` If installer cannot produce an install command (e.g. if installer is not installed) """ return self.installers[installer_key]
[docs] def get_installer_keys(self): """ :returns: list of registered installer keys """ return self.installers.keys()
[docs] def get_os_keys(self): """ :returns: list of OS keys that have registered with this context, ``[str]`` """ return self.os_installers.keys()
[docs] def add_os_installer_key(self, os_key, installer_key): """ Register an installer for the specified OS. This will fail with a :exc:`KeyError` if no :class:`Installer` can be found with the associated *installer_key*. :param os_key: Key for OS :param installer_key: Key for installer to add to OS :raises: :exc:`KeyError`: if installer for *installer_key* is not set. """ # validate, will throw KeyError self.get_installer(installer_key) if self.verbose: print('add installer [%s] to OS [%s]' % (installer_key, os_key)) if os_key in self.os_installers: self.os_installers[os_key].append(installer_key) else: self.os_installers[os_key] = [installer_key]
[docs] def get_os_installer_keys(self, os_key): """ Get list of installer keys registered for the specified OS. These keys can be resolved by calling :meth:`InstallerContext.get_installer`. :param os_key: Key for OS :raises: :exc:`KeyError`: if no information for OS *os_key* is registered. """ if os_key in self.os_installers: return self.os_installers[os_key][:] else: raise KeyError(os_key)
[docs] def set_default_os_installer_key(self, os_key, installer_key): """ Set the default OS installer to use for OS. :meth:`InstallerContext.add_os_installer` must have previously been called with the same arguments. :param os_key: Key for OS :param installer_key: Key for installer to add to OS :raises: :exc:`KeyError`: if installer for *installer_key* is not set or if OS for *os_key* has no associated installers. """ if os_key not in self.os_installers: raise KeyError('unknown OS: %s' % (os_key)) if not hasattr(installer_key, '__call__'): raise ValueError('version type should be a method') if not installer_key(self.os_detect) in self.os_installers[os_key]: raise KeyError('installer [%s] is not associated with OS [%s]. call add_os_installer_key() first' % (installer_key(self.os_detect), os_key)) if self.verbose: print('set default installer [%s] for OS [%s]' % (installer_key(self.os_detect), os_key,)) self.default_os_installer[os_key] = installer_key
[docs] def get_default_os_installer_key(self, os_key): """ Get the default OS installer key to use for OS, or ``None`` if there is no default. :param os_key: Key for OS :returns: :class:`Installer` :raises: :exc:`KeyError`: if no information for OS *os_key* is registered. """ if os_key not in self.os_installers: raise KeyError('unknown OS: %s' % (os_key)) try: installer_key = self.default_os_installer[os_key](self.os_detect) if installer_key not in self.os_installers[os_key]: raise KeyError('installer [%s] is not associated with OS [%s]. call add_os_installer_key() first' % (installer_key, os_key)) # validate, will throw KeyError self.get_installer(installer_key) return installer_key except KeyError: return None
[docs]class Installer(object): """ The :class:`Installer` API is designed around opaque *resolved* parameters. These parameters can be any type of sequence object, but they must obey set arithmetic. They should also implement ``__str__()`` methods so they can be pretty printed. """
[docs] def is_installed(self, resolved_item): """ :param resolved: resolved installation item. NOTE: this is a single item, not a list of items like the other APIs, ``opaque``. :returns: ``True`` if all of the *resolved* items are installed on the local system """ raise NotImplementedError('is_installed', resolved_item)
[docs] def get_install_command(self, resolved, interactive=True, reinstall=False, quiet=False): """ :param resolved: list of resolved installation items, ``[opaque]`` :param interactive: If `False`, disable interactive prompts, e.g. Pass through ``-y`` or equivalant to package manager. :param reinstall: If `True`, install everything even if already installed """ raise NotImplementedError('get_package_install_command', resolved, interactive, reinstall, quiet)
[docs] def get_depends(self, rosdep_args): """ :returns: list of dependencies on other rosdep keys. Only necessary if the package manager doesn't handle dependencies. """ return [] # Default return empty list
[docs] def resolve(self, rosdep_args_dict): """ :param rosdep_args_dict: argument dictionary to the rosdep rule for this package manager :returns: [resolutions]. resolved objects should be printable to a user, but are otherwise opaque. """ raise NotImplementedError('Base class resolve', rosdep_args_dict)
[docs] def unique(self, *resolved_rules): """ Combine the resolved rules into a unique list. This is meant to combine the results of multiple calls to :meth:`PackageManagerInstaller.resolve`. Example:: resolved1 = installer.resolve(args1) resolved2 = installer.resolve(args2) resolved = installer.unique(resolved1, resolved2) :param resolved_rules: resolved arguments. Resolved arguments must all be from this :class:`Installer` instance. """ raise NotImplementedError('Base class unique', resolved_rules)
[docs]class PackageManagerInstaller(Installer): """ General form of a package manager :class:`Installer` implementation that assumes: - installer rosdep args spec is a list of package names stored with the key "packages" - a detect function exists that can return a list of packages that are installed Also, if *supports_depends* is set to ``True``: - installer rosdep args spec can also include dependency specification with the key "depends" """ def __init__(self, detect_fn, supports_depends=False): """ :param supports_depends: package manager supports dependency key :param detect_fn: function that for a given list of packages determines the list of installed packages. """ self.detect_fn = detect_fn self.supports_depends = supports_depends self.as_root = True self.sudo_command = 'sudo -H' if hasattr(os, 'geteuid') and os.geteuid() != 0 else ''
[docs] def elevate_priv(self, cmd): """ Prepend *self.sudo_command* to the command if *self.as_root* is ``True``. :param list cmd: list of strings comprising the command :returns: a list of commands """ return (self.sudo_command.split() if self.as_root else []) + cmd
[docs] def resolve(self, rosdep_args): """ See :meth:`Installer.resolve()` """ packages = None if type(rosdep_args) == dict: packages = rosdep_args.get('packages', []) if isinstance(packages, str): packages = packages.split() elif isinstance(rosdep_args, str): packages = rosdep_args.split(' ') elif type(rosdep_args) == list: packages = rosdep_args else: raise InvalidData('Invalid rosdep args: %s' % (rosdep_args)) return packages
[docs] def unique(self, *resolved_rules): """ See :meth:`Installer.unique()` """ s = set() for resolved in resolved_rules: s.update(resolved) return sorted(list(s))
[docs] def get_packages_to_install(self, resolved, reinstall=False): """ Return a list of packages (out of *resolved*) that still need to get installed. """ if reinstall: return resolved if not resolved: return [] else: detected = self.detect_fn(resolved) return [x for x in resolved if x not in detected]
[docs] def is_installed(self, resolved_item): """ Check if a given package was installed. """ return not self.get_packages_to_install([resolved_item])
[docs] def get_version_strings(self): """ Return a list of version information strings. Where each string is of the form "<installer> <version string>". For example, ["apt-get x.y.z"] or ["pip x.y.z", "setuptools x.y.z"]. """ raise NotImplementedError('subclasses must implement get_version_strings method')
[docs] def get_install_command(self, resolved, interactive=True, reinstall=False, quiet=False): raise NotImplementedError('subclasses must implement', resolved, interactive, reinstall, quiet)
[docs] def get_depends(self, rosdep_args): """ :returns: list of dependencies on other rosdep keys. Only necessary if the package manager doesn't handle dependencies. """ if self.supports_depends and type(rosdep_args) == dict: return rosdep_args.get('depends', []) return [] # Default return empty list
def normalize_uninstalled_to_list(uninstalled): uninstalled_dependencies = [] for pkg_or_list in [v for k, v in uninstalled]: if isinstance(pkg_or_list, list): for pkg in pkg_or_list: uninstalled_dependencies.append(str(pkg)) else: uninstalled_dependencies.append(str(pkg)) return uninstalled_dependencies class RosdepInstaller(object): def __init__(self, installer_context, lookup): self.installer_context = installer_context self.lookup = lookup def get_uninstalled(self, resources, implicit=False, verbose=False): """ Get list of system dependencies that have not been installed as well as a list of errors from performing the resolution. This is a bulk API in order to provide performance optimizations in checking install state. :param resources: List of resource names (e.g. ROS package names), ``[str]]`` :param implicit: Install implicit (recursive) dependencies of resources. Default ``False``. :returns: (uninstalled, errors), ``({str: [opaque]}, {str: ResolutionError})``. Uninstalled is a dictionary with the installer_key as the key. :raises: :exc:`RosdepInternalError` """ installer_context = self.installer_context # resolutions have been unique()d if verbose: print('resolving for resources [%s]' % (', '.join(resources))) resolutions, errors = self.lookup.resolve_all(resources, installer_context, implicit=implicit) # for each installer, figure out what is left to install uninstalled = [] if resolutions == []: return uninstalled, errors for installer_key, resolved in resolutions: # py3k if verbose: print('resolution: %s [%s]' % (installer_key, ', '.join([str(r) for r in resolved]))) try: installer = installer_context.get_installer(installer_key) except KeyError as e: # lookup has to be buggy to cause this raise RosdepInternalError(e) try: packages_to_install = installer.get_packages_to_install(resolved) except Exception as e: rd_debug(traceback.format_exc()) raise RosdepInternalError(e, message='Bad installer [%s]: %s' % (installer_key, e)) # only create key if there is something to do if packages_to_install: uninstalled.append((installer_key, packages_to_install)) if verbose: print('uninstalled: [%s]' % (', '.join([str(p) for p in packages_to_install]))) return uninstalled, errors def install(self, uninstalled, interactive=True, simulate=False, continue_on_error=False, reinstall=False, verbose=False, quiet=False): """ Install the uninstalled rosdeps. This API is for the bulk workflow of rosdep (see example below). For a more targeted install API, see :meth:`RosdepInstaller.install_resolved`. :param uninstalled: uninstalled value from :meth:`RosdepInstaller.get_uninstalled`. Value is a dictionary mapping installer key to a dictionary with resolution data, ``{str: {str: vals}}`` :param interactive: If ``False``, suppress interactive prompts (e.g. by passing '-y' to ``apt``). :param simulate: If ``False`` simulate installation without actually executing. :param continue_on_error: If ``True``, continue installation even if an install fails. Otherwise, stop after first installation failure. :param reinstall: If ``True``, install dependencies if even already installed (default ``False``). :raises: :exc:`InstallFailed` if any rosdeps fail to install and *continue_on_error* is ``False``. :raises: :exc:`KeyError` If *uninstalled* value has invalid installer keys Example:: uninstalled, errors = installer.get_uninstalled(packages) installer.install(uninstalled) """ if verbose: print( 'install options: reinstall[%s] simulate[%s] interactive[%s]' % (reinstall, simulate, interactive) ) uninstalled_list = normalize_uninstalled_to_list(uninstalled) print('install: uninstalled keys are %s' % ', '.join(uninstalled_list)) # Squash uninstalled again, in case some dependencies were already installed squashed_uninstalled = [] previous_installer_key = None for installer_key, resolved in uninstalled: if previous_installer_key != installer_key: squashed_uninstalled.append((installer_key, [])) previous_installer_key = installer_key squashed_uninstalled[-1][1].extend(resolved) failures = [] for installer_key, resolved in squashed_uninstalled: try: self.install_resolved(installer_key, resolved, simulate=simulate, interactive=interactive, reinstall=reinstall, continue_on_error=continue_on_error, verbose=verbose, quiet=quiet) except InstallFailed as e: if not continue_on_error: raise else: # accumulate errors failures.extend(e.failures) if failures: raise InstallFailed(failures=failures) def install_resolved(self, installer_key, resolved, simulate=False, interactive=True, reinstall=False, continue_on_error=False, verbose=False, quiet=False): """ Lower-level API for installing a rosdep dependency. The rosdep keys have already been resolved to *installer_key* and *resolved* via :exc:`RosdepLookup` or other means. :param installer_key: Key for installer to apply to *resolved*, ``str`` :param resolved: Opaque resolution list from :class:`RosdepLookup`. :param interactive: If ``True``, allow interactive prompts (default ``True``) :param simulate: If ``True``, don't execute installation commands, just print to screen. :param reinstall: If ``True``, install dependencies if even already installed (default ``False``). :param verbose: If ``True``, print verbose output to screen (default ``False``) :param quiet: If ``True``, supress output except for errors (default ``False``) :raises: :exc:`InstallFailed` if any of *resolved* fail to install. """ installer_context = self.installer_context installer = installer_context.get_installer(installer_key) command = installer.get_install_command(resolved, interactive=interactive, reinstall=reinstall, quiet=quiet) if not command: if verbose: print('#No packages to install') return if simulate: print('#[%s] Installation commands:' % (installer_key)) for sub_command in command: if isinstance(sub_command[0], list): sub_cmd_len = len(sub_command) for i, cmd in enumerate(sub_command): print(" '%s' (alternative %d/%d)" % (' '.join(cmd), i + 1, sub_cmd_len)) else: print(' ' + ' '.join(sub_command)) # nothing left to do for simulation if simulate: return def run_command(command, installer_key, failures, verbose): # always echo commands to screen print_bold('executing command [%s]' % ' '.join(command)) result = subprocess.call(command) if verbose: print('command return code [%s]: %s' % (' '.join(command), result)) if result != 0: failures.append((installer_key, 'command [%s] failed' % (' '.join(command)))) return result # run each install command set and collect errors failures = [] for sub_command in command: if isinstance(sub_command[0], list): # list of alternatives alt_failures = [] for alt_command in sub_command: result = run_command(alt_command, installer_key, alt_failures, verbose) if result == 0: # one successsfull command is sufficient alt_failures = [] # clear failuers from other alternatives break failures.extend(alt_failures) else: result = run_command(sub_command, installer_key, failures, verbose) if result != 0: if not continue_on_error: raise InstallFailed(failures=failures) # test installation of each for r in resolved: if not installer.is_installed(r): failures.append((installer_key, 'Failed to detect successful installation of [%s]' % (r))) # finalize result if failures: raise InstallFailed(failures=failures) elif verbose: print('#successfully installed')