ad_plugins_rs/
gather.rs

1use std::sync::Arc;
2use std::sync::atomic::{AtomicU64, Ordering};
3
4use ad_core_rs::ndarray::NDArray;
5use ad_core_rs::ndarray_pool::NDArrayPool;
6use ad_core_rs::plugin::runtime::{
7    NDPluginProcess, ParamChangeResult, ParamUpdate, PluginParamSnapshot, ProcessResult,
8};
9use parking_lot::Mutex;
10
11/// Maximum number of gather input ports.
12pub const MAX_GATHER_PORTS: usize = 8;
13
14/// Per-port param indices for one gather source.
15#[derive(Debug, Clone, Copy, Default)]
16struct GatherPortParams {
17    /// Param index for GATHER_NDARRAY_PORT_N (Octet).
18    port_idx: Option<usize>,
19    /// Param index for GATHER_NDARRAY_ADDR_N (Int32).
20    addr_idx: Option<usize>,
21}
22
23/// Pure gather processing logic: merges arrays from multiple upstream ports
24/// into a single output stream.
25///
26/// Multi-source subscription is achieved at the IOC wiring level:
27/// `NDGatherConfigure` registers the same `NDArraySender` with multiple
28/// upstream `NDArrayOutput`s, so arrays from any configured source arrive
29/// on the plugin's single input channel.
30///
31/// The processor stores the configured source port names and addresses as
32/// params (GATHER_NDARRAY_PORT_1..8, GATHER_NDARRAY_ADDR_1..8) for
33/// introspection and runtime reconfiguration via PVs.
34pub struct GatherProcessor {
35    /// Total arrays received across all sources.
36    count: AtomicU64,
37    /// The configured sources, rewritten by a `GATHER_NDARRAY_*` param write
38    /// while the frame path may be running.
39    sources: Mutex<GatherSources>,
40    /// Param indices for per-port params.
41    port_params: [GatherPortParams; MAX_GATHER_PORTS],
42    /// Param index for GATHER_NUM_PORTS.
43    num_ports_idx: Option<usize>,
44}
45
46/// The configured source set: the port names, their addresses, and how many
47/// of the names are non-empty.
48#[derive(Default)]
49struct GatherSources {
50    /// Number of configured source ports (set during construction or param change).
51    num_ports: usize,
52    /// Configured source port names (indexed 0..MAX_GATHER_PORTS-1).
53    ports: [String; MAX_GATHER_PORTS],
54    /// Configured source addresses (indexed 0..MAX_GATHER_PORTS-1).
55    addrs: [i32; MAX_GATHER_PORTS],
56}
57
58impl GatherProcessor {
59    pub fn new() -> Self {
60        Self {
61            count: AtomicU64::new(0),
62            sources: Mutex::new(GatherSources::default()),
63            port_params: [GatherPortParams::default(); MAX_GATHER_PORTS],
64            num_ports_idx: None,
65        }
66    }
67
68    /// Create a GatherProcessor pre-configured with the given source port names.
69    pub fn with_ports(ports: &[&str]) -> Self {
70        let proc = Self::new();
71        let n = ports.len().min(MAX_GATHER_PORTS);
72        {
73            let mut src = proc.sources.lock();
74            src.num_ports = n;
75            for (i, &name) in ports.iter().take(n).enumerate() {
76                src.ports[i] = name.to_string();
77            }
78        }
79        proc
80    }
81
82    pub fn total_received(&self) -> u64 {
83        self.count.load(Ordering::Relaxed)
84    }
85
86    /// Number of configured source ports.
87    pub fn num_ports(&self) -> usize {
88        self.sources.lock().num_ports
89    }
90
91    /// Get the configured source port name for the given index (0-based).
92    pub fn source_port(&self, index: usize) -> String {
93        if index < MAX_GATHER_PORTS {
94            self.sources.lock().ports[index].clone()
95        } else {
96            String::new()
97        }
98    }
99}
100
101impl Default for GatherProcessor {
102    fn default() -> Self {
103        Self::new()
104    }
105}
106
107impl NDPluginProcess for GatherProcessor {
108    fn process_array(&self, array: &NDArray, _pool: &NDArrayPool) -> ProcessResult {
109        self.count.fetch_add(1, Ordering::Relaxed);
110        ProcessResult::arrays(vec![Arc::new(array.clone())])
111    }
112
113    fn plugin_type(&self) -> &str {
114        "NDPluginGather"
115    }
116
117    fn register_params(
118        &mut self,
119        base: &mut asyn_rs::port::PortDriverBase,
120    ) -> asyn_rs::error::AsynResult<()> {
121        use asyn_rs::param::ParamType;
122
123        // Register per-port params and store their indices
124        for i in 0..MAX_GATHER_PORTS {
125            let port_name = format!("GATHER_NDARRAY_PORT_{}", i + 1);
126            let addr_name = format!("GATHER_NDARRAY_ADDR_{}", i + 1);
127            base.create_param(&port_name, ParamType::Octet)?;
128            base.create_param(&addr_name, ParamType::Int32)?;
129            self.port_params[i].port_idx = base.find_param(&port_name);
130            self.port_params[i].addr_idx = base.find_param(&addr_name);
131        }
132
133        // Register aggregate param for number of configured ports
134        base.create_param("GATHER_NUM_PORTS", ParamType::Int32)?;
135        self.num_ports_idx = base.find_param("GATHER_NUM_PORTS");
136
137        Ok(())
138    }
139
140    fn on_param_change(&self, reason: usize, params: &PluginParamSnapshot) -> ParamChangeResult {
141        // Check if this is a GATHER_NDARRAY_PORT_N change
142        for i in 0..MAX_GATHER_PORTS {
143            if Some(reason) == self.port_params[i].port_idx {
144                if let Some(new_port) = params.value.as_string() {
145                    let mut src = self.sources.lock();
146                    src.ports[i] = new_port.to_string();
147                    // Recount active ports
148                    src.num_ports = src.ports.iter().filter(|s| !s.is_empty()).count();
149                    let n = src.num_ports as i32;
150                    drop(src);
151                    if let Some(idx) = self.num_ports_idx {
152                        return ParamChangeResult::updates(vec![ParamUpdate::int32(idx, n)]);
153                    }
154                }
155                return ParamChangeResult::empty();
156            }
157            if Some(reason) == self.port_params[i].addr_idx {
158                self.sources.lock().addrs[i] = params.value.as_i32();
159                return ParamChangeResult::empty();
160            }
161        }
162
163        ParamChangeResult::empty()
164    }
165}
166
167#[cfg(test)]
168mod tests {
169    use super::*;
170    use ad_core_rs::ndarray::{NDDataType, NDDimension};
171
172    #[test]
173    fn test_gather_processor_passthrough() {
174        let proc = GatherProcessor::new();
175        let pool = NDArrayPool::new(1_000_000);
176
177        let arr1 = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
178        let arr2 = NDArray::new(vec![NDDimension::new(4)], NDDataType::UInt8);
179
180        let result1 = proc.process_array(&arr1, &pool);
181        let result2 = proc.process_array(&arr2, &pool);
182
183        assert_eq!(result1.output_arrays.len(), 1);
184        assert_eq!(result2.output_arrays.len(), 1);
185        assert_eq!(proc.total_received(), 2);
186    }
187
188    #[test]
189    fn test_gather_with_ports() {
190        let proc = GatherProcessor::with_ports(&["SIM1", "SIM2", "SIM3"]);
191        assert_eq!(proc.num_ports(), 3);
192        assert_eq!(proc.source_port(0), "SIM1");
193        assert_eq!(proc.source_port(1), "SIM2");
194        assert_eq!(proc.source_port(2), "SIM3");
195        assert_eq!(proc.source_port(3), "");
196    }
197
198    #[test]
199    fn test_gather_multi_source_counting() {
200        let proc = GatherProcessor::with_ports(&["DRV1", "DRV2"]);
201        let pool = NDArrayPool::new(1_000_000);
202
203        // Simulate arrays arriving from different sources (all arrive on same channel)
204        for _ in 0..5 {
205            let arr = NDArray::new(vec![NDDimension::new(10)], NDDataType::UInt16);
206            proc.process_array(&arr, &pool);
207        }
208
209        assert_eq!(proc.total_received(), 5);
210    }
211
212    #[test]
213    fn test_gather_default() {
214        let proc = GatherProcessor::default();
215        assert_eq!(proc.total_received(), 0);
216        assert_eq!(proc.num_ports(), 0);
217    }
218
219    #[test]
220    fn test_gather_max_ports_clamped() {
221        // More ports than MAX should be clamped
222        let names: Vec<&str> = (0..12).map(|_| "PORT").collect();
223        let proc = GatherProcessor::with_ports(&names);
224        assert_eq!(proc.num_ports(), MAX_GATHER_PORTS);
225    }
226}
ad_plugins_rs/gather.rs

ad_plugins_rs/
gather.rs
