rove 0.3.0

System for real time spatial and timeseries quality control of weather data
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

//! Utilities for scheduling QC checks

use crate::{
    data_switch::{self, DataCache, DataSwitch, SpaceSpec, TimeSpec},
    harness::{self, CheckResult},
    pipeline::Pipeline,
};
use std::collections::HashMap;
use thiserror::Error;

/// Error type for Scheduler methods
#[derive(Error, Debug)]
#[non_exhaustive]
pub enum Error {
    /// The check harness returned an error
    #[error("failed to run check: {0}")]
    Runner(#[from] harness::Error),
    /// The method received an invalid argument
    #[error("invalid argument: {0}")]
    InvalidArg(&'static str),
    /// The [`DataSwitch`] returned an error
    #[error("data switch failed to find data: {0}")]
    DataSwitch(#[from] data_switch::Error),
}

/// Receiver type for QC runs
///
/// Holds information about test pipelines and data sources
#[derive(Debug)]
pub struct Scheduler {
    // this is pub so that the server can determine the number of checks in a pipeline to size
    // its channel with. can be made private if the server functionality is deprecated
    #[allow(missing_docs)]
    pub pipelines: HashMap<String, Pipeline>,
    data_switch: DataSwitch,
}

impl Scheduler {
    /// Instantiate a new scheduler
    pub fn new(pipelines: HashMap<String, Pipeline>, data_switch: DataSwitch) -> Self {
        Scheduler {
            pipelines,
            data_switch,
        }
    }

    /// Directly invoke a Pipeline on a Datacache. If you want the scheduler to fetch the Pipeline
    /// and DataCache for you, see [`validate_direct`](Scheduler::validate_direct).
    pub fn schedule_tests(pipeline: &Pipeline, data: DataCache) -> Result<Vec<CheckResult>, Error> {
        pipeline
            .steps
            .iter()
            .map(|step| harness::run_check(step, &data).map_err(Error::Runner))
            .collect()
    }

    /// Run a set of QC tests on some data
    ///
    /// `data_source` is the key identifying a connector in the
    /// [`DataSwitch`].
    /// `backing_sources` a list of keys similar to `data_source`, but data
    /// from these will only be used to QC data from `data_source` and will not
    /// themselves be QCed.
    /// `time_spec` and `space_spec` narrow down what data to QC, more info
    /// on what these mean and how to construct them can be found on their
    /// own doc pages.
    /// `test_pipeline` represents the pipeline of checks to be run. Available
    /// options of pipelines are defined at load time for the service, where
    /// pipelines are read from toml files.
    /// `extra_spec` is an extra identifier that gets passed to the relevant
    /// DataConnector. The format of `extra_spec` is connector-specific.
    ///
    /// # Errors
    ///
    /// Returned from the function if:
    /// - The pipeline named by in the `test_pipeline` argument is not recognized
    ///   by the system
    /// - The data_source string did not have a matching entry in the
    ///   Scheduler's DataSwitch
    ///
    /// In the the returned channel if:
    /// - The test harness encounters an error on during one of the QC tests.
    ///   This will also result in the channel being closed
    pub async fn validate_direct(
        &self,
        data_source: impl AsRef<str>,
        // TODO: we should actually use these
        _backing_sources: &[impl AsRef<str>],
        time_spec: &TimeSpec,
        space_spec: &SpaceSpec,
        // TODO: should we allow specifying multiple pipelines per call?
        test_pipeline: impl AsRef<str>,
        extra_spec: Option<&str>,
    ) -> Result<Vec<CheckResult>, Error> {
        let pipeline = self
            .pipelines
            .get(test_pipeline.as_ref())
            .ok_or(Error::InvalidArg("pipeline not recognised"))?;

        let data = match self
            .data_switch
            .fetch_data(
                data_source.as_ref(),
                space_spec,
                time_spec,
                pipeline.num_leading_required,
                pipeline.num_trailing_required,
                extra_spec,
            )
            .await
        {
            Ok(data) => data,
            Err(e) => {
                tracing::error!(%e);
                return Err(Error::DataSwitch(e));
            }
        };

        Scheduler::schedule_tests(pipeline, data)
    }
}