Feature: Schedule devices turning on/off

Cargo.lock

@@ -98,6 +98,7 @@ dependencies = [
  "serde_yaml",
  "thiserror",
  "tokio",
+ "tokio-cron-scheduler",
  "tracing",
  "tracing-subscriber",
  "wakey",
@@ -292,6 +293,17 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "cron"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ff76b51e4c068c52bfd2866e1567bee7c567ae8f24ada09fd4307019e25eab7"
+dependencies = [
+ "chrono",
+ "nom",
+ "once_cell",
+]
+
 [[package]]
 name = "crossbeam-channel"
 version = "0.5.8"
@@ -919,6 +931,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "num-derive"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "876a53fff98e03a936a674b29568b0e605f06b29372c2489ff4de23f1949743d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "num-traits"
 version = "0.2.16"
@@ -1637,6 +1660,21 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "tokio-cron-scheduler"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de2c1fd54a857b29c6cd1846f31903d0ae8e28175615c14a277aed45c58d8e27"
+dependencies = [
+ "chrono",
+ "cron",
+ "num-derive",
+ "num-traits",
+ "tokio",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "tokio-io-timeout"
 version = "1.2.0"
@@ -1877,6 +1915,15 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "uuid"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88ad59a7560b41a70d191093a945f0b87bc1deeda46fb237479708a1d6b6cdfc"
+dependencies = [
+ "getrandom",
+]
+
 [[package]]
 name = "valuable"
 version = "0.1.0"

Cargo.toml

@@ -40,6 +40,7 @@ serde_with = "3.2.0"
 enum_dispatch = "0.3.12"
 indexmap = { version = "2.0.0", features = ["serde"] }
 serde_yaml = "0.9.27"
+tokio-cron-scheduler = "0.9.4"
 
 [patch.crates-io]
 wakey = { git = "https://git.huizinga.dev/Dreaded_X/wakey" }

config/config.yml

@@ -117,8 +117,18 @@ devices:
             timeout: 60
 
 
-    bedroom_air_filter:
+    &air_filter bedroom_air_filter:
         !AirFilter
         name: "Air Filter"
         room: "Bedroom"
         topic: "pico/filter/test"
+
+# Run the air filter everyday for 19:00 to 20:00
+schedule:
+    0 0 19 * * *:
+        on:
+        - *air_filter
+
+    0 0 20 * * *:
+        off:
+        - *air_filter

config/zeus.dev.yml

@@ -89,7 +89,7 @@ devices:
         topic: "zigbee2mqtt/workbench/charger"
         timeout: 5
 
-    workbench_outlet:
+    &outlet workbench_outlet:
         !IkeaOutlet
         name: "Outlet"
         room: "Workbench"
@@ -121,4 +121,13 @@ devices:
         !AirFilter
         name: "Air Filter"
         room: "Bedroom"
-        topic: "pio/filter/test"
+        topic: "pico/filter/test"
+
+# schedule:
+#     0/30 * * * * *:
+#         on:
+#         - *outlet
+#
+#     15/30 * * * * *:
+#         off:
+#         - *outlet

src/config.rs

@@ -15,6 +15,7 @@ use crate::{
     device_manager::DeviceConfigs,
     devices::PresenceConfig,
     error::{ConfigParseError, MissingEnv},
+    schedule::Schedule,
 };
 
 #[derive(Debug, Deserialize)]
@@ -27,6 +28,7 @@ pub struct Config {
     pub ntfy: Option<NtfyConfig>,
     pub presence: PresenceConfig,
     pub devices: IndexMap<String, DeviceConfigs>,
+    pub schedule: Schedule,
 }
 
 #[derive(Debug, Clone, Deserialize)]

src/device_manager.rs

@@ -4,9 +4,11 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use enum_dispatch::enum_dispatch;
 use futures::future::join_all;
+use google_home::traits::OnOff;
 use rumqttc::{matches, AsyncClient, QoS};
 use serde::Deserialize;
 use tokio::sync::{RwLock, RwLockReadGuard};
+use tokio_cron_scheduler::{Job, JobScheduler};
 use tracing::{debug, error, instrument, trace};
 
 use crate::{
@@ -20,6 +22,7 @@ use crate::{
     event::OnNotification,
     event::OnPresence,
     event::{Event, EventChannel, OnMqtt},
+    schedule::{Action, Schedule},
 };
 
 pub struct ConfigExternal<'a> {
@@ -90,6 +93,55 @@ impl DeviceManager {
         device_manager
     }
 
+    // TODO: This function is currently extremely cursed...
+    pub async fn add_schedule(&self, schedule: Schedule) {
+        let sched = JobScheduler::new().await.unwrap();
+
+        for (when, actions) in schedule {
+            let manager = self.clone();
+            sched
+                .add(
+                    Job::new_async(when.as_str(), move |_uuid, _l| {
+                        let actions = actions.clone();
+                        let manager = manager.clone();
+
+                        Box::pin(async move {
+                            for (action, targets) in actions {
+                                for target in targets {
+                                    let device = manager.get(&target).await.unwrap();
+                                    match action {
+                                        Action::On => {
+                                            As::<dyn OnOff>::cast_mut(
+                                                device.write().await.as_mut(),
+                                            )
+                                            .unwrap()
+                                            .set_on(true)
+                                            .await
+                                            .unwrap();
+                                        }
+                                        Action::Off => {
+                                            As::<dyn OnOff>::cast_mut(
+                                                device.write().await.as_mut(),
+                                            )
+                                            .unwrap()
+                                            .set_on(false)
+                                            .await
+                                            .unwrap();
+                                        }
+                                    }
+                                }
+                            }
+                        })
+                    })
+                    .unwrap(),
+                )
+                .await
+                .unwrap();
+        }
+
+        sched.start().await.unwrap();
+    }
+
     pub async fn add(&self, device: Box<dyn Device>) {
         let id = device.get_id().into();
 

src/lib.rs

@@ -9,4 +9,5 @@ pub mod error;
 pub mod event;
 pub mod messages;
 pub mod mqtt;
+pub mod schedule;
 pub mod traits;

src/main.rs

@@ -64,6 +64,8 @@ async fn app() -> anyhow::Result<()> {
         device_manager.create(&id, device_config).await?;
     }
 
+    device_manager.add_schedule(config.schedule).await;
+
     let event_channel = device_manager.event_channel();
 
     // Create and add the presence system

src/schedule.rs

@@ -0,0 +1,17 @@
+use indexmap::IndexMap;
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize, Hash, PartialEq, Eq, Clone, Copy)]
+#[serde(rename_all = "snake_case")]
+pub enum Action {
+    On,
+    Off,
+}
+
+pub type Schedule = IndexMap<String, IndexMap<Action, Vec<String>>>;
+
+// #[derive(Debug, Deserialize)]
+// pub struct Schedule {
+//     pub when: String,
+//     pub actions: IndexMap<Action, Vec<String>>,
+// }