prepare of tn input collection

mems/examples/ds-common/src/lib.rs

 pub mod dyn_topo;
 pub mod static_topo;
+pub mod tn_input;
 
 pub const STATIC_TOPO_DF_NAME: &str = "static_topo";
 pub const TERMINAL_DF_NAME: &str = "terminal_cn_dev";

mems/examples/ds-common/src/tn_input.rs

+use std::collections::HashMap;
+
+use csv::StringRecordsIter;
+
+use mems::model::dev::MeasPhase;
+
+pub fn read_shunt_measures(records: &mut StringRecordsIter<&[u8]>)
+                           -> Result<HashMap<u64, (u64, MeasPhase)>, String> {
+    let mut meas = HashMap::new();
+    // 按行读取csv
+    let mut row = 0;
+    loop {
+        match records.next() {
+            Some(Ok(record)) => {
+                let mut col = 0;
+                let mut point = 0u64;
+                let mut terminal = 0u64;
+                for str in record.iter() {
+                    if col == 0 {
+                        if let Ok(id) = str.parse() {
+                            point = id;
+                        } else {
+                            return Err(format!("Wrong terminal input, row {row} col {col}"));
+                        }
+                    } else if col == 1 {
+                        if let Ok(id) = str.parse() {
+                            terminal = id;
+                        } else {
+                            return Err(format!("Wrong terminal input, row {row} col {col}"));
+                        }
+                    } else if col == 2 {
+                        meas.insert(point, (terminal, MeasPhase::from(str)));
+                    }
+                    col += 1;
+                    if col == 3 {
+                        break;
+                    }
+                }
+                if col != 3 {
+                    return Err(format!("Wrong terminal input, expected col at least 3, actual {col}"));
+                }
+            }
+            Some(Err(e)) => {
+                return Err(format!("Wrong terminal input, err: {:?}", e));
+            }
+            None => {
+                break;
+            }
+        }
+        row += 1;
+    }
+    Ok(meas)
+}
\ No newline at end of file

mems/examples/ds-pf-input/src/lib.rs

 use std::collections::{HashMap, HashSet};
 use arrow_schema::{DataType, Field, Schema};
 
-use ds_common::{DEV_TOPO_DF_NAME, DYN_TOPO_DF_NAME, POINT_DF_NAME, STATIC_TOPO_DF_NAME, TERMINAL_DF_NAME};
+use ds_common::{DEV_TOPO_DF_NAME, DYN_TOPO_DF_NAME, POINT_DF_NAME, SHUNT_MEAS_DF_NAME, STATIC_TOPO_DF_NAME, TERMINAL_DF_NAME};
 use ds_common::dyn_topo::{read_dev_topo, read_dyn_topo};
 use ds_common::static_topo::{read_point_terminal, read_static_topo, read_terminal_cn_dev};
+use ds_common::tn_input::read_shunt_measures;
 use eig_domain::DataUnit;
-use mems::model::{get_meas_from_plugin_input, get_wasm_result, PluginInput, PluginOutput};
+use mems::model::{get_df_from_in_plugin, get_meas_from_plugin_input, get_wasm_result, PluginInput, PluginOutput};
 use mems::model::dev::{MeasPhase, PsRsrType};
 
 #[no_mangle]
@@ -16,12 +17,8 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
         let input: PluginInput = serde_cbor::from_slice(slice).unwrap();
         input
     };
-    let from = 0;
+    let r2 = get_df_from_in_plugin(&input);
     let mut error = None;
-    let r1 = get_meas_from_plugin_input(&input);
-    if let Err(s) = &r1 {
-        error = Some(s.clone());
-    }
     // static topo
     // point, terminal
     let mut points: Vec<Vec<u64>> = vec![];
@@ -34,6 +31,13 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
     let mut dyn_topo: Vec<Vec<u64>>;
     // terminal, cn, tn, dev
     let mut dyn_dev_topo: Vec<Vec<u64>> = vec![];
