init

.env.example

@@ -0,0 +1,3 @@
+DATABASE_URL=mysql://user:pass@example.com/database
+USB_PORT=/dev/ttyUSB0
+INSERT_EVERY_NTH=10

.gitignore

@@ -0,0 +1,2 @@
+/target
+.env

Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "stromsensor"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+dotenvy = "0.15"
+serial = "0.4"
+sqlx = { version = "0.6", features = ["runtime-tokio-rustls", "mysql"], default-features=false }
+tokio = { version = "1", features = ["full"]}

src/main.rs

@@ -0,0 +1,120 @@
+use std::{
+    env, io,
+    time::{Duration, Instant},
+};
+
+use serial::{CharSize, Parity, SerialPort, StopBits};
+use sqlx::{Connection, MySqlConnection};
+
+const PATTERN_COUNT: usize = 1;
+const PATTERN: [[u8; 19]; PATTERN_COUNT] = [[
+    0x77, 0x07, 0x01, 0x00, 0x01, 0x08, 0x00, 0xFF, 0x64, 0x00, 0x01, 0x82, 0x01, 0x62, 0x1E, 0x52,
+    0xFF, 0x56, 0x00,
+]];
+
+#[tokio::main]
+async fn main() {
+    dotenvy::dotenv().ok();
+    let mut port =
+        serial::open(&env::var("USB_PORT").expect("USB_PORT environment variable")).unwrap();
+    let mut connection = MySqlConnection::connect(
+        &env::var("DATABASE_URL").expect("DATABASE_URL environment variable"),
+    )
+    .await
+    .unwrap();
+
+    interact(&mut port, &mut connection).await.unwrap();
+}
+
+async fn interact<T: SerialPort>(port: &mut T, connection: &mut MySqlConnection) -> io::Result<()> {
+    port.reconfigure(&|settings| {
+        settings.set_baud_rate(serial::Baud9600).unwrap();
+        settings.set_parity(Parity::ParityNone);
+        settings.set_char_size(CharSize::Bits8);
+        settings.set_stop_bits(StopBits::Stop1);
+        Ok(())
+    })
+    .unwrap();
+
+    port.set_timeout(Duration::from_millis(250)).unwrap();
+
+    let mut read_buf: [u8; 1024] = [0; 1024];
+    let mut queue: Vec<u8> = vec![];
+
+    let mut prev_value = 0;
+    let mut prev_time = Instant::now();
+
+    let mut brake = 0_u32;
+    let brake_target: u32 = env::var("INSERT_EVERY_NTH")
+        .expect("INSERT_EVERY_NTH environment variable")
+        .parse()
+        .expect("Number in INSERT_EVERY_NTH");
+
+    loop {
+        let read_result = port.read(&mut read_buf);
+        if let Ok(queue_length) = read_result {
+            queue.append(
+                &mut read_buf
+                    .iter()
+                    .take(queue_length)
+                    .copied()
+                    .collect::<Vec<u8>>(),
+            );
+        } else if !queue.is_empty() && brake >= brake_target {
+            let result = extract_values(&mut queue);
+
+            if prev_value > 0 {
+                let val =
+                    ((result[0] - prev_value) * 3600) as f64 / (prev_time.elapsed().as_secs_f64());
+                if val < 0.0 {
+                    println!("received invalid data, dropping it");
+                    continue;
+                }
+
+                println!(
+                    "Zählerstand: {} Wh, Aktueller Verbrauch: {} W",
+                    result[0] / 10,
+                    val / 10.0
+                );
+                sqlx::query("INSERT INTO counter_value (timestamp, value) VALUES (NOW(), ?)")
+                    .bind(result[0] / 10)
+                    .execute(&mut *connection)
+                    .await
+                    .unwrap();
+                sqlx::query("INSERT INTO current_power (timestamp, value) VALUES (NOW(), ?)")
+                    .bind(val as u32 / 10)
+                    .execute(&mut *connection)
+                    .await
+                    .unwrap();
+            }
+
+            prev_value = result[0];
+            prev_time = Instant::now();
+            brake = 0;
+        } else if !queue.is_empty() {
+            queue.clear();
+            brake += 1;
+        }
+    }
+}
+
+fn extract_values(queue: &mut Vec<u8>) -> [i32; PATTERN_COUNT] {
+    let mut pattern_index: [usize; PATTERN_COUNT] = [0; PATTERN_COUNT];
+    let mut value: [i32; PATTERN_COUNT] = [0; PATTERN_COUNT];
+    for queue_item in queue.iter() {
+        for i in 0..PATTERN_COUNT {
+            if pattern_index[i] >= 23 {
+            } else if pattern_index[i] >= 19 {
+                let e = *queue_item as i32;
+                value[i] += e << ((22 - pattern_index[i]) * 8);
+                pattern_index[i] += 1;
+            } else if *queue_item == PATTERN[i][pattern_index[i]] {
+                pattern_index[i] += 1;
+            } else {
+                pattern_index[i] = 0;
+            }
+        }
+    }
+    queue.clear();
+    value
+}

stromsensor.service

@@ -0,0 +1,10 @@
+[Unit]
+Name=Stromsensor
+Description=SML SmartMeter measurements to SQL
+After=network-online.target
+
+[Service]
+User=stromsensor
+User=stromsensor
+ExecStart=/opt/stromsensor
+EnvironmentFile=/opt/stromsensor-env