src.fairreckitlib.experiment.experiment_thread

  1"""This module contains functionality to execute the experiment pipelines on a thread.
  2
  3Classes:
  4
  5    ThreadExperiment: class that runs the experiment pipelines on a (closable) thread.
  6
  7Functions:
  8
  9    handle_experiment_event: handles incoming experiment events.
 10
 11This program has been developed by students from the bachelor Computer Science at
 12Utrecht University within the Software Project course.
 13© Copyright Utrecht University (Department of Information and Computing Sciences)
 14"""
 15
 16import time
 17from typing import Any, Callable, Dict
 18
 19from ..core.events.event_dispatcher import EventArgs
 20from ..core.threading.thread_base import ThreadBase
 21from .experiment_event import get_experiment_events, get_experiment_print_switch, \
 22    ExperimentThreadEventArgs, ON_BEGIN_EXPERIMENT_THREAD, ON_END_EXPERIMENT_THREAD
 23from .experiment_run import run_experiment_pipelines
 24
 25
 26class ThreadExperiment(ThreadBase):
 27    """Thread that runs the same experiment one or more times."""
 28
 29    def __init__(
 30            self,
 31            name: str,
 32            events: Dict[Any, Callable[[Any], None]]=None,
 33            verbose: bool=False,
 34            **kwargs):
 35        """Construct the ExperimentThread.
 36
 37        Args:
 38            name the name of the thread.
 39            events: events to dispatch for this thread.
 40            verbose: whether the thread should give verbose output.
 41
 42        Keyword Args:
 43            pipeline_config(ExperimentPipelineConfig): configuration of the experiment pipeline.
 44        """
 45        ThreadBase.__init__(self, name, verbose, **kwargs)
 46        # no external events specified
 47        if events is None:
 48            events = {}
 49
 50        # Add external events.
 51        for event_id in get_experiment_events():
 52            func_on_event = (handle_experiment_event, events.get(event_id))
 53            self.event_dispatcher.add_listener(event_id, self, func_on_event)
 54
 55    def on_run(self, **kwargs):
 56        """Run the experiment pipeline.
 57
 58        Keyword Args:
 59            pipeline_config(ExperimentPipelineConfig): configuration of the experiment pipeline.
 60        """
 61        pipeline_config = kwargs['pipeline_config']
 62
 63        self.event_dispatcher.dispatch(ExperimentThreadEventArgs(
 64            ON_BEGIN_EXPERIMENT_THREAD,
 65            pipeline_config.experiment_config.name,
 66            pipeline_config.num_runs,
 67            self.is_running()
 68        ))
 69
 70        start = time.time()
 71
 72        success = run_experiment_pipelines(pipeline_config, self.event_dispatcher, self.is_running)
 73
 74        end = time.time()
 75
 76        self.event_dispatcher.dispatch(ExperimentThreadEventArgs(
 77            ON_END_EXPERIMENT_THREAD,
 78            pipeline_config.experiment_config.name,
 79            pipeline_config.num_runs,
 80            self.is_running()
 81        ), elapsed_time=end - start, success=success)
 82
 83
 84def handle_experiment_event(
 85        experiment_thread: ThreadExperiment,
 86        event_args: EventArgs,
 87        **kwargs) -> None:
 88    """Handle incoming experiment events.
 89
 90    It is assumed that the event finished when the elapsed_time keyword argument is available.
 91
 92    Args:
 93        experiment_thread: the listening experiment thread.
 94        event_args: the event arguments to handle.
 95
 96    Keyword Args:
 97        elapsed_time(float): time that has passed since the event started, expressed in seconds.
 98    """
 99    if experiment_thread.verbose:
100        event_switch = get_experiment_print_switch(kwargs.get('elapsed_time'))
101        event_switch[event_args.event_id](event_args)