+    // key is point id, value is (terminal id, measure phase)
+    let mut point_terminal: HashMap<u64, (u64, MeasPhase)> = HashMap::with_capacity(0);
+    let mut with_static = false;
+    if let Err(s) = &r2 {
+        error = Some(s.clone());
+    } else {
+        let mut from = r2.unwrap();
         for i in 0..input.dfs_len.len() {
             let size = input.dfs_len[i] as usize;
             let end = from + size;
@@ -57,6 +61,7 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                     }
                 }
             } else if input.dfs[i] == STATIC_TOPO_DF_NAME {
+                with_static = true;
                 match read_static_topo(&mut records, None, Some(&mut dev_type)) {
                     Ok(_) => {},
                     Err(s) => error = Some(s),
@@ -71,9 +76,15 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                     Ok(v) => points = v,
                     Err(s) => error = Some(s),
                 }
+            } else if input.dfs[i] == SHUNT_MEAS_DF_NAME {
+                match read_shunt_measures(&mut records) {
+                    Ok(v) => point_terminal = v,
+                    Err(s) => error = Some(s),
+                }
+            }
         }
     }
-    if error.is_none() {
+    if error.is_some() {
         let output = PluginOutput {
             error_msg: error,
             schema: None,
@@ -81,7 +92,8 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
         };
         get_wasm_result(output)
     } else {
-        let mut csv_str = String::from("cn,tn\n");
+        if with_static {
+            let mut csv_str = String::from("point,terminal,phase\n");
             let type1 = PsRsrType::SyncGenerator as u16;
             let type2 = PsRsrType::Load as u16;
             let type3 = PsRsrType::ShuntCompensator as u16;
@@ -106,7 +118,32 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                     point_terminal.insert(point_id, (terminal, meas_phase[i].clone()));
                 }
             }
-
+            // build schema
+            let schema = Schema::new(vec![
+                Field::new("point", DataType::UInt64, false),
+                Field::new("terminal", DataType::UInt64, false),
+                Field::new("phase", DataType::Utf8, false),
+            ]);
+            let csv_bytes = vec![(SHUNT_MEAS_DF_NAME.to_string(), csv_str.into_bytes())];
+            let output = PluginOutput {
+                error_msg: None,
+                schema: Some(vec![schema]),
+                csv_bytes,
+            };
+            get_wasm_result(output)
+        } else {
+            let r1 = get_meas_from_plugin_input(&input);
+            if let Err(s) = &r1 {
+                error = Some(s.clone());
+            }
+            if error.is_some() {
+                let output = PluginOutput {
+                    error_msg: error,
+                    schema: None,
+                    csv_bytes: vec![],
+                };
+                get_wasm_result(output)
+            } else {
                 let (meas, units) = r1.unwrap();
                 for v in dyn_dev_topo {
                     let terminal = v[0];
@@ -117,7 +154,6 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                 // 开始处理开关量
                 for m in meas {
                     if let Some((terminal, phase)) = point_terminal.get(&m.point_id) {
-                if terminal_with_shunt_dev.contains(terminal) {
                         if let Some(unit) = units.get(&m.point_id) {
                             match unit {
                                 DataUnit::A => {}
@@ -134,19 +170,14 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                         }
                     }
                 }
-        }
                 let mut csv_str = String::from("cn,tn\n");
-        // build schema
-        let schema = Schema::new(vec![
-            Field::new("cn", DataType::UInt64, false),
-            Field::new("tn", DataType::UInt64, false),
-        ]);
-        let csv_bytes = vec![("".to_string(), csv_str.into_bytes())];
                 let output = PluginOutput {
-            error_msg: None,
-            schema: Some(vec![schema]),
-            csv_bytes,
+                    error_msg: error,
+                    schema: None,
+                    csv_bytes: vec![],
                 };
                 get_wasm_result(output)
             }
+        }
+    }
 }
\ No newline at end of file