modewise_results/
lib.rs

1//! A results folder: cached artifacts plus a record of every run, reopenable later.
2//!
3//! ```text
4//! <root>/store/                                   Zarr artifact store
5//! <root>/runs/<pipeline>/<recording>.json         run manifest per recording
6//! <root>/runs/<pipeline>/summary[-shard-i-of-n].json
7//! ```
8
9mod export;
10mod tables;
11
12pub use tables::write_feature_table;
13
14use std::collections::BTreeMap;
15use std::path::{Path, PathBuf};
16use std::sync::Arc;
17
18use modewise_pipeline::{
19    Artifact, ArtifactStore, InputSource, PipelineError, Plan, RecordingRun, RunManifest,
20    StoreError, run_batch,
21};
22use modewise_store_zarr::ZarrStore;
23use serde::{Deserialize, Serialize};
24use thiserror::Error;
25
26#[derive(Debug, Error)]
27pub enum ResultsError {
28    #[error("{path}: {message}")]
29    Io { path: String, message: String },
30    #[error(transparent)]
31    Store(#[from] StoreError),
32    #[error(transparent)]
33    Pipeline(#[from] PipelineError),
34    #[error(
35        "pipeline inputs {pipeline:?} are not all among the inputs the data provides {source_inputs:?}"
36    )]
37    InputMismatch {
38        pipeline: Vec<String>,
39        source_inputs: Vec<String>,
40    },
41    #[error("no run for recording {recording:?} of pipeline {pipeline:?}")]
42    MissingRun { pipeline: String, recording: String },
43    #[error("step {step:?} has keep = false, so its output {reference:?} was not stored")]
44    NotStored { step: String, reference: String },
45    #[error("no output {reference:?}; outputs are {available:?}")]
46    MissingOutput {
47        reference: String,
48        available: Vec<String>,
49    },
50}
51
52fn io(path: &Path, e: impl std::fmt::Display) -> ResultsError {
53    ResultsError::Io {
54        path: path.display().to_string(),
55        message: e.to_string(),
56    }
57}
58
59/// Which recordings of a batch succeeded, and why the others failed.
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct BatchSummary {
62    pub pipeline: String,
63    pub shard: Option<(usize, usize)>,
64    pub succeeded: Vec<String>,
65    pub failed: BTreeMap<String, String>,
66}
67
68pub struct ResultsDir {
69    root: PathBuf,
70    store: ZarrStore,
71}
72
73impl ResultsDir {
74    /// Opens the folder, creating it if needed.
75    pub fn open(root: &Path) -> Result<Self, ResultsError> {
76        let store = ZarrStore::open(&root.join("store"))?;
77        Ok(Self {
78            root: root.to_path_buf(),
79            store,
80        })
81    }
82
83    pub fn root(&self) -> &Path {
84        &self.root
85    }
86
87    pub fn store(&self) -> &ZarrStore {
88        &self.store
89    }
90
91    fn runs_dir(&self, pipeline: &str) -> PathBuf {
92        self.root.join("runs").join(file_name(pipeline))
93    }
94
95    pub fn save_run(
96        &self,
97        pipeline: &str,
98        recording: &str,
99        manifest: &RunManifest,
100    ) -> Result<(), ResultsError> {
101        let dir = self.runs_dir(pipeline);
102        std::fs::create_dir_all(&dir).map_err(|e| io(&dir, e))?;
103        let path = dir.join(format!("{}.json", file_name(recording)));
104        std::fs::write(&path, manifest.to_json()).map_err(|e| io(&path, e))
105    }
106
107    pub fn save_summary(&self, summary: &BatchSummary) -> Result<PathBuf, ResultsError> {
108        let dir = self.runs_dir(&summary.pipeline).join("summaries");
109        std::fs::create_dir_all(&dir).map_err(|e| io(&dir, e))?;
110        let name = match summary.shard {
111            Some((i, n)) => format!("shard-{i}-of-{n}.json"),
112            None => "all.json".to_string(),
113        };
114        let path = dir.join(name);
115        let text = serde_json::to_string_pretty(summary).map_err(|e| io(&path, e))?;
116        std::fs::write(&path, text).map_err(|e| io(&path, e))?;
117        Ok(path)
118    }
119
120    /// Runs `plan` over the recordings in `ids` (all when `None`), saving each run as it
121    /// finishes and a summary at the end. `on_done` sees each recording for progress.
122    pub fn run_batch(
123        &self,
124        plan: &Plan,
125        source: &dyn InputSource,
126        shard: Option<(usize, usize)>,
127        workers: usize,
128        on_done: &(dyn Fn(&RecordingRun) + Sync),
129    ) -> Result<BatchSummary, ResultsError> {
130        let mut wanted = plan.inputs().to_vec();
131        let mut given = source.input_names();
132        wanted.sort();
133        given.sort();
134        // Inputs the pipeline does not use (such as a channel table) are left out.
135        if !wanted.iter().all(|w| given.contains(w)) {
136            return Err(ResultsError::InputMismatch {
137                pipeline: wanted,
138                source_inputs: given,
139            });
140        }
141        let ids: Vec<String> = match shard {
142            Some((i, n)) => source
143                .recording_ids()
144                .into_iter()
145                .enumerate()
146                .filter(|(j, _)| j % n == i - 1)
147                .map(|(_, id)| id)
148                .collect(),
149            None => source.recording_ids(),
150        };
151        let save_failures = std::sync::Mutex::new(BTreeMap::new());
152        let save = |run: &RecordingRun| {
153            if let Ok(result) = &run.result
154                && let Err(e) = self.save_run(plan.name(), &run.id, &result.manifest)
155                && let Ok(mut failures) = save_failures.lock()
156            {
157                failures.insert(run.id.clone(), format!("saving run record failed: {e}"));
158            }
159            on_done(run);
160        };
161        let runs = run_batch(plan, source, &ids, &self.store, workers, &save)?;
162        let save_failures = save_failures.into_inner().unwrap_or_default();
163
164        let mut summary = BatchSummary {
165            pipeline: plan.name().to_string(),
166            shard,
167            succeeded: Vec::new(),
168            failed: BTreeMap::new(),
169        };
170        for run in runs {
171            let error = run
172                .result
173                .err()
174                .or_else(|| save_failures.get(&run.id).cloned());
175            match error {
176                None => summary.succeeded.push(run.id),
177                Some(e) => {
178                    summary.failed.insert(run.id, e);
179                }
180            }
181        }
182        self.save_summary(&summary)?;
183        Ok(summary)
184    }
185
186    pub fn pipelines(&self) -> Vec<String> {
187        list(&self.root.join("runs"), |p| p.is_dir())
188    }
189
190    /// Recordings with a saved run, sorted.
191    pub fn recordings(&self, pipeline: &str) -> Vec<String> {
192        list(&self.runs_dir(pipeline), |p| {
193            p.extension().is_some_and(|e| e == "json")
194        })
195        .into_iter()
196        .filter_map(|n| n.strip_suffix(".json").map(String::from))
197        .collect()
198    }
199
200    pub fn manifest(&self, pipeline: &str, recording: &str) -> Result<RunManifest, ResultsError> {
201        let path = self
202            .runs_dir(pipeline)
203            .join(format!("{}.json", file_name(recording)));
204        let text = std::fs::read_to_string(&path).map_err(|_| ResultsError::MissingRun {
205            pipeline: pipeline.to_string(),
206            recording: recording.to_string(),
207        })?;
208        serde_json::from_str(&text).map_err(|e| io(&path, e))
209    }
210
211    /// A step output, by `step.output`, or by `step` when it has one output.
212    pub fn output(
213        &self,
214        pipeline: &str,
215        recording: &str,
216        reference: &str,
217    ) -> Result<Arc<Artifact>, ResultsError> {
218        let manifest = self.manifest(pipeline, recording)?;
219        let outputs = manifest.outputs();
220        let prefix = format!("{reference}.");
221        let mut candidates = outputs
222            .iter()
223            .filter(|(name, _)| *name == reference || name.starts_with(&prefix));
224        let key = match (candidates.next(), candidates.next()) {
225            (Some((_, key)), None) => key,
226            _ => {
227                return Err(ResultsError::MissingOutput {
228                    reference: reference.to_string(),
229                    available: outputs.keys().cloned().collect(),
230                });
231            }
232        };
233        let step = reference.split('.').next().unwrap_or(reference);
234        if manifest.steps.iter().any(|s| s.id == step && !s.stored) {
235            return Err(ResultsError::NotStored {
236                step: step.to_string(),
237                reference: reference.to_string(),
238            });
239        }
240        self.store
241            .get(key)?
242            .ok_or_else(|| ResultsError::MissingOutput {
243                reference: reference.to_string(),
244                available: outputs.keys().cloned().collect(),
245            })
246    }
247
248    /// Writes one HDF5 file per recording to `dir`, for MATLAB and other tools.
249    pub fn export_hdf5(&self, pipeline: &str, dir: &Path) -> Result<Vec<PathBuf>, ResultsError> {
250        std::fs::create_dir_all(dir).map_err(|e| io(dir, e))?;
251        self.recordings(pipeline)
252            .iter()
253            .map(|recording| {
254                let manifest = self.manifest(pipeline, recording)?;
255                let path = dir.join(format!("{}.h5", file_name(recording)));
256                export::write_recording(&self.store, recording, &manifest, &path)?;
257                Ok(path)
258            })
259            .collect()
260    }
261}
262
263/// Keeps names safe as file names on every platform.
264fn file_name(name: &str) -> String {
265    name.chars()
266        .map(|c| {
267            if c.is_ascii_alphanumeric() || "-_.".contains(c) {
268                c
269            } else {
270                '_'
271            }
272        })
273        .collect()
274}
275
276fn list(dir: &Path, keep: impl Fn(&Path) -> bool) -> Vec<String> {
277    let mut names: Vec<String> = std::fs::read_dir(dir)
278        .into_iter()
279        .flatten()
280        .flatten()
281        .filter(|e| keep(&e.path()))
282        .map(|e| e.file_name().to_string_lossy().into_owned())
283        .collect();
284    names.sort();
285    names
286}
modewise_results/lib.rs

modewise_results/
lib.rs
