add ds-common module

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "eig-aoe",
     "eig-db",
     "mems",
+    "mems/examples/ds-common",
     "mems/examples/ds-static-topo",
     "mems/examples/ds-dev-ohm-cal",
     "mems/examples/ds-dyn-topo",

mems/examples/ds-3phase-pf/src/lib.rs

@@ -7,7 +7,7 @@ use ndarray::Array2;
 
 use mems::model::{get_wasm_result, PluginInput, PluginOutput};
 
-use crate::read::{read_dev_matrix, read_dev_topo, read_dyn_topo, read_tn_input};
+use crate::read::{read_dev_ohm, read_dev_topo, read_dyn_topo, read_tn_input};
 
 mod read;
 mod nlp;
@@ -66,7 +66,7 @@ pub unsafe fn run(ptr: i32, len: u32) -> u64 {
                 }
             }
         } else if input.dfs[i] == DEV_CONDUCTOR_DF_NAME {
-            match read_dev_matrix(&mut records) {
+            match read_dev_ohm(&mut records) {
                 Ok(v) => dev_conductor = v,
                 Err(s) => {
                     error = Some(s);

mems/examples/ds-3phase-pf/src/read.rs

@@ -80,8 +80,8 @@ pub(crate) fn read_dev_topo(records: &mut StringRecordsIter<&[u8]>)
     Ok(dev_topo)
 }
 
-pub(crate) fn read_dev_matrix(records: &mut StringRecordsIter<&[u8]>)
-                              -> Result<HashMap<u64, Vec<Array2<f64>>>, String> {
+pub(crate) fn read_dev_ohm(records: &mut StringRecordsIter<&[u8]>)
+                           -> Result<HashMap<u64, Vec<Array2<f64>>>, String> {
     let mut map = HashMap::new();
     let mut dev_id = 0u64;
     let mut matrix: Vec<f64> = Vec::with_capacity(MAT_SIZE);

mems/examples/ds-common/Cargo.toml

+[package]
+name = "ds-common"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+log = "0.4"
+serde_json = "1.0"
+csv = "1.3.0"
\ No newline at end of file

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

+use csv::StringRecordsIter;
+
+pub const DYN_TOPO_DF_NAME: &str = "dyn_topo";
+pub const DEV_TOPO_DF_NAME: &str = "dev_topo";
+
+pub fn read_dyn_topo(records: &mut StringRecordsIter<&[u8]>)
+                            -> Result<Vec<Vec<u64>>, String> {
+    let mut dyn_topo = Vec::new();
+    // 按行读取csv
+    let mut row = 0;
+    loop {
+        match records.next() {
+            Some(Ok(record)) => {
+                let mut col = 0;
+                dyn_topo.push(vec![0u64; 2]);
+                for str in record.iter() {
+                    if let Ok(id) = str.parse() {
+                        dyn_topo[row][col] = id;
+                    } else {
+                        return Err(format!("Wrong dynamic topology input, row {row} col {col}"));
+                    }
+                    col += 1;
+                    if col == 2 {
+                        break;
+                    }
+                }
+                if col != 2 {
+                    return Err(format!("Wrong dynamic topology input, expected col 2, actual {col}"));
+                }
+            }
+            Some(Err(e)) => {
+                return Err(format!("Wrong dynamic topology input, err: {:?}", e));
+            }
+            None => {
+                break;
+            }
+        }
+        row += 1;
+    }
+    Ok(dyn_topo)
+}
+pub fn read_dev_topo(records: &mut StringRecordsIter<&[u8]>)
+                            -> Result<Vec<Vec<u64>>, String> {
+    let mut dev_topo = Vec::new();
+    // 按行读取csv
+    let mut row = 0;
+    loop {
+        match records.next() {
+            Some(Ok(record)) => {
+                let mut col = 0;
+                dev_topo.push(vec![0u64; 4]);
+                for str in record.iter() {
+                    if let Ok(id) = str.parse() {
+                        dev_topo[row][col] = id;
+                    } else {
+                        return Err(format!("Wrong device topology, row {row} col {col}"));
+                    }
+                    col += 1;
+                    if col == 4 {
+                        break;
+                    }
+                }
+                if col != 4 {
+                    return Err(format!("Wrong device topology input, expected col 4, actual {col}"));
+                }
+            }
+            Some(Err(e)) => {
+                return Err(format!("Wrong device topology input, err: {:?}", e));
+            }
+            None => {
+                break;
+            }
+        }
+        row += 1;
+    }
+    Ok(dev_topo)
+}

mems/examples/ds-pf-input/Cargo.toml

@@ -8,6 +8,8 @@ crate-type = ["cdylib"]
 
 [dependencies]
 log = "0.4"
+csv = "1.3.0"
 serde_cbor = "0.11"
 arrow-schema = { version = "52.1", features = ["serde"] }
+ds-common = { path = "../ds-common" }
 mems = { path = "../../../mems" }
\ No newline at end of file

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

-pub fn add(left: usize, right: usize) -> usize {
-    left + right
-}
+use arrow_schema::{DataType, Field, Schema};
+use ds_common::{DEV_TOPO_DF_NAME, DYN_TOPO_DF_NAME, read_dev_topo, read_dyn_topo};
+use mems::model::{get_meas_from_plugin_input, get_wasm_result, PluginInput, PluginOutput};
 
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn it_works() {
-        let result = add(2, 2);
-        assert_eq!(result, 4);
+#[no_mangle]
+pub unsafe fn run(ptr: i32, len: u32) -> u64 {
+    // 从内存中获取字符串
+    let input = unsafe {
+        let slice = std::slice::from_raw_parts(ptr as _, len as _);
+        let input: PluginInput = serde_cbor::from_slice(slice).unwrap();
+        input
+    };
+    let from = 0;
+    let mut error = None;
+    let r1 = get_meas_from_plugin_input(&input);
+    if let Err(s) = &r1 {
+        error = Some(s.clone());
+    }
+    let mut dyn_topo: Vec<Vec<u64>>;
+    // terminal, cn, tn, dev
+    let mut dev_topo: Vec<Vec<u64>>;
+    for i in 0..input.dfs_len.len() {
+        let size = input.dfs_len[i] as usize;
+        let end = from + size;
+        let mut rdr = csv::ReaderBuilder::new().has_headers(true).from_reader(&input.bytes[from..end]);
+        let mut records = rdr.records();
+        // 对第i个边输入该节点的 dataframe 进行处理
+        if input.dfs[i] == DYN_TOPO_DF_NAME {
+            match read_dyn_topo(&mut records) {
+                Ok(v) => dyn_topo = v,
+                Err(s) => {
+                    error = Some(s);
+                    break;
+                }
+            }
+        } else if input.dfs[i] == DEV_TOPO_DF_NAME {
+            match read_dev_topo(&mut records) {
+                Ok(v) => dev_topo = v,
+                Err(s) => {
+                    error = Some(s);
+                    break;
+                }
+            }
+        }
+    }
+    if error.is_none() {
+        let output = PluginOutput {
+            error_msg: error,
+            schema: None,
+            csv_bytes: vec![],
+        };
+        get_wasm_result(output)
+    } else {
+        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,
+        };
+        get_wasm_result(output)
     }
-}
+}
\ No newline at end of file