modewise_operations/regions/
mod.rs

1//! Regional selection and summaries (docs/regions.md).
2
3mod subset;
4
5use std::collections::BTreeMap;
6
7use modewise_brain::{ChannelTable, Hemisphere, Pattern, Rule, Selection};
8use modewise_pipeline::{
9    Artifact, ArtifactKind, Inputs, Operation, OperationDescriptor, Outputs, Port, resolve_params,
10};
11use schemars::JsonSchema;
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14
15use crate::input;
16use subset::{
17    Average, Keep, average_connectivity, combine_channels, keep_connectivity, keep_table, labels,
18};
19
20const WITH_CHANNELS: &[ArtifactKind] = &[
21    ArtifactKind::Series,
22    ArtifactKind::Decomposition,
23    ArtifactKind::Instantaneous,
24    ArtifactKind::TimeFrequency,
25    ArtifactKind::Epochs,
26    ArtifactKind::FunctionalConnectivity,
27    ArtifactKind::Loadings,
28    ArtifactKind::Channels,
29];
30
31#[derive(Serialize, Deserialize, JsonSchema, Clone, Copy)]
32#[serde(rename_all = "lowercase")]
33enum Side {
34    Left,
35    Right,
36}
37
38#[derive(Serialize, Deserialize, JsonSchema, Default)]
39#[serde(deny_unknown_fields)]
40struct RuleParams {
41    /// Channel name, exactly or with `*` wildcards.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    name: Option<String>,
44    /// Region label, exactly or with `*` wildcards.
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    region: Option<String>,
47    /// `left` or `right`.
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    hemisphere: Option<Side>,
50    /// Tag values that must all match, for example `{ yeo17 = "Default*" }`.
51    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
52    tags: BTreeMap<String, String>,
53}
54
55impl RuleParams {
56    fn rule(&self) -> Rule {
57        Rule {
58            name: self.name.as_deref().map(Pattern::new),
59            region: self.region.as_deref().map(Pattern::new),
60            hemisphere: self.hemisphere.map(|s| match s {
61                Side::Left => Hemisphere::Left,
62                Side::Right => Hemisphere::Right,
63            }),
64            tags: self
65                .tags
66                .iter()
67                .map(|(k, v)| (k.clone(), Pattern::new(v)))
68                .collect(),
69        }
70    }
71}
72
73#[derive(Serialize, Deserialize, JsonSchema)]
74#[serde(deny_unknown_fields)]
75struct SelectParams {
76    /// Channels matching any rule are kept. A rule has any of `name`, `region`, `hemisphere` (`left`/`right`) and `tags` (a table); all of them must match. Text matches exactly, or with `*` wildcards.
77    rules: Vec<RuleParams>,
78}
79
80impl SelectParams {
81    fn selection(&self) -> Result<Selection, String> {
82        Selection::new(self.rules.iter().map(RuleParams::rule).collect()).map_err(|e| e.to_string())
83    }
84}
85
86fn channel_table(inputs: &Inputs) -> Result<&ChannelTable, String> {
87    match input(inputs, "channels")? {
88        Artifact::Channels(t) => Ok(t),
89        other => Err(format!(
90            "input \"channels\" must be a channel table, not a {}",
91            other.kind()
92        )),
93    }
94}
95
96fn data_labels(data: &Artifact) -> Result<Vec<String>, String> {
97    match data {
98        Artifact::Channels(t) => Ok(t.names()),
99        other => labels(other)
100            .map(<[String]>::to_vec)
101            .ok_or_else(|| format!("a {} has no channels", other.kind())),
102    }
103}
104
105/// Keeps the channels that match any rule; works before or after decomposition.
106pub struct SelectOperation;
107
108impl Operation for SelectOperation {
109    fn summary(&self) -> &'static str {
110        "Keeps the channels whose name, region, hemisphere or tags match the rules (docs/regions.md)."
111    }
112
113    fn params_schema(&self) -> serde_json::Value {
114        modewise_pipeline::schema_of::<SelectParams>()
115    }
116
117    fn descriptor(&self) -> OperationDescriptor {
118        OperationDescriptor {
119            name: "select",
120            version: env!("CARGO_PKG_VERSION"),
121            inputs: &[
122                Port {
123                    name: "data",
124                    kinds: WITH_CHANNELS,
125                },
126                Port {
127                    name: "channels",
128                    kinds: &[ArtifactKind::Channels],
129                },
130            ],
131            outputs: &[Port {
132                name: "selected",
133                kinds: WITH_CHANNELS,
134            }],
135        }
136    }
137
138    fn resolve(&self, params: Value) -> Result<Value, String> {
139        let (typed, resolved) = resolve_params::<SelectParams>(params)?;
140        typed.selection()?;
141        Ok(resolved)
142    }
143
144    fn run(&self, params: &Value, inputs: &Inputs) -> Result<Outputs, String> {
145        let selection = serde_json::from_value::<SelectParams>(params.clone())
146            .map_err(|e| e.to_string())?
147            .selection()?;
148        let data = input(inputs, "data")?;
149        let names = data_labels(data)?;
150        let positions = selection
151            .positions(channel_table(inputs)?, &names)
152            .map_err(|e| e.to_string())?;
153        let selected = match data {
154            Artifact::FunctionalConnectivity(c) => {
155                Artifact::FunctionalConnectivity(keep_connectivity(c, &positions)?)
156            }
157            Artifact::Channels(t) => Artifact::Channels(keep_table(t, &positions)?),
158            other => combine_channels(
159                other,
160                &Keep {
161                    positions: &positions,
162                    labels: &names,
163                },
164            )?,
165        };
166        Ok(Outputs::from([("selected".into(), selected)]))
167    }
168
169    fn output_kind(&self, port: &Port, inputs: &BTreeMap<String, ArtifactKind>) -> ArtifactKind {
170        inputs
171            .get("data")
172            .copied()
173            .unwrap_or_else(|| port.output_kind())
174    }
175}
176
177#[derive(Serialize, Deserialize, JsonSchema, Clone, Copy, PartialEq, Eq)]
178#[serde(rename_all = "lowercase")]
179enum GroupBy {
180    Region,
181    Hemisphere,
182    Tag,
183}
184
185#[derive(Serialize, Deserialize, JsonSchema)]
186#[serde(deny_unknown_fields)]
187struct AverageParams {
188    /// Group channels by `region`, `hemisphere` or `tag`.
189    by: GroupBy,
190    /// Tag key to group by when `by = "tag"`, for example `yeo17`.
191    #[serde(default, skip_serializing_if = "Option::is_none")]
192    tag: Option<String>,
193    /// Keep left and right apart: `PFCm_left`, `PFCm_right`.
194    #[serde(default)]
195    split_hemispheres: bool,
196}
197
198impl AverageParams {
199    fn check(&self) -> Result<(), String> {
200        match (self.by, &self.tag) {
201            (GroupBy::Tag, None) => Err("by = \"tag\" needs `tag`, the tag key to group by".into()),
202            (GroupBy::Region | GroupBy::Hemisphere, Some(_)) => {
203                Err("`tag` applies only with by = \"tag\"".into())
204            }
205            (GroupBy::Hemisphere, None) if self.split_hemispheres => {
206                Err("split_hemispheres has no effect when grouping by hemisphere".into())
207            }
208            _ => Ok(()),
209        }
210    }
211
212    /// Groups of positions in `names`, in order of first appearance. Channels without the
213    /// grouping description are left out.
214    fn groups(
215        &self,
216        table: &ChannelTable,
217        names: &[String],
218    ) -> Result<Vec<(String, Vec<usize>)>, String> {
219        let infos = table.lookup(names).map_err(|e| e.to_string())?;
220        let mut groups: Vec<(String, Vec<usize>)> = Vec::new();
221        for (i, c) in infos.iter().enumerate() {
222            let key = match self.by {
223                GroupBy::Region => c.region().map(|r| r.as_str().to_string()),
224                GroupBy::Hemisphere => c.hemisphere().map(|h| h.as_str().to_string()),
225                GroupBy::Tag => self
226                    .tag
227                    .as_deref()
228                    .and_then(|t| c.tag(t))
229                    .map(str::to_string),
230            };
231            let Some(mut key) = key else { continue };
232            if self.split_hemispheres
233                && let Some(h) = c.hemisphere()
234            {
235                key = format!("{key}_{}", h.as_str());
236            }
237            match groups.iter_mut().find(|(k, _)| *k == key) {
238                Some((_, members)) => members.push(i),
239                None => groups.push((key, vec![i])),
240            }
241        }
242        if groups.is_empty() {
243            return Err("no channel has the grouping description".into());
244        }
245        Ok(groups)
246    }
247}
248
249/// Averages channels within each region, hemisphere or tag group.
250pub struct RegionAverageOperation;
251
252impl Operation for RegionAverageOperation {
253    fn summary(&self) -> &'static str {
254        "Averages channels by region, hemisphere or tag (functional connectivity in Fisher z) (docs/regions.md)."
255    }
256
257    fn params_schema(&self) -> serde_json::Value {
258        modewise_pipeline::schema_of::<AverageParams>()
259    }
260
261    fn descriptor(&self) -> OperationDescriptor {
262        const AVERAGEABLE: &[ArtifactKind] = &[
263            ArtifactKind::Series,
264            ArtifactKind::Decomposition,
265            ArtifactKind::TimeFrequency,
266            ArtifactKind::Epochs,
267            ArtifactKind::FunctionalConnectivity,
268        ];
269        OperationDescriptor {
270            name: "region_average",
271            version: env!("CARGO_PKG_VERSION"),
272            inputs: &[
273                Port {
274                    name: "data",
275                    kinds: AVERAGEABLE,
276                },
277                Port {
278                    name: "channels",
279                    kinds: &[ArtifactKind::Channels],
280                },
281            ],
282            outputs: &[Port {
283                name: "averaged",
284                kinds: AVERAGEABLE,
285            }],
286        }
287    }
288
289    fn resolve(&self, params: Value) -> Result<Value, String> {
290        let (typed, resolved) = resolve_params::<AverageParams>(params)?;
291        typed.check()?;
292        Ok(resolved)
293    }
294
295    fn run(&self, params: &Value, inputs: &Inputs) -> Result<Outputs, String> {
296        let p: AverageParams = serde_json::from_value(params.clone()).map_err(|e| e.to_string())?;
297        let data = input(inputs, "data")?;
298        if let Artifact::Epochs(e) = data
299            && matches!(e.source(), modewise_epochs::EpochSource::Instantaneous(_))
300        {
301            return Err(
302                "averaging instantaneous frequency across channels is not meaningful".into(),
303            );
304        }
305        let groups = p.groups(channel_table(inputs)?, &data_labels(data)?)?;
306        let averaged = match data {
307            Artifact::FunctionalConnectivity(c) => {
308                Artifact::FunctionalConnectivity(average_connectivity(c, &groups)?)
309            }
310            other => combine_channels(other, &Average { groups })?,
311        };
312        Ok(Outputs::from([("averaged".into(), averaged)]))
313    }
314
315    fn output_kind(&self, port: &Port, inputs: &BTreeMap<String, ArtifactKind>) -> ArtifactKind {
316        inputs
317            .get("data")
318            .copied()
319            .unwrap_or_else(|| port.output_kind())
320    }
321}
modewise_operations/regions/mod.rs

modewise_operations/regions/
mod.rs
