Add waiting for next period
This commit is contained in:
		
							parent
							
								
									6f2deb38e1
								
							
						
					
					
						commit
						0b0350da0e
					
				
					 5 changed files with 105 additions and 37 deletions
				
			
		| 
						 | 
				
			
			@ -1,11 +1,12 @@
 | 
			
		|||
use std::time::Duration;
 | 
			
		||||
 | 
			
		||||
use actix::Addr;
 | 
			
		||||
use chrono::Local;
 | 
			
		||||
use chrono::{Local, Timelike};
 | 
			
		||||
use emgauwa_lib::constants::RELAYS_RETRY_TIMEOUT;
 | 
			
		||||
use emgauwa_lib::errors::EmgauwaError;
 | 
			
		||||
use emgauwa_lib::models::Controller;
 | 
			
		||||
use emgauwa_lib::types::RelayStates;
 | 
			
		||||
use emgauwa_lib::types::{RelayStates, Weekday};
 | 
			
		||||
use emgauwa_lib::utils::printable_relay_states;
 | 
			
		||||
use futures::pin_mut;
 | 
			
		||||
use tokio::time;
 | 
			
		||||
use tokio::time::timeout;
 | 
			
		||||
| 
						 | 
				
			
			@ -26,58 +27,35 @@ pub async fn run_relays_loop(app_state: Addr<AppState>) {
 | 
			
		|||
}
 | 
			
		||||
 | 
			
		||||
