thoth.adviser package

Submodules

thoth.adviser.anneal module

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

class thoth.adviser.anneal.AdaptiveSimulatedAnnealing(*, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, hill_climbing: bool = False, library_usage: Optional[Dict[str, Any]] = NOTHING, graph: thoth.storages.graph.postgres.GraphDatabase = NOTHING, limit: int = 1000, count: int = 1000, temperature_function: Callable[[int, float, int, int], float] = <bound method ASATemperatureFunction.exp of <class 'thoth.adviser.temperature.ASATemperatureFunction'>>, seed: Optional[int] = None, beam_width=-1, keep_temperature_history=None, limit_latest_versions=-1, solver: Optional[thoth.adviser.solver.PythonPackageGraphSolver] = None, context: Optional[thoth.adviser.context.Context] = None, fully_specified_runtime_environment: Optional[bool] = None, temperature_history: List[Tuple[float, bool, float, int]] = NOTHING)[source]

Bases: object

Implementation of adaptive simulated annealing looking for stacks based on the scoring function.

DEFAULT_BEAM_WIDTH = -1
DEFAULT_COUNT = 1000
DEFAULT_LIMIT = 1000
DEFAULT_LIMIT_LATEST_VERSIONS = -1
anneal(*, with_devel: bool = True) → thoth.adviser.report.Report[source]

Perform adaptive simulated annealing and report back a report of run.

anneal_products(*, with_devel: bool = True) → Generator[[thoth.adviser.product.Product, None], None][source]

Run simulated annealing, produce product.

beam_width
classmethod compute_on_project(*, beam_width: Optional[int] = None, count: int = 1000, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit: int = 1000, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, recommendation_type: thoth.adviser.enums.RecommendationType, seed: Optional[int] = None, temperature_function: Callable[[int, float, int, int], float] = <bound method ASATemperatureFunction.exp of <class 'thoth.adviser.temperature.ASATemperatureFunction'>>) → thoth.adviser.report.Report[source]

Compute recommendations on the given project.

property context

Return context bound to this annealing.

count
get_temperature_history() → List[Tuple[float, bool, float, int]][source]

Retrieve temperature history from the last run.

graph
hill_climbing
keep_temperature_history
library_usage
limit
limit_latest_versions
pipeline
project
seed
property solver

Get solver instance - solver implemented on top of graph database.

temperature_function

thoth.adviser.beam module

Implementation of Beam for beam searching part of adviser.

class thoth.adviser.beam.Beam(width: Optional[int] = None, states: List[thoth.adviser.state.State] = NOTHING)[source]

Bases: object

Beam implementation.

The implementation of beam respecting beam width:

Beam is internally implemented on top of heap queue to optimize inserts and respect beam width in O(log(N)).

add_state(state: thoth.adviser.state.State) → None[source]

Add candidate to the internal candidates listing.

get(idx: int) → thoth.adviser.state.State[source]

Get i-th element from the beam, keep it in the beam.

iter_states() → List[thoth.adviser.state.State][source]

Iterate over states, do not respect their score in order of iteration.

pop(idx: Optional[int] = None) → thoth.adviser.state.State[source]

Pop i-th element from the beam and remove it from the beam (this is actually toppop).

If index is not provided, pop largest element kept in the beam.

property size

Get the current size of beam.

property states

Return a list of candidates stored in the beam.

top() → thoth.adviser.state.State[source]

Return the highest rated state as kept in the beam.

width

thoth.adviser.boot module

A base class for implementing boot units.

class thoth.adviser.boot.Boot(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: thoth.adviser.unit.Unit

Boot base class implementation.

abstract run() → None[source]

Main entry-point for boot units to filter and score packages.

thoth.adviser.cli module

Thoth-adviser CLI.

thoth.adviser.context module

Pipeline context carried during annealing.

class thoth.adviser.context.Context(*, package_versions: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion] = NOTHING, sources: Dict[str, thoth.python.source.Source] = NOTHING, advised_runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = None, stack_info: Optional[List[Dict[str, Any]]] = NOTHING, accepted_final_states_count: int = 0, discarded_final_states_count: int = 0)[source]

