dspk_app_core/domain/
session.rs

1//! The session document: the whole working session as one versioned JSON document, so the app and an agent (a future MCP server, as Paper's MCP edits the open
2//! file) read and change the same thing.
3//!
4//! A session holds a processing graph ([`GraphDoc`]: recordings, ops, views, writes, as the
5//! Pipeline canvas shows them) and the views compiled from it. Each view is one recording (file,
6//! source within it, optional channel subset read as one grouped array), the op chain on its path
7//! (dspk-pipeline nodes, applied to every channel of the view), how it is displayed, and what
8//! plugins keep for it (e.g. electrophysiology: the linked sorting). The graph is the truth: the
9//! views are rewritten from it after every command (they stay in the file for agents to read).
10//!
11//! The document only changes through [`crate::domain::command::SessionCommand`]s; the app
12//! re-renders from it and saves it after each change.
13
14use std::collections::BTreeMap;
15use std::path::{Path, PathBuf};
16
17use dspk_pipeline::{Node, OutputFormat, PipelineSpec, SPEC_VERSION};
18
19use super::epoch::{EpochDoc, EventSetDoc, Placement};
20use super::graph::{GraphDoc, Recipes};
21use schemars::JsonSchema;
22use serde::{Deserialize, Serialize};
23
24/// The document version this app reads and writes (`dspk_session` field).
25pub const SESSION_VERSION: u32 = 1;
26
27/// The whole session.
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
29#[serde(deny_unknown_fields)]
30pub struct SessionDoc {
31    /// Document version; must equal [`SESSION_VERSION`].
32    pub dspk_session: u32,
33    /// The processing graph (the Pipeline canvas).
34    #[serde(default, skip_serializing_if = "GraphDoc::is_empty")]
35    pub graph: GraphDoc,
36    /// The recipes the graph uses, by name (copies of library recipes, or made here).
37    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
38    pub recipes: Recipes,
39    /// Marked spans of the recordings.
40    #[serde(default, skip_serializing_if = "Vec::is_empty")]
41    pub epochs: Vec<EpochDoc>,
42    /// Where recordings sit on the session's time.
43    #[serde(default, skip_serializing_if = "Vec::is_empty")]
44    pub placements: Vec<Placement>,
45    /// Event sets drawn over the recordings' traces.
46    #[serde(default, skip_serializing_if = "Vec::is_empty")]
47    pub event_sets: Vec<EventSetDoc>,
48    /// Every view, in the order they are listed: compiled from `graph` (a document with views
49    /// and no graph gets one on load).
50    #[serde(default)]
51    pub views: Vec<ViewDoc>,
52    /// Id of the view shown (and edited by default).
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub active_view: Option<String>,
55}
56
57impl Default for SessionDoc {
58    fn default() -> Self {
59        Self { dspk_session: SESSION_VERSION, graph: GraphDoc::default(), recipes: Recipes::new(), epochs: Vec::new(), placements: Vec::new(), event_sets: Vec::new(), views: Vec::new(), active_view: None }
60    }
61}
62
63/// One view: a recording, its op chain, its display and its results.
64#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
65#[serde(deny_unknown_fields)]
66pub struct ViewDoc {
67    /// Unique within the session (`v1`, `v2`, …).
68    pub id: String,
69    /// Shown in tabs and lists.
70    pub name: String,
71    pub recording: RecordingRef,
72    /// Processing applied to every channel of the view, in order (no `read` / `write`).
73    #[serde(default)]
74    pub ops: Vec<OpDoc>,
75    #[serde(default)]
76    pub display: DisplayDoc,
77    /// What plugins keep for this view, by plugin id (e.g. `synapse`: the linked
78    /// `.sorting.zarr`); each plugin owns its own value.
79    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
80    #[schemars(with = "BTreeMap<String, serde_json::Value>")]
81    pub plugins: BTreeMap<String, serde_json::Value>,
82}
83
84/// The recording of a view: one source of one file, optionally a channel subset.
85#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
86#[serde(deny_unknown_fields)]
87pub struct RecordingRef {
88    pub path: PathBuf,
89    /// Source id within the file; default: the file's default source.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub source: Option<String>,
92    /// Channels of the source read by this view, in this order; default: all.
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub channels: Option<Vec<usize>>,
95}
96
97/// One op of a view's chain: a dspk-pipeline node with an id the commands address.
98#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
99#[serde(deny_unknown_fields)]
100pub struct OpDoc {
101    /// Unique within the view (`bandpass`, `cmr`, `op3`, …).
102    pub id: String,
103    /// Off: kept in the chain, skipped when processing (compare with and without).
104    #[serde(default = "yes")]
105    pub enabled: bool,
106    pub node: Node,
107}
108
109fn yes() -> bool {
110    true
111}
112
113/// How a view is drawn.
114#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
115#[serde(deny_unknown_fields)]
116pub struct DisplayDoc {
117    /// `processed`: the op chain's output (live preview); `raw`: the recording as stored.
118    #[serde(default)]
119    pub show: Show,
120    /// Amplitude gain (1 = nominal).
121    #[serde(default = "default_gain")]
122    pub gain: f32,
123    /// Traces on screen at once.
124    #[serde(default = "default_lanes")]
125    pub lanes: usize,
126    /// Visible time window, seconds from the recording's start: `[start, end]`.
127    #[serde(default, skip_serializing_if = "Option::is_none")]
128    pub window: Option<[f64; 2]>,
129    /// What the view is in Explore: the stream (time views), or its events' windows stacked.
130    #[serde(default, skip_serializing_if = "is_default")]
131    pub layout: ViewLayout,
132    /// Epochs layout: the window around each event, an index into the Analytics windows
133    /// (`screens::analytics::WINDOWS_MS`, default −200 … +800 ms).
134    #[serde(default = "default_epochs_window")]
135    pub epochs_window: usize,
136    /// Epochs layout: where the events come from.
137    #[serde(default, skip_serializing_if = "is_default")]
138    pub epochs_from: EpochsFrom,
139}
140
141/// What a view is in Explore.
142#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
143#[serde(rename_all = "snake_case")]
144pub enum ViewLayout {
145    /// Time views of the stream (traces, heatmap, raster).
146    #[default]
147    Stream,
148    /// Each event's window stacked per channel, mean ± SD (the chain ends with Crossings).
149    Epochs,
150}
151
152/// Where an Epochs view's events come from.
153#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
154#[serde(rename_all = "snake_case")]
155pub enum EpochsFrom {
156    /// The event set a Write of the same chain saved, else the live ones of the visible window.
157    #[default]
158    Auto,
159    /// Found now in the visible window (follows the level while it is tuned).
160    Live,
161    /// The saved event set of the chain's Write.
162    Saved,
163}
164
165fn is_default<T: Default + PartialEq>(v: &T) -> bool {
166    *v == T::default()
167}
168
169/// Default window of an Epochs view (−200 … +800 ms).
170pub const DEFAULT_EPOCHS_WINDOW: usize = 2;
171
172fn default_epochs_window() -> usize {
173    DEFAULT_EPOCHS_WINDOW
174}
175
176/// Default amplitude gain of a view.
177pub const DEFAULT_GAIN: f32 = 1.0;
178/// Default traces on screen of a view.
179pub const DEFAULT_LANES: usize = 8;
180
181fn default_gain() -> f32 {
182    DEFAULT_GAIN
183}
184fn default_lanes() -> usize {
185    DEFAULT_LANES
186}
187
188impl Default for DisplayDoc {
189    fn default() -> Self {
190        Self { show: Show::default(), gain: DEFAULT_GAIN, lanes: DEFAULT_LANES, window: None, layout: ViewLayout::Stream, epochs_window: DEFAULT_EPOCHS_WINDOW, epochs_from: EpochsFrom::Auto }
191    }
192}
193
194#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
195#[serde(rename_all = "snake_case")]
196pub enum Show {
197    #[default]
198    Processed,
199    Raw,
200}
201
202/// Why a document could not be read.
203#[derive(Debug, thiserror::Error)]
204pub enum SessionError {
205    #[error("session document: {0}")]
206    Parse(String),
207    #[error("session document version {0} is not supported (this app reads version {SESSION_VERSION})")]
208    Version(u32),
209    #[error("{0}")]
210    Io(String),
211}
212
213impl SessionDoc {
214    pub fn from_json(text: &str) -> Result<Self, SessionError> {
215        serde_json::from_str::<Self>(text).map_err(|e| SessionError::Parse(e.to_string()))?.checked()
216    }
217
218    fn checked(mut self) -> Result<Self, SessionError> {
219        if self.dspk_session != SESSION_VERSION {
220            return Err(SessionError::Version(self.dspk_session));
221        }
222        self.normalize();
223        Ok(self)
224    }
225
226    /// A graph for a document with views only, then the views from the graph. Views still named
227    /// the old default way (`file · /source/path`) take their source's name.
228    pub fn normalize(&mut self) {
229        if self.graph.is_empty() && !self.views.is_empty() {
230            self.graph = GraphDoc::from_views(&self.views);
231        }
232        self.compile();
233        let renames: Vec<(String, String)> = self.views.iter().filter(|v| v.name == legacy_view_name(&v.recording)).map(|v| (v.id.clone(), view_name(&v.recording))).collect();
234        if !renames.is_empty() {
235            for (id, name) in renames {
236                if let Some(super::graph::NodeKind::View { name: old, .. }) = self.graph.nodes.iter_mut().find(|n| n.id == id).map(|n| &mut n.kind) {
237                    *old = name;
238                }
239            }
240            self.compile();
241        }
242    }
243
244    /// The views from the graph; the active view stays when it is still one, else the first.
245    pub fn compile(&mut self) {
246        self.views = self.graph.compile(&self.recipes);
247        if !self.active_view.as_ref().is_some_and(|a| self.views.iter().any(|v| v.id == *a)) {
248            self.active_view = self.views.first().map(|v| v.id.clone());
249        }
250    }
251
252    /// The document as written by [`Self::from_json`] (graph and compiled views).
253    pub fn normalized(mut self) -> Self {
254        self.normalize();
255        self
256    }
257
258    /// Reads a document (JSON).
259    pub fn load(path: &Path) -> Result<Self, SessionError> {
260        let text = std::fs::read_to_string(path).map_err(|e| SessionError::Io(format!("{}: {e}", crate::domain::paths::short_path(&path))))?;
261        Self::from_json(&text)
262    }
263
264    /// The canonical form: pretty JSON with every default written out.
265    pub fn to_json(&self) -> String {
266        serde_json::to_string_pretty(self).expect("a session always serializes")
267    }
268
269    /// Writes the canonical JSON (through a temporary file, so a reader never sees half of it).
270    pub fn save(&self, path: &Path) -> std::io::Result<()> {
271        if let Some(dir) = path.parent() {
272            std::fs::create_dir_all(dir)?;
273        }
274        let tmp = path.with_extension("json.tmp");
275        std::fs::write(&tmp, self.to_json())?;
276        std::fs::rename(tmp, path)
277    }
278
279    /// JSON Schema of the document (for agents and editors).
280    pub fn json_schema() -> String {
281        serde_json::to_string_pretty(&schemars::schema_for!(SessionDoc)).expect("a schema always serializes")
282    }
283
284    pub fn view(&self, id: &str) -> Option<&ViewDoc> {
285        self.views.iter().find(|v| v.id == id)
286    }
287
288    pub fn view_mut(&mut self, id: &str) -> Option<&mut ViewDoc> {
289        self.views.iter_mut().find(|v| v.id == id)
290    }
291
292    pub fn active(&self) -> Option<&ViewDoc> {
293        self.active_view.as_deref().and_then(|id| self.view(id))
294    }
295
296    /// A view id not used yet (`v1`, `v2`, …), by any view or node.
297    pub fn new_view_id(&self) -> String {
298        fresh_id("v", |id| self.view(id).is_some() || self.graph.contains(id))
299    }
300
301    /// An op id not used yet by any node, from the op's name (`iir`, `iir1`, …).
302    pub fn new_op_id(&self, node: &Node) -> String {
303        let id = self.graph.fresh_id(op_name(node));
304        if self.view(&id).is_none() {
305            return id;
306        }
307        fresh_id(op_name(node), |id| self.graph.contains(id) || self.view(id).is_some())
308    }
309}
310
311impl ViewDoc {
312    pub fn op(&self, id: &str) -> Option<&OpDoc> {
313        self.ops.iter().find(|o| o.id == id)
314    }
315
316    /// An op id not used yet in this view, from the op's name (`iir`, `iir2`, …).
317    pub fn new_op_id(&self, node: &Node) -> String {
318        let base = op_name(node);
319        if self.op(base).is_none() {
320            return base.to_string();
321        }
322        fresh_id(base, |id| self.op(id).is_some())
323    }
324
325    /// The enabled ops as stages, in order.
326    pub fn enabled_nodes(&self) -> impl Iterator<Item = &Node> {
327        self.ops.iter().filter(|o| o.enabled).map(|o| &o.node)
328    }
329
330    /// The pipeline that writes this view (its recording through its enabled ops) to `output`.
331    pub fn pipeline(&self, output: PathBuf, format: OutputFormat) -> PipelineSpec {
332        let mut nodes = BTreeMap::new();
333        let mut edges = Vec::new();
334        nodes.insert(
335            "read".to_string(),
336            Node::Read {
337                path: Some(self.recording.path.clone()),
338                inputs: None,
339                source: self.recording.source.clone(),
340                channels: self.recording.channels.clone(),
341                start: None,
342                stop: None,
343            },
344        );
345        let mut last = "read".to_string();
346        for op in self.ops.iter().filter(|o| o.enabled) {
347            // Op ids are unique in a view; `read` / `write` are reserved by the commands
348            nodes.insert(op.id.clone(), op.node.clone());
349            edges.push((last, op.id.clone()));
350            last = op.id.clone();
351        }
352        nodes.insert("write".to_string(), Node::Write { path: output, format, compression: None, chunk_samples: None, parallel: None });
353        edges.push((last, "write".to_string()));
354        PipelineSpec { dspk_pipeline: SPEC_VERSION, name: Some(self.name.clone()), nodes, edges, execution: Default::default() }
355    }
356}
357
358/// The family of an op (a recipe's default label).
359pub fn op_family(node: &Node) -> &'static str {
360    match node {
361        Node::Iir { .. } | Node::Fir { .. } | Node::Sos { .. } | Node::Notch { .. } => "filter",
362        Node::Car {} | Node::Cmr {} | Node::Laplacian { .. } | Node::Whiten { .. } => "reference",
363        Node::Median { .. } | Node::GaussianSmooth { .. } => "smooth",
364        Node::Remap { .. } => "channels",
365        _ => "transform",
366    }
367}
368
369/// The spec name of a node (`iir`, `cmr`, …).
370pub fn op_name(node: &Node) -> &'static str {
371    match node {
372        Node::Read { .. } => "read",
373        Node::Write { .. } => "write",
374        Node::Scale { .. } => "scale",
375        Node::SubtractBaseline { .. } => "subtract_baseline",
376        Node::Clamp { .. } => "clamp",
377        Node::Iir { .. } => "iir",
378        Node::Notch { .. } => "notch",
379        Node::Sos { .. } => "sos",
380        Node::Fir { .. } => "fir",
381        Node::Resample { .. } => "resample",
382        Node::Remap { .. } => "remap",
383        Node::Car {} => "car",
384        Node::Cmr {} => "cmr",
385        Node::GaussianSmooth { .. } => "gaussian_smooth",
386        Node::Median { .. } => "median",
387        Node::TeagerKaiser { .. } => "teager_kaiser",
388        Node::Rectify {} => "rectify",
389        Node::Square {} => "square",
390        Node::Sqrt {} => "sqrt",
391        Node::Log10 {} => "log10",
392        Node::Laplacian { .. } => "laplacian",
393        Node::Whiten { .. } => "whiten",
394        Node::Crossings { .. } => "crossings",
395    }
396}
397
398fn fresh_id(prefix: &str, taken: impl Fn(&str) -> bool) -> String {
399    (1..).map(|i| format!("{prefix}{i}")).find(|id| !taken(id)).expect("ids are unbounded")
400}
401
402#[cfg(test)]
403mod tests {
404    use super::*;
405    use dspk_pipeline::{Band, Design, Mode, Start};
406
407    fn bandpass(lo: f64, hi: f64) -> Node {
408        Node::Iir { band: Band::Bandpass, cutoff: vec![lo, hi], order: 3, design: Design::Butterworth, ripple_db: None, mode: Mode::ForwardBackward, start: Start::Rest }
409    }
410
411    fn doc() -> SessionDoc {
412        let view = ViewDoc {
413            id: "v1".into(),
414            name: "probe A".into(),
415            recording: RecordingRef { path: "/data/rec.bin".into(), source: None, channels: Some(vec![0, 1, 2]) },
416            ops: vec![
417                OpDoc { id: "iir".into(), enabled: true, node: bandpass(300.0, 6000.0) },
418                OpDoc { id: "cmr".into(), enabled: false, node: Node::Cmr {} },
419            ],
420            display: DisplayDoc::default(),
421            plugins: BTreeMap::new(),
422        };
423        SessionDoc { views: vec![view], active_view: Some("v1".into()), ..SessionDoc::default() }.normalized()
424    }
425
426    #[test]
427    fn test_json_round_trip() {
428        let d = doc();
429        assert_eq!(d.graph.nodes.len(), 4, "source, two ops, view");
430        assert_eq!(SessionDoc::from_json(&d.to_json()).unwrap(), d);
431    }
432
433    #[test]
434    fn test_defaults_are_filled() {
435        let text = r#"{"dspk_session": 1, "active_view": "v1",
436            "views": [{"id": "v1", "name": "probe A", "recording": {"path": "/data/rec.bin"},
437                       "ops": [{"id": "hp", "node": {"op": "iir", "band": "highpass", "cutoff": [300.0]}}]}]}"#;
438        let d = SessionDoc::from_json(text).unwrap();
439        assert_eq!(d.views[0].ops[0].id, "hp");
440        assert!(d.views[0].ops[0].enabled);
441        assert_eq!(d.views[0].display, DisplayDoc::default());
442    }
443
444    #[test]
445    fn test_unknown_version_and_fields_fail() {
446        assert!(matches!(SessionDoc::from_json(r#"{"dspk_session": 2}"#), Err(SessionError::Version(2))));
447        assert!(SessionDoc::from_json(r#"{"dspk_session": 1, "viewz": []}"#).is_err());
448    }
449
450    #[test]
451    fn test_pipeline_of_a_view_skips_disabled_ops() {
452        let spec = doc().views[0].pipeline("/out/x.zarr".into(), OutputFormat::Zarr);
453        let chain = spec.chain().unwrap();
454        assert_eq!(chain.stages.iter().map(|(id, _)| *id).collect::<Vec<_>>(), ["iir"]);
455        match chain.read.1 {
456            Node::Read { channels, .. } => assert_eq!(channels.as_deref(), Some(&[0, 1, 2][..])),
457            _ => unreachable!(),
458        }
459    }
460
461    #[test]
462    fn test_ids() {
463        let d = doc();
464        assert_eq!(d.new_view_id(), "v2");
465        assert_eq!(d.views[0].new_op_id(&bandpass(1.0, 2.0)), "iir1");
466        assert_eq!(d.views[0].new_op_id(&Node::Car {}), "car");
467        assert_eq!(d.new_op_id(&bandpass(1.0, 2.0)), "iir1");
468    }
469
470    #[test]
471    fn test_schema_names_the_document() {
472        assert!(SessionDoc::json_schema().contains("dspk_session"));
473    }
474}
475
476/// The name of a view of `r`: its source's name (the last part of the source's path: `EMG` for
477/// `/acquisition/EMG`), or the file's name for a file of one source.
478pub fn view_name(r: &RecordingRef) -> String {
479    match r.source.as_deref().filter(|s| !s.is_empty() && *s != dspk_io::sources::MAIN) {
480        Some(s) => source_name(s),
481        None => file_name(&r.path),
482    }
483}
484
485/// A source's short name: the last part of its path (`EMG` for `/acquisition/EMG`).
486pub fn source_name(id: &str) -> String {
487    id.rsplit('/').find(|p| !p.is_empty()).unwrap_or(id).to_string()
488}
489
490fn file_name(path: &Path) -> String {
491    path.file_name().map_or_else(|| path.display().to_string(), |n| n.to_string_lossy().into_owned())
492}
493
494/// The default view name before views were named by source (`file · /source/path`).
495fn legacy_view_name(r: &RecordingRef) -> String {
496    match &r.source {
497        Some(s) => format!("{} · {s}", file_name(&r.path)),
498        None => file_name(&r.path),
499    }
500}
dspk_app_core/domain/session.rs

dspk_app_core/domain/
session.rs
