modewise_operations/
epoch.rs

1use modewise_epochs::{EpochParams, EpochSource, Epochs, Window};
2use modewise_pipeline::{
3    Artifact, ArtifactKind, Inputs, Operation, OperationDescriptor, Outputs, Port, resolve_params,
4};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9use crate::input;
10
11#[derive(Serialize, Deserialize, JsonSchema, Clone, Copy)]
12#[serde(untagged)]
13enum WindowParam {
14    Named(WindowName),
15    Fixed { tmin_s: f64, tmax_s: f64 },
16}
17
18#[derive(Serialize, Deserialize, JsonSchema, Clone, Copy)]
19#[serde(rename_all = "snake_case")]
20enum WindowName {
21    Duration,
22}
23
24#[derive(Serialize, Deserialize, JsonSchema)]
25#[serde(deny_unknown_fields)]
26struct Params {
27    /// Required: shift added to event times (haemodynamic delay in fMRI; 0 for EEG).
28    lag_s: f64,
29    /// `"duration"` uses each event's own length (blocks); `{ tmin_s = -0.2, tmax_s = 0.5 }` uses a fixed window around each onset (trials).
30    #[serde(default = "duration")]
31    window: WindowParam,
32    /// Keep only these event labels; all by default.
33    #[serde(default)]
34    conditions: Option<Vec<String>>,
35    /// A mode or frequency is resolvable in an epoch when frequency × epoch length ≥ this, and, for maps with a cone of influence, the whole epoch is outside it (docs/epochs.md).
36    #[serde(default = "one")]
37    min_cycles: f64,
38    /// Merge consecutive same-label events (trials) into blocks when gaps are at most this long.
39    #[serde(default)]
40    merge_gap_s: Option<f64>,
41}
42
43fn duration() -> WindowParam {
44    WindowParam::Named(WindowName::Duration)
45}
46
47fn one() -> f64 {
48    1.0
49}
50
51impl Params {
52    fn epoch_params(&self) -> Result<EpochParams, String> {
53        if let Some(gap) = self.merge_gap_s
54            && !(gap.is_finite() && gap >= 0.0)
55        {
56            return Err(format!("merge_gap_s must be finite and >= 0, got {gap}"));
57        }
58        let window = match self.window {
59            WindowParam::Named(WindowName::Duration) => Window::Duration,
60            WindowParam::Fixed { tmin_s, tmax_s } => Window::Fixed { tmin_s, tmax_s },
61        };
62        let params = EpochParams {
63            lag_s: self.lag_s,
64            window,
65            conditions: self.conditions.clone(),
66            min_cycles: self.min_cycles,
67        };
68        params.validate().map_err(|e| e.to_string())?;
69        Ok(params)
70    }
71}
72
73/// Cuts a full-run series or decomposition into epochs around events (rules: docs/epochs.md).
74pub struct EpochOperation;
75
76impl Operation for EpochOperation {
77    fn summary(&self) -> &'static str {
78        "Cuts windows around events out of a full-run series, decomposition, instantaneous attributes or time–frequency map, with an explicit haemodynamic lag (docs/epochs.md)."
79    }
80
81    fn params_schema(&self) -> serde_json::Value {
82        modewise_pipeline::schema_of::<Params>()
83    }
84
85    fn descriptor(&self) -> OperationDescriptor {
86        OperationDescriptor {
87            name: "epoch",
88            version: env!("CARGO_PKG_VERSION"),
89            inputs: &[
90                Port {
91                    name: "data",
92                    kinds: &[
93                        ArtifactKind::Series,
94                        ArtifactKind::Decomposition,
95                        ArtifactKind::Instantaneous,
96                        ArtifactKind::TimeFrequency,
97                    ],
98                },
99                Port {
100                    name: "events",
101                    kinds: &[ArtifactKind::Events],
102                },
103            ],
104            outputs: &[Port {
105                name: "epochs",
106                kinds: &[ArtifactKind::Epochs],
107            }],
108        }
109    }
110
111    fn resolve(&self, params: Value) -> Result<Value, String> {
112        let (typed, resolved) = resolve_params::<Params>(params)?;
113        typed.epoch_params()?;
114        Ok(resolved)
115    }
116
117    fn run(&self, params: &Value, inputs: &Inputs) -> Result<Outputs, String> {
118        let p: Params = serde_json::from_value(params.clone()).map_err(|e| e.to_string())?;
119        let source = match input(inputs, "data")? {
120            Artifact::Series(s) => EpochSource::Series(s.clone()),
121            Artifact::Decomposition(d) => EpochSource::Decomposition(d.clone()),
122            Artifact::Instantaneous(m) => EpochSource::Instantaneous(m.clone()),
123            Artifact::TimeFrequency(m) => EpochSource::TimeFrequency(m.clone()),
124            other => {
125                return Err(format!(
126                    "input \"data\" cannot be epoched: it is {}",
127                    other.kind()
128                ));
129            }
130        };
131        let Artifact::Events(events) = input(inputs, "events")? else {
132            return Err("input \"events\" must be events".into());
133        };
134        let events = match p.merge_gap_s {
135            Some(gap) => events.merge_consecutive(gap),
136            None => events.clone(),
137        };
138        let epochs = Epochs::cut(source, &events, p.epoch_params()?).map_err(|e| e.to_string())?;
139        Ok(Outputs::from([("epochs".into(), Artifact::Epochs(epochs))]))
140    }
141}
modewise_operations/epoch.rs

modewise_operations/
epoch.rs
