modewise_operations/
emd.rs

1use modewise_decomposition::ModeDecomposer;
2use modewise_emd::{Ceemdan, Eemd, Emd, Memd, Stop};
3use modewise_pipeline::{
4    Artifact, ArtifactKind, Inputs, Operation, OperationDescriptor, Outputs, Port, resolve_params,
5};
6use schemars::JsonSchema;
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9
10use crate::input;
11
12#[derive(Serialize, Deserialize, JsonSchema, Clone, Copy, PartialEq)]
13#[serde(rename_all = "snake_case")]
14enum StopName {
15    Rilling,
16    SNumber,
17    Fixed,
18}
19
20#[derive(Serialize, Deserialize, JsonSchema)]
21#[serde(deny_unknown_fields)]
22struct Params {
23    /// When sifting stops: `rilling` (thresholds sd, sd2, tol), `s_number` (count) or `fixed` (siftings); `fixed` for eemd, `rilling` otherwise.
24    #[serde(default, skip_serializing_if = "Option::is_none")]
25    stop: Option<StopName>,
26    /// rilling: the envelope mean relative to the mode amplitude must stay below this on all but a fraction tol of samples.
27    #[serde(default, skip_serializing_if = "Option::is_none")]
28    sd: Option<f64>,
29    /// rilling: the same ratio must stay below this everywhere (0.5; memd 0.75).
30    #[serde(default, skip_serializing_if = "Option::is_none")]
31    sd2: Option<f64>,
32    /// rilling: the fraction of samples allowed above sd (0.05; memd 0.075).
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    tol: Option<f64>,
35    /// s_number: consecutive siftings with extrema and zero crossings within one (2 by default).
36    #[serde(default, skip_serializing_if = "Option::is_none")]
37    count: Option<usize>,
38    /// fixed: siftings per IMF (10 by default).
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    siftings: Option<usize>,
41    /// Stop after this many IMFs (the rest is the residual); all by default, at most ⌊log₂ n⌋ − 1 for eemd.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    max_imfs: Option<usize>,
44    /// memd only: projection directions; use at least twice the channels.
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    num_directions: Option<usize>,
47    /// eemd and ceemdan: noise realisations.
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    trials: Option<usize>,
50    /// eemd only: noise SD relative to the signal's.
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    noise_sd: Option<f64>,
53    /// ceemdan only: noise scale ε.
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    epsilon: Option<f64>,
56    /// eemd and ceemdan: noise seed.
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    seed: Option<u64>,
59}
60
61#[derive(Clone, Copy, PartialEq)]
62enum Variant {
63    Emd,
64    Memd,
65    Eemd,
66    Ceemdan,
67}
68
69fn only(name: &str, owner: &str) -> String {
70    format!("`{name}` only applies to {owner}")
71}
72
73impl Params {
74    /// Fills the defaults of the method and its stop rule, and rejects settings that do not apply.
75    fn complete(mut self, variant: Variant) -> Result<Self, String> {
76        let default_stop = if variant == Variant::Eemd {
77            StopName::Fixed
78        } else {
79            StopName::Rilling
80        };
81        let stop = *self.stop.get_or_insert(default_stop);
82        let rilling = [self.sd, self.sd2, self.tol];
83        let foreign = match stop {
84            StopName::Rilling => self.count.is_some() || self.siftings.is_some(),
85            StopName::SNumber => rilling.iter().any(Option::is_some) || self.siftings.is_some(),
86            StopName::Fixed => rilling.iter().any(Option::is_some) || self.count.is_some(),
87        };
88        if foreign {
89            return Err("only the chosen stop's settings apply: sd, sd2 and tol for rilling, count for s_number, siftings for fixed".into());
90        }
91        match stop {
92            StopName::Rilling => {
93                let default = if variant == Variant::Memd {
94                    Stop::RILLING_MEMD
95                } else {
96                    Stop::RILLING_EMD
97                };
98                if let Stop::Rilling { sd, sd2, tol } = default {
99                    self.sd.get_or_insert(sd);
100                    self.sd2.get_or_insert(sd2);
101                    self.tol.get_or_insert(tol);
102                }
103            }
104            StopName::SNumber => {
105                self.count.get_or_insert(2);
106            }
107            StopName::Fixed => {
108                self.siftings.get_or_insert(10);
109            }
110        }
111        let noisy = matches!(variant, Variant::Eemd | Variant::Ceemdan);
112        if variant == Variant::Memd {
113            self.num_directions.get_or_insert(64);
114        } else if self.num_directions.is_some() {
115            return Err(only("num_directions", "memd"));
116        }
117        if noisy {
118            self.trials.get_or_insert(100);
119            self.seed.get_or_insert(0);
120        } else if self.trials.is_some() || self.seed.is_some() {
121            return Err("`trials` and `seed` only apply to eemd and ceemdan".into());
122        }
123        if variant == Variant::Eemd {
124            self.noise_sd.get_or_insert(0.2);
125        } else if self.noise_sd.is_some() {
126            return Err(only("noise_sd", "eemd"));
127        }
128        if variant == Variant::Ceemdan {
129            self.epsilon.get_or_insert(0.2);
130        } else if self.epsilon.is_some() {
131            return Err(only("epsilon", "ceemdan"));
132        }
133        Ok(self)
134    }
135
136    fn stop(&self) -> Stop {
137        match self.stop.unwrap_or(StopName::Rilling) {
138            StopName::Rilling => Stop::Rilling {
139                sd: self.sd.unwrap_or_default(),
140                sd2: self.sd2.unwrap_or_default(),
141                tol: self.tol.unwrap_or_default(),
142            },
143            StopName::SNumber => Stop::SNumber {
144                count: self.count.unwrap_or_default(),
145            },
146            StopName::Fixed => Stop::Fixed {
147                siftings: self.siftings.unwrap_or_default(),
148            },
149        }
150    }
151
152    fn method(&self, variant: Variant) -> Result<Box<dyn ModeDecomposer + Send + Sync>, String> {
153        let (stop, max_imfs) = (self.stop(), self.max_imfs);
154        let (trials, seed) = (
155            self.trials.unwrap_or_default(),
156            self.seed.unwrap_or_default(),
157        );
158        let method: Result<Box<dyn ModeDecomposer + Send + Sync>, _> = match variant {
159            Variant::Emd => Emd::new(stop, max_imfs).map(|m| Box::new(m) as _),
160            Variant::Memd => Memd::new(stop, max_imfs, self.num_directions.unwrap_or_default())
161                .map(|m| Box::new(m) as _),
162            Variant::Eemd => {
163                let noise_sd = self.noise_sd.unwrap_or_default();
164                Eemd {
165                    stop,
166                    max_imfs,
167                    trials,
168                    noise_sd,
169                    seed,
170                }
171                .validated()
172                .map(|m| Box::new(m) as _)
173            }
174            Variant::Ceemdan => {
175                let epsilon = self.epsilon.unwrap_or_default();
176                Ceemdan {
177                    stop,
178                    max_imfs,
179                    trials,
180                    epsilon,
181                    seed,
182                }
183                .validated()
184                .map(|m| Box::new(m) as _)
185            }
186        };
187        method.map_err(|e| e.to_string())
188    }
189}
190
191/// The shared schema narrowed to what `variant` accepts, with its defaults filled in by the
192/// same completion that `resolve` uses.
193fn variant_schema(variant: Variant) -> Value {
194    let mut schema = modewise_pipeline::schema_of::<Params>();
195    let applies = |field: &str| match field {
196        "num_directions" => variant == Variant::Memd,
197        "trials" | "seed" => matches!(variant, Variant::Eemd | Variant::Ceemdan),
198        "noise_sd" => variant == Variant::Eemd,
199        "epsilon" => variant == Variant::Ceemdan,
200        _ => true,
201    };
202    let defaults = serde_json::from_value::<Params>(Value::Object(Default::default()))
203        .ok()
204        .and_then(|p| p.complete(variant).ok())
205        .and_then(|p| serde_json::to_value(p).ok())
206        .unwrap_or_default();
207    if let Some(properties) = schema.get_mut("properties").and_then(Value::as_object_mut) {
208        properties.retain(|field, _| applies(field));
209        for (field, spec) in properties.iter_mut() {
210            if let (Some(default), Some(spec)) = (defaults.get(field), spec.as_object_mut()) {
211                spec.insert("default".into(), default.clone());
212            }
213        }
214    }
215    schema
216}
217
218/// `emd`, `eemd` and `ceemdan` (each channel on its own) and `memd` (multivariate).
219pub struct EmdOperation;
220pub struct MemdOperation;
221pub struct EemdOperation;
222pub struct CeemdanOperation;
223
224fn descriptor(name: &'static str) -> OperationDescriptor {
225    OperationDescriptor {
226        name,
227        version: env!("CARGO_PKG_VERSION"),
228        inputs: &[Port {
229            name: "signal",
230            kinds: &[ArtifactKind::Series],
231        }],
232        outputs: &[Port {
233            name: "modes",
234            kinds: &[ArtifactKind::Decomposition],
235        }],
236    }
237}
238
239fn resolve(variant: Variant, params: Value) -> Result<Value, String> {
240    let (typed, _) = resolve_params::<Params>(params)?;
241    let typed = typed.complete(variant)?;
242    typed.method(variant)?;
243    serde_json::to_value(&typed).map_err(|e| e.to_string())
244}
245
246fn run(variant: Variant, params: &Value, inputs: &Inputs) -> Result<Outputs, String> {
247    let method = serde_json::from_value::<Params>(params.clone())
248        .map_err(|e| e.to_string())?
249        .complete(variant)?
250        .method(variant)?;
251    let Artifact::Series(signal) = input(inputs, "signal")? else {
252        return Err("input \"signal\" must be a series".into());
253    };
254    let modes = method.decompose(signal).map_err(|e| e.to_string())?;
255    Ok(Outputs::from([(
256        "modes".into(),
257        Artifact::Decomposition(modes),
258    )]))
259}
260
261macro_rules! emd_family {
262    ($operation:ident, $name:literal, $variant:expr, $summary:literal) => {
263        impl Operation for $operation {
264            fn summary(&self) -> &'static str {
265                $summary
266            }
267
268            fn params_schema(&self) -> serde_json::Value {
269                variant_schema($variant)
270            }
271
272            fn descriptor(&self) -> OperationDescriptor {
273                descriptor($name)
274            }
275
276            fn resolve(&self, params: Value) -> Result<Value, String> {
277                resolve($variant, params)
278            }
279
280            fn run(&self, params: &Value, inputs: &Inputs) -> Result<Outputs, String> {
281                run($variant, params, inputs)
282            }
283        }
284    };
285}
286
287emd_family!(
288    EmdOperation,
289    "emd",
290    Variant::Emd,
291    "Empirical mode decomposition of each channel on its own (docs/emd.md)."
292);
293emd_family!(
294    MemdOperation,
295    "memd",
296    Variant::Memd,
297    "Multivariate EMD: the same IMFs in every channel, from projections on many directions (docs/emd.md)."
298);
299emd_family!(
300    EemdOperation,
301    "eemd",
302    Variant::Eemd,
303    "Ensemble EMD: the mean of EMDs of the signal plus seeded white noise (docs/emd.md)."
304);
305emd_family!(
306    CeemdanOperation,
307    "ceemdan",
308    Variant::Ceemdan,
309    "CEEMDAN, improved form: noise-assisted EMD whose modes sum to the signal (docs/emd.md)."
310);
modewise_operations/emd.rs

modewise_operations/
emd.rs
