Added debug bridge to publish state on mqtt

config/ares.dev.toml

@@ -15,6 +15,9 @@ topic = "${NTFY_TOPIC}"
 [presence]
 topic = "automation_dev/presence/+/#"
 
+[debug_bridge]
+topic = "automation_dev/debug"
+
 [light_sensor]
 topic = "zigbee2mqtt_dev/living/light"
 min = 23_000

config/config.toml

@@ -19,6 +19,9 @@ ip = "10.0.0.146"
 login = "${HUE_TOKEN}"
 flags = { presence = 41, darkness = 43 }
 
+[debug_bridge]
+topic = "automation/debug"
+
 [light_sensor]
 topic = "zigbee2mqtt/living/light"
 min = 23_000

config/zeus.dev.toml

@@ -20,6 +20,9 @@ ip = "10.0.0.146"
 login = "${HUE_TOKEN}"
 flags = { presence = 41, darkness = 43 }
 
+[debug_bridge]
+topic = "automation_dev/debug"
+
 [light_sensor]
 topic = "zigbee2mqtt_dev/living/light"
 min = 23_000

src/config.rs

@@ -19,8 +19,9 @@ pub struct Config {
     pub presence: MqttDeviceConfig,
     pub light_sensor: LightSensorConfig,
     pub hue_bridge: Option<HueBridgeConfig>,
+    pub debug_bridge: Option<DebugBridgeConfig>,
     #[serde(default)]
-    pub devices: HashMap<String, Device>
+    pub devices: HashMap<String, Device>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -99,6 +100,11 @@ pub struct HueBridgeConfig {
     pub flags: Flags,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct DebugBridgeConfig {
+    pub topic: String,
+}
+
 #[derive(Debug, Clone, Deserialize)]
 pub struct InfoConfig {
     pub name: String,

src/debug_bridge.rs

@@ -0,0 +1,69 @@
+use async_trait::async_trait;
+use rumqttc::AsyncClient;
+use tracing::warn;
+
+use crate::{config::DebugBridgeConfig, presence::{OnPresence, self}, light_sensor::{OnDarkness, self}, mqtt::{PresenceMessage, DarknessMessage}};
+
+struct DebugBridge {
+    topic: String,
+    client: AsyncClient,
+}
+
+impl DebugBridge {
+    pub fn new(topic: String, client: AsyncClient) -> Self {
+        Self { topic, client }
+    }
+}
+
+pub fn start(mut presence_rx: presence::Receiver, mut light_sensor_rx: light_sensor::Receiver, config: DebugBridgeConfig, client: AsyncClient) {
+    let mut debug_bridge = DebugBridge::new(config.topic, client);
+
+    tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                res = presence_rx.changed() => {
+                    if !res.is_ok() {
+                        break;
+                    }
+
+                    let presence = *presence_rx.borrow();
+                    debug_bridge.on_presence(presence).await;
+                }
+                res = light_sensor_rx.changed() => {
+                    if !res.is_ok() {
+                        break;
+                    }
+
+                    let darkness = *light_sensor_rx.borrow();
+                    debug_bridge.on_darkness(darkness).await;
+                }
+            }
+        }
+
+        unreachable!("Did not expect this");
+    });
+}
+
+#[async_trait]
+impl OnPresence for DebugBridge {
+    async fn on_presence(&mut self, presence: bool) {
+        let message = PresenceMessage::new(presence);
+        let topic = self.topic.clone() + "/presence";
+        self.client.publish(topic, rumqttc::QoS::AtLeastOnce, true, serde_json::to_string(&message).unwrap())
+            .await
+            .map_err(|err| warn!("Failed to update presence on {}/presence: {err}", self.topic))
+            .ok();
+    }
+}
+
+#[async_trait]
+impl OnDarkness for DebugBridge {
+    async fn on_darkness(&mut self, dark: bool) {
+        let message = DarknessMessage::new(dark);
+        let topic = self.topic.clone() + "/darkness";
+        self.client.publish(topic, rumqttc::QoS::AtLeastOnce, true, serde_json::to_string(&message).unwrap())
+            .await
+            .map_err(|err| warn!("Failed to update presence on {}/presence: {err}", self.topic))
+            .ok();
+    }
+}

src/devices/contact_sensor.rs

