src.fairreckitlib.data.pipeline.data_run

 1"""This module contains functionality to run the data pipeline.
 2
 3Classes:
 4
 5    DataPipelineConfig: configuration class to run the data pipelines.
 6
 7Functions:
 8
 9    run_data_pipelines: run the pipeline using dataset configurations.
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
16from dataclasses import dataclass
17from typing import Callable, List
18
19from ...core.config.config_factories import GroupFactory
20from ...core.events.event_dispatcher import EventDispatcher
21from ...core.events.event_error import ON_FAILURE_ERROR, ErrorEventArgs
22from ..set.dataset_registry import DataRegistry
23from .data_config import DataMatrixConfig
24from .data_pipeline import DataPipeline, DataTransition
25
26
27@dataclass
28class DataPipelineConfig:
29    """Data Pipeline Configuration.
30
31    output_dir: the directory to store the output.
32    data_registry: the registry with available datasets.
33    data_factory: the factory with available data modifier factories.
34    data_config_list: the dataset matrix configurations to compute.
35    """
36
37    output_dir: str
38    data_registry: DataRegistry
39    data_factory: GroupFactory
40    data_config_list: List[DataMatrixConfig]
41
42
43def run_data_pipelines(
44        pipeline_config: DataPipelineConfig,
45        event_dispatcher: EventDispatcher,
46        is_running: Callable[[], bool]) -> List[DataTransition]:
47    """Run a Data Pipeline several times according to the specified data pipeline configuration.
48
49    Args:
50        pipeline_config: the configuration on how to run the data pipelines.
51        event_dispatcher: used to dispatch data/IO events when running the pipeline.
52        is_running: function that returns whether the pipelines
53            are still running. Stops early when False is returned.
54
55    Returns:
56        a list of DataTransition's.
57    """
58    data_result = []
59
60    data_pipeline = DataPipeline(pipeline_config.data_factory, event_dispatcher)
61    for data_config in pipeline_config.data_config_list:
62        dataset = pipeline_config.data_registry.get_set(data_config.dataset)
63        if dataset is None:
64            event_dispatcher.dispatch(ErrorEventArgs(
65                ON_FAILURE_ERROR,
66                'Failure: to get dataset ' + data_config.dataset + ' from registry'
67            ))
68            continue
69
70        if data_config.matrix not in dataset.get_available_matrices():
71            event_dispatcher.dispatch(ErrorEventArgs(
72                ON_FAILURE_ERROR,
73                'Failure: to get matrix ' + data_config.matrix + ' from ' + data_config.dataset
74            ))
75            continue
76
77        try:
78            data_transition = data_pipeline.run(
79                pipeline_config.output_dir,
80                dataset,
81                data_config,
82                is_running
83            )
84        except (FileNotFoundError, RuntimeError):
85            continue
86
87        data_result.append(data_transition)
88        if not is_running():
89            return data_result
90
91    return data_result