ad_core_rs/plugin/
file_controller.rs

1use std::path::PathBuf;
2use std::sync::Arc;
3
4use crate::error::{ADError, ADResult};
5use crate::ndarray::{NDArray, NDDataType, NDDimension};
6
7use super::file_base::{NDFileMode, NDFileWriter, NDPluginFileBase};
8use super::runtime::{
9    ParamChangeResult, ParamChangeValue, ParamUpdate, PluginParamSnapshot, ProcessResult,
10};
11
12/// Param indices for file plugin control (looked up once at registration time).
13#[derive(Default)]
14pub struct FileParamIndices {
15    pub file_path: Option<usize>,
16    pub file_name: Option<usize>,
17    pub file_number: Option<usize>,
18    pub file_template: Option<usize>,
19    pub auto_increment: Option<usize>,
20    pub write_file: Option<usize>,
21    pub read_file: Option<usize>,
22    pub write_mode: Option<usize>,
23    pub num_capture: Option<usize>,
24    pub capture: Option<usize>,
25    pub auto_save: Option<usize>,
26    pub create_dir: Option<usize>,
27    pub file_path_exists: Option<usize>,
28    pub write_status: Option<usize>,
29    pub write_message: Option<usize>,
30    pub full_file_name: Option<usize>,
31    pub file_temp_suffix: Option<usize>,
32    pub num_captured: Option<usize>,
33    pub lazy_open: Option<usize>,
34    pub delete_driver_file: Option<usize>,
35    pub free_capture: Option<usize>,
36}
37
38/// Generic file plugin controller that wraps any NDFileWriter with the full
39/// C ADCore NDPluginFile control-plane logic: auto_save, capture, stream,
40/// temp_suffix rename, create_dir, param updates, error reporting.
41///
42/// Each file format plugin (TIFF, HDF5, JPEG) creates one of these and
43/// delegates `process_array`, `register_params`, and `on_param_change` to it.
44pub struct FilePluginController<W: NDFileWriter> {
45    pub file_base: NDPluginFileBase,
46    pub writer: W,
47    pub params: FileParamIndices,
48    pub auto_save: bool,
49    pub capture_active: bool,
50    pub lazy_open: bool,
51    pub delete_driver_file: bool,
52    pub latest_array: Option<Arc<NDArray>>,
53}
54
55impl<W: NDFileWriter> FilePluginController<W> {
56    pub fn new(writer: W) -> Self {
57        Self {
58            file_base: NDPluginFileBase::new(),
59            writer,
60            params: FileParamIndices::default(),
61            auto_save: false,
62            capture_active: false,
63            lazy_open: false,
64            delete_driver_file: false,
65            latest_array: None,
66        }
67    }
68
69    /// Look up all standard file param indices from the port driver base.
70    pub fn register_params(
71        &mut self,
72        base: &mut asyn_rs::port::PortDriverBase,
73    ) -> asyn_rs::error::AsynResult<()> {
74        self.params.file_path = base.find_param("FILE_PATH");
75        self.params.file_name = base.find_param("FILE_NAME");
76        self.params.file_number = base.find_param("FILE_NUMBER");
77        self.params.file_template = base.find_param("FILE_TEMPLATE");
78        self.params.auto_increment = base.find_param("AUTO_INCREMENT");
79        self.params.write_file = base.find_param("WRITE_FILE");
80        self.params.read_file = base.find_param("READ_FILE");
81        self.params.write_mode = base.find_param("WRITE_MODE");
82        self.params.num_capture = base.find_param("NUM_CAPTURE");
83        self.params.capture = base.find_param("CAPTURE");
84        self.params.auto_save = base.find_param("AUTO_SAVE");
85        self.params.create_dir = base.find_param("CREATE_DIR");
86        self.params.file_path_exists = base.find_param("FILE_PATH_EXISTS");
87        self.params.write_status = base.find_param("WRITE_STATUS");
88        self.params.write_message = base.find_param("WRITE_MESSAGE");
89        self.params.full_file_name = base.find_param("FULL_FILE_NAME");
90        self.params.file_temp_suffix = base.find_param("FILE_TEMP_SUFFIX");
91        self.params.num_captured = base.find_param("NUM_CAPTURED");
92        self.params.lazy_open = base.find_param("FILE_LAZY_OPEN");
93        self.params.delete_driver_file = base.find_param("DELETE_DRIVER_FILE");
94        self.params.free_capture = base.find_param("FREE_CAPTURE");
95        Ok(())
96    }
97
98    /// Process an incoming array: auto_save, capture buffering, stream write.
99    pub fn process_array(&mut self, array: &NDArray) -> ProcessResult {
100        let mut proc_result = ProcessResult::empty();
101        let array = Arc::new(array.clone());
102        self.latest_array = Some(array.clone());
103
104        let result = match self.file_base.mode() {
105            NDFileMode::Single => {
106                if self.auto_save {
107                    self.write_single(array)
108                } else {
109                    Ok(())
110                }
111            }
112            NDFileMode::Capture => {
113                if self.capture_active {
114                    self.file_base.capture_array(array);
115                    self.push_num_captured_update(&mut proc_result.param_updates);
116                    if self.file_base.num_captured() >= self.file_base.num_capture_target() {
117                        if self.auto_save {
118                            if let Err(err) = self.file_base.flush_capture(&mut self.writer) {
119                                Err(err)
120                            } else {
121                                self.push_full_file_name_update(&mut proc_result.param_updates);
122                                self.push_num_captured_update(&mut proc_result.param_updates);
123                                self.capture_active = false;
124                                Ok(())
125                            }
126                        } else {
127                            self.capture_active = false;
128                            Ok(())
129                        }
130                    } else {
131                        Ok(())
132                    }
133                } else {
134                    Ok(())
135                }
136            }
137            NDFileMode::Stream => {
138                if self.capture_active {
139                    let r = self.file_base.process_array(array, &mut self.writer);
140                    let target = self.file_base.num_capture_target();
141                    if r.is_ok() && target > 0 && self.file_base.num_captured() >= target {
142                        if let Err(e) = self.file_base.close_stream(&mut self.writer) {
143                            return ProcessResult::sink(self.error_updates(
144                                false,
145                                false,
146                                e.to_string(),
147                            ));
148                        }
149                        self.capture_active = false;
150                        self.push_full_file_name_update(&mut proc_result.param_updates);
151                        self.push_num_captured_update(&mut proc_result.param_updates);
152                    }
153                    r
154                } else {
155                    Ok(())
156                }
157            }
158        };
159
160        if result.is_ok() {
161            proc_result.param_updates.extend(self.success_updates());
162            if self.file_base.mode() == NDFileMode::Single && self.auto_save {
163                self.push_full_file_name_update(&mut proc_result.param_updates);
164            }
165            if self.file_base.mode() == NDFileMode::Stream && self.capture_active {
166                self.push_full_file_name_update(&mut proc_result.param_updates);
167            }
168        } else if let Err(err) = result {
169            proc_result.param_updates = self.error_updates(false, false, err.to_string());
170        }
171        proc_result
172    }
173
174    /// Handle a control-plane param change. Returns true if the reason was handled.
175    pub fn on_param_change(
176        &mut self,
177        reason: usize,
178        params: &PluginParamSnapshot,
179    ) -> ParamChangeResult {
180        let mut updates = Vec::new();
181
182        if Some(reason) == self.params.file_path {
183            if let ParamChangeValue::Octet(s) = &params.value {
184                let normalized = normalize_file_path(s);
185                self.file_base.file_path = normalized.clone();
186                let exists =
187                    std::path::Path::new(normalized.trim_end_matches(std::path::MAIN_SEPARATOR))
188                        .is_dir();
189                if let Some(idx) = self.params.file_path_exists {
190                    updates.push(ParamUpdate::Int32 {
191                        reason: idx,
192                        addr: 0,
193                        value: if exists { 1 } else { 0 },
194                    });
195                }
196            }
197        } else if Some(reason) == self.params.file_name {
198            if let ParamChangeValue::Octet(s) = &params.value {
199                self.file_base.file_name = s.clone();
200            }
201        } else if Some(reason) == self.params.file_number {
202            self.file_base.file_number = params.value.as_i32();
203        } else if Some(reason) == self.params.file_template {
204            if let ParamChangeValue::Octet(s) = &params.value {
205                self.file_base.file_template = s.clone();
206            }
207        } else if Some(reason) == self.params.auto_increment {
208            self.file_base.auto_increment = params.value.as_i32() != 0;
209        } else if Some(reason) == self.params.auto_save {
210            self.auto_save = params.value.as_i32() != 0;
211        } else if Some(reason) == self.params.write_mode {
212            self.file_base
213                .set_mode(NDFileMode::from_i32(params.value.as_i32()));
214        } else if Some(reason) == self.params.num_capture {
215            self.file_base
216                .set_num_capture(params.value.as_i32().max(1) as usize);
217        } else if Some(reason) == self.params.create_dir {
218            self.file_base.create_dir = params.value.as_i32();
219        } else if Some(reason) == self.params.file_temp_suffix {
220            if let ParamChangeValue::Octet(s) = &params.value {
221                self.file_base.temp_suffix = s.clone();
222            }
223        } else if Some(reason) == self.params.write_file {
224            if params.value.as_i32() != 0 {
225                let result = match self.file_base.mode() {
226                    NDFileMode::Single => {
227                        if let Some(array) = self.latest_array.clone() {
228                            self.write_single(array)
229                        } else {
230                            Err(ADError::UnsupportedConversion(
231                                "no array available for write".into(),
232                            ))
233                        }
234                    }
235                    NDFileMode::Capture => self.file_base.flush_capture(&mut self.writer),
236                    NDFileMode::Stream => {
237                        if let Some(array) = self.latest_array.clone() {
238                            self.file_base.process_array(array, &mut self.writer)
239                        } else {
240                            Err(ADError::UnsupportedConversion(
241                                "no array available for write".into(),
242                            ))
243                        }
244                    }
245                };
246                match result {
247                    Ok(()) => {
248                        updates.extend(self.success_updates());
249                        self.push_num_captured_update(&mut updates);
250                        self.push_full_file_name_update(&mut updates);
251                    }
252                    Err(err) => {
253                        return ParamChangeResult::updates(self.error_updates(
254                            false,
255                            true,
256                            err.to_string(),
257                        ));
258                    }
259                }
260            }
261        } else if Some(reason) == self.params.read_file {
262            if params.value.as_i32() != 0 {
263                let result = (|| -> ADResult<Arc<NDArray>> {
264                    let path = PathBuf::from(self.file_base.create_file_name());
265                    self.writer.open_file(
266                        &path,
267                        NDFileMode::Single,
268                        &NDArray::new(vec![NDDimension::new(1)], NDDataType::UInt8),
269                    )?;
270                    let array = Arc::new(self.writer.read_file()?);
271                    self.writer.close_file()?;
272                    self.latest_array = Some(array.clone());
273                    Ok(array)
274                })();
275                match result {
276                    Ok(array) => {
277                        updates.extend(self.success_updates());
278                        self.push_full_file_name_update(&mut updates);
279                        return ParamChangeResult::combined(vec![array], updates);
280                    }
281                    Err(err) => {
282                        return ParamChangeResult::updates(self.error_updates(
283                            true,
284                            false,
285                            err.to_string(),
286                        ));
287                    }
288                }
289            }
290        } else if Some(reason) == self.params.lazy_open {
291            self.lazy_open = params.value.as_i32() != 0;
292        } else if Some(reason) == self.params.delete_driver_file {
293            self.delete_driver_file = params.value.as_i32() != 0;
294        } else if Some(reason) == self.params.free_capture {
295            if params.value.as_i32() != 0 {
296                self.file_base.clear_capture();
297                self.push_num_captured_update(&mut updates);
298            }
299        } else if Some(reason) == self.params.capture {
300            if params.value.as_i32() != 0 {
301                match self.file_base.mode() {
302                    NDFileMode::Single => {
303                        self.capture_active = false;
304                        return ParamChangeResult::updates(self.error_updates(
305                            false,
306                            false,
307                            "ERROR: capture not supported in Single mode".into(),
308                        ));
309                    }
310                    NDFileMode::Capture => {
311                        self.file_base.clear_capture();
312                        self.capture_active = true;
313                        self.push_num_captured_update(&mut updates);
314                    }
315                    NDFileMode::Stream => {
316                        self.capture_active = true;
317                        self.push_num_captured_update(&mut updates);
318                    }
319                }
320            } else {
321                if self.file_base.mode() == NDFileMode::Stream {
322                    if let Err(err) = self.file_base.close_stream(&mut self.writer) {
323                        return ParamChangeResult::updates(self.error_updates(
324                            false,
325                            false,
326                            err.to_string(),
327                        ));
328                    }
329                }
330                self.capture_active = false;
331            }
332        }
333
334        ParamChangeResult::updates(updates)
335    }
336
337    // ── helpers ──
338
339    fn write_single(&mut self, array: Arc<NDArray>) -> ADResult<()> {
340        self.file_base.ensure_directory()?;
341        self.file_base.process_array(array, &mut self.writer)
342    }
343
344    fn success_updates(&self) -> Vec<ParamUpdate> {
345        let mut updates = Vec::new();
346        if let Some(idx) = self.params.file_number {
347            updates.push(ParamUpdate::Int32 {
348                reason: idx,
349                addr: 0,
350                value: self.file_base.file_number,
351            });
352        }
353        if let Some(idx) = self.params.write_status {
354            updates.push(ParamUpdate::Int32 {
355                reason: idx,
356                addr: 0,
357                value: 0,
358            });
359        }
360        if let Some(idx) = self.params.write_message {
361            updates.push(ParamUpdate::Octet {
362                reason: idx,
363                addr: 0,
364                value: String::new(),
365            });
366        }
367        if let Some(idx) = self.params.write_file {
368            updates.push(ParamUpdate::Int32 {
369                reason: idx,
370                addr: 0,
371                value: 0,
372            });
373        }
374        if let Some(idx) = self.params.capture {
375            updates.push(ParamUpdate::Int32 {
376                reason: idx,
377                addr: 0,
378                value: if self.capture_active { 1 } else { 0 },
379            });
380        }
381        if let Some(idx) = self.params.read_file {
382            updates.push(ParamUpdate::Int32 {
383                reason: idx,
384                addr: 0,
385                value: 0,
386            });
387        }
388        updates
389    }
390
391    fn push_num_captured_update(&self, updates: &mut Vec<ParamUpdate>) {
392        if let Some(idx) = self.params.num_captured {
393            updates.push(ParamUpdate::Int32 {
394                reason: idx,
395                addr: 0,
396                value: self.file_base.num_captured() as i32,
397            });
398        }
399    }
400
401    fn push_full_file_name_update(&self, updates: &mut Vec<ParamUpdate>) {
402        if let Some(idx) = self.params.full_file_name {
403            updates.push(ParamUpdate::Octet {
404                reason: idx,
405                addr: 0,
406                value: self.file_base.last_written_name().to_string(),
407            });
408        }
409    }
410
411    fn error_updates(
412        &self,
413        read_reason: bool,
414        write_reason: bool,
415        message: String,
416    ) -> Vec<ParamUpdate> {
417        let mut updates = Vec::new();
418        if write_reason {
419            if let Some(idx) = self.params.write_file {
420                updates.push(ParamUpdate::Int32 {
421                    reason: idx,
422                    addr: 0,
423                    value: 0,
424                });
425            }
426        }
427        if read_reason {
428            if let Some(idx) = self.params.read_file {
429                updates.push(ParamUpdate::Int32 {
430                    reason: idx,
431                    addr: 0,
432                    value: 0,
433                });
434            }
435        }
436        if let Some(idx) = self.params.write_status {
437            updates.push(ParamUpdate::Int32 {
438                reason: idx,
439                addr: 0,
440                value: 1,
441            });
442        }
443        if let Some(idx) = self.params.write_message {
444            updates.push(ParamUpdate::Octet {
445                reason: idx,
446                addr: 0,
447                value: message,
448            });
449        }
450        updates
451    }
452}
453
454fn normalize_file_path(path: &str) -> String {
455    if path.is_empty() || path.ends_with(std::path::MAIN_SEPARATOR) {
456        path.to_string()
457    } else {
458        format!("{path}{}", std::path::MAIN_SEPARATOR)
459    }
460}
ad_core_rs/plugin/file_controller.rs

ad_core_rs/plugin/
file_controller.rs