@@ -97,7 +97,10 @@ impl OnMqtt for ContactSensor {
             // This is to prevent the house from being marked as present for however long the
             // timeout is set when leaving the house
             if !self.overall_presence {
-                self.client.publish(topic.clone(), rumqttc::QoS::AtLeastOnce, false, serde_json::to_string(&PresenceMessage::new(true)).unwrap()).await.map_err(|err| warn!("Failed to publish presence on {topic}: {err}")).ok();
+                self.client.publish(topic.clone(), rumqttc::QoS::AtLeastOnce, false, serde_json::to_string(&PresenceMessage::new(true)).unwrap())
+                    .await
+                    .map_err(|err| warn!("Failed to publish presence on {topic}: {err}"))
+                    .ok();
             }
         } else {
             // Once the door is closed again we start a timeout for removing the presence
@@ -109,7 +112,10 @@ impl OnMqtt for ContactSensor {
                     debug!(id, "Starting timeout ({timeout:?}) for contact sensor...");
                     tokio::time::sleep(timeout).await;
                     debug!(id, "Removing door device!");
-                    client.publish(topic.clone(), rumqttc::QoS::AtLeastOnce, false, "").await.map_err(|err| warn!("Failed to publish presence on {topic}: {err}")).ok();
+                    client.publish(topic.clone(), rumqttc::QoS::AtLeastOnce, false, "")
+                        .await
+                        .map_err(|err| warn!("Failed to publish presence on {topic}: {err}"))
+                        .ok();
                 })
             );
         }

src/devices/ikea_outlet.rs

@@ -38,7 +38,10 @@ async fn set_on(client: AsyncClient, topic: &str, on: bool) {
     let message = OnOffMessage::new(on);
 
     // @TODO Handle potential errors here
-    client.publish(topic.to_owned() + "/set", rumqttc::QoS::AtLeastOnce, false, serde_json::to_string(&message).unwrap()).await.map_err(|err| warn!("Failed to update state on {topic}: {err}")).ok();
+    client.publish(topic.to_owned() + "/set", rumqttc::QoS::AtLeastOnce, false, serde_json::to_string(&message).unwrap())
+        .await
+        .map_err(|err| warn!("Failed to update state on {topic}/set: {err}"))
+        .ok();
 }
 
 impl Device for IkeaOutlet {

src/lib.rs

@@ -8,3 +8,4 @@ pub mod light_sensor;
 pub mod hue_bridge;
 pub mod auth;
 pub mod error;
+pub mod debug_bridge;

src/main.rs

@@ -10,7 +10,7 @@ use automation::{
     hue_bridge,
     light_sensor, mqtt::Mqtt,
     ntfy,
-    presence, error::ApiError,
+    presence, error::ApiError, debug_bridge,
 };
 use dotenvy::dotenv;
 use rumqttc::{AsyncClient, MqttOptions, Transport};
@@ -95,11 +95,16 @@ async fn app() -> Result<(), Box<dyn std::error::Error>> {
         ntfy::start(presence.clone(), &ntfy_config);
     }
 
-    // Start he hue bridge if it is configured
+    // Start the hue bridge if it is configured
     if let Some(hue_bridge_config) = config.hue_bridge {
         hue_bridge::start(presence.clone(), light_sensor.clone(), hue_bridge_config);
     }
 
+    // Start the debug bridge if it is configured
+    if let Some(debug_bridge_config) = config.debug_bridge {
+        debug_bridge::start(presence.clone(), light_sensor.clone(), debug_bridge_config, client.clone());
+    }
+
     // Actually start listening for mqtt message,
     // we wait until all the setup is done, as otherwise we might miss some messages
     mqtt.start();

src/mqtt.rs

@@ -1,3 +1,5 @@
+use std::time::{UNIX_EPOCH, SystemTime};
+
 use async_trait::async_trait;
 use serde::{Serialize, Deserialize};
 use tracing::{debug, warn};
@@ -125,12 +127,13 @@ impl TryFrom<&Publish> for RemoteMessage {
 
 #[derive(Debug, Deserialize, Serialize)]
 pub struct PresenceMessage {
-    state: bool
+    state: bool,
+    updated: Option<u128>,
 }
 
 impl PresenceMessage {
     pub fn new(state: bool) -> Self {
-        Self { state }
+        Self { state, updated: Some(SystemTime::now().duration_since(UNIX_EPOCH).expect("Time is after UNIX EPOCH").as_millis()) }
     }
 
     pub fn present(&self) -> bool {
@@ -187,3 +190,28 @@ impl TryFrom<&Publish> for ContactMessage {
     }
 }
 
+#[derive(Debug, Deserialize, Serialize)]
+pub struct DarknessMessage {
+    state: bool,
+    updated: Option<u128>,
+}
+
+impl DarknessMessage {
+    pub fn new(state: bool) -> Self {
+        Self { state, updated: Some(SystemTime::now().duration_since(UNIX_EPOCH).expect("Time is after UNIX EPOCH").as_millis()) }
+    }
+
+    pub fn present(&self) -> bool {
+        self.state
+    }
+}
+
+impl TryFrom<&Publish> for DarknessMessage {
+    type Error = anyhow::Error;
+
+    fn try_from(message: &Publish) -> Result<Self, Self::Error> {
+        serde_json::from_slice(&message.payload)
+            .or(Err(anyhow::anyhow!("Invalid message payload received: {:?}", message.payload)))
+    }
+}
+