Evaluator (Implementation)

alts.modules.evaluator

Core Module

class LogAllEvaluator

Bases: LogingEvaluator

Description

This evaluator logs combines the LogStreamEvaluator, LogProcessEvaluator and LogResultEvaluator.

file_name: str = 'all_data'

folder: str = 'log'

lpev: LogProcessEvaluator

lrev: LogResultEvaluator

lsev: LogStreamEvaluator

post_init(self) → None

Description

Initializes the LogStream-, LogProcess- and LogResult- evaluators

register(self, experiment) → None

Description

Registers the experiment with all 3 above listed evaluators.

class LogOracleEvaluator

Bases: LogingEvaluator

Description

This evaluator logs all queries processed by the oracle.

file_name: str = 'oracle_data'

folder: str = 'log'

log_data(self) → None

Description

Logs all saved queries to an `.npy` file.

register(self, experiment) → None

Description

Modifies the experiment to log all queries added to the oracle process.

Requires the experiment’s oracles to be a POracles.

Parameters:

experiment (Experiments) – The experiment to be evaluated

Raises:

TypeError if self.experiment.oracles is not a POracles

save_query(self, queries) → None

Description

Saves the given query with the previously saved queries.

Parameters:

queries (Tuple[NDArray[Shape["query_nr, ... query_dim"], Number]) – New queries going to the oracle process

class LogProcessEvaluator

Bases: LogingEvaluator

Description

This evaluator logs all data points added to the experiment’s data pools process.

file_name: str = 'process'

folder: str = 'log'

log_data(self) → None

Description

Logs all saved data points to an `.npy` file if there is at least one data point.

register(self, experiment) → None

Description

Modifies the experiment to log all data points added to the data pools process.

Requires the experiment’s data pools to be a ProcessDataPools.

Parameters:

experiment (Experiments) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a ProcessDataPools

save_process(self, data) → None

Description

Saves the given data point with the previously saved data points.

Parameters:

queries (Tuple[Tuple[NDArray[Shape["query_nr, ... query_dim"], Number], Tuple[NDArray[Shape["result_nr, ... result_dim"], Number]]) – New data points going to the data pools process

class LogResultEvaluator(*args: Any, **kwargs: Any)

Bases: LogingEvaluator

LogProcessEvaluator() | Description | This evaluator logs all results added to the experiment’s data pools results.

file_name: str = 'result'

folder: str = 'log'

log_data(self) → None

Description

Logs all saved results to an `.npy` file if there is at least one result.

register(self, experiment) → None

Description

Modifies the experiment to log all results added to the data pools results.

Requires the experiment’s data pools to be a ResultDataPools.

Parameters:

experiment (Experiments) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a ResultDataPools

save_result(self, data) → None

Description

Saves the given result with the previously saved results.

Parameters:

queries (Tuple[Tuple[NDArray[Shape["query_nr, ... query_dim"], Number], Tuple[NDArray[Shape["result_nr, ... result_dim"], Number]]) – New results going to the data pools results

class LogStreamEvaluator

Bases: LogingEvaluator

Description

This evaluator logs all data points added to the experiment’s data pools stream.

file_name: str = 'stream'

folder: str = 'log'

log_data(self) → None

Description

Logs all saved data points to an `.npy` file if there is at least one data point.

register(self, experiment) → None

Description

Modifies the experiment to log all data points added to the data pools stream.

Requires the experiment’s data pools to be a StreamDataPools.

Parameters:

experiment (Experiments) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a StreamDataPools

save_stream(self, data) → None

Description

Saves the given data point with the previously saved data points.

Parameters:

queries (Tuple[Tuple[NDArray[Shape["query_nr, ... query_dim"], Number], Tuple[NDArray[Shape["result_nr, ... result_dim"], Number]]) – New data points going to the data pools stream

class LogTVPGTEvaluator

Bases: LogingEvaluator

Description

The Log Time Varying Process Ground Truth Evaluator —

file_name: str = 'gt_data'

folder: str = 'log'

log_data(self) → None

Description

Logs all saved data points (ground truths) to an `.npy` file if there is at least one data point.

register(self, experiment) → None

Description

Modifies the experiment to log all results coming form the process’ update.

Requires the experiment’s process to be a DelayedProcess.

Parameters:

experiment (Experiments) – The experiment to be evaluated

Raises:

TypeError if self.experiment.process is not a DelayedProcess

save_gt(self, data) → None

Description

Saves the given data (ground truths) points with the previously saved data points.

Parameters:

queries (Tuple[Tuple[NDArray[Shape["query_nr, ... query_dim"], Number], Tuple[NDArray[Shape["result_nr, ... result_dim"], Number]]) – New results going to the data pools results

class PlotAllDataPointsEvaluator(*args: Any, **kwargs: Any)

Bases: LogingEvaluator

PlotALlDataPointsEvaluator() | Description | This evaluator plots all of the experiment’s data points after the experiment has concluded.

data_pools: ResultDataPools

fig_name: str = 'AllData'

folder: str = 'fig'

interactive: bool = False

log_data(self) → None

Description

Plots all of the experiment’s data points.

register(self, experiment) → None

Description

Modifies the experiment to plot all data points after running the experiment.

Requires the experiment’s data pool to be a ResultDataPool.

Parameters:

experiment (Experiment) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a ResultDataPools

class PlotNewDataPointsEvaluator

Bases: LogingEvaluator

Description

This evaluator plots all of the experiment’s new data points continuously as they arrive.

fig_name: str = 'Data'

folder: str = 'fig'

interactive: bool = False

plot_new_data_points(self, data_points) → None

Description

Adds the given data points to the saved ones and updates the plot.

Parameters:

data_points (Tuple[NDArray[Shape["query_nr, ... query_dim"], Number], NDArray[Shape["query_nr, ... result_dim"], Number]]) – New data points to be plotted

queries: NDArray[Shape['query_nr, ... query_dim'], Number] = None

register(self, experiment) → None

Description

Modifies the experiment to plot new data points before adding them to the experiment’s result data pool.

Requires the experiment’s data pool to be a ResultDataPool.

Parameters:

experiment (Experiment) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a ResultDataPools

results: NDArray[Shape['query_nr, ... result_dim'], Number] = None

class PlotQueryDistEvaluator

Bases: LogingEvaluator

Description

This evaluator plots all of the experiment’s new queries continuously as they are made as a histogram.

fig_name: str = 'Query distribution'

folder: str = 'fig'

interactive: bool = False

plot_query_dist(self, queries) → None

Description

Plots the given queries on a new frame of the histogram.

Parameters:

queries (Tuple[NDArray[Shape["query_nr, ... query_dim"], Number]) – New queries going to the query queue

queries: NDArray[Shape['query_nr, ... query_dim'], Number] = None

register(self, experiment) → None

Description

Modifies the experiment to plot the new queries as new frames of the histogram.

Requires the experiment’s oracles to be a POracles.

Parameters:

experiment (Experiment) – The experiment to be evaluated

Raises:

TypeError if self.experiment.oracles is not a POracles

class PlotSampledQueriesEvaluator

Bases: LogingEvaluator

Description

This evaluator plots the experiment’s selected queries.

fig_name: str = 'Sampled queries'

folder: str = 'fig'

interactive: bool = True

plot_queries(self, queries) → None

Description

Plots the given queries.

Parameters:

queries (Tuple[NDArray[Shape["query_nr, ... query_dim"], Number]) – New queries going to the query queue

register(self, experiment) → None

Description

Modifies the experiment to plot the queries passing its selection criteria.

Parameters:

experiment (Experiments) – The experiment to be evaluated

class PrintExpTimeEvaluator

Bases: Evaluator

Description

This evaluator measures how long the experiment takes to run.

end_time(self) → None

Description

Substracts the start time from the current time and prints the resulting time.

register(self, experiment) → None

Description

Modifies the experiment to print how long it took to run.

Parameters:

experiment (Experiment) – The experiment to be evaluated

start_time(self) → None

Description

Prints an anouncement of time measurement. Saves the current time as the start time.

class PrintNewDataPointsEvaluator

Bases: Evaluator

Description

This evaluator keeps track of the experiment’s results.

register(self, experiment) → None

Description

Modifies the experiment to print new data points before adding them to the experiment’s result data pool.

Requires the experiment’s data pool to be a ResultDataPool.

Parameters:

experiment (Experiment) – The experiment to be evaluated

Raises:

TypeError if self.experiment.data_pools is not a ResultDataPools

class PrintQueryEvaluator

Bases: Evaluator

Description

This evaluator keeps track of the experiment’s queries.

print_query(self, queries) → None

Description

Prints the given queries.

Parameters:

queries (Tuple[NDArray[Shape["query_nr, ... query_dim"], Number]) – New queries going to the query queue

register(self, experiment) → None

Description

Modifies the experiment to print new queries before adding them to the experiment’s query queue.

Requires the experiment’s oracle to be a POracles.

Parameters:

experiment (Experiment) – The experiment to be evaluated

Raises:

TypeError if self.experiment.oracles is not a POracles

class PrintTimeSourceEvaluator

Bases: Evaluator

Description

This Evaluator keeps track of the experiment’s internal time.

end_time(self, time) → None

Description

Prints the given time.

Parameters:

time (int) – The experiment’s current internal time

register(self, experiment) → None

Description

Modifies the experiment’s time source to print the current internal time at each time step.

Parameters:

experiment (Experiment) – The experiment to be evaluated