Bases: object

Context carried during adviser’s pipeline run.

It’s suitable to cache entries such as PackageVersion to optimize memory usage and optimize overhead needed - for example for parsing version strings (this is lazily pre-cached in PackageVersion).

property accepted_final_states_count

Return number of accepted final states as produced by the pipeline.

advised_runtime_environment
property discarded_final_states_count

Return number of discarded final states (discarded by strides) during the pipeline run.

get_package_version(package_tuple: Tuple[str, str, str]) → thoth.python.package_version.PackageVersion[source]

Get the given package version registered to the context.

inc_accepted_final_states_count() → None[source]

Increment counter of accepted final states count.

inc_discarded_final_states_count() → None[source]

Increment counter of discarded final states count.

package_versions
register_package_tuple(package_tuple: Tuple[str, str, str], *, develop: bool, markers: Optional[str] = None, extras: Optional[List[str]] = None) → thoth.python.package_version.PackageVersion[source]

Register the given package tuple to pipeline context and return its package version representative.

register_package_version(package_version: thoth.python.package_version.PackageVersion) → bool[source]

Register the given package version to the context.

sources
stack_info

thoth.adviser.dependency_monkey module

Routines for dependency monkey and its output handling.

class thoth.adviser.dependency_monkey.DependencyMonkey(*, asa: thoth.adviser.anneal.AdaptiveSimulatedAnnealing, stack_output: str = '-', context: Optional[Dict[Any, Any]] = NOTHING, dry_run: bool = False, decision_type: thoth.adviser.enums.DecisionType = <DecisionType.ALL: 1>)[source]

Bases: object

Dependency monkey creates stacks based on the configuration using ASA.

anneal(*, with_devel: bool = True) → thoth.adviser.dependency_monkey.DependencyMonkeyReport[source]

Perform simulated annealing and run dependency monkey on products.

asa
context
decision_type
dry_run
stack_output
class thoth.adviser.dependency_monkey.DependencyMonkeyReport(skipped: int = 0, responses: List[str] = NOTHING, temperature_history: Optional[List[Tuple[float, bool, float, int]]] = None)[source]

Bases: object

Report produced by a Dependency Monkey run.

add_response(response: str) → None[source]

Add a new response to response listing.

inc_skipped() → None[source]

Increment skipped stacks (e.g. due to API unavailability errors).

plot_history(output_file: Optional[str] = None) → matplotlib.figure.Figure[source]

Plot history of temperature function.

set_temperature_history(temperature_history: List[Tuple[float, bool, float, int]]) → None[source]

Mark the temperature history during annealing.

to_dict() → Dict[str, Any][source]

Convert report to a dict representation suitable for serialization.

thoth.adviser.digests_fetcher module

Fetcher for fetching digests from the graph database.

class thoth.adviser.digests_fetcher.GraphDigestsFetcher(graph: thoth.storages.graph.postgres.GraphDatabase = NOTHING)[source]

Bases: thoth.python.digests_fetcher.DigestsFetcherBase

Fetch digests from the graph database.

fetch_digests(package_name: str, package_version: str) → Dict[str, List[Dict[str, str]]][source]

Fetch digests for the given package in specified version, consider only enabled indexes.

graph

thoth.adviser.enums module

Enum types used in thoth-adviser code.

class thoth.adviser.enums.DecisionType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Type of decision used in Dependency Monkey for generating stacks.

