From 0efca3cadbad5517a59884576ddb90cfe7ac30f8 Mon Sep 17 00:00:00 2001 From: Diego Russo Date: Mon, 30 May 2022 13:34:14 +0100 Subject: Add MLIA codebase Add MLIA codebase including sources and tests. Change-Id: Id41707559bd721edd114793618d12ccd188d8dbd --- src/mlia/core/context.py | 218 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 src/mlia/core/context.py (limited to 'src/mlia/core/context.py') diff --git a/src/mlia/core/context.py b/src/mlia/core/context.py new file mode 100644 index 0000000..8b3dd2c --- /dev/null +++ b/src/mlia/core/context.py @@ -0,0 +1,218 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Context module. + +This module contains functionality related to the Context. +Context is an object that describes advisor working environment +and requested behavior (advice categories, input configuration +parameters). +""" +import logging +from abc import ABC +from abc import abstractmethod +from pathlib import Path +from typing import Any +from typing import List +from typing import Mapping +from typing import Optional +from typing import Union + +from mlia.core.common import AdviceCategory +from mlia.core.events import DefaultEventPublisher +from mlia.core.events import EventHandler +from mlia.core.events import EventPublisher +from mlia.core.helpers import ActionResolver +from mlia.core.helpers import APIActionResolver + +logger = logging.getLogger(__name__) + + +class Context(ABC): + """Abstract class for the execution context.""" + + @abstractmethod + def get_model_path(self, model_filename: str) -> Path: + """Return path for the intermediate/optimized models. + + During workflow execution different parts of the advisor + require creating intermediate files for models. + + This method allows to provide paths where those models + could be saved. + + :param model_filename: filename of the model + """ + + @property + @abstractmethod + def event_publisher(self) -> EventPublisher: + """Return event publisher.""" + + @property + @abstractmethod + def event_handlers(self) -> Optional[List[EventHandler]]: + """Return list of the event_handlers.""" + + @property + @abstractmethod + def advice_category(self) -> Optional[AdviceCategory]: + """Return advice category.""" + + @property + @abstractmethod + def config_parameters(self) -> Optional[Mapping[str, Any]]: + """Return configuration parameters.""" + + @property + @abstractmethod + def action_resolver(self) -> ActionResolver: + """Return action resolver.""" + + @abstractmethod + def update( + self, + *, + advice_category: AdviceCategory, + event_handlers: List[EventHandler], + config_parameters: Mapping[str, Any], + ) -> None: + """Update context parameters.""" + + def category_enabled(self, category: AdviceCategory) -> bool: + """Check if category enabled.""" + return category == self.advice_category + + def any_category_enabled(self, *categories: AdviceCategory) -> bool: + """Return true if any category is enabled.""" + return self.advice_category in categories + + def register_event_handlers(self) -> None: + """Register event handlers.""" + self.event_publisher.register_event_handlers(self.event_handlers) + + +class ExecutionContext(Context): + """Execution context.""" + + def __init__( + self, + *, + advice_category: Optional[AdviceCategory] = None, + config_parameters: Optional[Mapping[str, Any]] = None, + working_dir: Optional[Union[str, Path]] = None, + event_handlers: Optional[List[EventHandler]] = None, + event_publisher: Optional[EventPublisher] = None, + verbose: bool = False, + logs_dir: str = "logs", + models_dir: str = "models", + action_resolver: Optional[ActionResolver] = None, + ) -> None: + """Init execution context. + + :param advice_category: requested advice category + :param config_parameters: dictionary like object with input parameters + :param working_dir: path to the directory that will be used as a place + to store temporary files, logs, models. If not provided then + current working directory will be used instead + :param event_handlers: optional list of event handlers + :param event_publisher: optional event publisher instance. If not provided + then default implementation of event publisher will be used + :param verbose: enable verbose output + :param logs_dir: name of the directory inside working directory where + log files will be stored + :param models_dir: name of the directory inside working directory where + temporary models will be stored + :param action_resolver: instance of the action resolver that could make + advice actionable + """ + self._advice_category = advice_category + self._config_parameters = config_parameters + + self._working_dir_path = Path.cwd() + if working_dir: + self._working_dir_path = Path(working_dir) + self._working_dir_path.mkdir(exist_ok=True) + + self._event_handlers = event_handlers + self._event_publisher = event_publisher or DefaultEventPublisher() + self.verbose = verbose + self.logs_dir = logs_dir + self.models_dir = models_dir + self._action_resolver = action_resolver or APIActionResolver() + + @property + def advice_category(self) -> Optional[AdviceCategory]: + """Return advice category.""" + return self._advice_category + + @advice_category.setter + def advice_category(self, advice_category: AdviceCategory) -> None: + """Setter for the advice category.""" + self._advice_category = advice_category + + @property + def config_parameters(self) -> Optional[Mapping[str, Any]]: + """Return configuration parameters.""" + return self._config_parameters + + @config_parameters.setter + def config_parameters(self, config_parameters: Optional[Mapping[str, Any]]) -> None: + """Setter for the configuration parameters.""" + self._config_parameters = config_parameters + + @property + def event_handlers(self) -> Optional[List[EventHandler]]: + """Return list of the event handlers.""" + return self._event_handlers + + @event_handlers.setter + def event_handlers(self, event_handlers: List[EventHandler]) -> None: + """Setter for the event handlers.""" + self._event_handlers = event_handlers + + @property + def event_publisher(self) -> EventPublisher: + """Return event publisher.""" + return self._event_publisher + + @property + def action_resolver(self) -> ActionResolver: + """Return action resolver.""" + return self._action_resolver + + def get_model_path(self, model_filename: str) -> Path: + """Return path for the model.""" + models_dir_path = self._working_dir_path / self.models_dir + models_dir_path.mkdir(exist_ok=True) + + return models_dir_path / model_filename + + @property + def logs_path(self) -> Path: + """Return path to the logs directory.""" + return self._working_dir_path / self.logs_dir + + def update( + self, + *, + advice_category: AdviceCategory, + event_handlers: List[EventHandler], + config_parameters: Mapping[str, Any], + ) -> None: + """Update context parameters.""" + self._advice_category = advice_category + self._event_handlers = event_handlers + self._config_parameters = config_parameters + + def __str__(self) -> str: + """Return string representation.""" + category = ( + "" if self.advice_category is None else self.advice_category.name + ) + + return ( + f"ExecutionContext: working_dir={self._working_dir_path}, " + f"advice_category={category}, " + f"config_parameters={self.config_parameters}, " + f"verbose={self.verbose}" + ) -- cgit v1.2.1