Add mqtt client to plant watering logic instread of udp

Cargo.toml

@@ -3,6 +3,9 @@ name = "homebot"
 version = "0.1.0"
 edition = "2021"
 
+[profile.release]
+strip = "debuginfo"
+
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
@@ -12,7 +15,7 @@ serde_json = "1.0"
 anyhow = "1.0"
 serde = { version = "1.0.2", features = ["derive"] }
 actix-web = { version = "4", features = ["openssl"] }
-openssl = "0.10"
+openssl = { version = "0.10", features = ["vendored"] }
 toml = "0.8.8"
 serde_with = { version = "3.0.0", features = ["chrono"] }
 chrono = "0.4.26"
@@ -35,6 +38,7 @@ llm-chain-qdrant = "0.13.0"
 qdrant-client = "1.4.0"
 html2text = "0.11.0"
 influxdb = { version = "0.7.1", features = ["derive"] }
+rumqttc = "0.23.0"
 
 [[bin]]
 name = "homebot"
@@ -48,3 +52,6 @@ path = "src/lib/lib.rs"
 actix-rt = "2.8.0"
 httpmock = "0.7.0"
 tempfile = "3.7.1"
+
+[package.metadata.cross.target.aarch64-unknown-linux-gnu]
+xargo = false

deploy.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+
+set -eux
+cross build --release --target aarch64-unknown-linux-gnu
+
+#stop the service on the pi
+ssh [email protected] sudo systemctl stop homebot.service
+
+#copy the binary
+scp target/aarch64-unknown-linux-gnu/release/homebot [email protected]:~/homebot/
+
+#start the service
+ssh [email protected] sudo systemctl start homebot.service

src/lib/plant.rs

@@ -1,21 +1,21 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use crate::Bot;
-use actix_web::cookie::time::format_description::parse;
 use anyhow::Result;
 use chrono::{DateTime, Utc};
+use influxdb::Client;
 use influxdb::InfluxDbWriteable;
-use influxdb::{Client, Query, ReadQuery, Timestamp};
+use rumqttc::{AsyncClient, Event};
+use rumqttc::{MqttOptions, Packet};
 use serde::Deserialize;
-use tokio::net::UdpSocket;
 
 #[derive(Deserialize, Debug)]
 pub struct PlantData {
     moisture: u32,
-    is_watering: bool,
 }
 
-struct Plant {
+struct _Plant {
     name: String,
     moisture_level: u32,
     last_watering: DateTime<Utc>,
@@ -25,11 +25,11 @@ struct Plant {
 struct PlantReading {
     time: DateTime<Utc>,
     moisture: u32,
-    is_watering: bool,
     #[influxdb(tag)]
     plant_name: String,
 }
 
+#[allow(dead_code)]
 pub struct PlantServer {
     host: String,
     port: u32,
@@ -38,50 +38,78 @@ pub struct PlantServer {
 }
 
 impl PlantServer {
+    const MAX_DRY: u32 = 1900;
+    const MIN_WET: u32 = 1500;
+
     pub fn new(host: &str, chat_id: &str, port: u32, db_token: &str) -> Self {
         Self {
             host: host.to_string(),
             chat_id: chat_id.to_string(),
             port,
-            db_client: Client::new("http://localhost:8086", "homebucket").with_token(db_token),
+            db_client: Client::new("http://192.168.2.132:8086", "homebucket").with_token(db_token),
         }
     }
 
     pub async fn start(&self, bot: Arc<impl Bot>) -> Result<()> {
-        let socket = UdpSocket::bind(format!("{}:{}", self.host, self.port)).await?;
-        let mut buf = [0u8; 2048];
-
         let mut avg_moisture: Vec<u32> = vec![];
+        let mut mqttoptions = MqttOptions::new("homebot", "192.168.2.214", 1883);
+        mqttoptions.set_keep_alive(Duration::from_secs(5));
+        let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+
+        client
+            .subscribe("plants/coleus/moisture", rumqttc::QoS::AtMostOnce)
+            .await?;
 
         loop {
-            let (len, _) = socket.recv_from(&mut buf).await?;
-            let data = &buf[..len];
+            let notification: Event = eventloop.poll().await?;
 
-            match serde_json::from_slice::<PlantData>(data) {
-                Ok(parsed_json) => {
-                    let write_query = PlantReading {
-                        time: Utc::now(),
-                        moisture: parsed_json.moisture,
-                        is_watering: parsed_json.is_watering,
-                        plant_name: "flowery".to_string(),
-                    }
-                    .into_query("moisture");
+            if let Event::Incoming(Packet::Publish(data)) = notification {
+                match serde_json::from_slice::<PlantData>(&data.payload) {
+                    Ok(parsed_json) => {
+                        let write_query = PlantReading {
+                            time: Utc::now(),
+                            moisture: parsed_json.moisture,
+                            plant_name: "flowery".to_string(),
+                        }
+                        .into_query("moisture");
 
-                    self.db_client.query(write_query).await?;
+                        self.db_client.query(write_query).await?;
 
-                    avg_moisture.push(parsed_json.moisture);
-                    tracing::info!("Received {:?}", parsed_json);
-                    if avg_moisture.len() == 12 {
-                        bot.send_message(
-                            &self.chat_id,
-                            &format!("Moisture now is {}.", avg_moisture.iter().sum::<u32>() / 12),
-                        )
-                        .await?;
-                        avg_moisture.clear();
+                        avg_moisture.push(parsed_json.moisture);
+                        tracing::info!("Received {:?}", parsed_json);
+                        if avg_moisture.len() == 12 {
+                            let avg = avg_moisture.iter().sum::<u32>() / 12;
+                            match avg {
+                                Self::MAX_DRY.. => {
+                                    // Inform the user in telegram
+                                    bot.send_message(
+                                        &self.chat_id,
+                                        &format!("Moisture now is {}. Watering the plant!", avg),
+                                    )
+                                    .await?;
+                                    // water the plant
+                                    client
+                                        .publish(
+                                            "plants/coleus/water",
+                                            rumqttc::QoS::AtLeastOnce,
+                                            false,
+                                            "true",
+                                        )
+                                        .await?
+                                }
+                                0..=Self::MIN_WET => {
+                                    bot.send_message(&self.chat_id, "Coleus is too wet!")
+                                        .await?;
+                                    tracing::info!("Coleus is too wet!")
+                                }
+                                _ => (),
+                            };
+                            avg_moisture.clear();
+                        }
+                    }
+                    Err(e) => {
+                        tracing::error!("Error parsing json {}", e);
                     }
-                }
-                Err(e) => {
-                    tracing::error!("Error parsing json {}", e);
                 }
             }
         }