Rename active_schedule to override_schedule and add EmgauwaNow

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<u32>,
 }
 
+#[derive(Message)]
+#[rtype(result = "Result<(), EmgauwaError>")]
+pub struct RelayOverrideSchedule {
+	pub relay_number: i64,
+	pub schedule: Option<DbSchedule>,
+	pub weekday: Weekday,
+}
+
 #[derive(Message)]
 #[rtype(result = "Controller")]
 pub struct GetThis {}
@@ -105,7 +114,7 @@ impl Handler<UpdateRelayStates> 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<RelayPulse> for AppState {
 	}
 }
 
+impl Handler<RelayOverrideSchedule> 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<GetThis> for AppState {
 	type Result = Controller;
 

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<AppState>) -> 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<AppState>) -> 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 {
+	this.relays.iter().for_each(|r| {
-		relay_states.push(None);
+		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<AppState>,
+	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>) -> Duration {
+fn get_next_duration(
+	this: &mut Controller,
+	duration_override: &mut Option<Duration>,
+	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<AppState>,
 	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;

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<AppState>,
+	relay_number: i64,
+	schedule: Option<DbSchedule>,
+	weekday: Weekday,
+) -> Result<(), EmgauwaError> {
+	app_state
+		.send(app_state::RelayOverrideSchedule {
+			relay_number,
+			weekday,
+			schedule,
+		})
+		.await
+		.map_err(EmgauwaError::from)?
+}

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<Sqlite>,
+	pool: Pool<Sqlite>,
-    app_state: &Addr<AppState>,
+	app_state: &Addr<AppState>,
-    message_result: Result<Message, tungstenite::Error>,
+	message_result: Result<Message, tungstenite::Error>,
 ) {
-    let msg = match message_result {
+	let msg = match message_result {
-        Ok(msg) => msg,
+		Ok(msg) => msg,
-        Err(err) => {
+		Err(err) => {
-            log::error!("Error reading message: {}", err);
+			log::error!("Error reading message: {}", err);
-            return;
+			return;
-        }
+		}
-    };
+	};
-    if let Message::Text(text) = msg {
+	if let Message::Text(text) = msg {
-        match serde_json::from_str(&text) {
+		match serde_json::from_str(&text) {
-            Ok(action) => {
+			Ok(action) => {
-                log::debug!("Received action: {:?}", action);
+				log::debug!("Received action: {:?}", action);
-                let mut pool_conn = match pool.acquire().await {
+				let mut pool_conn = match pool.acquire().await {
-                    Ok(conn) => conn,
+					Ok(conn) => conn,
-                    Err(err) => {
+					Err(err) => {
-                        log::error!("Failed to acquire database connection: {:?}", err);
+						log::error!("Failed to acquire database connection: {:?}", err);
-                        return;
+						return;
-                    }
+					}
-                };
+				};
-                let action_res = handle_action(&mut pool_conn, app_state, action).await;
+				let action_res = handle_action(&mut pool_conn, app_state, action).await;
-                if let Err(e) = action_res {
+				if let Err(e) = action_res {
-                    log::error!("Error handling action: {:?}", e);
+					log::error!("Error handling action: {:?}", e);
-                }
+				}
-            }
+			}
-            Err(e) => {
+			Err(e) => {
-                log::error!("Error deserializing action: {:?}", e);
+				log::error!("Error deserializing action: {:?}", e);
-            }
+			}
-        }
+		}
-    }
+	}
 }
 
 pub async fn handle_action(
-    conn: &mut PoolConnection<Sqlite>,
+	conn: &mut PoolConnection<Sqlite>,
-    app_state: &Addr<AppState>,
+	app_state: &Addr<AppState>,
-    action: ControllerWsAction,
+	action: ControllerWsAction,
 ) -> Result<(), EmgauwaError> {
-    let this = app_state_get_this(app_state).await?;
+	let this = app_state_get_this(app_state).await?;
 
-    match action {
+	match action {
-        ControllerWsAction::Controller(controller) => {
+		ControllerWsAction::Controller(controller) => {
-            handle_controller(conn, &this, controller).await?
+			handle_controller(conn, &this, controller).await?
-        }
+		}
-        ControllerWsAction::Relays(relays) => handle_relays(conn, &this, relays).await?,
+		ControllerWsAction::Relays(relays) => handle_relays(conn, app_state, &this, relays).await?,
-        ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?,
+		ControllerWsAction::Schedules(schedules) => handle_schedules(conn, schedules).await?,
-        ControllerWsAction::RelayPulse((relay_num, duration)) => {
+		ControllerWsAction::RelayPulse((relay_num, duration)) => {
-            handle_relay_pulse(app_state, relay_num, duration).await?
+			handle_relay_pulse(app_state, relay_num, duration).await?
-        }
+		}
-        _ => return Ok(()),
+		_ => return Ok(()),
-    };
+	};
 
-    utils::app_state_reload(app_state).await
+	utils::app_state_reload(app_state).await
 }
 
 async fn handle_controller(
-    conn: &mut PoolConnection<Sqlite>,
+	conn: &mut PoolConnection<Sqlite>,
-    this: &Controller,
+	this: &Controller,
-    controller: Controller,
+	controller: Controller,
 ) -> Result<(), EmgauwaError> {
-    if controller.c.uid != this.c.uid {
+	if controller.c.uid != this.c.uid {
-        return Err(EmgauwaError::Other(String::from(
+		return Err(EmgauwaError::Other(String::from(
-            "Controller UID mismatch during update",
+			"Controller UID mismatch during update",
-        )));
+		)));
-    }
+	}
-    DbController::get_by_uid(conn, &controller.c.uid)
+	DbController::get_by_uid(conn, &controller.c.uid)
-        .await?
+		.await?
-        .ok_or(DatabaseError::NotFound)?
+		.ok_or(DatabaseError::NotFound)?
-        .update(conn, controller.c.name.as_str(), this.c.relay_count)
+		.update(conn, controller.c.name.as_str(), this.c.relay_count)
-        .await?;
+		.await?;
 
-    Ok(())
+	Ok(())
 }
 
 async fn handle_schedules(
-    conn: &mut PoolConnection<Sqlite>,
+	conn: &mut PoolConnection<Sqlite>,
-    schedules: Vec<DbSchedule>,
+	schedules: Vec<DbSchedule>,
 ) -> Result<(), EmgauwaError> {
-    let mut handled_uids = vec![
+	let mut handled_uids = vec![
-        // on and off schedules are always present and should not be updated
+		// on and off schedules are always present and should not be updated
-        ScheduleUid::On,
+		ScheduleUid::On,
-        ScheduleUid::Off,
+		ScheduleUid::Off,
-    ];
+	];
-    for schedule in schedules {
+	for schedule in schedules {
-        if handled_uids.contains(&schedule.uid) {
+		if handled_uids.contains(&schedule.uid) {
-            continue;
+			continue;
-        }
+		}
-        handled_uids.push(schedule.uid.clone());
+		handled_uids.push(schedule.uid.clone());
 
-        log::debug!("Handling schedule: {:?}", schedule);
+		log::debug!("Handling schedule: {:?}", schedule);
-        let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?;
+		let schedule_db = DbSchedule::get_by_uid(conn, &schedule.uid).await?;
 
-        if let Some(schedule_db) = schedule_db {
+		if let Some(schedule_db) = schedule_db {
-            schedule_db
+			schedule_db
-                .update(conn, schedule.name.as_str(), &schedule.periods)
+				.update(conn, schedule.name.as_str(), &schedule.periods)
-                .await?;
+				.await?;
-        } else {
+		} else {
-            DbSchedule::create(
+			DbSchedule::create(
-                conn,
+				conn,
-                schedule.uid.clone(),
+				schedule.uid.clone(),
-                schedule.name.as_str(),
+				schedule.name.as_str(),
-                &schedule.periods,
+				&schedule.periods,
-            )
+			)
-                .await?;
+			.await?;
-        }
+		}
-    }
+	}
 
-    Ok(())
+	Ok(())
 }
 
 async fn handle_relays(
-    conn: &mut PoolConnection<Sqlite>,
+	conn: &mut PoolConnection<Sqlite>,
-    this: &Controller,
+	app_state: &Addr<AppState>,
-    relays: Vec<Relay>,
+	this: &Controller,
+	relays: Vec<Relay>,
 ) -> Result<(), EmgauwaError> {
-    for relay in relays {
+	for relay in relays {
-        if relay.controller.uid != this.c.uid {
+		if relay.controller.uid != this.c.uid {
-            return Err(EmgauwaError::Other(String::from(
+			return Err(EmgauwaError::Other(String::from(
-                "Controller UID mismatch during relay update",
+				"Controller UID mismatch during relay update",
-            )));
+			)));
-        }
+		}
-        let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number)
+		let db_relay = DbRelay::get_by_controller_and_num(conn, &this.c, relay.r.number)
-            .await?
+			.await?
-            .ok_or(DatabaseError::NotFound)?;
+			.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
+		let mut schedules = Vec::new(); // We need to get the schedules from the database to have the right IDs
-        for schedule in relay.schedules {
+		for schedule in &relay.schedules {
-            schedules.push(
+			schedules.push(
-                DbSchedule::get_by_uid(conn, &schedule.uid)
+				DbSchedule::get_by_uid(conn, &schedule.uid)
-                    .await?
+					.await?
-                    .ok_or(DatabaseError::NotFound)?,
+					.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<AppState>,
+	app_state: &Addr<AppState>,
-    relay_num: i64,
+	relay_num: i64,
-    duration: Option<u32>,
+	duration: Option<u32>,
 ) -> Result<(), EmgauwaError> {
-    utils::app_state_relay_pulse(app_state, relay_num, duration).await
+	utils::app_state_relay_pulse(app_state, relay_num, duration).await
 }