Cargo.lock

@@ -239,7 +239,7 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
 
 [[package]]
 name = "charge-controller-supervisor"
-version = "1.9.9-pre-30"
+version = "1.9.9-pre-27"
 dependencies = [
  "bitflags 2.7.0",
  "chrono",
@@ -2205,7 +2205,7 @@ dependencies = [
 
 [[package]]
 name = "tesla-charge-controller"
-version = "1.9.9-pre-30"
+version = "1.9.9-pre-27"
 dependencies = [
  "chrono",
  "clap",

Cargo.toml

@@ -4,7 +4,7 @@ default-members = ["charge-controller-supervisor"]
 resolver = "2"
 
 [workspace.package]
-version = "1.9.9-pre-30"
+version = "1.9.9-pre-27"
 
 [workspace.lints.clippy]
 pedantic = "warn"

charge-controller-supervisor/src/controller.rs

@@ -8,9 +8,8 @@ pub struct Controller {
     interval: std::time::Duration,
     inner: ControllerInner,
     data: std::sync::Arc<ControllerData>,
-    follow_voltage: bool,
-    voltage_rx: tokio::sync::watch::Receiver<VoltageCommand>,
-    voltage_tx: Option<tokio::sync::watch::Sender<VoltageCommand>>,
+    voltage_rx: Option<tokio::sync::mpsc::UnboundedReceiver<VoltageCommand>>,
+    voltage_tx: Option<MultiTx>,
     settings_last_read: Option<std::time::Instant>,
 }
 
@@ -54,15 +53,16 @@ pub enum ControllerSettings {
 
 #[derive(Clone, Copy, Debug)]
 pub enum VoltageCommand {
-    None,
     Set(f64),
 }
 
 impl Controller {
     pub async fn new(
         config: crate::config::ChargeControllerConfig,
-        voltage_rx: tokio::sync::watch::Receiver<VoltageCommand>,
-    ) -> eyre::Result<Self> {
+    ) -> eyre::Result<(
+        Self,
+        Option<tokio::sync::mpsc::UnboundedSender<VoltageCommand>>,
+    )> {
         let inner = match config.variant {
             crate::config::ChargeControllerVariant::Tristar => ControllerInner::Tristar(
                 tristar::Tristar::new(&config.name, &config.transport).await?,
@@ -81,16 +81,25 @@ impl Controller {
 
         let data = std::sync::Arc::new(ControllerData::new());
 
-        Ok(Self {
-            name: config.name,
-            interval: std::time::Duration::from_secs(config.watch_interval_seconds),
-            inner,
-            data,
-            voltage_rx,
-            voltage_tx: None,
-            settings_last_read: None,
-            follow_voltage: config.follow_primary,
-        })
+        let (voltage_tx, voltage_rx) = if config.follow_primary {
+            let (a, b) = tokio::sync::mpsc::unbounded_channel();
+            (Some(a), Some(b))
+        } else {
+            (None, None)
+        };
+
+        Ok((
+            Self {
+                name: config.name,
+                interval: std::time::Duration::from_secs(config.watch_interval_seconds),
+                inner,
+                data,
+                voltage_rx,
+                voltage_tx: None,
+                settings_last_read: None,
+            },
+            voltage_tx,
+        ))
     }
 
     pub fn get_data_ptr(&self) -> std::sync::Arc<ControllerData> {
@@ -112,7 +121,7 @@ impl Controller {
                     target
                 );
 
-                tx.send(VoltageCommand::Set(target))?;
+                tx.send_to_all(VoltageCommand::Set(target));
             }
         }
 
@@ -137,12 +146,12 @@ impl Controller {
     }
 
     pub fn name(&self) -> &str {
-        self.name.as_str()
+        &self.name
     }
 
-    pub fn set_tx_to_secondary(&mut self, tx: tokio::sync::watch::Sender<VoltageCommand>) {
+    pub fn set_tx_to_secondary(&mut self, tx: MultiTx) {
         assert!(
-            !self.follow_voltage,
+            self.voltage_rx.is_none(),
             "trying to set {} as primary when it is also a secondary!",
             self.name
         );
@@ -150,22 +159,14 @@ impl Controller {
         self.voltage_tx = Some(tx);
     }
 
-    pub fn get_rx(&mut self) -> &mut tokio::sync::watch::Receiver<VoltageCommand> {
-        &mut self.voltage_rx
+    pub fn get_rx(&mut self) -> Option<&mut tokio::sync::mpsc::UnboundedReceiver<VoltageCommand>> {
+        self.voltage_rx.as_mut()
     }
 
     pub async fn process_command(&mut self, command: VoltageCommand) -> eyre::Result<()> {
         match command {
             VoltageCommand::Set(target_voltage) => {
-                if self.follow_voltage {
-                    self.inner.set_target_voltage(target_voltage).await
-                } else {
-                    Ok(())
-                }
-            }
-            VoltageCommand::None => {
-                // todo: disable voltage control
-                Ok(())
+                self.inner.set_target_voltage(target_voltage).await
             }
         }
     }
@@ -177,7 +178,19 @@ impl Controller {
     }
 }
 
-#[expect(clippy::large_enum_variant)]
+#[derive(Clone)]
+pub struct MultiTx(pub Vec<tokio::sync::mpsc::UnboundedSender<VoltageCommand>>);
+
+impl MultiTx {
+    pub fn send_to_all(&self, command: VoltageCommand) {
+        for sender in &self.0 {
+            if let Err(e) = sender.send(command) {
+                log::error!("failed to send command {command:?}: {e:?}");
+            }
+        }
+    }
+}
+
 pub enum ControllerInner {
     Pl(pl::Pli),
     Tristar(tristar::Tristar),

charge-controller-supervisor/src/controller/tristar/modbus_wrapper.rs

@@ -62,12 +62,13 @@ impl<T> TryInsert for Option<T> {
 
 #[derive(Default, Debug)]
 struct Counters {
-    gateway: usize,
     timeout: usize,
     protocol: usize,
 }
 
-const MAX_ERRORS: usize = 2;
+const MAX_TIMEOUTS: usize = 3;
+
+const MAX_PROTOCOL_ERRORS: usize = 3;
 
 impl Counters {
     fn reset(&mut self) {
@@ -75,12 +76,10 @@ impl Counters {
     }
 
     const fn any_above_max(&self) -> bool {
-        self.gateway > MAX_ERRORS || self.timeout > MAX_ERRORS || self.protocol > MAX_ERRORS
+        self.timeout > MAX_TIMEOUTS || self.protocol > MAX_PROTOCOL_ERRORS
     }
 }
 
-const NUM_TRIES: usize = 3;
-
 impl ModbusTimeout {
     pub async fn new(transport_settings: crate::config::Transport) -> eyre::Result<Self> {
         let context = Some(connect(&transport_settings).await?);
@@ -91,65 +90,52 @@ impl ModbusTimeout {
         })
     }
 
-    async fn with_context<R, D: Copy>(
+    async fn with_context<R, D>(
         &mut self,
         f: ContextFn<R, D>,
         data: D,
     ) -> eyre::Result<Result<R, tokio_modbus::ExceptionCode>> {
-        let mut last_err = None;
-        for _ in 0..NUM_TRIES {
-            if let Ok(context) = self
-                .context
-                .get_or_try_insert_with(async || connect(&self.transport_settings).await)
-                .await
-            {
-                let res = tokio::time::timeout(MODBUS_TIMEOUT, f(context, data)).await;
-                match res {
-                    Ok(Ok(Err(e)))
-                        if e == tokio_modbus::ExceptionCode::GatewayTargetDevice
-                            || e == tokio_modbus::ExceptionCode::GatewayPathUnavailable =>
-                    {
-                        log::warn!("gateway error: {e:?}");
-                        last_err = Some(e.into());
-                        self.counters.gateway += 1;
-                    }
-                    Ok(Ok(v)) => {
-                        self.counters.reset();
-                        return Ok(v);
-                    }
-                    Ok(Err(tokio_modbus::Error::Protocol(e))) => {
-                        // protocol error
-                        log::warn!("protocol error: {e:?}");
-                        last_err = Some(e.into());
-                        self.counters.protocol += 1;
-                    }
-                    Ok(Err(tokio_modbus::Error::Transport(e))) => {
-                        // transport error
-                        log::warn!("reconnecting due to transport error: {e:?}");
-                        last_err = Some(e.into());
-                        self.context = None;
-                    }
-                    Err(_) => {
-                        // timeout
-                        last_err = Some(eyre::eyre!("timeout"));
-                        self.counters.timeout += 1;
-                    }
-                }
-                if self.counters.any_above_max() {
-                    self.context = None;
-                    log::warn!(
-                        "reconnecting due to multiple errors without a successful operation: {:?}",
-                        self.counters
-                    );
+        if let Ok(context) = self
+            .context
+            .get_or_try_insert_with(async || connect(&self.transport_settings).await)
+            .await
+        {
+            let res = tokio::time::timeout(MODBUS_TIMEOUT, f(context, data)).await;
+            match res {
+                Ok(Ok(v)) => {
                     self.counters.reset();
+                    return Ok(v);
+                }
+
+                Ok(Err(tokio_modbus::Error::Protocol(e))) => {
+                    // protocol error
+                    log::warn!("protocol error: {e:?}");
+                    self.counters.protocol += 1;
+                }
+                Ok(Err(tokio_modbus::Error::Transport(e))) => {
+                    // transport error
+                    log::warn!("reconnecting due to transport error: {e:?}");
+                    self.context = None;
+                }
+                Err(_) => {
+                    // timeout
+                    self.counters.timeout += 1;
                 }
-            } else {
-                // failed to reconnect
-                return Err(eyre::eyre!("failed to reconnect to controller"));
             }
+            if self.counters.any_above_max() {
+                self.context = None;
+                log::warn!(
+                    "reconnecting due to multiple errors without a successful operation: {:?}",
+                    self.counters
+                );
+                self.counters.reset();
+            }
+        } else {
+            // failed to reconnect
+            return Err(eyre::eyre!("failed to reconnect to controller"));
         }
 
-        Err(last_err.unwrap_or_else(|| eyre::eyre!("unknown last error????")))
+        Err(eyre::eyre!(":("))
     }
 
     pub async fn write_single_register(

charge-controller-supervisor/src/main.rs

@@ -108,26 +108,29 @@ async fn watch(args: Args) -> eyre::Result<()> {
 
         let mut controllers = Vec::new();
         let mut map = std::collections::HashMap::new();
-
-        let (voltage_tx, voltage_rx) =
-            tokio::sync::watch::channel(controller::VoltageCommand::None);
+        let mut follow_voltage_tx = Vec::new();
 
         for config in &config.charge_controllers {
             let n = config.name.clone();
-            match controller::Controller::new(config.clone(), voltage_rx.clone()).await {
-                Ok(v) => {
+            match controller::Controller::new(config.clone()).await {
+                Ok((v, voltage_tx)) => {
                     map.insert(n, v.get_data_ptr());
                     controllers.push(v);
+                    if let Some(voltage_tx) = voltage_tx {
+                        follow_voltage_tx.push(voltage_tx);
+                    }
                 }
                 Err(e) => log::error!("couldn't connect to {}: {e:?}", n),
             }
         }
 
+        let follow_voltage_tx = controller::MultiTx(follow_voltage_tx);
+
         if let Some(primary) = controllers
             .iter_mut()
             .find(|c| c.name() == config.primary_charge_controller)
         {
-            primary.set_tx_to_secondary(voltage_tx.clone());
+            primary.set_tx_to_secondary(follow_voltage_tx.clone());
         }
 
         drop(config);
@@ -139,7 +142,7 @@ async fn watch(args: Args) -> eyre::Result<()> {
 
         (
             storage::AllControllers::new(map),
-            voltage_tx,
+            follow_voltage_tx,
             controller_tasks,
         )
     };
@@ -150,7 +153,6 @@ async fn watch(args: Args) -> eyre::Result<()> {
         follow_voltage_tx,
     ));
     let server_task = tokio::task::spawn(server.launch());
-    log::warn!("...started!");
 
     tokio::select! {
         v = controller_tasks.next() => {
@@ -178,20 +180,22 @@ async fn watch(args: Args) -> eyre::Result<()> {
 
 async fn run_loop(mut controller: controller::Controller) -> eyre::Result<()> {
     let mut timeout = tokio::time::interval(controller.timeout_interval());
-    timeout.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
     loop {
-        let rx = controller.get_rx();
-        tokio::select! {
-            _ = timeout.tick() => {
-                do_refresh(&mut controller).await;
-            }
-            Ok(()) = rx.changed() => {
-                let command = *rx.borrow();
-                if let Err(e) = controller.process_command(command).await {
-                    log::error!("controller {} failed to process command: {e}", controller.name());
+        if let Some(rx) = controller.get_rx() {
+            tokio::select! {
+                _ = timeout.tick() => {
+                    do_refresh(&mut controller).await;
+                }
+                Some(command) = rx.recv() => {
+                    if let Err(e) = controller.process_command(command).await {
+                        log::error!("controller {} failed to process command: {e}", controller.name());
+                    }
                 }
             }
+        } else {
+            timeout.tick().await;
+            do_refresh(&mut controller).await;
         }
     }
 }

charge-controller-supervisor/src/web.rs

@@ -7,14 +7,14 @@ mod static_handler;
 pub struct ServerState {
     primary_name: String,
     data: AllControllers,
-    tx_to_controllers: tokio::sync::watch::Sender<crate::controller::VoltageCommand>,
+    tx_to_controllers: crate::controller::MultiTx,
 }
 
 impl ServerState {
     pub fn new(
         primary_name: &impl ToString,
         data: AllControllers,
-        tx_to_controllers: tokio::sync::watch::Sender<crate::controller::VoltageCommand>,
+        tx_to_controllers: crate::controller::MultiTx,
     ) -> Self {
         let primary_name = primary_name.to_string();
         Self {
@@ -200,23 +200,20 @@ async fn enable_control() {
 }
 
 #[post("/control/disable")]
-async fn disable_control(state: &State<ServerState>) -> Result<(), ServerError> {
+async fn disable_control(state: &State<ServerState>) {
     log::warn!("disabling control");
     crate::config::write_to_config()
         .await
         .enable_secondary_control = false;
     state
         .tx_to_controllers
-        .send(crate::controller::VoltageCommand::None)?;
-    Ok(())
+        .send_to_all(crate::controller::VoltageCommand::Set(-1.0));
 }
-
 enum ServerError {
     Prometheus,
     NotFound,
     InvalidPrimaryName,
     NoData,
-    ControllerTx,
 }
 
 impl From<prometheus::Error> for ServerError {
@@ -225,20 +222,12 @@ impl From<prometheus::Error> for ServerError {
     }
 }
 
-impl<T> From<tokio::sync::watch::error::SendError<T>> for ServerError {
-    fn from(_: tokio::sync::watch::error::SendError<T>) -> Self {
-        Self::ControllerTx
-    }
-}
-
 impl<'a> rocket::response::Responder<'a, 'a> for ServerError {
     fn respond_to(self, _: &'a rocket::Request<'_>) -> rocket::response::Result<'a> {
         Err(match self {
             Self::NotFound => rocket::http::Status::NotFound,
             Self::InvalidPrimaryName => rocket::http::Status::ServiceUnavailable,
-            Self::ControllerTx | Self::NoData | Self::Prometheus => {
-                rocket::http::Status::InternalServerError
-            }
+            Self::NoData | Self::Prometheus => rocket::http::Status::InternalServerError,
         })
     }
 }

rust-toolchain.toml

@@ -1,3 +1,2 @@
 [toolchain]
-channel = "nightly-2025-01-16"
-targets = ["aarch64-unknown-linux-musl"]
+channel = "nightly"

tesla-charge-controller/src/api/http.rs

@@ -95,7 +95,6 @@ impl Vehicle {
         Ok(state.charge_state)
     }
 
-    #[expect(dead_code, reason = "active charge control not yet implemented")]
     pub async fn set_charging_amps(&self, charging_amps: i64) -> eyre::Result<()> {
         self.client
             .post(format!(

tesla-charge-controller/src/api/mod.rs

@@ -42,11 +42,11 @@ impl Car {
         }
     }
 
-    pub const fn vehicle(&self) -> &http::Vehicle {
+    pub fn vehicle(&self) -> &http::Vehicle {
         &self.vehicle
     }
 
-    pub const fn state(&self) -> &tokio::sync::RwLock<CarState> {
+    pub fn state(&self) -> &tokio::sync::RwLock<CarState> {
         &self.state
     }
 }
@@ -153,7 +153,7 @@ impl ChargeState {
     }
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
 pub enum ChargingState {
     Charging,
     Stopped,

tesla-charge-controller/src/config.rs

@@ -60,7 +60,7 @@ impl ConfigWatcher {
 async fn overwrite_config(config: Config) -> eyre::Result<()> {
     *CONFIG
         .get()
-        .ok_or_else(|| eyre::eyre!("could not get config"))?
+        .ok_or(eyre::eyre!("could not get config"))?
         .write()
         .await = config;
     Ok(())

tesla-charge-controller/src/control.rs

@@ -4,7 +4,6 @@ pub struct VehicleController {
     control_state: ChargeRateControllerState,
 }
 
-#[expect(dead_code, reason = "not all states are currently in use")]
 pub enum ChargeRateControllerState {
     Inactive,
     Charging { rate_amps: i64 },
@@ -15,7 +14,7 @@ pub enum InterfaceRequest {
 }
 
 impl VehicleController {
-    pub const fn new(
+    pub fn new(
         car: std::sync::Arc<crate::api::Car>,
         requests: tokio::sync::mpsc::UnboundedReceiver<InterfaceRequest>,
     ) -> Self {
@@ -51,10 +50,7 @@ impl VehicleController {
         }
         match self.control_state {
             ChargeRateControllerState::Inactive => {
-                let car_state = self.car.state().read().await;
-                let state = car_state.charge_state().await;
-
-                if let Some(state) = state {
+                if let Some(state) = self.car.state().read().await.charge_state().await {
                     if state.is_charging() {
                         self.control_state = ChargeRateControllerState::Charging {
                             rate_amps: state.charge_amps,
@@ -62,14 +58,10 @@ impl VehicleController {
                     }
                 }
             }
-            ChargeRateControllerState::Charging { rate_amps: _ } => todo!(),
+            ChargeRateControllerState::Charging { rate_amps } => todo!(),
         }
     }
 
-    #[expect(
-        clippy::needless_pass_by_ref_mut,
-        reason = "this will eventually need to mutate self"
-    )]
     pub async fn process_requests(&mut self, req: InterfaceRequest) {
         if let Err(e) = match req {
             InterfaceRequest::FlashLights => self.car.vehicle().flash_lights().await,

tesla-charge-controller/src/main.rs

@@ -1,5 +1,3 @@
-#![allow(clippy::significant_drop_tightening)]
-
 use std::path::PathBuf;
 
 use clap::Parser;

tesla-charge-controller/src/server/mod.rs

@@ -27,7 +27,7 @@ pub struct ServerState {
 }
 
 impl ServerState {
-    pub const fn new(car: Arc<Car>, api_requests: UnboundedSender<InterfaceRequest>) -> Self {
+    pub fn new(car: Arc<Car>, api_requests: UnboundedSender<InterfaceRequest>) -> Self {
         Self { car, api_requests }
     }
 }

tesla-charge-controller/src/server/static_handler.rs

@@ -52,12 +52,10 @@ impl Handler for UiStatic {
                             data: v.contents().to_vec(),
                             name: p,
                         })
-                        .or_else(|| {
-                            UI_DIR_FILES.get_file(&plus_index).map(|v| RawHtml {
-                                data: v.contents().to_vec(),
-                                name: plus_index,
-                            })
-                        });
+                        .or(UI_DIR_FILES.get_file(&plus_index).map(|v| RawHtml {
+                            data: v.contents().to_vec(),
+                            name: plus_index,
+                        }));
                     file.respond_to(req).or_forward((data, Status::NotFound))
                 }
             }