modewise_pipeline/
spec.rs

1use std::collections::BTreeMap;
2
3use serde::Deserialize;
4use serde_json::{Map, Value};
5
6use crate::{ArtifactKind, PipelineError};
7
8/// A pipeline as the user writes it, usually in TOML.
9#[derive(Debug, Clone, Deserialize)]
10#[serde(deny_unknown_fields)]
11pub struct PipelineSpec {
12    pub pipeline: PipelineHeader,
13    #[serde(default, rename = "step")]
14    pub steps: Vec<StepSpec>,
15}
16
17#[derive(Debug, Clone, Deserialize)]
18#[serde(deny_unknown_fields)]
19pub struct PipelineHeader {
20    pub name: String,
21    /// Data supplied when the pipeline runs: a list of names (all series), or a table of
22    /// name = kind, for example `{ bold = "series", events = "events" }`.
23    #[serde(default)]
24    pub inputs: InputDecl,
25}
26
27#[derive(Debug, Clone, Deserialize)]
28#[serde(untagged)]
29pub enum InputDecl {
30    Series(Vec<String>),
31    Typed(BTreeMap<String, ArtifactKind>),
32}
33
34impl Default for InputDecl {
35    fn default() -> Self {
36        Self::Series(Vec::new())
37    }
38}
39
40impl InputDecl {
41    /// Every input with its kind.
42    pub fn kinds(&self) -> Vec<(String, ArtifactKind)> {
43        match self {
44            Self::Series(names) => names
45                .iter()
46                .map(|n| (n.clone(), ArtifactKind::Series))
47                .collect(),
48            Self::Typed(map) => map.iter().map(|(n, k)| (n.clone(), *k)).collect(),
49        }
50    }
51}
52
53#[derive(Debug, Clone, Deserialize)]
54#[serde(deny_unknown_fields)]
55pub struct StepSpec {
56    pub id: String,
57    pub op: String,
58    #[serde(default)]
59    pub input: InputSpec,
60    #[serde(default)]
61    pub params: Map<String, Value>,
62    /// Store this step's outputs. With `false` they live only in memory during a run: steps that
63    /// use them still run and are cached, but large intermediates (for example modes before band
64    /// powers) take no disk space. They are recomputed only when a later step needs them.
65    #[serde(default = "keep_by_default")]
66    pub keep: bool,
67}
68
69fn keep_by_default() -> bool {
70    true
71}
72
73/// `input = "x"` for single-input operations, or `input = { port = "x" }`.
74/// A reference is a pipeline input name, a step id, or `step.output`.
75#[derive(Debug, Clone, Deserialize)]
76#[serde(untagged)]
77pub enum InputSpec {
78    Single(String),
79    Named(BTreeMap<String, String>),
80}
81
82impl Default for InputSpec {
83    fn default() -> Self {
84        Self::Named(BTreeMap::new())
85    }
86}
87
88impl PipelineSpec {
89    pub fn from_toml(text: &str) -> Result<Self, PipelineError> {
90        toml::from_str(text).map_err(|e| PipelineError::Spec(e.to_string()))
91    }
92
93    pub fn from_json(text: &str) -> Result<Self, PipelineError> {
94        serde_json::from_str(text).map_err(|e| PipelineError::Spec(e.to_string()))
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use super::*;
101
102    #[test]
103    fn parses_single_and_named_inputs() {
104        let spec = PipelineSpec::from_toml(
105            r#"
106            [pipeline]
107            name = "demo"
108            inputs = ["bold"]
109
110            [[step]]
111            id = "modes"
112            op = "mvmd"
113            input = "bold"
114            params = { num_modes = 4, alpha = 1000.0 }
115
116            [[step]]
117            id = "mix"
118            op = "combine"
119            input = { left = "modes", right = "bold" }
120            "#,
121        )
122        .unwrap();
123        assert_eq!(
124            spec.pipeline.inputs.kinds(),
125            [("bold".to_string(), ArtifactKind::Series)]
126        );
127        assert!(matches!(&spec.steps[0].input, InputSpec::Single(r) if r == "bold"));
128        assert_eq!(spec.steps[0].params["num_modes"], 4);
129        assert!(matches!(&spec.steps[1].input, InputSpec::Named(m) if m.len() == 2));
130    }
131
132    #[test]
133    fn parses_json() {
134        let spec = PipelineSpec::from_json(
135            r#"{"pipeline": {"name": "j", "inputs": ["x"]},
136                "step": [{"id": "a", "op": "scale", "input": "x", "params": {"factor": 2}}]}"#,
137        )
138        .unwrap();
139        assert_eq!(spec.steps[0].params["factor"], 2);
140    }
141
142    #[test]
143    fn parses_typed_inputs() {
144        let spec = PipelineSpec::from_toml(
145            "[pipeline]\nname = \"t\"\ninputs = { bold = \"series\", trials = \"events\" }\n",
146        )
147        .unwrap();
148        assert_eq!(
149            spec.pipeline.inputs.kinds(),
150            [
151                ("bold".to_string(), ArtifactKind::Series),
152                ("trials".to_string(), ArtifactKind::Events)
153            ]
154        );
155        assert!(
156            PipelineSpec::from_toml("[pipeline]\nname = \"t\"\ninputs = { bold = \"sound\" }\n")
157                .is_err()
158        );
159    }
160
161    #[test]
162    fn rejects_invalid_json() {
163        assert!(matches!(
164            PipelineSpec::from_json("{not json"),
165            Err(PipelineError::Spec(_))
166        ));
167    }
168
169    #[test]
170    fn rejects_unknown_fields() {
171        let err = PipelineSpec::from_toml("[pipeline]\nname = \"x\"\ntypo = 1\n").unwrap_err();
172        assert!(matches!(err, PipelineError::Spec(_)));
173    }
174}
modewise_pipeline/spec.rs

modewise_pipeline/
spec.rs
