Source code for idmtools.registry.platform_specification

PlatformSpecification provided definition for the platform plugin specification, hooks, and plugin manager.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.
# Define our platform specific specifications
import typing
from abc import ABC
from logging import getLogger, DEBUG
import pluggy
from idmtools.registry import PluginSpecification
from idmtools.registry.plugin_specification import PLUGIN_REFERENCE_NAME
from idmtools.registry.utils import load_plugin_map
from idmtools.utils.decorators import SingletonMixin

if typing.TYPE_CHECKING:
    from idmtools.entities.iplatform import IPlatform
example_configuration_spec = pluggy.HookspecMarker(PLUGIN_REFERENCE_NAME)
get_platform_spec = pluggy.HookspecMarker(PLUGIN_REFERENCE_NAME)
get_platform_type_spec = pluggy.HookspecMarker(PLUGIN_REFERENCE_NAME)
example_configuration_impl = pluggy.HookimplMarker(PLUGIN_REFERENCE_NAME)
get_platform_impl = pluggy.HookimplMarker(PLUGIN_REFERENCE_NAME)
get_platform_type_impl = pluggy.HookimplMarker(PLUGIN_REFERENCE_NAME)
logger = getLogger(__name__)

[docs]class PlatformSpecification(PluginSpecification, ABC): """ PlatformSpecification for Platform Plugins. """
[docs] @classmethod def get_name(cls, strip_all: bool = True) -> str: """ Get name of plugin. By default we remove the PlatformSpecification portion. Args: strip_all: When true, PlatformSpecification is stripped from name. When false only Specification is Stripped Returns: Name of plugin """ if strip_all: ret = cls.__name__.replace("PlatformSpecification", '') else: ret = cls.__name__.replace('Specification', '') return ret
[docs] @example_configuration_spec def example_configuration(self): """ Example configuration for the platform. This is useful in help or error messages. Returns: Example configuration """ raise NotImplementedError("Plugin did not implement example_configuration")
[docs] @get_platform_spec def get(self, configuration: dict) -> 'IPlatform': """ Return a new platform using the passed in configuration. Args: configuration: The INI configuration file to use. Returns: The new platform. """ raise NotImplementedError("Plugin did not implement get")
[docs] @get_platform_type_spec def get_type(self) -> typing.Type['IPlatform']: """ Get type of the Platform type. """ pass
[docs] def get_configuration_aliases(self) -> typing.Dict[str, typing.Dict]: """ Get a list of configuration aliases for the platform. A configuration alias should be in the form of "name" -> "Spec, Config Options Dict" where name is the alias the user will use, and the config options is a dictionary of config options to be passed to the item Returns: """ return {}
[docs]class PlatformPlugins(SingletonMixin): """ PlatformPlugin registry. """
[docs] def __init__(self, strip_all: bool = True) -> None: """ Initialize the Platform Registry. When strip all is false, the full plugin name will be used for names in map. Args: strip_all: Whether to strip common parts of name from plugins in plugin map """ self._plugins = typing.cast(typing.Dict[str, PlatformSpecification], load_plugin_map('idmtools_platform', PlatformSpecification, strip_all)) self._aliases: typing.Dict[str, typing.Tuple[PlatformSpecification, typing.Dict]] = dict() for _name, spec in self._plugins.items(): for alias, details in spec.get_configuration_aliases().items(): if alias.upper() in self._aliases or alias.upper() in self._plugins: logger.debug(f"Conflicting alias found: {alias.upper()} from {spec.get_name()}") if logger.isEnabledFor(DEBUG): logger.debug(f"Found Platform Configuration Alias: {alias}") self._aliases[alias.upper()] = (spec, details)
[docs] def get_plugins(self) -> typing.Set[PlatformSpecification]: """ Get platform plugins. Returns: Platform plugins """ return set(self._plugins.values())
[docs] def get_aliases(self) -> typing.Dict[str, typing.Tuple[PlatformSpecification, typing.Dict]]: """ Get Platform Configuration Aliases for Platform Plugin. Returns: Platform CConfiguration Aliases """ return self._aliases
[docs] def get_plugin_map(self) -> typing.Dict[str, PlatformSpecification]: """ Get plugin map for Platform Plugins. Returns: Plugin map """ return self._plugins