diff --git a/src/app_state.rs b/src/app_state.rs index c47cd9b..624621e 100644 --- a/src/app_state.rs +++ b/src/app_state.rs @@ -3,9 +3,10 @@ use std::time::{Duration, Instant}; use actix::{Actor, Context, Handler, Message}; use emgauwa_common::constants; +use emgauwa_common::db::DbSchedule; use emgauwa_common::errors::EmgauwaError; use emgauwa_common::models::Controller; -use emgauwa_common::types::RelayStates; +use emgauwa_common::types::{RelayStates, Weekday}; use futures::executor::block_on; use sqlx::{Pool, Sqlite}; use tokio::sync::Notify; @@ -30,6 +31,14 @@ pub struct RelayPulse { pub duration: Option, } +#[derive(Message)] +#[rtype(result = "Result<(), EmgauwaError>")] +pub struct RelayOverrideSchedule { + pub relay_number: i64, + pub schedule: Option, + pub weekday: Weekday, +} + #[derive(Message)] #[rtype(result = "Controller")] pub struct GetThis {} @@ -105,7 +114,7 @@ impl Handler for AppState { .iter_mut() .zip(msg.relay_states.iter()) .for_each(|(driver, state)| { - if let Err(e) = driver.set(state.unwrap_or(false)) { + if let Err(e) = driver.set(state.is_on.unwrap_or(false)) { log::error!("Error setting relay: {}", e); } }); @@ -148,6 +157,35 @@ impl Handler for AppState { } } +impl Handler for AppState { + type Result = Result<(), EmgauwaError>; + + fn handle(&mut self, msg: RelayOverrideSchedule, _ctx: &mut Self::Context) -> Self::Result { + let relay_num = msg.relay_number; + let schedule = msg.schedule; + let weekday = msg.weekday; + + let relay = self + .this + .relays + .iter_mut() + .find(|r| r.r.number == relay_num) + .ok_or(EmgauwaError::Other(String::from("Relay not found")))?; + + log::debug!( + "Overriding schedule for relay {} to '{}' on day {}", + relay_num, + schedule.as_ref().map_or("NONE", |s| &s.name), + weekday + ); + + relay.override_schedule = Some(schedule); + relay.override_schedule_weekday = weekday; + + Ok(()) + } +} + impl Handler for AppState { type Result = Controller; diff --git a/src/relay_loop.rs b/src/relay_loop.rs index 3247f7f..972824b 100644 --- a/src/relay_loop.rs +++ b/src/relay_loop.rs @@ -1,11 +1,11 @@ -use std::time::{Duration, Instant}; +use std::time::Duration; use actix::Addr; -use chrono::{Local, Timelike}; +use chrono::Timelike; use emgauwa_common::constants::RELAYS_RETRY_TIMEOUT; use emgauwa_common::errors::EmgauwaError; use emgauwa_common::models::Controller; -use emgauwa_common::types::{RelayStates, Weekday}; +use emgauwa_common::types::{EmgauwaNow, RelayState, RelayStates, Weekday}; use emgauwa_common::utils::printable_relay_states; use futures::pin_mut; use tokio::time; @@ -32,42 +32,44 @@ async fn run_relays(app_state: &Addr) -> Result<(), EmgauwaError> { let mut last_weekday = emgauwa_common::utils::get_weekday(); let mut this = utils::app_state_get_this(app_state).await?; let mut relay_states: RelayStates = Vec::new(); + let now = EmgauwaNow::now(); + init_relay_states(&mut relay_states, &this); - calc_relay_states(&mut relay_states, &mut this, app_state).await?; + calc_relay_states(&mut relay_states, &mut this, app_state, &now).await?; let mut duration_override = None; loop { + let now = EmgauwaNow::now(); log::debug!( "Relay loop at {}: {}", - Local::now().naive_local().time(), + now.time, printable_relay_states(&this.get_relay_states()) ); let notifier_future = notifier.notified(); pin_mut!(notifier_future); let mut changed = timeout( - get_next_duration(&this, &mut duration_override), + get_next_duration(&mut this, &mut duration_override, &now), &mut notifier_future, ) .await .is_ok(); - check_weekday(app_state, &mut last_weekday, &mut changed).await?; + check_weekday(app_state, &mut last_weekday, &mut changed, &now).await?; if changed { log::debug!("Reloading controller in relay loop"); this = utils::app_state_get_this(app_state).await?; } - let now_pulse = Instant::now(); duration_override = this .relays .iter_mut() - .filter_map(|relay| match relay.check_pulsing(&now_pulse) { + .filter_map(|relay| match relay.check_pulsing(&now.instant) { None => None, Some(pulse) => { - let dur = pulse - now_pulse; + let dur = pulse - now.instant; log::debug!( "Pulsing relay {} for {}s until {:?} ", relay.r.number, @@ -79,50 +81,57 @@ async fn run_relays(app_state: &Addr) -> Result<(), EmgauwaError> { }) .min(); - calc_relay_states(&mut relay_states, &mut this, app_state).await?; + calc_relay_states(&mut relay_states, &mut this, app_state, &now).await?; } } fn init_relay_states(relay_states: &mut RelayStates, this: &Controller) { relay_states.clear(); - for _ in 0..this.c.relay_count { - relay_states.push(None); - } + this.relays.iter().for_each(|r| { + relay_states.push(RelayState { + active_schedule: r.active_schedule.clone(), + is_on: None + }); + }); } async fn calc_relay_states( relay_states: &mut RelayStates, this: &mut Controller, app_state: &Addr, + now: &EmgauwaNow, ) -> Result<(), EmgauwaError> { - let now = Local::now().time(); - let now_pulse = Instant::now(); - this.relays .iter_mut() .zip(relay_states.iter_mut()) .for_each(|(relay, state)| { + relay.reload_active_schedule(now.weekday); relay.is_on = Some( - relay.active_schedule.is_on(&now) || relay.check_pulsing(&now_pulse).is_some(), + relay.active_schedule.is_on(&now.time) + || relay.check_pulsing(&now.instant).is_some(), ); - *state = relay.is_on; + + state.is_on = relay.is_on; + state.active_schedule = relay.active_schedule.clone(); }); utils::app_state_update_relays_on(app_state, relay_states.clone()).await } -fn get_next_duration(this: &Controller, duration_override: &mut Option) -> Duration { +fn get_next_duration( + this: &mut Controller, + duration_override: &mut Option, + now: &EmgauwaNow, +) -> Duration { if let Some(duration) = duration_override { log::debug!("Duration override. Waiting for {}s", duration.as_secs()); return *duration; } - let now = Local::now().time(); - let now_in_s = now.num_seconds_from_midnight(); let next_timestamp = this - .get_next_time(&now) + .check_next_time(now) .map_or(86400, |t| t.num_seconds_from_midnight()); - let duration_to_next = Duration::from_secs((next_timestamp - now_in_s) as u64); + let duration_to_next = Duration::from_secs((next_timestamp - now.time_in_s()) as u64); log::debug!( "Next timestamp: {}; Waiting for {}s", @@ -137,8 +146,9 @@ async fn check_weekday( app_state: &Addr, last_weekday: &mut Weekday, changed: &mut bool, + now: &EmgauwaNow, ) -> Result<(), EmgauwaError> { - let current_weekday = emgauwa_common::utils::get_weekday(); + let current_weekday = now.weekday; if current_weekday.ne(last_weekday) { log::debug!("Weekday changed"); *last_weekday = current_weekday; diff --git a/src/utils.rs b/src/utils.rs index c6d2830..2080c50 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,9 +1,10 @@ use std::sync::Arc; use actix::Addr; +use emgauwa_common::db::DbSchedule; use emgauwa_common::errors::EmgauwaError; use emgauwa_common::models::Controller; -use emgauwa_common::types::RelayStates; +use emgauwa_common::types::{RelayStates, Weekday}; use tokio::sync::Notify; use crate::app_state; @@ -64,3 +65,19 @@ pub async fn app_state_relay_pulse( .await .map_err(EmgauwaError::from)? } + +pub async fn app_state_relay_override_schedule( + app_state: &Addr, + relay_number: i64, + schedule: Option, + weekday: Weekday, +) -> Result<(), EmgauwaError> { + app_state + .send(app_state::RelayOverrideSchedule { + relay_number, + weekday, + schedule, + }) + .await + .map_err(EmgauwaError::from)? +} diff --git a/src/ws/handlers.rs b/src/ws/handlers.rs index 7af953a..20ada8c 100644 --- a/src/ws/handlers.rs +++ b/src/ws/handlers.rs @@ -1,166 +1,178 @@ use actix::Addr; -use sqlx::{Pool, Sqlite}; -use sqlx::pool::PoolConnection; -use tokio_tungstenite::tungstenite; -use tokio_tungstenite::tungstenite::Message; use emgauwa_common::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; use emgauwa_common::errors::{DatabaseError, EmgauwaError}; use emgauwa_common::models::{Controller, Relay}; use emgauwa_common::types::{ControllerWsAction, ScheduleUid}; +use sqlx::pool::PoolConnection; +use sqlx::{Pool, Sqlite}; +use tokio_tungstenite::tungstenite; +use tokio_tungstenite::tungstenite::Message; + use crate::app_state::AppState; use crate::utils; use crate::utils::app_state_get_this; pub async fn handle_message( - pool: Pool, - app_state: &Addr, - message_result: Result, + pool: Pool, + app_state: &Addr, + message_result: Result, ) { - let msg = match message_result { - Ok(msg) => msg, - Err(err) => { - log::error!("Error reading message: {}", err); - return; - } - }; - if let Message::Text(text) = msg { - match serde_json::from_str(&text) { - Ok(action) => { - log::debug!("Received action: {:?}", action); - let mut pool_conn = match pool.acquire().await { - Ok(conn) => conn, - Err(err) => { - log::error!("Failed to acquire database connection: {:?}", err); - return; - } - }; - let action_res = handle_action(&mut pool_conn, app_state, action).await; - if let Err(e) = action_res { - log::error!("Error handling action: {:?}", e); - } - } - Err(e) => { - log::error!("Error deserializing action: {:?}", e); - } - } - } + let msg = match message_result { + Ok(msg) => msg, + Err(err) => { + log::error!("Error reading message: {}", err); + return; + } + }; + if let Message::Text(text) = msg { + match serde_json::from_str(&text) { + Ok(action) => { + log::debug!("Received action: {:?}", action); + let mut pool_conn = match pool.acquire().await { + Ok(conn) => conn, + Err(err) => { + log::error!("Failed to acquire database connection: {:?}", err); + return; + } + }; + let action_res = handle_action(&mut pool_conn, app_state, action).await; + if let Err(e) = action_res { + log::error!("Error handling action: {:?}", e); + } + } + Err(e) => { + log::error!("Error deserializing action: {:?}", e); + } + } + } } pub async fn handle_action( - conn: &mut PoolConnection, - app_state: &Addr, - action: ControllerWsAction, + conn: &mut PoolConnection, + app_state: &Addr, + action: ControllerWsAction, ) -> Result<(), EmgauwaError> { - let this = app_state_get_this(app_state).await?; + let this = app_state_get_this(app_state).await?; - match action { - ControllerWsAction::Controller(controller) => { - handle_controller(conn, &this, controller).await? - } - ControllerWsAction::Relays(relays) => handle_relays(conn, &this, relays).await?, - ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?, - ControllerWsAction::RelayPulse((relay_num, duration)) => { - handle_relay_pulse(app_state, relay_num, duration).await? - } - _ => return Ok(()), - }; + match action { + ControllerWsAction::Controller(controller) => { + handle_controller(conn, &this, controller).await? + } + ControllerWsAction::Relays(relays) => handle_relays(conn, app_state, &this, relays).await?, + ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?, + ControllerWsAction::RelayPulse((relay_num, duration)) => { + handle_relay_pulse(app_state, relay_num, duration).await? + } + _ => return Ok(()), + }; - utils::app_state_reload(app_state).await + utils::app_state_reload(app_state).await } async fn handle_controller( - conn: &mut PoolConnection, - this: &Controller, - controller: Controller, + conn: &mut PoolConnection, + this: &Controller, + controller: Controller, ) -> Result<(), EmgauwaError> { - if controller.c.uid != this.c.uid { - return Err(EmgauwaError::Other(String::from( - "Controller UID mismatch during update", - ))); - } - DbController::get_by_uid(conn, &controller.c.uid) - .await? - .ok_or(DatabaseError::NotFound)? - .update(conn, controller.c.name.as_str(), this.c.relay_count) - .await?; + if controller.c.uid != this.c.uid { + return Err(EmgauwaError::Other(String::from( + "Controller UID mismatch during update", + ))); + } + DbController::get_by_uid(conn, &controller.c.uid) + .await? + .ok_or(DatabaseError::NotFound)? + .update(conn, controller.c.name.as_str(), this.c.relay_count) + .await?; - Ok(()) + Ok(()) } async fn handle_schedules( - conn: &mut PoolConnection, - schedules: Vec, + conn: &mut PoolConnection, + schedules: Vec, ) -> Result<(), EmgauwaError> { - let mut handled_uids = vec![ - // on and off schedules are always present and should not be updated - ScheduleUid::On, - ScheduleUid::Off, - ]; - for schedule in schedules { - if handled_uids.contains(&schedule.uid) { - continue; - } - handled_uids.push(schedule.uid.clone()); + let mut handled_uids = vec![ + // on and off schedules are always present and should not be updated + ScheduleUid::On, + ScheduleUid::Off, + ]; + for schedule in schedules { + if handled_uids.contains(&schedule.uid) { + continue; + } + handled_uids.push(schedule.uid.clone()); - log::debug!("Handling schedule: {:?}", schedule); - let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?; + log::debug!("Handling schedule: {:?}", schedule); + let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?; - if let Some(schedule_db) = schedule_db { - schedule_db - .update(conn, schedule.name.as_str(), &schedule.periods) - .await?; - } else { - DbSchedule::create( - conn, - schedule.uid.clone(), - schedule.name.as_str(), - &schedule.periods, - ) - .await?; - } - } + if let Some(schedule_db) = schedule_db { + schedule_db + .update(conn, schedule.name.as_str(), &schedule.periods) + .await?; + } else { + DbSchedule::create( + conn, + schedule.uid.clone(), + schedule.name.as_str(), + &schedule.periods, + ) + .await?; + } + } - Ok(()) + Ok(()) } async fn handle_relays( - conn: &mut PoolConnection, - this: &Controller, - relays: Vec, + conn: &mut PoolConnection, + app_state: &Addr, + this: &Controller, + relays: Vec, ) -> Result<(), EmgauwaError> { - for relay in relays { - if relay.controller.uid != this.c.uid { - return Err(EmgauwaError::Other(String::from( - "Controller UID mismatch during relay update", - ))); - } - let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number) - .await? - .ok_or(DatabaseError::NotFound)?; + for relay in relays { + if relay.controller.uid != this.c.uid { + return Err(EmgauwaError::Other(String::from( + "Controller UID mismatch during relay update", + ))); + } + let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number) + .await? + .ok_or(DatabaseError::NotFound)?; - db_relay.update(conn, relay.r.name.as_str()).await?; + db_relay.update(conn, relay.r.name.as_str()).await?; - handle_schedules(conn, relay.schedules.clone()).await?; + handle_schedules(conn, relay.schedules.clone()).await?; - let mut schedules = Vec::new(); // We need to get the schedules from the database to have the right IDs - for schedule in relay.schedules { - schedules.push( - DbSchedule::get_by_uid(conn, &schedule.uid) - .await? - .ok_or(DatabaseError::NotFound)?, - ); - } + let mut schedules = Vec::new(); // We need to get the schedules from the database to have the right IDs + for schedule in &relay.schedules { + schedules.push( + DbSchedule::get_by_uid(conn, &schedule.uid) + .await? + .ok_or(DatabaseError::NotFound)?, + ); + } - DbJunctionRelaySchedule::set_schedules(conn, &db_relay, schedules.iter().collect()).await?; - } + if let Some((override_schedule, weekday)) = relay.unwrap_override_schedule() { + utils::app_state_relay_override_schedule( + app_state, + relay.r.number, + override_schedule.clone(), + weekday, + ) + .await?; + } - Ok(()) + DbJunctionRelaySchedule::set_schedules(conn, &db_relay, schedules.iter().collect()).await?; + } + + Ok(()) } async fn handle_relay_pulse( - app_state: &Addr, - relay_num: i64, - duration: Option, + app_state: &Addr, + relay_num: i64, + duration: Option, ) -> Result<(), EmgauwaError> { - utils::app_state_relay_pulse(app_state, relay_num, duration).await + utils::app_state_relay_pulse(app_state, relay_num, duration).await }