feat(guizhou): add guizhou example

Cargo.lock

@@ -124,7 +124,9 @@ checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401"
 dependencies = [
 dependencies = [
  "android-tzdata",
  "android-tzdata",
  "iana-time-zone",
  "iana-time-zone",
+ "js-sys",
  "num-traits",
  "num-traits",
+ "wasm-bindgen",
  "windows-targets",
  "windows-targets",
 ]
 ]
 
 
@@ -293,6 +295,19 @@ dependencies = [
 ]
 ]
 
 
 [[package]]
 [[package]]
+name = "ds-guizhou"
+version = "0.1.0"
+dependencies = [
+ "arrow-schema",
+ "bytes",
+ "chrono",
+ "ciborium",
+ "csv",
+ "log",
+ "mems",
+]
+
+[[package]]
 name = "ds-static-topo"
 name = "ds-static-topo"
 version = "0.1.0"
 version = "0.1.0"
 dependencies = [
 dependencies = [

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "mems/examples/ds-dyn-topo",
     "mems/examples/ds-dyn-topo",
     "mems/examples/ds-pf-input",
     "mems/examples/ds-pf-input",
     "mems/examples/ds-3phase-pf",
     "mems/examples/ds-3phase-pf",
+    "mems/examples/ds-guizhou",
 ]
 ]
 
 
 [workspace.package]
 [workspace.package]

mems/examples/ds-guizhou/Cargo.toml

+[package]
+name = "ds-guizhou"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+log = "0.4"
+ciborium = "0.2"
+arrow-schema = { version = "52.1", features = ["serde"] }
+mems = { path = "../../../mems" }
+csv = "1.3.0"
+bytes = "1.6.1"
+chrono = "0.4.38"
+

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

+use arrow_schema::{DataType, TimeUnit, Field, Schema};
+use bytes::{Buf, BufMut, BytesMut};
+use log::info;
+use mems::model::{PluginInput, PluginOutput};
+use chrono::{Local, Days, Timelike};
+
+static mut OUTPUT: Vec<u8> = vec![];
+#[no_mangle]
+pub unsafe fn run(ptr: i32, len: u32) -> u64 {
+    info!("Read plugin input firstly");
+    // 从内存中获取字符串
+    let input = unsafe {
+        let slice = std::slice::from_raw_parts(ptr as _, len as _);
+        let input: PluginInput = ciborium::from_reader(slice).unwrap();
+        input
+    };
+    let schema = Schema::new(vec![
+        Field::new("datetime", DataType::Timestamp(TimeUnit::Millisecond, Some("+08:00".into())), false),
+        Field::new("value", DataType::Float64, false),
+    ]);
+
+    let mut csv_str = String::from("datetime,value\n");
+
+    let mut rdr = csv::ReaderBuilder::new().has_headers(true).from_reader(&*input.bytes);
+    let records = rdr.records();
+
+    let now = Local::now();
+    let today = now.date_naive();
+    let startday = if now.hour() < 1 {
+        today
+    } else {
+        today.checked_add_days(Days::new(1)).unwrap()
+    };
+
+    let timestamp = startday.and_hms_opt(0, 0, 0)
+        .unwrap()
+        .and_local_timezone(Local)
+        .unwrap()
+        .timestamp_millis();
+
+    for (i, record) in records.enumerate() {
+        if let Ok(f) = record {
+            let s = f.get(0).unwrap().trim();
+            let value = s.parse::<f64>().unwrap();
+            let date = timestamp + (i*15*60*1000) as i64;
+            csv_str.push_str(&format!("{date}, {value}\n"));
+        }
+    }
+    log::info!("{csv_str}");
+
+    let csv_bytes = vec![("".to_string(), csv_str.into_bytes())];
+    let output = PluginOutput {
+            error_msg: None,
+            schema: Some(vec![schema]),
+            csv_bytes,
+        };
+    #[allow(static_mut_refs)]
+    ciborium::into_writer(&output, &mut OUTPUT).unwrap();
+    let offset = OUTPUT.as_ptr() as i32;
+    let len = OUTPUT.len() as u32;
+    let mut bytes = BytesMut::with_capacity(8);
+    bytes.put_i32(offset);
+    bytes.put_u32(len);
+    return bytes.get_u64();
+}