Add handler for relay states

This commit is contained in:
Tobias Reisinger 2024-04-25 14:10:32 +02:00
parent 82f2d49dc6
commit 55617dbd7c
Signed by: serguzim
GPG key ID: 13AD60C237A28DFE
8 changed files with 53 additions and 20 deletions

View file

@ -3,6 +3,7 @@ use std::sync::Arc;
use actix::{Actor, Context, Handler, Message}; use actix::{Actor, Context, Handler, Message};
use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::errors::EmgauwaError;
use emgauwa_lib::models::Controller; use emgauwa_lib::models::Controller;
use emgauwa_lib::types::RelayStates;
use futures::executor::block_on; use futures::executor::block_on;
use sqlx::{Pool, Sqlite}; use sqlx::{Pool, Sqlite};
use tokio::sync::Notify; use tokio::sync::Notify;
@ -13,8 +14,8 @@ pub struct Reload {}
#[derive(Message)] #[derive(Message)]
#[rtype(result = "()")] #[rtype(result = "()")]
pub struct UpdateRelaysOn { pub struct UpdateRelayStates {
pub relays_are_on: Vec<Option<bool>>, pub relay_states: RelayStates,
} }
#[derive(Message)] #[derive(Message)]
@ -74,17 +75,11 @@ impl Handler<Reload> for AppState {
} }
} }
impl Handler<UpdateRelaysOn> for AppState { impl Handler<UpdateRelayStates> for AppState {
type Result = (); type Result = ();
fn handle(&mut self, msg: UpdateRelaysOn, _ctx: &mut Self::Context) -> Self::Result { fn handle(&mut self, msg: UpdateRelayStates, _ctx: &mut Self::Context) -> Self::Result {
self.this self.this.apply_relay_states(&msg.relay_states);
.relays
.iter_mut()
.zip(msg.relays_are_on.iter())
.for_each(|(relay, is_on)| {
relay.is_on = *is_on;
});
self.notify_relay_change(); self.notify_relay_change();
} }

View file

@ -5,6 +5,7 @@ use chrono::Local;
use emgauwa_lib::constants::RELAYS_RETRY_TIMEOUT; use emgauwa_lib::constants::RELAYS_RETRY_TIMEOUT;
use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::errors::EmgauwaError;
use emgauwa_lib::models::Controller; use emgauwa_lib::models::Controller;
use emgauwa_lib::types::RelayStates;
use futures::pin_mut; use futures::pin_mut;
use tokio::time; use tokio::time;
use tokio::time::timeout; use tokio::time::timeout;
@ -30,7 +31,7 @@ async fn run_relays(app_state: &Addr<AppState>) -> Result<(), EmgauwaError> {
let mut last_weekday = emgauwa_lib::utils::get_weekday(); let mut last_weekday = emgauwa_lib::utils::get_weekday();
let mut this = utils::app_state_get_this(app_state).await?; let mut this = utils::app_state_get_this(app_state).await?;
let mut relay_states: Vec<Option<bool>> = Vec::new(); let mut relay_states: RelayStates = Vec::new();
init_relay_states(&mut relay_states, &this); init_relay_states(&mut relay_states, &this);
loop { loop {
@ -80,7 +81,7 @@ async fn run_relays(app_state: &Addr<AppState>) -> Result<(), EmgauwaError> {
} }
} }
fn init_relay_states(relay_states: &mut Vec<Option<bool>>, this: &Controller) { fn init_relay_states(relay_states: &mut RelayStates, this: &Controller) {
relay_states.clear(); relay_states.clear();
for _ in 0..this.c.relay_count { for _ in 0..this.c.relay_count {
relay_states.push(None); relay_states.push(None);

View file

@ -3,6 +3,7 @@ use std::sync::Arc;
use actix::Addr; use actix::Addr;
use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::errors::EmgauwaError;
use emgauwa_lib::models::Controller; use emgauwa_lib::models::Controller;
use emgauwa_lib::types::RelayStates;
use tokio::sync::Notify; use tokio::sync::Notify;
use crate::app_state; use crate::app_state;
@ -42,11 +43,11 @@ pub async fn app_state_reload(app_state: &Addr<AppState>) -> Result<(), EmgauwaE
pub async fn app_state_update_relays_on( pub async fn app_state_update_relays_on(
app_state: &Addr<AppState>, app_state: &Addr<AppState>,
relay_states: Vec<Option<bool>>, relay_states: RelayStates,
) -> Result<(), EmgauwaError> { ) -> Result<(), EmgauwaError> {
app_state app_state
.send(app_state::UpdateRelaysOn { .send(app_state::UpdateRelayStates {
relays_are_on: relay_states, relay_states: relay_states,
}) })
.await .await
.map_err(EmgauwaError::from) .map_err(EmgauwaError::from)

View file

@ -78,7 +78,9 @@ async fn read_app_state(
loop { loop {
notifier.notified().await; notifier.notified().await;
log::debug!("Relay change detected"); log::debug!("Relay change detected");
let ws_action = ControllerWsAction::Register(app_state_get_this(&app_state).await?); let this = app_state_get_this(&app_state).await?;
let relay_states = this.get_relay_states();
let ws_action = ControllerWsAction::RelayStates((this.c.uid, relay_states));
let ws_action_json = serde_json::to_string(&ws_action)?; let ws_action_json = serde_json::to_string(&ws_action)?;
tx.unbounded_send(Message::text(ws_action_json)) tx.unbounded_send(Message::text(ws_action_json))
@ -122,9 +124,6 @@ async fn handle_message(
log::error!("Error deserializing action: {:?}", e); log::error!("Error deserializing action: {:?}", e);
} }
}, },
Message::Ping(_) => {
log::debug!("Received ping");
}
_ => {} _ => {}
} }
} }

View file

@ -1,7 +1,10 @@
use std::hash::{Hash, Hasher};
use actix::{Actor, AsyncContext}; use actix::{Actor, AsyncContext};
use emgauwa_lib::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; use emgauwa_lib::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule};
use emgauwa_lib::errors::{DatabaseError, EmgauwaError}; use emgauwa_lib::errors::{DatabaseError, EmgauwaError};
use emgauwa_lib::models::{Controller, FromDbModel}; use emgauwa_lib::models::{Controller, FromDbModel};
use emgauwa_lib::types::{ControllerUid, RelayStates};
use futures::executor::block_on; use futures::executor::block_on;
use sqlx::pool::PoolConnection; use sqlx::pool::PoolConnection;
use sqlx::Sqlite; use sqlx::Sqlite;
@ -81,4 +84,18 @@ impl ControllerWs {
log::debug!("Done registering controller"); log::debug!("Done registering controller");
Ok(()) Ok(())
} }
pub fn handle_relay_states(
&mut self,
ctx: &mut <ControllerWs as Actor>::Context,
controller_uid: ControllerUid,
relay_states: RelayStates,
) -> Result<(), EmgauwaError> {
log::debug!(
"Received relay states: {:?} for {}",
relay_states,
controller_uid
);
Ok(())
}
} }

