Party mode

backend/src/main.rs

@@ -1,6 +1,7 @@
 mod collector;
 mod persistence;
 mod tasks;
+mod util;
 
 use clap::Parser;
 use collector::CollectorConfig;

backend/src/tasks/lights.rs

@@ -1,22 +1,22 @@
 use std::collections::HashMap;
 
-use chrono::{DateTime, Datelike, Local, NaiveTime, Weekday};
-use common::{ClientMessage, ServerMessage};
-use lighter_lib::{BulbColor, BulbId};
+use common::{BulbPrefs, ClientMessage, ScriptId, ServerMessage};
+use lighter_lib::BulbId;
 use lighter_manager::manager::{BulbCommand, BulbManager, BulbSelector};
+use lighter_manager::provider::mqtt::BulbsMqtt;
 use serde::{Deserialize, Serialize};
-use std::time::Duration;
 use tokio::select;
-use tokio::sync::{broadcast, mpsc};
-use tokio::task::{spawn, JoinHandle};
-use tokio::time::sleep;
 
 use crate::persistence::PersistenceFile;
-use crate::{ClientRequest, State};
+use crate::State;
+
+use self::scripts::{LightScript, Party, Waker};
+
+pub mod scripts;
 
 #[derive(Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
 struct LightsState {
-    wake_schedule: HashMap<BulbId, HashMap<Weekday, NaiveTime>>,
+    script_prefs: HashMap<ScriptId, HashMap<BulbId, BulbPrefs>>,
 }
 
 pub async fn lights_task(state: &State) {
@@ -29,36 +29,43 @@ pub async fn lights_task(state: &State) {
         .await
         .expect("Failed to open lights config");
 
-    let (cmd, bulb_states) = BulbManager::launch(config.bulbs.clone(), config.mqtt.clone())
+    let provider = BulbsMqtt::new(config.bulbs.clone(), config.mqtt.clone());
+    let manager = BulbManager::launch(config.bulbs.clone(), provider)
         .await
         .expect("Failed to launch bulb manager");
 
-    let mut wake_tasks: HashMap<(BulbId, Weekday), JoinHandle<()>> = lights_state
-        .get()
-        .wake_schedule
-        .iter()
-        .flat_map(|(bulb, schedule)| schedule.iter().map(move |(day, time)| (bulb, day, time)))
-        .map(|(bulb, day, time)| {
-            let handle = spawn(wake_task(
-                state.client_message.clone(),
-                cmd.clone(),
-                bulb.clone(),
-                *day,
-                *time,
-            ));
+    let mut scripts: HashMap<ScriptId, Box<dyn LightScript + Send>> = Default::default();
+    scripts.insert(
+        "waker".to_string(),
+        Box::new(Waker::create(manager.clone())),
+    );
+    scripts.insert(
+        "party".to_string(),
+        Box::new(Party::create(manager.clone())),
+    );
 
-            ((bulb.clone(), *day), handle)
-        })
-        .collect();
+    for (script, prefs) in &lights_state.get().script_prefs {
+        let Some(script) = scripts.get_mut(script) else {
+            continue;
+        };
+
+        for (bulb, prefs) in prefs {
+            for (name, value) in &prefs.kvs {
+                script.set_param(bulb, name, value.clone())
+            }
+        }
+    }
 
     loop {
-        let notify = bulb_states.notify_on_change();
         select! {
-            _ = notify => {
-                let lights_state = lights_state.get();
-                for (id, mode) in bulb_states.bulbs().await.clone().into_iter() {
-                    let wake_schedule = lights_state.wake_schedule.get(&id).cloned().unwrap_or_default();
-                    let msg = ServerMessage::BulbState { id, mode, wake_schedule };
+            _ = manager.notify_on_change() => {
+                for (id, mode) in manager.bulbs().await.clone().into_iter() {
+                    let prefs = scripts.iter_mut()
+                        .map(|(script, prefs)|
+                            (script.clone(), prefs.get_params(&id)))
+                        .collect();
+
+                    let msg = ServerMessage::BulbState { id, mode, prefs };
                     if let Err(e) = server_message.send(msg) {
                         error!("broadcast channel error: {e}");
                         return;
@@ -73,58 +80,42 @@ pub async fn lights_task(state: &State) {
 
                 match request.message {
                     ClientMessage::SetBulbColor { id, color } =>  {
-                        if let Err(e) = cmd.send(BulbCommand::SetColor(BulbSelector::Id(id), color)).await {
-                            error!("bulb manager error: {e}");
-                        }
+                        manager.send_command(BulbCommand::SetColor(BulbSelector::Id(id), color)).await;
                     }
                     ClientMessage::SetBulbPower { id, power } =>  {
-                        if let Err(e) = cmd.send(BulbCommand::SetPower(BulbSelector::Id(id), power)).await {
-                            error!("bulb manager error: {e}");
-                        }
+                        manager.send_command(BulbCommand::SetPower(BulbSelector::Id(id), power)).await;
                     }
                     ClientMessage::GetBulbs => {
                         if let Err(e) = request.response.send(ServerMessage::BulbMap(config.bulb_map.clone())).await {
                             error!("GetBulbs response channel error: {e}");
                             return;
                         }
-                        let lights_state = lights_state.get();
-                        for (id, mode) in bulb_states.bulbs().await.clone().into_iter() {
-                            let wake_schedule = lights_state.wake_schedule.get(&id).cloned().unwrap_or_default();
-                            let msg = ServerMessage::BulbState { id, mode, wake_schedule };
+                        for (id, mode) in manager.bulbs().await.clone().into_iter() {
+                            let prefs = scripts.iter_mut()
+                                .map(|(script, prefs)|
+                                    (script.clone(), prefs.get_params(&id)))
+                                .collect();
+
+                            let msg = ServerMessage::BulbState { id, mode, prefs };
                             if let Err(e) = request.response.send(msg).await {
                                 error!("GetBulbs response channel error: {e}");
                                 return;
                             }
                         }
                     }
-                    ClientMessage::SetBulbWakeTime { id, day, time } => {
-                        if let Err(e) = lights_state.update(|lights_state| {
-                            let schedule = lights_state.wake_schedule.entry(id.clone()).or_default();
-                            if let Some(time) = time {
-                                schedule.insert(day, time);
-                            }
-                            else {
-                                schedule.remove(&day);
-                            }
-                        }).await {
-                            error!("Failed to save wake schedule: {e}");
+                    ClientMessage::SetBulbPref { bulb, script, name, value } => {
+                        let Some(s) = scripts.get_mut(&script) else {
+                            continue;
                         };
+                        s.set_param(&bulb, &name, value.clone());
 
-                        if let Some(time) = time {
-                            let handle = spawn(wake_task(
-                                state.client_message.clone(),
-                                cmd.clone(),
-                                id.clone(),
-                                day,
-                                time,
-                            ));
-
-                            if let Some(old_handle) = wake_tasks.insert((id, day), handle) {
-                                old_handle.abort();
-                            }
-                        } else if let Some(old_handle) = wake_tasks.remove(&(id, day)) {
-                            old_handle.abort();
-                        }
+                        // TODO handle error
+                        lights_state.update(move |state| {
+                            state.script_prefs
+                                .entry(script).or_default()
+                                .entry(bulb).or_default()
+                                .kvs.insert(name, value);
+                        }).await.expect("failed to persist lights state");
                     }
                     _ => {}
                 }
@@ -132,112 +123,3 @@ pub async fn lights_task(state: &State) {
         }
     }
 }
-
-async fn wake_task(
-    client_messages: broadcast::Sender<ClientRequest>,
-    cmd: mpsc::Sender<BulbCommand>,
-    id: BulbId,
-    day: Weekday,
-    time: NaiveTime,
-) {
-    let mut alarm = next_alarm(Local::now(), day, time);
-
-    loop {
-        info!("sleeping until {alarm}");
-        sleep((alarm - Local::now()).to_std().unwrap()).await;
-
-        // slowly turn up brightness of bulb
-        for brightness in (1..=75).map(|i| (i as f32) * 0.01) {
-            select! {
-                // abort if the client pokes the bulb
-                _ = wait_for_bulb_command(&id, client_messages.subscribe()) => break,
-                _ = sleep(Duration::from_secs(12)) => {}
-            };
-
-            if cmd
-                .send(BulbCommand::SetColor(
-                    BulbSelector::Id(id.clone()),
-                    BulbColor::Kelvin {
-                        t: 0.0,
-                        b: brightness,
-                    },
-                ))
-                .await
-                .is_err()
-            {
-                return;
-            };
-        }
-
-        alarm = next_alarm(Local::now(), day, time);
-    }
-}
-
-/// Get the next alarm, from a weekday+time schedule.
-fn next_alarm(now: DateTime<Local>, day: Weekday, time: NaiveTime) -> DateTime<Local> {
-    let day_of_alarm = day.num_days_from_monday() as i64;
-    let day_now = now.weekday().num_days_from_monday() as i64;
-
-    let alarm = now + chrono::Duration::days(day_of_alarm - day_now);
-    let mut alarm = alarm
-        .date_naive()
-        .and_time(time)
-        .and_local_timezone(Local)
-        .unwrap();
-
-    if alarm <= now {
-        alarm += chrono::Duration::weeks(1);
-    }
-
-    alarm
-}
-
-/// Wait until we receive a client request that mutates the given bulb
-async fn wait_for_bulb_command(
-    bulb_id: &BulbId,
-    mut client_messages: broadcast::Receiver<ClientRequest>,
-) {
-    loop {
-        match client_messages.recv().await {
-            Err(_) => return,
-            Ok(request) => match request.message {
-                ClientMessage::SetBulbColor { id, .. }
-                | ClientMessage::SetBulbPower { id, .. }
-                | ClientMessage::SetBulbWakeTime { id, .. }
-                    if &id == bulb_id =>
-                {
-                    break
-                }
-                _ => continue,
-            },
-        }
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use chrono::{offset::TimeZone, Local, NaiveTime, Weekday};
-
-    use super::next_alarm;
-
-    #[test]
-    fn test_alarm_date() {
-        const FMT: &str = "%Y-%m-%d %H:%M";
-        let now = Local.datetime_from_str("2022-10-18 15:30", FMT).unwrap();
-        let test_values = [
-            (Weekday::Tue, (16, 30), "2022-10-18 16:30"),
-            (Weekday::Tue, (14, 30), "2022-10-25 14:30"),
-            (Weekday::Wed, (15, 30), "2022-10-19 15:30"),
-            (Weekday::Mon, (15, 30), "2022-10-24 15:30"),
-        ];
-
-        for (day, (hour, min), expected) in test_values {
-            let expected = Local.datetime_from_str(expected, FMT).unwrap();
-
-            assert_eq!(
-                next_alarm(now, day, NaiveTime::from_hms(hour, min, 0)),
-                expected
-            );
-        }
-    }
-}

backend/src/tasks/lights/scripts/mod.rs

@@ -0,0 +1,12 @@
+use common::{BulbPrefs, Param};
+use lighter_lib::BulbId;
+
+mod party;
+mod waker;
+pub use party::Party;
+pub use waker::Waker;
+
+pub trait LightScript {
+    fn get_params(&mut self, bulb: &BulbId) -> BulbPrefs;
+    fn set_param(&mut self, bulb: &BulbId, name: &str, value: Param);
+}

backend/src/tasks/lights/scripts/party.rs

@@ -0,0 +1,79 @@
+use std::{collections::HashMap, time::Duration};
+
+use common::{BulbPrefs, Param};
+use lighter_lib::{BulbColor, BulbId};
+use lighter_manager::manager::{BulbCommand, BulbManager, BulbSelector};
+use rand::random;
+use tokio::{spawn, time::sleep};
+
+use crate::util::DeadMansHandle;
+
+use super::LightScript;
+
+pub struct Party {
+    manager: BulbManager,
+    party_tasks: HashMap<BulbId, DeadMansHandle>,
+}
+
+impl Party {
+    pub fn create(manager: BulbManager) -> Self {
+        Party {
+            manager,
+            party_tasks: Default::default(),
+        }
+    }
+}
+
+impl LightScript for Party {
+    fn get_params(&mut self, bulb: &BulbId) -> BulbPrefs {
+        let enabled = self.party_tasks.get(bulb).is_some();
+
+        BulbPrefs {
+            kvs: [("Party".to_string(), Param::Toggle(enabled))]
+                .into_iter()
+                .collect(),
+        }
+    }
+
+    fn set_param(&mut self, bulb: &BulbId, name: &str, param: Param) {
+        if name != "Party" {
+            error!("invalit param name");
+            return;
+        }
+
+        // TODO: should be toggle
+        let Param::Toggle(enabled) = param else {
+            error!("invalit param kind");
+            return;
+        };
+
+        if !enabled {
+            self.party_tasks.remove(bulb);
+        } else {
+            let task = spawn(party_task(self.manager.clone(), bulb.clone()));
+            self.party_tasks
+                .insert(bulb.clone(), DeadMansHandle(task.abort_handle()));
+        }
+    }
+}
+
+async fn party_task(manager: BulbManager, id: BulbId) {
+    manager
+        .until_interrupted(id.clone(), async {
+            let mut h: f32 = random();
+            loop {
+                sleep(Duration::from_millis(50)).await;
+
+                h += 0.01;
+                if h > 1.0 {
+                    h = 0.0;
+                }
+
+                let color = BulbColor::HSB { h, s: 1.0, b: 1.0 };
+                manager
+                    .send_command(BulbCommand::SetColor(BulbSelector::Id(id.clone()), color))
+                    .await;
+            }
+        })
+        .await;
+}

backend/src/tasks/lights/scripts/waker.rs

@@ -0,0 +1,188 @@
+use std::{collections::HashMap, time::Duration};
+
+use chrono::{DateTime, Datelike, Local, NaiveTime, Weekday};
+use common::{BulbPrefs, Param};
+use lighter_lib::{BulbColor, BulbId};
+use lighter_manager::manager::{BulbCommand, BulbManager, BulbSelector};
+use tokio::{spawn, time::sleep};
+
+use crate::util::DeadMansHandle;
+
+use super::LightScript;
+
+pub struct Waker {
+    manager: BulbManager,
+    wake_times: HashMap<BulbId, HashMap<Weekday, NaiveTime>>,
+    wake_tasks: HashMap<(BulbId, Weekday), DeadMansHandle>,
+}
+
+impl Waker {
+    pub fn create(manager: BulbManager) -> Self {
+        Waker {
+            manager,
+            wake_times: Default::default(),
+            wake_tasks: Default::default(),
+        }
+    }
+}
+
+const TIME_FMT: &str = "%H:%M";
+
+impl LightScript for Waker {
+    fn get_params(&mut self, bulb: &BulbId) -> BulbPrefs {
+        let settings = self.wake_times.entry(bulb.clone()).or_default();
+
+        let kvs = DAYS_OF_WEEK
+            .iter()
+            .map(|day| {
+                let time = match settings.get(day) {
+                    Some(time) => time.format(TIME_FMT).to_string(),
+                    None => String::new(),
+                };
+                (format!("{day:?}"), Param::String(time))
+            })
+            .collect();
+
+        BulbPrefs { kvs }
+    }
+
+    fn set_param(&mut self, bulb: &BulbId, name: &str, time: super::Param) {
+        let settings = self.wake_times.entry(bulb.clone()).or_default();
+
+        let Param::String(time) = time else {
+            error!("invalit param kind");
+            return;
+        };
+
+        let time = NaiveTime::parse_from_str(&time, TIME_FMT)
+            .map(Some)
+            .unwrap_or(None);
+
+        let weekday = match name {
+            "Mon" => Weekday::Mon,
+            "Tue" => Weekday::Tue,
+            "Wed" => Weekday::Wed,
+            "Thu" => Weekday::Thu,
+            "Fri" => Weekday::Fri,
+            "Sat" => Weekday::Sat,
+            "Sun" => Weekday::Sun,
+            _ => {
+                error!("invalit param name");
+                return;
+            }
+        };
+
+        let Some(time) = time else {
+            settings.remove(&weekday);
+            self.wake_tasks.remove(&(bulb.clone(), weekday));
+            return;
+        };
+
+        settings.insert(weekday, time);
+        let task = spawn(wake_task(self.manager.clone(), bulb.clone(), weekday, time));
+        self.wake_tasks
+            .insert((bulb.clone(), weekday), DeadMansHandle(task.abort_handle()));
+    }
+}
+
+const DAYS_OF_WEEK: &[Weekday] = &[
+    Weekday::Mon,
+    Weekday::Tue,
+    Weekday::Wed,
+    Weekday::Thu,
+    Weekday::Fri,
+    Weekday::Sat,
+    Weekday::Sun,
+];
+
+async fn wake_task(manager: BulbManager, id: BulbId, day: Weekday, time: NaiveTime) {
+    let mut alarm = next_alarm(Local::now(), day, time);
+
+    loop {
+        info!("waking lamp {id:?} at {alarm}");
+        sleep((alarm - Local::now()).to_std().unwrap()).await;
+
+        if let Some(bulb) = manager.bulbs().await.get(&id) {
+            // don't wake the bulb if it's already turned on
+            if bulb.power {
+                continue;
+            }
+        } else {
+            warn!("bulb {id:?} does not exist");
+            return;
+        };
+
+        info!("waking lamp {id:?}");
+        let r = manager
+            .until_interrupted(id.clone(), async {
+                // slowly turn up brightness of bulb
+                for brightness in (1..=75).map(|i| (i as f32) * 0.01) {
+                    //sleep(Duration::from_secs(12)).await;
+                    sleep(Duration::from_millis(500)).await;
+
+                    manager
+                        .send_command(BulbCommand::SetColor(
+                            BulbSelector::Id(id.clone()),
+                            BulbColor::Kelvin {
+                                t: 0.0,
+                                b: brightness,
+                            },
+                        ))
+                        .await
+                }
+            })
+            .await;
+
+        if r.is_none() {
+            info!("interrupted waking lamp {id:?}");
+        }
+
+        alarm = next_alarm(Local::now(), day, time);
+    }
+}
+
+/// Get the next alarm, from a weekday+time schedule.
+fn next_alarm(now: DateTime<Local>, day: Weekday, time: NaiveTime) -> DateTime<Local> {
+    let day_of_alarm = day.num_days_from_monday() as i64;
+    let day_now = now.weekday().num_days_from_monday() as i64;
+
+    let alarm = now + chrono::Duration::days(day_of_alarm - day_now);
+    let mut alarm = alarm
+        .date_naive()
+        .and_time(time)
+        .and_local_timezone(Local)
+        .unwrap();
+
+    if alarm <= now {
+        alarm += chrono::Duration::weeks(1);
+    }
+
+    alarm
+}
+
+#[cfg(test)]
+mod test {
+    use super::next_alarm;
+    use chrono::{offset::TimeZone, Local, NaiveTime, Weekday};
+
+    #[test]
+    fn test_alarm_date() {
+        const FMT: &str = "%Y-%m-%d %H:%M";
+        let now = Local.datetime_from_str("2022-10-18 15:30", FMT).unwrap();
+        let test_values = [
+            (Weekday::Tue, (16, 30), "2022-10-18 16:30"),
+            (Weekday::Tue, (14, 30), "2022-10-25 14:30"),
+            (Weekday::Wed, (15, 30), "2022-10-19 15:30"),
+            (Weekday::Mon, (15, 30), "2022-10-24 15:30"),
+        ];
+
+        for (day, (hour, min), expected) in test_values {
+            let expected = Local.datetime_from_str(expected, FMT).unwrap();
+
+            assert_eq!(
+                next_alarm(now, day, NaiveTime::from_hms(hour, min, 0)),
+                expected
+            );
+        }
+    }
+}

backend/src/util.rs

@@ -0,0 +1,15 @@
+use tokio::task::AbortHandle;
+
+pub struct DeadMansHandle(pub AbortHandle);
+
+impl From<AbortHandle> for DeadMansHandle {
+    fn from(abort: AbortHandle) -> Self {
+        DeadMansHandle(abort)
+    }
+}
+
+impl Drop for DeadMansHandle {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}