tokio-modbus instead of libmodbus-rs

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tesla-charge-controller"
-version = "1.4.0"
+version = "1.5.0-pre"
 edition = "2021"
 license = "MITNFA"
 description = "Controls Tesla charge rate based on solar charge data"
@@ -28,7 +28,8 @@ prometheus = "0.13"
 env_logger = "0.10"
 log = "0.4"
 serialport = "4.3"
-libmodbus-rs = "0.8.3"
+tokio-modbus = "0.13.1"
+tokio-serial = "5.4.4"
 if_chain = "1.0.2"
 notify-debouncer-mini = { version = "0.4.1", default-features = false }
 lazy_static = "1.4"

src/api_interface.rs

@@ -470,7 +470,9 @@ impl TeslaInterface {
             .map(|v| {
                 if self.monitored_values.conn_charge_cable != v.conn_charge_cable {
                     log::warn!("Current conn charge cable: \"{}\"", v.conn_charge_cable);
-                    self.monitored_values.conn_charge_cable = v.conn_charge_cable.clone();
+                    self.monitored_values
+                        .conn_charge_cable
+                        .clone_from(&v.conn_charge_cable);
                 }
                 v.into()
             });
@@ -507,7 +509,9 @@ impl TeslaInterface {
             .and_then(|v| {
                 if self.monitored_values.climate_keeper != v.climate_keeper_mode {
                     log::warn!("Current climate keeper mode: \"{}\"", v.climate_keeper_mode);
-                    self.monitored_values.climate_keeper = v.climate_keeper_mode.clone();
+                    self.monitored_values
+                        .climate_keeper
+                        .clone_from(&v.climate_keeper_mode);
                 }
                 v.try_into().ok()
             });

src/charge_controllers/tristar.rs

@@ -1,5 +1,5 @@
-use libmodbus_rs::{Modbus, ModbusClient, ModbusRTU};
 use prometheus::core::{AtomicI64, GenericGauge};
+use tokio_modbus::client::Reader;
 
 use crate::{
     charge_controllers::gauges::*,
@@ -44,7 +44,7 @@ impl Scaling {
 pub struct Tristar {
     state: TristarState,
     port_name: String,
-    modbus: Modbus,
+    modbus: tokio_modbus::client::Context,
     data_in: [u16; RAM_ARRAY_SIZE],
     charge_state_gauges: ChargeStateGauges,
     consecutive_errors: usize,
@@ -233,16 +233,12 @@ impl ChargeStateGauges {
     }
 }
 
-const MAX_CONSECUTIVE_ERRORS: usize = 5;
-
 impl Tristar {
-    pub fn new(serial_port: String, baud_rate: i32) -> Result<Self, TristarError> {
-        let parity = 'N';
-        let data_bit = 8;
-        let stop_bit = 2;
-        let mut modbus = Modbus::new_rtu(&serial_port, baud_rate, parity, data_bit, stop_bit)?;
-        modbus.set_slave(DEVICE_ID)?;
-        modbus.connect()?;
+    pub fn new(serial_port: String, baud_rate: u32) -> Result<Self, TristarError> {
+        let modbus_serial =
+            tokio_serial::SerialStream::open(&tokio_serial::new(&serial_port, baud_rate))?;
+        let slave = tokio_modbus::Slave(DEVICE_ID);
+        let modbus = tokio_modbus::client::rtu::attach_slave(modbus_serial, slave);
         let charge_state_gauges = ChargeStateGauges::new(&serial_port);
         Ok(Self {
             state: Default::default(),
@@ -254,9 +250,10 @@ impl Tristar {
         })
     }
 
-    pub fn refresh(&mut self) {
+    pub async fn refresh(&mut self) {
         if let Some(new_state) = self
             .get_data()
+            .await
             .some_or_print_with("reading tristar state")
             .map(|scaling| TristarState::from_ram(scaling, &self.data_in))
         {
@@ -297,21 +294,24 @@ impl Tristar {
 
             self.charge_state_gauges.set(new_state.charge_state);
             self.state = new_state;
-        } else {
-            self.consecutive_errors += 1;
-            if self.consecutive_errors >= MAX_CONSECUTIVE_ERRORS {
-                self.modbus.close();
-                if let Err(e) = self.modbus.connect() {
-                    log::error!("error reconnecting to modbus device: {e:?}");
-                }
-            }
         }
+        // else {
+        //     self.consecutive_errors += 1;
+        //     if self.consecutive_errors >= MAX_CONSECUTIVE_ERRORS {
+        //         self.modbus.close();
+        //         if let Err(e) = self.modbus.connect() {
+        //             log::error!("error reconnecting to modbus device: {e:?}");
+        //         }
+        //     }
+        // }
     }
 
-    fn get_data(&mut self) -> Result<Scaling, TristarError> {
-        self.modbus
-            .read_registers(0x0000, RAM_DATA_SIZE + 1, &mut self.data_in)?;
-        let scaling = Scaling::from(&self.data_in);
+    async fn get_data(&mut self) -> Result<Scaling, TristarError> {
+        let data = self
+            .modbus
+            .read_holding_registers(0x0000, RAM_DATA_SIZE + 1)
+            .await??;
+        let scaling = Scaling::from(&data);
         Ok(scaling)
     }
 }

src/config.rs

@@ -157,7 +157,7 @@ pub enum ChargeControllerConfig {
     },
     Tristar {
         serial_port: String,
-        baud_rate: i32,
+        baud_rate: u32,
         watch_interval_seconds: u64,
     },
 }

src/errors.rs

@@ -104,14 +104,12 @@ pub enum TeslaStateParseError {
 
 #[derive(Error, Debug)]
 pub enum TristarError {
-    #[error("modbus error")]
-    Modbus(libmodbus_rs::prelude::Error),
-}
-
-impl From<libmodbus_rs::prelude::Error> for TristarError {
-    fn from(value: libmodbus_rs::prelude::Error) -> Self {
-        Self::Modbus(value)
-    }
+    #[error(transparent)]
+    Modbus(#[from] tokio_modbus::Error),
+    #[error(transparent)]
+    ModbusException(#[from] tokio_modbus::Exception),
+    #[error(transparent)]
+    Serial(#[from] tokio_serial::Error),
 }
 
 #[derive(Error, Debug)]

src/main.rs

@@ -72,8 +72,7 @@ async fn main() {
             let (api_requests, mut api_receiver) = tokio::sync::mpsc::unbounded_channel();
             // and to the pli thread
             let (pli_requests, mut pli_receiver) = tokio::sync::mpsc::unbounded_channel();
-            // and to the charge rate controller thread
-            let (tcrc_requests, mut tcrc_receiver) = tokio::sync::mpsc::unbounded_channel();
+
             // try to spawn the pli loop
             let pli = {
                 let config = access_config();
@@ -146,7 +145,7 @@ async fn main() {
                                 );
                                 loop {
                                     interval.tick().await;
-                                    tristar.refresh();
+                                    tristar.refresh().await;
                                 }
                             })
                         }),
@@ -173,7 +172,7 @@ async fn main() {
                 pl_state,
                 api_requests,
                 pli_requests,
-                tcrc_requests,
+                // tcrc_requests,
             ));
 
             if let Some((mut interface, mut tesla_charge_rate_controller)) = interface_and_tcrc {
@@ -222,10 +221,6 @@ async fn main() {
                             api_message = api_receiver.recv() => match api_message {
                                 Some(message) => interface.process_request(message).await,
                                 None => panic!("Tesla send channel dropped")
-                            },
-                            tcrc_message = tcrc_receiver.recv() => match tcrc_message {
-                                Some(message) => tesla_charge_rate_controller.process_request(message),
-                                None => panic!("TCRC send channel dropped")
                             }
                         }
                     }

src/server/mod.rs

@@ -16,7 +16,7 @@ use crate::{
     charge_controllers::pl::{PlState, PliRequest, RegulatorState},
     config::{access_config, write_to_config},
     errors::{PrintErrors, ServerError},
-    tesla_charge_rate::{TcrcRequest, TcrcState},
+    tesla_charge_rate::TcrcState,
     types::CarState,
 };
 
@@ -30,7 +30,6 @@ pub struct ServerState {
     pub pl_state: Option<Arc<RwLock<PlState>>>,
     pub api_requests: UnboundedSender<InterfaceRequest>,
     pub pli_requests: UnboundedSender<PliRequest>,
-    pub tcrc_requests: UnboundedSender<TcrcRequest>,
     waiting_for_auth: std::sync::Mutex<Option<(String, std::time::Instant)>>,
 }
 
@@ -41,7 +40,6 @@ impl ServerState {
         pl_state: Option<Arc<RwLock<PlState>>>,
         api_requests: UnboundedSender<InterfaceRequest>,
         pli_requests: UnboundedSender<PliRequest>,
-        tcrc_requests: UnboundedSender<TcrcRequest>,
     ) -> Self {
         Self {
             car_state,
@@ -49,7 +47,7 @@ impl ServerState {
             pl_state,
             api_requests,
             pli_requests,
-            tcrc_requests,
+            // tcrc_requests,
             waiting_for_auth: std::sync::Mutex::new(None),
         }
     }