View file

@ -54,6 +54,9 @@ impl ControllerWs {
) { ) {
let action_res = match action { let action_res = match action {
ControllerWsAction::Register(controller) => self.handle_register(conn, ctx, controller), ControllerWsAction::Register(controller) => self.handle_register(conn, ctx, controller),
ControllerWsAction::RelayStates((controller_uid, relay_states)) => {
self.handle_relay_states(ctx, controller_uid, relay_states)
}
_ => Ok(()), _ => Ok(()),
}; };
if let Err(e) = action_res { if let Err(e) = action_res {

View file

@ -7,6 +7,7 @@ use sqlx::Sqlite;
use crate::db::DbController; use crate::db::DbController;
use crate::errors::{DatabaseError, EmgauwaError}; use crate::errors::{DatabaseError, EmgauwaError};
use crate::models::{convert_db_list_cache, FromDbModel, Relay}; use crate::models::{convert_db_list_cache, FromDbModel, Relay};
use crate::types::RelayStates;
#[derive(Serialize, Deserialize, Debug, Clone, MessageResponse)] #[derive(Serialize, Deserialize, Debug, Clone, MessageResponse)]
pub struct Controller { pub struct Controller {
@ -48,4 +49,17 @@ impl Controller {
} }
Ok(()) Ok(())
} }
pub fn apply_relay_states(&mut self, relay_states: &RelayStates) {
self.relays
.iter_mut()
.zip(relay_states.iter())
.for_each(|(relay, is_on)| {
relay.is_on = *is_on;
});
}
pub fn get_relay_states(&self) -> RelayStates {
self.relays.iter().map(|r| r.is_on).collect()
}
} }

View file

@ -14,6 +14,8 @@ use crate::models::{Controller, Relay};
pub type Weekday = i64; pub type Weekday = i64;
pub type RelayStates = Vec<Option<bool>>;
#[derive(Debug, Serialize, Deserialize, Message)] #[derive(Debug, Serialize, Deserialize, Message)]
#[rtype(result = "Result<(), EmgauwaError>")] #[rtype(result = "Result<(), EmgauwaError>")]
pub enum ControllerWsAction { pub enum ControllerWsAction {
@ -22,4 +24,5 @@ pub enum ControllerWsAction {
Schedules(Vec<DbSchedule>), Schedules(Vec<DbSchedule>),
Relays(Vec<Relay>), Relays(Vec<Relay>),
Controller(Controller), Controller(Controller),
RelayStates((ControllerUid, RelayStates)),
} }