Added low battery notification and made mqtt message parsing more robust (#1)

automation_devices/src/contact_sensor.rs

@@ -36,6 +36,9 @@ pub struct Config {
 
     #[device_config(from_lua, default)]
     pub callback: ActionCallback<ContactSensor, bool>,
+    #[device_config(from_lua, default)]
+    pub battery_callback: ActionCallback<ContactSensor, f32>,
+
     #[device_config(from_lua)]
     pub client: WrappedAsyncClient,
 }
@@ -149,14 +152,15 @@ impl OnMqtt for ContactSensor {
             return;
         }
 
-        let is_closed = match ContactMessage::try_from(message) {
+        let message = match ContactMessage::try_from(message) {
-            Ok(state) => state.is_closed(),
+            Ok(message) => message,
             Err(err) => {
                 error!(id = self.get_id(), "Failed to parse message: {err}");
                 return;
             }
         };
 
+        if let Some(is_closed) = message.contact {
             if is_closed == self.state().await.is_closed {
                 return;
             }
@@ -166,4 +170,9 @@ impl OnMqtt for ContactSensor {
             debug!(id = self.get_id(), "Updating state to {is_closed}");
             self.state_mut().await.is_closed = is_closed;
         }
+
+        if let Some(battery) = message.battery {
+            self.config.battery_callback.call(self, &battery).await;
+        }
+    }
 }

automation_devices/src/hue_bridge.rs

@@ -99,7 +99,7 @@ impl AddAdditionalMethods for HueBridge {
     {
         methods.add_async_method(
             "set_flag",
-            |lua, this, (flag, value): (mlua::Value, bool)| async move {
+            async |lua, this, (flag, value): (mlua::Value, bool)| {
                 let flag: Flag = lua.from_value(flag)?;
 
                 this.set_flag(flag, value).await;

automation_devices/src/hue_switch.rs

@@ -31,9 +31,12 @@ pub struct Config {
 
     #[device_config(from_lua, default)]
     pub right_hold_callback: ActionCallback<HueSwitch, ()>,
+
+    #[device_config(from_lua, default)]
+    pub battery_callback: ActionCallback<HueSwitch, f32>,
 }
 
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Copy, Clone, Deserialize)]
 #[serde(rename_all = "snake_case")]
 enum Action {
     LeftPress,
@@ -48,7 +51,8 @@ enum Action {
 
 #[derive(Debug, Clone, Deserialize)]
 struct State {
-    action: Action,
+    action: Option<Action>,
+    battery: Option<f32>,
 }
 
 #[derive(Debug, Clone, LuaDevice)]
@@ -84,14 +88,20 @@ impl OnMqtt for HueSwitch {
     async fn on_mqtt(&self, message: Publish) {
         // Check if the message is from the device itself or from a remote
         if matches(&message.topic, &self.config.mqtt.topic) {
-            let action = match serde_json::from_slice::<State>(&message.payload) {
+            let message = match serde_json::from_slice::<State>(&message.payload) {
-                Ok(message) => message.action,
+                Ok(message) => message,
                 Err(err) => {
                     warn!(id = Device::get_id(self), "Failed to parse message: {err}");
                     return;
                 }
             };
-            debug!(id = Device::get_id(self), "Remote action = {:?}", action);
+
+            if let Some(action) = message.action {
+                debug!(
+                    id = Device::get_id(self),
+                    ?message.action,
+                    "Action received",
+                );
 
                 match action {
                     Action::LeftPressRelease => self.config.left_callback.call(self, &()).await,
@@ -112,5 +122,10 @@ impl OnMqtt for HueSwitch {
                     _ => {}
                 }
             }
+
+            if let Some(battery) = message.battery {
+                self.config.battery_callback.call(self, &battery).await;
+            }
+        }
     }
 }

automation_devices/src/ikea_remote.rs

@@ -25,6 +25,8 @@ pub struct Config {
 
     #[device_config(from_lua)]
     pub callback: ActionCallback<IkeaRemote, bool>,
+    #[device_config(from_lua, default)]
+    pub battery_callback: ActionCallback<IkeaRemote, f32>,
 }
 
 #[derive(Debug, Clone, LuaDevice)]
@@ -60,13 +62,15 @@ impl OnMqtt for IkeaRemote {
     async fn on_mqtt(&self, message: Publish) {
         // Check if the message is from the deviec itself or from a remote
         if matches(&message.topic, &self.config.mqtt.topic) {
-            let action = match RemoteMessage::try_from(message) {
+            let message = match RemoteMessage::try_from(message) {
-                Ok(message) => message.action(),
+                Ok(message) => message,
                 Err(err) => {
                     error!(id = Device::get_id(self), "Failed to parse message: {err}");
                     return;
                 }
             };
+
+            if let Some(action) = message.action {
                 debug!(id = Device::get_id(self), "Remote action = {:?}", action);
 
                 let on = if self.config.single_button {
@@ -87,5 +91,10 @@ impl OnMqtt for IkeaRemote {
                     self.config.callback.call(self, &on).await;
                 }
             }
+
+            if let Some(battery) = message.battery {
+                self.config.battery_callback.call(self, &battery).await;
+            }
+        }
     }
 }

automation_devices/src/ntfy.rs

@@ -170,7 +170,7 @@ impl AddAdditionalMethods for Ntfy {
     {
         methods.add_async_method(
             "send_notification",
-            |lua, this, notification: mlua::Value| async move {
+            async |lua, this, notification: mlua::Value| {
                 let notification: Notification = lua.from_value(notification)?;
 
                 this.send(notification).await;

automation_lib/src/device_manager.rs

@@ -73,9 +73,7 @@ impl DeviceManager {
         match event {
             Event::MqttMessage(message) => {
                 let devices = self.devices.read().await;
-                let iter = devices.iter().map(|(id, device)| {
+                let iter = devices.iter().map(async |(id, device)| {
-                    let message = message.clone();
-                    async move {
                     let device: Option<&dyn OnMqtt> = device.cast();
                     if let Some(device) = device {
                         // let subscribed = device
@@ -85,11 +83,10 @@ impl DeviceManager {
                         //
                         // if subscribed {
                         trace!(id, "Handling");
-                            device.on_mqtt(message).await;
+                        device.on_mqtt(message.clone()).await;
                         trace!(id, "Done");
                         // }
                     }
-                    }
                 });
 
                 join_all(iter).await;
@@ -100,7 +97,7 @@ impl DeviceManager {
 
 impl mlua::UserData for DeviceManager {
     fn add_methods<M: mlua::UserDataMethods<Self>>(methods: &mut M) {
-        methods.add_async_method("add", |_lua, this, device: Box<dyn Device>| async move {
+        methods.add_async_method("add", async |_lua, this, device: Box<dyn Device>| {
             this.add(device).await;
 
             Ok(())
@@ -108,7 +105,7 @@ impl mlua::UserData for DeviceManager {
 
         methods.add_async_method(
             "schedule",
-            |lua, this, (schedule, f): (String, mlua::Function)| async move {
+            async |lua, this, (schedule, f): (String, mlua::Function)| {
                 debug!("schedule = {schedule}");
                 // This creates a function, that returns the actual job we want to run
                 let create_job = {

automation_lib/src/helpers/timeout.rs

@@ -29,7 +29,7 @@ impl mlua::UserData for Timeout {
 
         methods.add_async_method(
             "start",
-            |_lua, this, (timeout, callback): (f32, ActionCallback<mlua::Value, bool>)| async move {
+            async |_lua, this, (timeout, callback): (f32, ActionCallback<mlua::Value, bool>)| {
                 if let Some(handle) = this.state.write().await.handle.take() {
                     handle.abort();
                 }
@@ -50,7 +50,7 @@ impl mlua::UserData for Timeout {
             },
         );
 
-        methods.add_async_method("cancel", |_lua, this, ()| async move {
+        methods.add_async_method("cancel", async |_lua, this, ()| {
             debug!("Canceling timeout callback");
 
             if let Some(handle) = this.state.write().await.handle.take() {
@@ -60,7 +60,7 @@ impl mlua::UserData for Timeout {
             Ok(())
         });
 
-        methods.add_async_method("is_waiting", |_lua, this, ()| async move {
+        methods.add_async_method("is_waiting", async |_lua, this, ()| {
             debug!("Canceling timeout callback");
 
             if let Some(handle) = this.state.read().await.handle.as_ref() {

automation_lib/src/lua/traits.rs

@@ -7,13 +7,13 @@ pub trait OnOff {
     where
         Self: Sized + google_home::traits::OnOff + 'static,
     {
-        methods.add_async_method("set_on", |_lua, this, on: bool| async move {
+        methods.add_async_method("set_on", async |_lua, this, on: bool| {
             this.deref().set_on(on).await.unwrap();
 
             Ok(())
         });
 
-        methods.add_async_method("on", |_lua, this, ()| async move {
+        methods.add_async_method("on", async |_lua, this, ()| {
             Ok(this.deref().on().await.unwrap())
         });
     }
@@ -25,13 +25,13 @@ pub trait Brightness {
     where
         Self: Sized + google_home::traits::Brightness + 'static,
     {
-        methods.add_async_method("set_brightness", |_lua, this, brightness: u8| async move {
+        methods.add_async_method("set_brightness", async |_lua, this, brightness: u8| {
             this.set_brightness(brightness).await.unwrap();
 
             Ok(())
         });
 
-        methods.add_async_method("brightness", |_lua, this, _: ()| async move {
+        methods.add_async_method("brightness", async |_lua, this, _: ()| {
             Ok(this.brightness().await.unwrap())
         });
     }
@@ -45,7 +45,7 @@ pub trait ColorSetting {
     {
         methods.add_async_method(
             "set_color_temperature",
-            |_lua, this, temperature: u32| async move {
+            async |_lua, this, temperature: u32| {
                 this.set_color(google_home::traits::Color { temperature })
                     .await
                     .unwrap();
@@ -54,7 +54,7 @@ pub trait ColorSetting {
             },
         );
 
-        methods.add_async_method("color_temperature", |_lua, this, ()| async move {
+        methods.add_async_method("color_temperature", async |_lua, this, ()| {
             Ok(this.color().await.temperature)
         });
     }
@@ -66,16 +66,13 @@ pub trait OpenClose {
     where
         Self: Sized + google_home::traits::OpenClose + 'static,
     {
-        methods.add_async_method(
+        methods.add_async_method("set_open_percent", async |_lua, this, open_percent: u8| {
-            "set_open_percent",
-            |_lua, this, open_percent: u8| async move {
             this.set_open_percent(open_percent).await.unwrap();
 
             Ok(())
-            },
+        });
-        );
 
-        methods.add_async_method("open_percent", |_lua, this, _: ()| async move {
+        methods.add_async_method("open_percent", async |_lua, this, _: ()| {
             Ok(this.open_percent().await.unwrap())
         });
     }

automation_lib/src/messages.rs

@@ -68,13 +68,8 @@ pub enum RemoteAction {
 // Message used to report the action performed by a remote
 #[derive(Debug, Deserialize)]
 pub struct RemoteMessage {
-    action: RemoteAction,
+    pub action: Option<RemoteAction>,
-}
+    pub battery: Option<f32>,
-
-impl RemoteMessage {
-    pub fn action(&self) -> RemoteAction {
-        self.action
-    }
 }
 
 impl TryFrom<Publish> for RemoteMessage {
@@ -144,13 +139,8 @@ impl TryFrom<Publish> for BrightnessMessage {
 // Message to report the state of a contact sensor
 #[derive(Debug, Deserialize)]
 pub struct ContactMessage {
-    contact: bool,
+    pub contact: Option<bool>,
-}
+    pub battery: Option<f32>,
-
-impl ContactMessage {
-    pub fn is_closed(&self) -> bool {
-        self.contact
-    }
 }
 
 impl TryFrom<Publish> for ContactMessage {

automation_lib/src/mqtt.rs

@@ -27,7 +27,7 @@ impl mlua::UserData for WrappedAsyncClient {
     fn add_methods<M: mlua::UserDataMethods<Self>>(methods: &mut M) {
         methods.add_async_method(
             "send_message",
-            |_lua, this, (topic, message): (String, mlua::Value)| async move {
+            async |_lua, this, (topic, message): (String, mlua::Value)| {
                 let message = serde_json::to_string(&message).unwrap();
 
                 debug!("message = {message}");

automation_macro/src/impl_device.rs

@@ -45,7 +45,7 @@ impl Impl {
         quote! {
             impl mlua::UserData for #name #generics {
                 fn add_methods<M: mlua::UserDataMethods<Self>>(methods: &mut M) {
-                    methods.add_async_function("new", |_lua, config| async {
+                    methods.add_async_function("new", async |_lua, config| {
                         let device: Self = LuaDeviceCreate::create(config)
                             .await
                             .map_err(mlua::ExternalError::into_lua_err)?;
@@ -58,7 +58,7 @@ impl Impl {
                         Ok(b)
                     });
 
-                    methods.add_async_method("get_id", |_lua, this, _: ()| async move { Ok(this.get_id()) });
+                    methods.add_async_method("get_id", async |_lua, this, _: ()| { Ok(this.get_id()) });
 
                     #(
                         #traits::add_methods(methods);

config.lua

@@ -34,6 +34,37 @@ local ntfy = Ntfy.new({
 })
 automation.device_manager:add(ntfy)
 
+local low_battery = {}
+local function check_battery(device, battery)
+	local id = device:get_id()
+	if battery < 15 then
+		print("Device '" .. id .. "' has low battery: " .. tostring(battery))
+		low_battery[id] = battery
+	else
+		low_battery[id] = nil
+	end
+end
+automation.device_manager:schedule("0 0 21 */1 * *", function()
+	-- Don't send notifications if there are now devices with low battery
+	if next(low_battery) == nil then
+		print("No devices with low battery")
+		return
+	end
+
+	local lines = {}
+	for name, battery in pairs(low_battery) do
+		table.insert(lines, name .. ": " .. tostring(battery) .. "%")
+	end
+	local message = table.concat(lines, "\n")
+
+	ntfy:send_notification({
+		title = "Low battery",
+		message = message,
+		tags = { "battery" },
+		priority = "default",
+	})
+end)
+
 local on_presence = {
 	add = function(self, f)
 		self[#self + 1] = f
@@ -171,6 +202,7 @@ automation.device_manager:add(HueSwitch.new({
 	right_hold_callback = function()
 		living_lights_relax:set_on(true)
 	end,
+	battery_callback = check_battery,
 }))
 
 automation.device_manager:add(WakeOnLAN.new({
@@ -222,6 +254,7 @@ automation.device_manager:add(IkeaRemote.new({
 			end
 		end
 	end,
+	battery_callback = check_battery,
 }))
 
 local function kettle_timeout()
@@ -260,6 +293,7 @@ automation.device_manager:add(IkeaRemote.new({
 	topic = mqtt_z2m("bedroom/remote"),
 	single_button = true,
 	callback = set_kettle,
+	battery_callback = check_battery,
 }))
 
 automation.device_manager:add(IkeaRemote.new({
@@ -269,6 +303,7 @@ automation.device_manager:add(IkeaRemote.new({
 	topic = mqtt_z2m("kitchen/remote"),
 	single_button = true,
 	callback = set_kettle,
+	battery_callback = check_battery,
 }))
 
 local function off_timeout(duration)
@@ -356,6 +391,7 @@ automation.device_manager:add(IkeaRemote.new({
 			workbench_light:set_on(false)
 		end
 	end,
+	battery_callback = check_battery,
 }))
 
 local hallway_top_light = HueGroup.new({
@@ -373,6 +409,7 @@ automation.device_manager:add(HueSwitch.new({
 	left_callback = function()
 		hallway_top_light:set_on(not hallway_top_light:on())
 	end,
+	battery_callback = check_battery,
 }))
 automation.device_manager:add(HueSwitch.new({
 	name = "SwitchTop",
@@ -382,6 +419,7 @@ automation.device_manager:add(HueSwitch.new({
 	left_callback = function()
 		hallway_top_light:set_on(not hallway_top_light:on())
 	end,
+	battery_callback = check_battery,
 }))
 
 local hallway_light_automation = {
@@ -488,6 +526,7 @@ automation.device_manager:add(IkeaRemote.new({
 	callback = function(_, on)
 		hallway_light_automation:switch_callback(on)
 	end,
+	battery_callback = check_battery,
 }))
 local hallway_frontdoor = ContactSensor.new({
 	name = "Frontdoor",
@@ -503,6 +542,7 @@ local hallway_frontdoor = ContactSensor.new({
 		hallway_light_automation:door_callback(open)
 		frontdoor_presence(open)
 	end,
+	battery_callback = check_battery,
 })
 automation.device_manager:add(hallway_frontdoor)
 hallway_light_automation.door = hallway_frontdoor
@@ -516,6 +556,7 @@ local hallway_trash = ContactSensor.new({
 	callback = function(_, open)
 		hallway_light_automation:trash_callback(open)
 	end,
+	battery_callback = check_battery,
 })
 automation.device_manager:add(hallway_trash)
 hallway_light_automation.trash = hallway_trash
@@ -564,6 +605,7 @@ automation.device_manager:add(HueSwitch.new({
 	left_hold_callback = function()
 		bedroom_lights_relax:set_on(true)
 	end,
+	battery_callback = check_battery,
 }))
 
 automation.device_manager:add(ContactSensor.new({
@@ -572,24 +614,28 @@ automation.device_manager:add(ContactSensor.new({
 	sensor_type = "Door",
 	topic = mqtt_z2m("living/balcony"),
 	client = mqtt_client,
+	battery_callback = check_battery,
 }))
 automation.device_manager:add(ContactSensor.new({
 	name = "Window",
 	room = "Living Room",
 	topic = mqtt_z2m("living/window"),
 	client = mqtt_client,
+	battery_callback = check_battery,
 }))
 automation.device_manager:add(ContactSensor.new({
 	name = "Window",
 	room = "Bedroom",
 	topic = mqtt_z2m("bedroom/window"),
 	client = mqtt_client,
+	battery_callback = check_battery,
 }))
 automation.device_manager:add(ContactSensor.new({
 	name = "Window",
 	room = "Guest Room",
 	topic = mqtt_z2m("guest/window"),
 	client = mqtt_client,
+	battery_callback = check_battery,
 }))
 
 local storage_light = LightBrightness.new({
@@ -614,6 +660,7 @@ automation.device_manager:add(ContactSensor.new({
 			storage_light:set_on(false)
 		end
 	end,
+	battery_callback = check_battery,
 }))
 
 automation.device_manager:schedule("0 0 19 * * *", function()

google_home/google_home/src/fulfillment.rs

@@ -40,8 +40,7 @@ impl GoogleHome {
         let intent = request.inputs.into_iter().next();
 
         let payload: OptionFuture<_> = intent
-            .map(|intent| async move {
+            .map(async |intent| match intent {
-                match intent {
                 Intent::Sync => ResponsePayload::Sync(self.sync(devices).await),
                 Intent::Query(payload) => {
                     ResponsePayload::Query(self.query(payload, devices).await)
@@ -49,7 +48,6 @@ impl GoogleHome {
                 Intent::Execute(payload) => {
                     ResponsePayload::Execute(self.execute(payload, devices).await)
                 }
-                }
             })
             .into();
 
@@ -64,7 +62,7 @@ impl GoogleHome {
         devices: &HashMap<String, Box<T>>,
     ) -> sync::Payload {
         let mut resp_payload = sync::Payload::new(&self.user_id);
-        let f = devices.values().map(|device| async move {
+        let f = devices.values().map(async |device| {
             if let Some(device) = device.as_ref().cast() {
                 Some(Device::sync(device).await)
             } else {
@@ -86,7 +84,7 @@ impl GoogleHome {
             .devices
             .into_iter()
             .map(|device| device.id)
-            .map(|id| async move {
+            .map(async |id| {
                 // NOTE: Requires let_chains feature
                 let device = if let Some(device) = devices.get(id.as_str())
                     && let Some(device) = device.as_ref().cast()
@@ -115,9 +113,7 @@ impl GoogleHome {
     ) -> execute::Payload {
         let resp_payload = Arc::new(Mutex::new(response::execute::Payload::new()));
 
-        let f = payload.commands.into_iter().map(|command| {
+        let f = payload.commands.into_iter().map(async |command| {
-            let resp_payload = resp_payload.clone();
-            async move {
             let mut success = response::execute::Command::new(execute::Status::Success);
             success.states = Some(execute::States {
                 online: true,
@@ -134,9 +130,7 @@ impl GoogleHome {
                 .devices
                 .into_iter()
                 .map(|device| device.id)
-                    .map(|id| {
+                .map(async |id| {
-                        let execution = command.execution.clone();
-                        async move {
                     if let Some(device) = devices.get(id.as_str())
                         && let Some(device) = device.as_ref().cast()
                     {
@@ -146,14 +140,13 @@ impl GoogleHome {
 
                         // NOTE: We can not use .map here because async =(
                         let mut results = Vec::new();
-                                for cmd in &execution {
+                        for cmd in &command.execution {
                             results.push(Device::execute(device, cmd.clone()).await);
                         }
 
                         // Convert vec of results to a result with a vec and the first
                         // encountered error
-                                let results =
+                        let results = results.into_iter().collect::<Result<Vec<_>, ErrorCode>>();
-                                    results.into_iter().collect::<Result<Vec<_>, ErrorCode>>();
 
                         // TODO: We only get one error not all errors
                         if let Err(err) = results {
@@ -164,7 +157,6 @@ impl GoogleHome {
                     } else {
                         (id.clone(), Err(DeviceError::DeviceNotFound.into()))
                     }
-                        }
                 });
 
             let a = join_all(f).await;
@@ -193,7 +185,6 @@ impl GoogleHome {
                 cmd.error_code = Some(error);
                 resp_payload.add_command(cmd);
             }
-            }
         });
 
         join_all(f).await;