first commit

.env.example

@@ -0,0 +1,5 @@
+DB_USER=bd_user
+DB_PASSWORD=bd_password
+DB_HOST=localhost
+DB_PORT=3306
+DB_NAME=dn_name

.gitignore

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

Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "rust_agreagator"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+chrono = "0.4"
+dotenv = "0.15"
+mysql = "25.0.1"
+serde = { version = "1.0", features = ["derive"] }
+serde_derive = "1.0"

src/main.rs

@@ -0,0 +1,130 @@
+extern crate mysql;
+extern crate serde;
+extern crate serde_derive;
+
+use dotenv::dotenv;
+use std::env;
+use mysql::*;
+use mysql::prelude::*;
+use serde::Deserialize;
+extern crate chrono;
+
+use chrono::{Duration, Utc, NaiveDate};
+use chrono::prelude::*;
+
+#[derive(Debug, Deserialize)]
+struct Sensor {
+    service: String,
+    capteur: String,
+    type_donnee: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct Donnee {
+    id: i32,
+    service: String,
+    capteur: String,
+    type_donnee: String,
+    donnee: String,
+    date_donnee: String,
+}
+
+impl From<Row> for Donnee {
+    fn from(row: Row) -> Self {
+        Donnee {
+            id: row.get("id").unwrap_or_default(),
+            service: row.get("service").unwrap_or_default(),
+            capteur: row.get("capteur").unwrap_or_default(),
+            type_donnee: row.get("type").unwrap_or_default(),
+            donnee: row.get("donnee").unwrap_or_default(),
+            date_donnee: row.get("date_donnee").unwrap_or_default(),
+        }
+    }
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    dotenv().ok();
+
+    let db_user = env::var("DB_USER").expect("DB_USER must be set");
+    let db_password = env::var("DB_PASSWORD").expect("DB_PASSWORD must be set");
+    let db_host = env::var("DB_HOST").expect("DB_HOST must be set");
+    let db_port: u16 = env::var("DB_PORT").expect("DB_PORT must be set").parse().unwrap();
+    let db_name = env::var("DB_NAME").expect("DB_NAME must be set");
+
+    let url = format!("mysql://{}:{}@{}:{}/{}", db_user, db_password, db_host, db_port, db_name);
+    let pool = Pool::new(url.as_str())?;
+    let mut conn = pool.get_conn()?;
+
+    let mut date = (Utc::now() - Duration::days(7)).naive_utc().date();
+    let args: Vec<String> = env::args().collect();
+    if args.len() >= 2 {
+        date = NaiveDate::parse_from_str(&args[1], "%Y-%m-%d").unwrap();
+    }
+
+    let _ = aggregate_values(&mut conn, date);
+
+    Ok(())
+}
+
+fn aggregate_values(conn: &mut PooledConn, date: NaiveDate) -> Result<(), Box<dyn std::error::Error>>  {
+
+    let start_time = Utc.with_ymd_and_hms(date.year(), date.month(), date.day(), 00, 00, 00).unwrap() - Duration::minutes(15);
+    dbg!(date, start_time);
+    let end_time = start_time + Duration::days(1);
+    let mut current_time = start_time;
+
+    let sensor = Sensor {
+        service: "maison".to_string(),
+        capteur: "bureau".to_string(),
+        type_donnee: "co2".to_string(),
+    };
+
+    while current_time <= end_time {
+        let start_time_loop = current_time.to_string();
+        let end_time_loop = (current_time + Duration::minutes(29) + Duration::seconds(59)).to_string();
+        let query = format!("SELECT * FROM donnees WHERE capteur = '{}' AND type = '{}' AND (date_donnee BETWEEN '{}' AND '{}');", sensor.capteur, sensor.type_donnee, start_time_loop, end_time_loop);
+        let selected_rows: Vec<Row> = conn.query(query)?;
+
+        let donnees: Vec<Donnee> = selected_rows.into_iter().map(Donnee::from).collect();
+
+        let mut value: i32 = 0;
+        let mut count = 0;
+        let mut ids: Vec<String> = Vec::new();
+        for donnee in donnees {
+            value = value + donnee.donnee.parse::<i32>().unwrap();
+            count = count + 1;
+            ids.push(donnee.id.to_string());
+        }
+
+        if count > 1 {
+            let average = value / count;
+            let entry_date = current_time + Duration::minutes(15);
+            let donnee = Donnee {
+                id: 0,
+                service: sensor.service.to_string(),
+                capteur: sensor.capteur.to_string(),
+                type_donnee: sensor.type_donnee.to_string(),
+                donnee: average.to_string(),
+                date_donnee: entry_date.format("%Y-%m-%d %H:%M:%S").to_string(),
+            };
+
+            let _ = insert_value(conn, donnee);
+
+            let query = "DELETE FROM donnees WHERE id IN (".to_owned() + &*ids.join(", ") + ");";
+            let _: Vec<Row> = conn.query(query)?;
+        }
+        current_time = current_time + Duration::minutes(30);
+    }
+
+    Ok(())
+}
+
+fn insert_value(conn: &mut PooledConn, donnee: Donnee) -> Result<(), mysql::Error> {
+    let query = format!(
+        "INSERT INTO donnees (service, capteur, type, donnee, date_donnee) VALUES ('{}', '{}', '{}', '{}', '{}');",
+        donnee.service, donnee.capteur, donnee.type_donnee, donnee.donnee, donnee.date_donnee
+    );
+    conn.exec_drop(query, ())?;
+
+    Ok(())
+}