src.fairreckitlib.experiment.experiment_event

  1"""This module contains all event ids, event args and a print switch for the experiment pipeline.
  2
  3Constants:
  4
  5    ON_BEGIN_EXPERIMENT_PIPELINE: id of the event that is used when the experiment pipeline starts.
  6    ON_END_EXPERIMENT_PIPELINE: id of the event that is used when the experiment pipeline ends.
  7    ON_BEGIN_THREAD_EXPERIMENT: id of the event that is used when the experiment thread starts.
  8    ON_END_THREAD_EXPERIMENT: id of the event that is used when the experiment thread ends.
  9
 10Classes:
 11
 12    ExperimentEventArgs: event args related to an experiment.
 13    ExperimentThreadEventArgs: event args related to an experiment thread.
 14
 15Functions:
 16
 17    get_experiment_events: list of experiment pipeline event IDs.
 18    get_experiment_event_print_switch: switch to print experiment pipeline event arguments by ID.
 19
 20This program has been developed by students from the bachelor Computer Science at
 21Utrecht University within the Software Project course.
 22© Copyright Utrecht University (Department of Information and Computing Sciences)
 23"""
 24
 25from dataclasses import dataclass
 26from typing import Dict, Callable, List
 27
 28from ..core.events.event_dispatcher import EventArgs
 29from ..core.events.event_error import get_error_events, get_error_event_print_switch
 30from ..core.io.event_io import get_io_events, get_io_event_print_switch
 31from ..data.pipeline.data_event import get_data_events, get_data_event_print_switch
 32from ..evaluation.pipeline.evaluation_event import get_eval_events, get_eval_event_print_switch
 33from ..model.pipeline.model_event import get_model_events, get_model_event_print_switch
 34
 35ON_BEGIN_EXPERIMENT_PIPELINE = 'Experiment.on_begin_pipeline'
 36ON_END_EXPERIMENT_PIPELINE = 'Experiment.on_end_pipeline'
 37ON_BEGIN_EXPERIMENT_THREAD = 'Experiment.on_begin_thread'
 38ON_END_EXPERIMENT_THREAD = 'Experiment.on_end_thread'
 39
 40
 41@dataclass
 42class ExperimentEventArgs(EventArgs):
 43    """Experiment Event Arguments.
 44
 45    event_id: the unique ID that classifies the experiment event.
 46    experiment_name: the name of the experiment.
 47    """
 48
 49    experiment_name: str
 50
 51
 52@dataclass
 53class ExperimentThreadEventArgs(ExperimentEventArgs):
 54    """Experiment Thread Event Arguments.
 55
 56    event_id: the unique ID that classifies the experiment event.
 57    experiment_name: the name of the experiment.
 58    num_runs: the amount of times the experiment will run.
 59    is_running: whether the experiment thread is running or aborted.
 60    """
 61
 62    num_runs: int
 63    is_running: bool=True
 64
 65
 66def get_experiment_events() -> List[str]:
 67    """Get a list of experiment pipeline event IDs.
 68
 69    Returns:
 70        a list of unique experiment pipeline event IDs.
 71    """
 72    events = [
 73        # ExperimentEventArgs
 74        ON_BEGIN_EXPERIMENT_PIPELINE,
 75        ON_END_EXPERIMENT_PIPELINE,
 76        # ExperimentThreadEventArgs
 77        ON_BEGIN_EXPERIMENT_THREAD,
 78        ON_END_EXPERIMENT_THREAD
 79    ]
 80    events += get_error_events()
 81    events += get_io_events()
 82    events += get_data_events()
 83    events += get_model_events()
 84    events += get_eval_events()
 85    return events
 86
 87
 88def get_experiment_print_switch(elapsed_time: float=None) -> Dict[str, Callable[[EventArgs], None]]:
 89    """Get a switch that prints experiment pipeline event IDs.
 90
 91    Returns:
 92        the print experiment pipeline event switch.
 93    """
 94    event_switch = {
 95        # experiment thread events
 96        ON_BEGIN_EXPERIMENT_THREAD: lambda args:
 97            print('Starting', args.num_runs, 'experiment(s) with name', args.experiment_name),
 98        ON_END_EXPERIMENT_THREAD: lambda args:
 99            print('Finished', args.num_runs, 'experiment(s) with name', args.experiment_name,
100                  f'in {elapsed_time:1.4f}s'),
101        # experiment pipeline events
102        ON_BEGIN_EXPERIMENT_PIPELINE: lambda args:
103            print('Starting Experiment:', args.experiment_name),
104        ON_END_EXPERIMENT_PIPELINE: lambda args:
105            print('Finished Experiment:', args.experiment_name,
106                  f'in {elapsed_time:1.4f}s')
107    }
108
109    # merge error/IO/pipeline event switches
110    event_switch.update(get_error_event_print_switch())
111    event_switch.update(get_io_event_print_switch())
112    event_switch.update(get_data_event_print_switch(elapsed_time))
113    event_switch.update(get_model_event_print_switch(elapsed_time))
114    event_switch.update(get_eval_event_print_switch(elapsed_time))
115
116    return event_switch