async fn run_relays(app_state: &Addr<AppState>) -> Result<(), EmgauwaError> {
 | 
			
		||||
	let default_duration = Duration::new(10, 0);
 | 
			
		||||
	let notifier = &*app_state_get_controller_notifier(app_state).await?;
 | 
			
		||||
 | 
			
		||||
	let mut last_weekday = emgauwa_lib::utils::get_weekday();
 | 
			
		||||
	let mut this = utils::app_state_get_this(app_state).await?;
 | 
			
		||||
	let mut relay_states: RelayStates = Vec::new();
 | 
			
		||||
	init_relay_states(&mut relay_states, &this);
 | 
			
		||||
	calc_relay_states(&mut relay_states, &mut this, app_state).await?;
 | 
			
		||||
 | 
			
		||||
	loop {
 | 
			
		||||
		log::debug!(
 | 
			
		||||
			"Relay loop at {}: {}",
 | 
			
		||||
			Local::now().naive_local().time(),
 | 
			
		||||
			printable_relay_states(&this.get_relay_states())
 | 
			
		||||
		);
 | 
			
		||||
 | 
			
		||||
		let notifier_future = notifier.notified();
 | 
			
		||||
		pin_mut!(notifier_future);
 | 
			
		||||
		let timeout_result = timeout(default_duration, &mut notifier_future).await;
 | 
			
		||||
		let mut changed = timeout_result.is_ok();
 | 
			
		||||
		let mut changed = timeout(get_next_duration(&this), &mut notifier_future)
 | 
			
		||||
			.await
 | 
			
		||||
			.is_ok();
 | 
			
		||||
 | 
			
		||||
		let current_weekday = emgauwa_lib::utils::get_weekday();
 | 
			
		||||
		if current_weekday != last_weekday {
 | 
			
		||||
			log::debug!("Weekday changed");
 | 
			
		||||
			last_weekday = current_weekday;
 | 
			
		||||
			utils::app_state_reload(app_state).await?;
 | 
			
		||||
			changed = true;
 | 
			
		||||
		}
 | 
			
		||||
		check_weekday(app_state, &mut last_weekday, &mut changed).await?;
 | 
			
		||||
 | 
			
		||||
		if changed {
 | 
			
		||||
			log::debug!("Reloading controller in relay loop");
 | 
			
		||||
			this = utils::app_state_get_this(app_state).await?;
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		let mut relay_debug = String::new();
 | 
			
		||||
		let now = Local::now().time();
 | 
			
		||||
		this.relays
 | 
			
		||||
			.iter()
 | 
			
		||||
			.zip(relay_states.iter_mut())
 | 
			
		||||
			.for_each(|(relay, state)| {
 | 
			
		||||
				*state = Some(relay.active_schedule.is_on(&now));
 | 
			
		||||
 | 
			
		||||
				relay_debug.push_str(&format!(
 | 
			
		||||
					"{}{}: {} ; ",
 | 
			
		||||
					if relay.active_schedule.is_on(&now) {
 | 
			
		||||
						"+"
 | 
			
		||||
					} else {
 | 
			
		||||
						"-"
 | 
			
		||||
					},
 | 
			
		||||
					relay.r.name,
 | 
			
		||||
					relay.active_schedule.name
 | 
			
		||||
				));
 | 
			
		||||
			});
 | 
			
		||||
		log::debug!(
 | 
			
		||||
			"Relay loop at {}: {}",
 | 
			
		||||
			Local::now().naive_local().time(),
 | 
			
		||||
			relay_debug
 | 
			
		||||
		);
 | 
			
		||||
		utils::app_state_update_relays_on(app_state, relay_states.clone()).await?;
 | 
			
		||||
		calc_relay_states(&mut relay_states, &mut this, app_state).await?;
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -87,3 +65,54 @@ fn init_relay_states(relay_states: &mut RelayStates, this: &Controller) {
 | 
			
		|||
		relay_states.push(None);
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
async fn calc_relay_states(
 | 
			
		||||
	relay_states: &mut RelayStates,
 | 
			
		||||
	this: &mut Controller,
 | 
			
		||||
	app_state: &Addr<AppState>,
 | 
			
		||||
) -> Result<(), EmgauwaError> {
 | 
			
		||||
	let now = Local::now().time();
 | 
			
		||||
 | 
			
		||||
	this.relays
 | 
			
		||||
		.iter_mut()
 | 
			
		||||
		.zip(relay_states.iter_mut())
 | 
			
		||||
		.for_each(|(relay, state)| {
 | 
			
		||||
			relay.is_on = Some(relay.active_schedule.is_on(&now));
 | 
			
		||||
			*state = relay.is_on;
 | 
			
		||||
		});
 | 
			
		||||
	utils::app_state_update_relays_on(app_state, relay_states.clone()).await
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
fn get_next_duration(this: &Controller) -> Duration {
 | 
			
		||||
	let now = Local::now().time();
 | 
			
		||||
	let now_in_s = now.num_seconds_from_midnight();
 | 
			
		||||
	let next_timestamp = this
 | 
			
		||||
		.get_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);
 | 
			
		||||
 | 
			
		||||
	log::debug!(
 | 
			
		||||
		"Next timestamp: {}; Waiting for {}s",
 | 
			
		||||
		next_timestamp,
 | 
			
		||||
		duration_to_next.as_secs()
 | 
			
		||||
	);
 | 
			
		||||
 | 
			
		||||
	duration_to_next
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
async fn check_weekday(
 | 
			
		||||
	app_state: &Addr<AppState>,
 | 
			
		||||
	last_weekday: &mut Weekday,
 | 
			
		||||
	changed: &mut bool,
 | 
			
		||||
) -> Result<(), EmgauwaError> {
 | 
			
		||||
	let current_weekday = emgauwa_lib::utils::get_weekday();
 | 
			
		||||
	if current_weekday.ne(last_weekday) {
 | 
			
		||||
		log::debug!("Weekday changed");
 | 
			
		||||
		*last_weekday = current_weekday;
 | 
			
		||||
		utils::app_state_reload(app_state).await?;
 | 
			
		||||
		*changed = true;
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	Ok(())
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -54,6 +54,25 @@ impl Period {
 | 
			
		|||
	pub fn is_on(&self, now: &NaiveTime) -> bool {
 | 
			
		||||
		self.start.eq(&self.end) || (self.start.le(now) && self.end.gt(now))
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	pub fn get_next_time(&self, now: &NaiveTime) -> Option<NaiveTime> {
 | 
			
		||||
		if self.start.eq(&self.end) {
 | 
			
		||||
			// this period is always on
 | 
			
		||||
			return None;
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		let start_after_now = self.start.gt(now);
 | 
			
		||||
		let end_after_now = self.end.gt(now);
 | 
			
		||||
		let start_before_end = self.start.lt(&self.end);
 | 
			
		||||
 | 
			
		||||
		return match (start_after_now, end_after_now, start_before_end) {
 | 
			
		||||
			(false, false, _) => None,              // both before now
 | 
			
		||||
			(true, false, _) => Some(self.start),   // only start after now
 | 
			
		||||
			(false, true, _) => Some(self.end),     // only end after now
 | 
			
		||||
			(true, true, true) => Some(self.start), // both after now but start first
 | 
			
		||||
			(true, true, false) => Some(self.end),  // both after now but end first
 | 
			
		||||
		};
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
impl Type<Sqlite> for DbPeriods {
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -198,4 +198,12 @@ impl DbSchedule {
 | 
			
		|||
	pub fn is_on(&self, now: &NaiveTime) -> bool {
 | 
			
		||||
		self.periods.0.iter().any(|period| period.is_on(now))
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	pub fn get_next_time(&self, now: &NaiveTime) -> Option<NaiveTime> {
 | 
			
		||||
		self.periods
 | 
			
		||||
			.0
 | 
			
		||||
			.iter()
 | 
			
		||||
			.filter_map(|period| period.get_next_time(now))
 | 
			
		||||
			.min()
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -1,4 +1,5 @@
 | 
			
		|||
use actix::MessageResponse;
 | 
			
		||||
use chrono::NaiveTime;
 | 
			
		||||
use futures::executor::block_on;
 | 
			
		||||
use serde_derive::{Deserialize, Serialize};
 | 
			
		||||
use sqlx::pool::PoolConnection;
 | 
			
		||||
| 
						 | 
				
			
			@ -62,4 +63,11 @@ impl Controller {
 | 
			
		|||
	pub fn get_relay_states(&self) -> RelayStates {
 | 
			
		||||
		self.relays.iter().map(|r| r.is_on).collect()
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	pub fn get_next_time(&self, now: &NaiveTime) -> Option<NaiveTime> {
 | 
			
		||||
		self.relays
 | 
			
		||||
			.iter()
 | 
			
		||||
			.filter_map(|r| r.active_schedule.get_next_time(now))
 | 
			
		||||
			.min()
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -79,4 +79,8 @@ impl Relay {
 | 
			
		|||
	pub fn is_on(&self, now: &NaiveTime) -> bool {
 | 
			
		||||
		self.active_schedule.is_on(now)
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	pub fn get_next_time(&self, now: &NaiveTime) -> Option<NaiveTime> {
 | 
			
		||||
		self.active_schedule.get_next_time(now)
 | 
			
		||||
	}
 | 
			
		||||
}
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue