platypush/platypush/utils/manifest.py

import enum
import inspect
import json
import logging
import os
import pathlib
import shutil
import yaml

from abc import ABC, abstractmethod
from typing import Optional, Iterable, Mapping, Callable, Type

supported_package_managers = {
    'pacman': 'pacman -S',
    'apt': 'apt-get install',
}

_available_package_manager = None


class ManifestType(enum.Enum):
    PLUGIN = 'plugin'
    BACKEND = 'backend'


class Manifest(ABC):
    """
    Base class for plugin/backend manifests.
    """
    def __init__(self, package: str, description: Optional[str] = None,
                 install: Optional[Iterable[str]] = None, events: Optional[Mapping] = None, **_):
        self.description = description
        self.install = install or {}
        self.events = events or {}
        self.logger = logging.getLogger(__name__)
        self.package = package
        self.component_name = '.'.join(package.split('.')[2:])
        self.component = None

    @classmethod
    @property
    @abstractmethod
    def component_getter(self) -> Callable[[str], object]:
        raise NotImplementedError

    @classmethod
    def from_file(cls, filename: str) -> "Manifest":
        with open(str(filename), 'r') as f:
            manifest = yaml.safe_load(f).get('manifest', {})

        assert 'type' in manifest, f'Manifest file {filename} has no type field'
        comp_type = ManifestType(manifest.pop('type'))
        manifest_class = _manifest_class_by_type[comp_type]
        return manifest_class(**manifest)

    @classmethod
    def from_class(cls, clazz) -> "Manifest":
        return cls.from_file(os.path.dirname(inspect.getfile(clazz)))

    @classmethod
    def from_component(cls, comp) -> "Manifest":
        return cls.from_class(comp.__class__)

    def get_component(self):
        try:
            self.component = self.component_getter(self.component_name)
        except Exception as e:
            self.logger.warning(f'Could not load {self.component_name}: {e}')

        return self.component

    def __repr__(self):
        return json.dumps({
            'description': self.description,
            'install': self.install,
            'events': self.events,
            'type': _manifest_type_by_class[self.__class__].value,
            'package': self.package,
            'component_name': self.component_name,
        })


class PluginManifest(Manifest):
    @classmethod
    @property
    def component_getter(self):
        from platypush.context import get_plugin
        return get_plugin


class BackendManifest(Manifest):
    @classmethod
    @property
    def component_getter(self):
        from platypush.context import get_backend
        return get_backend


_manifest_class_by_type: Mapping[ManifestType, Type[Manifest]] = {
    ManifestType.PLUGIN: PluginManifest,
    ManifestType.BACKEND: BackendManifest,
}

_manifest_type_by_class: Mapping[Type[Manifest], ManifestType] = {
    cls: t for t, cls in _manifest_class_by_type.items()
}


def scan_manifests(base_class: Type) -> Iterable[str]:
    for mf in pathlib.Path(os.path.dirname(inspect.getfile(base_class))).rglob('manifest.yaml'):
        yield str(mf)


def get_manifests(base_class: Type) -> Iterable[Manifest]:
    return [
        Manifest.from_file(mf)
        for mf in scan_manifests(base_class)
    ]


def get_components(base_class: Type) -> Iterable:
    manifests = get_manifests(base_class)
    components = {mf.get_component() for mf in manifests}
    return {comp for comp in components if comp is not None}


def get_manifests_from_conf(conf_file: Optional[str] = None) -> Mapping[str, Manifest]:
    import platypush
    from platypush.config import Config

    conf_args = []
    if conf_file:
        conf_args.append(conf_file)

    Config.init(*conf_args)
    app_dir = os.path.dirname(inspect.getfile(platypush))
    manifest_files = set()

    for name in Config.get_backends().keys():
        manifest_files.add(os.path.join(app_dir, 'backend', *name.split('.'), 'manifest.yaml'))

    for name in Config.get_plugins().keys():
        manifest_files.add(os.path.join(app_dir, 'plugins', *name.split('.'), 'manifest.yaml'))

    return {
        manifest_file: Manifest.from_file(manifest_file)
        for manifest_file in manifest_files
    }


def get_dependencies_from_conf(conf_file: Optional[str] = None) -> Mapping[str, Iterable[str]]:
    manifests = get_manifests_from_conf(conf_file)
    deps = {
        'pip': set(),
        'packages': set(),
        'exec': set(),
    }

    for manifest in manifests.values():
        deps['pip'].update(manifest.install.get('pip', set()))
        deps['exec'].update(manifest.install.get('exec', set()))
        has_requires_packages = len([
            section for section in manifest.install.keys()
            if section in supported_package_managers
        ]) > 0

        if has_requires_packages:
            pkg_manager = get_available_package_manager()
            deps['packages'].update(manifest.install.get(pkg_manager, set()))

    return deps


def get_install_commands_from_conf(conf_file: Optional[str] = None) -> Mapping[str, str]:
    deps = get_dependencies_from_conf(conf_file)
    return {
        'pip': f'pip install {" ".join(deps["pip"])}',
        'exec': deps["exec"],
        'packages': f'{supported_package_managers[_available_package_manager]} {" ".join(deps["packages"])}'
        if deps['packages'] else None,
    }


def get_available_package_manager() -> str:
    global _available_package_manager
    if _available_package_manager:
        return _available_package_manager

    available_package_managers = [
        pkg_manager for pkg_manager in supported_package_managers.keys()
        if shutil.which(pkg_manager)
    ]

    assert available_package_managers, (
        'Your OS does not provide any of the supported package managers. '
        f'Supported package managers: {supported_package_managers.keys()}'
    )

    _available_package_manager = available_package_managers[0]
    return _available_package_manager