ALL = 1
RANDOM = 2
class thoth.adviser.enums.Ecosystem[source]

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1
class thoth.adviser.enums.PythonRecommendationOutput[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1
REQUIREMENTS = 2
class thoth.adviser.enums.RecommendationType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3
STABLE = 1
TESTING = 2

thoth.adviser.exceptions module

Exception hierarchy used in the whole adviser implementation.

exception thoth.adviser.exceptions.AdviserException[source]

Bases: Exception

A base adviser exception in adviser’s exception hierarchy.

exception thoth.adviser.exceptions.AdviserPipelineException[source]

Bases: thoth.adviser.exceptions.AdviserException

A base class for implementing pipeline specific exceptions.

exception thoth.adviser.exceptions.BootError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline boot unit fails unexpectedly.

exception thoth.adviser.exceptions.CannotProduceStack[source]

Bases: thoth.adviser.exceptions.AdviserException

Raised if adviser or dependency monkey cannot produce stack.

exception thoth.adviser.exceptions.EagerStopPipeline[source]

Bases: thoth.adviser.exceptions.AdviserPipelineException

Raised to signalize premature end of pipeline.

exception thoth.adviser.exceptions.InternalError[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on bugs in the code base.

exception thoth.adviser.exceptions.NoHistoryKept[source]

Bases: thoth.adviser.exceptions.AdviserException

Raised if a user asks for history, but history was not kept (e.g. temperature function history in annealing).

exception thoth.adviser.exceptions.NotAcceptable[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised during stack generation when the given action would produce un-acceptable solution.

exception thoth.adviser.exceptions.NotFound[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if the given resource was not found.

exception thoth.adviser.exceptions.PipelineUnitError[source]

Bases: thoth.adviser.exceptions.AdviserPipelineException

An exception raised when there is an error during pipeline run, unexpectedly.

exception thoth.adviser.exceptions.PipfileParseError[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on invalid Pipfile or Pipfile.lock.

exception thoth.adviser.exceptions.SieveError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline sieve unit fails unexpectedly.

exception thoth.adviser.exceptions.StepError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline step unit fails unexpectedly.

exception thoth.adviser.exceptions.StrideError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline stride unit fails unexpectedly.

exception thoth.adviser.exceptions.UnableLock[source]

Bases: thoth.adviser.exceptions.AdviserException

Raised if it is unable to lock dependencies given the set of constraints.

exception thoth.adviser.exceptions.UnresolvedDependencies(*args: Any, unresolved: List[str])[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if dependencies were not resolved and cannot produce stack.

unresolved
exception thoth.adviser.exceptions.UnsupportedConfiguration[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on unsupported configuration by recommendation engine.

exception thoth.adviser.exceptions.VersionIdentifierError[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if the given version identifier is not a semver identifier.

exception thoth.adviser.exceptions.WrapError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline stride unit fails unexpectedly.

thoth.adviser.pipeline_builder module

Implementation of pipeline builder - create pipeline configuration based on the project and its library usage.

class thoth.adviser.pipeline_builder.PipelineBuilder[source]

Bases: object

Builder responsible for creating pipeline configuration from the project and its library usage.

classmethod get_adviser_pipeline_config(*, recommendation_type: thoth.adviser.enums.RecommendationType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get adviser’s pipeline configuration.

classmethod get_dependency_monkey_config(*, decision_type: thoth.adviser.enums.DecisionType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get dependency-monkey’s pipeline configuration.

class thoth.adviser.pipeline_builder.PipelineBuilderContext(*, graph: thoth.storages.graph.postgres.GraphDatabase = None, project: thoth.python.project.Project = None, library_usage: Optional[Dict[str, Any]] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING, boots_included: Set[Any] = NOTHING, sieves_included: Set[Any] = NOTHING, steps_included: Set[Any] = NOTHING, strides_included: Set[Any] = NOTHING, wraps_included: Set[Any] = NOTHING)[source]

Bases: object

A context passed to units to determine if they want to participate in a pipeline.

add_unit(unit: thoth.adviser.unit.Unit) → None[source]

Add the given unit to pipeline configuration.

boots
decision_type
graph
is_adviser_pipeline() → bool[source]

Check if the pipeline built is meant for adviser.

is_dependency_monkey_pipeline() → bool[source]

Check if the pipeline built is meant for Dependency Monkey.

is_included(unit_class: type) → bool[source]

Check if the given pipeline unit is already included in the pipeline configuration.

library_usage
project
recommendation_type
sieves
steps
strides
wraps

thoth.adviser.pipeline_config module

A base class for implementing sieves.

class thoth.adviser.pipeline_config.PipelineConfig(boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING)[source]

Bases: object

A configuration of a pipline for dependency-monkey and for adviser.

boots
sieves
steps
strides
to_dict() → Dict[str, List[Dict[str, Any]]][source]

Return this pipeline configuration in a dict representation.

wraps

thoth.adviser.plot module

Routines for dependency monkey and its output handling.

thoth.adviser.plot.plot_history(temperature_history: Optional[List[Tuple[float, bool, float, int]]], output_file: Optional[str] = None) → matplotlib.figure.Figure[source]

Plot temperature history during annealing.

thoth.adviser.product module

Representation of an advised stack.

class thoth.adviser.product.Product(project: thoth.python.project.Project, score: float, justification: List[Dict[str, str]])[source]

Bases: object

A representation of an advised stack.

classmethod from_final_state(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product[source]

Instantiate advised stack from final state produced by adviser’s pipeline.

justification
project
score
to_dict() → Dict[str, Any][source]

Convert this instance into a dictionary.

thoth.adviser.report module

A class for adviser’s report - output of an adviser run.

class thoth.adviser.report.Report(*, count: int, pipeline: thoth.adviser.pipeline_config.PipelineConfig, advised_runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = None, stack_info: Optional[List[Dict[str, Any]]] = None, temperature_history: Optional[List[Tuple[float, bool, float, int]]] = None, heapq: List[thoth.adviser.product.Product] = NOTHING)[source]

Bases: object

A report stating output of an adviser run.

add_product(product: thoth.adviser.product.Product) → None[source]

Add adviser pipeline product to report.

count
iter_products() → List[thoth.adviser.product.Product][source]

Iterate over products stored in this report, respect their scores.

pipeline
plot_history(output_file: Optional[str] = None) → matplotlib.figure.Figure[source]

Plot history of temperature function.

product_count() → int[source]

Get number of products stored in the report.

set_advised_runtime_environment(runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment) → None[source]

Mark advised runtime environment.

set_stack_info(stack_info: List[Dict[str, Any]]) → None[source]

Set stack information.

set_temperature_history(temperature_history: List[Tuple[float, bool, float, int]]) → None[source]

Mark the temperature history during annealing.

to_dict() → Dict[str, Any][source]

Convert pipeline report to a dict representation.

thoth.adviser.run module

Utilities which wrap adviser’s and dependency monkey’s and run them in a subprocess.

This is useful for a cluster deployment where we restrict resources allocated for an adviser or dependency monkey run. Annealing is run as a subprocess to the main process - if resources are exhausted (CPU time or memory allocated), the subprocess is killed (either by liveness probe or by OOM killer) and the main process can still produce an error message.

thoth.adviser.run.subprocess_run(asa: Callable[Union[thoth.adviser.report.Report, thoth.adviser.dependency_monkey.DependencyMonkeyReport]], print_func: Callable[[float, Union[Dict[str, Any], List[Any]]], None], result_dict: Dict[str, Any], plot_history: Optional[str] = None) → int[source]

Run the given function (partial annealing method) in a subprocess and output the produced report.

thoth.adviser.sieve module

A base class for implementing steps.

class thoth.adviser.sieve.Sieve(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: thoth.adviser.unit.Unit

Sieve base class implementation.

abstract run(package_version: thoth.python.package_version.PackageVersion) → Optional[Tuple[Optional[float], Optional[List[Dict[str, str]]]]][source]

Main entry-point for sieves to filter and score packages.

thoth.adviser.solver module

Definition of package resolution based on precomputed data available in graph.

There are defined primitives required for offline resolution. This off-line resolution is done based on data aggregated in the graph database so thereis no need to perform resolving by actually installing Python packages (as this version resolution is dynamic in case of Python).

class thoth.adviser.solver.GraphReleasesFetcher(runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = NOTHING, *, graph: thoth.storages.graph.postgres.GraphDatabase)[source]

Bases: thoth.solver.python.base.ReleasesFetcher

Fetch releases for packages from the graph database.

fetch_releases(package_name: str)[source]

Fetch releases for the given package name.

graph
runtime_environment
class thoth.adviser.solver.PackageVersionDependencyParser[source]

Bases: thoth.solver.python.base.DependencyParser

Parse an instance of PackageVersion to Dependency object needed by solver.

parse(dependencies: List[thoth.python.package_version.PackageVersion]) → Generator[[packaging.requirements.Requirement, None], None][source]

Parse the given list of PackageVersion objects.

class thoth.adviser.solver.PythonGraphSolver(*, dependency_parser: thoth.adviser.solver.PackageVersionDependencyParser, releases_fetcher: thoth.adviser.solver.GraphReleasesFetcher)[source]

Bases: thoth.solver.python.base.Solver

Solve Python dependencies based on data available in the graph database.

dependency_parser
releases_fetcher
class thoth.adviser.solver.PythonPackageGraphSolver(*, graph: thoth.storages.graph.postgres.GraphDatabase, runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = NOTHING, package_versions: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion] = NOTHING, sources: Dict[str, thoth.python.source.Source] = NOTHING, solver: thoth.adviser.solver.PythonGraphSolver = None)[source]

Bases: object

A wrapper to manipulate with Python packages using pure PackageVersion object interface.

graph
runtime_environment
solve(dependencies: List[thoth.python.package_version.PackageVersion], graceful: bool = True) → Dict[str, List[thoth.python.package_version.PackageVersion]][source]

Solve the given dependencies and return object representation of packages.

property solver

Retrieve solver instance resolving using graph database.

thoth.adviser.state module

A state of not fully resolved software stack in adviser’s recommendations implementation.

class thoth.adviser.state.State(score: float = 0.0, unresolved_dependencies: Dict[str, Tuple[str, str, str]] = NOTHING, resolved_dependencies: Dict[str, Tuple[str, str, str]] = NOTHING, justification: List[Dict[str, str]] = NOTHING)[source]

Bases: object

Implementation of an adviser’s state in state space.

add_justification(justification: List[Dict[str, str]]) → None[source]

Add new entries to the justification field.

add_unresolved_dependency(package_tuple: Tuple[str, str, str]) → None[source]

Add unresolved dependency into the beam.

clone() → thoth.adviser.state.State[source]

Return a swallow copy of this state that can be used as a next state.

is_final() → bool[source]

Check if the given state is a final state.

property justification

Get justification for the current state.

resolved_dependencies
score
unresolved_dependencies

thoth.adviser.step module

A base class for implementing steps.

class thoth.adviser.step.Step(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: thoth.adviser.unit.Unit

Step base class implementation.

abstract run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → Optional[Tuple[Optional[float], Optional[List[Dict[str, str]]]]][source]

Main entry-point for steps to filter and score packages.

thoth.adviser.stride module

A base class for implementing strides.

class thoth.adviser.stride.Stride(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: thoth.adviser.unit.Unit

Stride base class implementation.

abstract run(state: thoth.adviser.state.State) → Optional[Tuple[float, List[Dict[str, str]]]][source]

Main entry-point for strides.

thoth.adviser.temperature module

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

class thoth.adviser.temperature.ASATemperatureFunction[source]

Bases: object

Temperature function used in simulated annealing.

Some of the functions were picked from:

classmethod exp(iteration: int, t0: float, count: int, limit: int) → float[source]

Exponential temperature function.

thoth.adviser.unit module

A base class for implementing pipeline units - strides and steps.

class thoth.adviser.unit.Unit(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: object

A base class for implementing pipeline units - strides and steps.

PARAMETERS_DEFAULT = {}
PARAMETERS_SCHEMA = None
classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[[None, None], None][source]

Assign context to all units.

classmethod compute_expanded_parameters(parameters_dict: Optional[Dict[str, Any]]) → Dict[str, Any][source]

Compute parameters as they would be computed based on unit configuration.

property context

Get context in which the unit runs in.

classmethod get_aicoe_configuration(package_version: thoth.python.package_version.PackageVersion) → Optional[Dict[str, Any]][source]

Get AICoE specific configuration encoded in the AICoE index URL.

graph
classmethod is_aicoe_release(package_version: thoth.python.package_version.PackageVersion) → bool[source]

Check if the given package-version is AICoE release.

library_usage
property name

Name of a stack pipeline.

property parameters

Get parameters of instantiated pipeline step.

project
classmethod should_include(context: PipelineBuilderContext) → Optional[Dict[str, Any]][source]

Check if the given pipeline unit should be included in the given pipeline configuration.

to_dict() → Dict[str, Any][source]

Turn this pipeline step into its dictionary representation.

update_parameters(parameters_dict: Dict[str, Any]) → None[source]

Set parameters for a stack pipeline.

If setting parameters fails due to schema checks, parameters are kept in an invalid state.

thoth.adviser.wrap module

A base class for implementing wrap units.

class thoth.adviser.wrap.Wrap(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]] = NOTHING, parameters: Dict[str, Any] = NOTHING, name: str = None, context: Optional[thoth.adviser.context.Context] = None)[source]

Bases: thoth.adviser.unit.Unit

Wrap base class implementation.

abstract run(state: thoth.adviser.state.State) → None[source]

Main entry-point for wrap units to filter and score packages.

Module contents

Thoth’s adviser for recommending Python stacks.

class thoth.adviser.AdaptiveSimulatedAnnealing(*, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, hill_climbing: bool = False, library_usage: Optional[Dict[str, Any]] = NOTHING, graph: thoth.storages.graph.postgres.GraphDatabase = NOTHING, limit: int = 1000, count: int = 1000, temperature_function: Callable[[int, float, int, int], float] = <bound method ASATemperatureFunction.exp of <class 'thoth.adviser.temperature.ASATemperatureFunction'>>, seed: Optional[int] = None, beam_width=-1, keep_temperature_history=None, limit_latest_versions=-1, solver: Optional[thoth.adviser.solver.PythonPackageGraphSolver] = None, context: Optional[thoth.adviser.context.Context] = None, fully_specified_runtime_environment: Optional[bool] = None, temperature_history: List[Tuple[float, bool, float, int]] = NOTHING)[source]

Bases: object

Implementation of adaptive simulated annealing looking for stacks based on the scoring function.

DEFAULT_BEAM_WIDTH = -1
DEFAULT_COUNT = 1000
DEFAULT_LIMIT = 1000
DEFAULT_LIMIT_LATEST_VERSIONS = -1
anneal(*, with_devel: bool = True) → thoth.adviser.report.Report[source]

Perform adaptive simulated annealing and report back a report of run.

anneal_products(*, with_devel: bool = True) → Generator[[thoth.adviser.product.Product, None], None][source]

Run simulated annealing, produce product.

beam_width
classmethod compute_on_project(*, beam_width: Optional[int] = None, count: int = 1000, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit: int = 1000, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, recommendation_type: thoth.adviser.enums.RecommendationType, seed: Optional[int] = None, temperature_function: Callable[[int, float, int, int], float] = <bound method ASATemperatureFunction.exp of <class 'thoth.adviser.temperature.ASATemperatureFunction'>>) → thoth.adviser.report.Report[source]

Compute recommendations on the given project.

property context

Return context bound to this annealing.

count
get_temperature_history() → List[Tuple[float, bool, float, int]][source]

Retrieve temperature history from the last run.

graph
hill_climbing
keep_temperature_history
library_usage
limit
limit_latest_versions
pipeline
project
seed
property solver

Get solver instance - solver implemented on top of graph database.

temperature_function
class thoth.adviser.ASATemperatureFunction[source]

Bases: object

Temperature function used in simulated annealing.

Some of the functions were picked from:

classmethod exp(iteration: int, t0: float, count: int, limit: int) → float[source]

Exponential temperature function.

class thoth.adviser.Beam(width: Optional[int] = None, states: List[thoth.adviser.state.State] = NOTHING)[source]

Bases: object

Beam implementation.

The implementation of beam respecting beam width:

Beam is internally implemented on top of heap queue to optimize inserts and respect beam width in O(log(N)).

add_state(state: thoth.adviser.state.State) → None[source]

Add candidate to the internal candidates listing.

get(idx: int) → thoth.adviser.state.State[source]

Get i-th element from the beam, keep it in the beam.

iter_states() → List[thoth.adviser.state.State][source]

Iterate over states, do not respect their score in order of iteration.

pop(idx: Optional[int] = None) → thoth.adviser.state.State[source]

Pop i-th element from the beam and remove it from the beam (this is actually toppop).

If index is not provided, pop largest element kept in the beam.

property size

Get the current size of beam.

property states

Return a list of candidates stored in the beam.

top() → thoth.adviser.state.State[source]

Return the highest rated state as kept in the beam.

width
class thoth.adviser.DecisionType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Type of decision used in Dependency Monkey for generating stacks.

ALL = 1
RANDOM = 2
class thoth.adviser.Ecosystem[source]

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1
class thoth.adviser.PipelineBuilder[source]

Bases: object

Builder responsible for creating pipeline configuration from the project and its library usage.

classmethod get_adviser_pipeline_config(*, recommendation_type: thoth.adviser.enums.RecommendationType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get adviser’s pipeline configuration.

classmethod get_dependency_monkey_config(*, decision_type: thoth.adviser.enums.DecisionType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get dependency-monkey’s pipeline configuration.

class thoth.adviser.PipelineConfig(boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING)[source]

Bases: object

A configuration of a pipline for dependency-monkey and for adviser.

boots
sieves
steps
strides
to_dict() → Dict[str, List[Dict[str, Any]]][source]

Return this pipeline configuration in a dict representation.

wraps
class thoth.adviser.Product(project: thoth.python.project.Project, score: float, justification: List[Dict[str, str]])[source]

Bases: object

A representation of an advised stack.

classmethod from_final_state(*, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product[source]

Instantiate advised stack from final state produced by adviser’s pipeline.

justification
project
score
to_dict() → Dict[str, Any][source]

Convert this instance into a dictionary.

class thoth.adviser.PythonRecommendationOutput[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1
REQUIREMENTS = 2
class thoth.adviser.RecommendationType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3
STABLE = 1
TESTING = 2
class thoth.adviser.Report(*, count: int, pipeline: thoth.adviser.pipeline_config.PipelineConfig, advised_runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = None, stack_info: Optional[List[Dict[str, Any]]] = None, temperature_history: Optional[List[Tuple[float, bool, float, int]]] = None, heapq: List[thoth.adviser.product.Product] = NOTHING)[source]

Bases: object

A report stating output of an adviser run.

add_product(product: thoth.adviser.product.Product) → None[source]

Add adviser pipeline product to report.

count
iter_products() → List[thoth.adviser.product.Product][source]

Iterate over products stored in this report, respect their scores.

pipeline
plot_history(output_file: Optional[str] = None) → matplotlib.figure.Figure[source]

Plot history of temperature function.

product_count() → int[source]

Get number of products stored in the report.

set_advised_runtime_environment(runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment) → None[source]

Mark advised runtime environment.

set_stack_info(stack_info: List[Dict[str, Any]]) → None[source]

Set stack information.

set_temperature_history(temperature_history: List[Tuple[float, bool, float, int]]) → None[source]

Mark the temperature history during annealing.

to_dict() → Dict[str, Any][source]

Convert pipeline report to a dict representation.