Source code for thoth.adviser.predictors.annealing

#!/usr/bin/env python3
# thoth-adviser
# Copyright(C) 2019 - 2021 Fridolin Pokorny
#
# This program is free software: you can redistribute it and / or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""Implementation of Adaptive Simulated Annealing (ASA) used to resolve software stacks."""

from typing import Any
from typing import List
from typing import Optional
from typing import Tuple
from typing import cast
from typing import TYPE_CHECKING
import logging
import random
import math

import attr

from ..context import Context
from ..exceptions import NoHistoryKept
from ..predictor import Predictor
from ..state import State

if TYPE_CHECKING:
    import matplotlib

_LOGGER = logging.getLogger(__name__)


[docs]@attr.s(slots=True) class AdaptiveSimulatedAnnealing(Predictor): """Implementation of adaptive simulated annealing looking for stacks based on the scoring function.""" temperature_coefficient = attr.ib(type=float, default=0.999, kw_only=True) _temperature_history = attr.ib( type=List[Tuple[Optional[float], Optional[bool], Optional[float], int]], factory=list, kw_only=True, ) _temperature = attr.ib(type=float, kw_only=True, default=0.0) def _temperature_function(self, t0: float, context: Context) -> float: """Temperature function used to compute new temperature.""" k = (context.accepted_final_states_count + math.log(context.iteration + 1)) / context.limit temperature = t0 * cast( float, self.temperature_coefficient**k ) # https://github.com/python/typeshed/issues/285 _LOGGER.debug( "New temperature for (iteration=%d, t0=%g, accepted final states=%d, limit=%d, beam size= %d, k=%f) = %g", context.iteration, t0, context.accepted_final_states_count, context.limit, context.beam.size, k, temperature, ) return max(temperature, 0.0) @staticmethod def _compute_acceptance_probability(top_score: float, neighbour_score: float, temperature: float) -> float: """Check the probability of acceptance the given solution to expansion.""" if temperature == 0.0: # This can happen as we consider also iteration in the exp function - if we # drop bellow 0, acceptance probability drops to 0. return 0.0 if neighbour_score > top_score: return 1.0 acceptance_probability = math.exp((neighbour_score - top_score) / temperature) _LOGGER.debug( "Acceptance probability for (top_score=%g, neighbour_score=%g, temperature=%g) = %g", top_score, neighbour_score, temperature, acceptance_probability, ) return acceptance_probability
[docs] def run(self) -> Tuple[State, Tuple[str, str, str]]: """Run adaptive simulated annealing on top of beam.""" self._temperature = self._temperature_function(self._temperature, self.context) # Expand highest promising by default. state = self.context.beam.max() # Pick a random state to be expanded if accepted. probable_state_idx = random.randrange(1, self.context.beam.size) if self.context.beam.size > 1 else 0 probable_state = self.context.beam.get(probable_state_idx) acceptance_probability = self._compute_acceptance_probability( state.score, probable_state.score, self._temperature ) if probable_state_idx != 0 and acceptance_probability >= random.random(): # Skip to probable state, do not use the top rated state. _LOGGER.debug( "Performing transition to a neighbour state with score %g", probable_state.score, ) # state_expansion_idx = probable_state_idx state = probable_state unresolved_dependency_tuple = state.get_random_unresolved_dependency(prefer_recent=False) else: _LOGGER.debug("Expanding TOP rated state with score %g", state.score) unresolved_dependency_tuple = state.get_random_unresolved_dependency(prefer_recent=True) if self.keep_history: self._temperature_history.append( ( self._temperature, state is self.context.beam.max(), acceptance_probability, self.context.accepted_final_states_count, ) ) return state, unresolved_dependency_tuple
[docs] def pre_run(self) -> None: """Initialize before the actual annealing run.""" self._temperature_history.clear() self._temperature = float(self.context.limit)
@staticmethod def _make_patch_spines_invisible(ax: Any) -> None: """Make spines invisible.""" ax.set_frame_on(True) ax.patch.set_visible(False) for sp in ax.spines.values(): sp.set_visible(False)
[docs] def plot(self) -> "matplotlib.figure.Figure": """Plot temperature history of adaptive simulated annealing.""" # Code adjusted based on: # https://matplotlib.org/3.1.1/gallery/ticks_and_spines/multiple_yaxis_with_spines.html if self._temperature_history is None: raise NoHistoryKept("No history datapoints kept") import matplotlib.pyplot as plt from matplotlib.font_manager import FontProperties x = [i for i in range(len(self._temperature_history))] # Top rated candidate was chosen. y1 = [i[0] if i[1] is True else None for i in self._temperature_history] # A neighbour candidate was chosen. y2 = [i[0] if i[1] is False else None for i in self._temperature_history] # Acceptance probability - as the probability in 0 - 1, lets make it larger - scale to temperature size. y3 = [i[2] for i in self._temperature_history] # Number of products. y4 = [i[3] for i in self._temperature_history] fig, host = plt.subplots() fig.subplots_adjust(right=0.75) par1 = host.twinx() par2 = host.twinx() # Offset the right spine of par1 and par2. The ticks and label have already been # placed on the right by twinx above. par1.spines["right"].set_position(("axes", 1.050)) par2.spines["right"].set_position(("axes", 1.225)) # Having been created by twinx, par1 par2 have their frame off, so the line of its # detached spine is invisible. First, activate the frame but make the patch # and spines invisible. self._make_patch_spines_invisible(par1) self._make_patch_spines_invisible(par2) # Second, show the right spine. par1.spines["right"].set_visible(True) par2.spines["right"].set_visible(True) host.spines["right"].set_visible(False) host.spines["top"].set_visible(False) host.plot(x, y1, ".g", label="Expansion of a highest rated candidate") host.plot(x, y2, ",r", label="Expansion of a neighbour candidate") (p3,) = par1.plot(x, y3, ",b", label="Acceptance probability for a neighbour candidate") (p4,) = par2.plot(x, y4, ",y", label="Number of products produced in the pipeline") host.set_xlabel("iteration") host.set_ylabel("temperature") par1.set_ylabel("acceptance probability") par2.set_ylabel("product count") host.yaxis.label.set_color("black") par1.yaxis.label.set_color(p3.get_color()) par2.yaxis.label.set_color(p4.get_color()) tkw = dict(size=4, width=1.5) host.tick_params(axis="y", colors="black", **tkw) par1.tick_params(axis="y", colors=p3.get_color(), **tkw) par2.tick_params(axis="y", colors=p4.get_color(), **tkw) host.tick_params(axis="x", **tkw) font_prop = FontProperties() font_prop.set_size("small") fig.legend( loc="upper center", bbox_to_anchor=(0.50, 1.00), ncol=2, fancybox=True, shadow=True, prop=font_prop, ) return fig