tokio-modbus instead of libmodbus-rs

.gitea/workflows/build.yaml

@@ -0,0 +1,29 @@
+name: Build .deb
+
+on: workflow_call
+
+env:
+  SCCACHE_GHA_ENABLED: "true"
+  RUSTC_WRAPPER: "sccache"
+
+jobs:
+  Build-Deb:
+    runs-on: aarch64
+    steps:
+      - name: Run sccache-cache
+        uses: https://github.com/mozilla-actions/sccache-action@v0.0.5
+      - name: Check out repository code
+        uses: actions/checkout@v3
+        with:
+          submodules: "recursive"
+      - name: Build
+        run: "cargo build --release --target=aarch64-unknown-linux-gnu"
+      - name: Build .deb
+        run: "cargo deb --target=aarch64-unknown-linux-gnu"
+      - name: Upload artifacts
+        uses: actions/upload-artifact@v4
+        with:
+          name: release
+          path: |
+            ./target/aarch64-unknown-linux-gnu/debian/*.deb
+            ./target/aarch64-unknown-linux-gnu/release/tesla-charge-controller

.gitea/workflows/deb.yaml

@@ -1,25 +1,21 @@
-name: Build .deb on release
+name: Build and release .deb
-run-name: Building .deb of latest release and adding to apt repo
 
 on:
   push:
     tags:
-      - "*"
+      - "v[0-9]+.[0-9]+.[0-9]+"
 
 jobs:
-  Build-Deb:
+  Build:
+    uses: ./.gitea/workflows/build.yaml
+
+  Release:
     runs-on: aarch64
     steps:
-      - name: Check out repository code
+      - name: Download artifacts
-        uses: actions/checkout@v3
+        uses: actions/download-artifact@v4
-        with:
-          submodules: "recursive"
-      - name: Build
-        run: "cargo build --release --target=aarch64-unknown-linux-gnu"
-      - name: Build .deb
-        run: "cargo deb --target=aarch64-unknown-linux-gnu"
       - name: Add .deb to apt repository
-        run: "curl --user alex:${{ secrets.PACKAGING_TOKEN }} --upload-file $(ls -t ./target/aarch64-unknown-linux-gnu/debian/*.deb | head -1) https://git.alexjanka.com/api/packages/alex/debian/pool/testing/main/upload"
+        run: "curl --user alex:${{ secrets.PACKAGING_TOKEN }} --upload-file $(ls -t ./release/*.deb | head -1) https://git.alexjanka.com/api/packages/alex/debian/pool/testing/main/upload"
       - name: "Release package"
         id: use-go-action
         uses: https://gitea.com/actions/release-action@main

.gitea/workflows/pre.yaml

@@ -0,0 +1,10 @@
+name: Build prerelease .deb
+
+on:
+  push:
+    tags:
+      - "*-pre"
+
+jobs:
+  Build:
+    uses: ./.gitea/workflows/build.yaml

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tesla-charge-controller"
-version = "1.2.5"
+version = "1.5.0-pre"
 edition = "2021"
 license = "MITNFA"
 description = "Controls Tesla charge rate based on solar charge data"
@@ -28,7 +28,13 @@ 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"
+rand = "0.8"
+reqwest = "0.12.4"
+
+# [patch."https://git.alexjanka.com/alex/tesla-common"]
+# tesla-common = { path = "../tesla-common" }

src/api_interface.rs

@@ -261,12 +261,13 @@ struct MonitoredValues {
     climate_keeper: String,
 }
 
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Debug)]
 // these are the messages that the webserver can send the api thread
 pub enum InterfaceRequest {
     FlashLights,
     StopCharge,
     SetChargeRate(i64),
+    NewAuth(AuthInfo),
 }
 
 const KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60);
@@ -278,6 +279,13 @@ impl TeslaInterface {
         api.refresh().await?;
         let last_refresh = Instant::now();
         info!("Refreshed auth key");
+        AuthInfo {
+            access_token: api.access_token.clone(),
+            refresh_token: api.refresh_token.clone(),
+        }
+        .save(Some(&auth_path))
+        .map_err(tesla_common::SaveError::from)?;
+        info!("Saved auth key");
         let vehicle = api
             .products()
             .await?
@@ -339,6 +347,13 @@ impl TeslaInterface {
                 Ok(_) => {}
                 Err(e) => log::error!("Error setting charge rate: {e:?}"),
             },
+            InterfaceRequest::NewAuth(new) => {
+                self.api.access_token = new.access_token;
+                self.api.refresh_token = new.refresh_token;
+                if let Err(e) = self.save_key() {
+                    log::error!("failed to save key: {e:?}");
+                }
+            }
         }
     }
 
@@ -455,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()
             });
@@ -492,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> {
+    pub fn new(serial_port: String, baud_rate: u32) -> Result<Self, TristarError> {
-        let parity = 'N';
+        let modbus_serial =
-        let data_bit = 8;
+            tokio_serial::SerialStream::open(&tokio_serial::new(&serial_port, baud_rate))?;
-        let stop_bit = 2;
+        let slave = tokio_modbus::Slave(DEVICE_ID);
-        let mut modbus = Modbus::new_rtu(&serial_port, baud_rate, parity, data_bit, stop_bit)?;
+        let modbus = tokio_modbus::client::rtu::attach_slave(modbus_serial, slave);
-        modbus.set_slave(DEVICE_ID)?;
-        modbus.connect()?;
         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> {
+    async fn get_data(&mut self) -> Result<Scaling, TristarError> {
-        self.modbus
+        let data = self
-            .read_registers(0x0000, RAM_DATA_SIZE + 1, &mut self.data_in)?;
+            .modbus
-        let scaling = Scaling::from(&self.data_in);
+            .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

@@ -38,6 +38,18 @@ pub enum ServerError {
     InvalidParameters,
     #[error("prometheus")]
     Prometheus(#[from] prometheus::Error),
+    #[error("uri")]
+    Uri,
+    #[error(transparent)]
+    Auth(#[from] AuthKeyError),
+    #[error(transparent)]
+    Channel(#[from] tokio::sync::mpsc::error::SendError<crate::api_interface::InterfaceRequest>),
+}
+
+impl From<rocket::http::uri::Error<'_>> for ServerError {
+    fn from(_: rocket::http::uri::Error<'_>) -> Self {
+        Self::Uri
+    }
 }
 
 impl<T> From<PoisonError<T>> for ServerError {
@@ -49,10 +61,13 @@ impl<T> From<PoisonError<T>> for ServerError {
 impl<'a> Responder<'a, 'a> for ServerError {
     fn respond_to(self, _: &'a rocket::Request<'_>) -> rocket::response::Result<'a> {
         Err(match self {
-            ServerError::Lock => rocket::http::Status::InternalServerError,
             ServerError::NoData => rocket::http::Status::ServiceUnavailable,
-            Self::InvalidParameters => rocket::http::Status::BadRequest,
+            ServerError::InvalidParameters => rocket::http::Status::BadRequest,
-            ServerError::Prometheus(_) => rocket::http::Status::InternalServerError,
+            ServerError::Auth(_) => rocket::http::Status::Unauthorized,
+            ServerError::Channel(_)
+            | ServerError::Uri
+            | ServerError::Lock
+            | ServerError::Prometheus(_) => rocket::http::Status::InternalServerError,
         })
     }
 }
@@ -89,14 +104,12 @@ pub enum TeslaStateParseError {
 
 #[derive(Error, Debug)]
 pub enum TristarError {
-    #[error("modbus error")]
+    #[error(transparent)]
-    Modbus(libmodbus_rs::prelude::Error),
+    Modbus(#[from] tokio_modbus::Error),
-}
+    #[error(transparent)]
-
+    ModbusException(#[from] tokio_modbus::Exception),
-impl From<libmodbus_rs::prelude::Error> for TristarError {
+    #[error(transparent)]
-    fn from(value: libmodbus_rs::prelude::Error) -> Self {
+    Serial(#[from] tokio_serial::Error),
-        Self::Modbus(value)
-    }
 }
 
 #[derive(Error, Debug)]
@@ -106,3 +119,24 @@ pub enum ConfigError {
     #[error("io")]
     Io(#[from] std::io::Error),
 }
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum AuthKeyError {
+    #[error("Error returned from Tesla")]
+    TeslaError(TeslaError),
+    #[error(transparent)]
+    SerdeError(#[from] serde_json::Error),
+    #[error(transparent)]
+    Reqwest(#[from] reqwest::Error),
+    #[error("Secrets not registered")]
+    Secrets,
+}
+
+#[derive(serde::Deserialize, Debug)]
+#[allow(dead_code)]
+pub(crate) struct TeslaError {
+    error: String,
+    error_description: String,
+    #[serde(rename = "referenceID")]
+    reference_id: String,
+}

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;
                                 }
                             })
                         }),
@@ -167,14 +166,14 @@ async fn main() {
                 .map(|(_, tcrc)| tcrc.tcrc_state.clone())
                 .unwrap_or_default();
 
-            let server_handle = server::launch_server(server::ServerState {
+            let server_handle = server::launch_server(server::ServerState::new(
                 car_state,
-                pl_state,
                 tcrc_state,
+                pl_state,
                 api_requests,
                 pli_requests,
-                tcrc_requests,
+                // tcrc_requests,
-            });
+            ));
 
             if let Some((mut interface, mut tesla_charge_rate_controller)) = interface_and_tcrc {
                 // spawn the api / charge rate control loop
@@ -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/authentication.rs

@@ -0,0 +1,76 @@
+use tesla_common::AuthInfo;
+
+use crate::errors::AuthKeyError;
+
+#[derive(serde::Deserialize, Debug)]
+#[allow(dead_code)]
+struct AuthKeySuccess {
+    access_token: String,
+    refresh_token: String,
+    id_token: String,
+    expires_in: usize,
+    state: String,
+    token_type: String,
+}
+
+impl AuthKeySuccess {
+    fn get_keys(&self) -> AuthInfo {
+        AuthInfo {
+            access_token: tesla_common::AccessToken(self.access_token.clone()),
+            refresh_token: Some(tesla_common::RefreshToken(self.refresh_token.clone())),
+        }
+    }
+}
+
+#[derive(serde::Deserialize, Debug)]
+#[serde(untagged)]
+enum AuthKeyResponse {
+    Ok(AuthKeySuccess),
+    Error(crate::errors::TeslaError),
+}
+
+pub async fn register_auth_key(code: String) -> Result<AuthInfo, AuthKeyError> {
+    if let Some(secrets) = super::SECRETS.as_ref() {
+        let mut headers = reqwest::header::HeaderMap::new();
+        headers.insert(
+            "Content-Type",
+            "application/x-www-form-urlencoded".parse().unwrap(),
+        );
+
+        let client = reqwest::Client::builder()
+            .redirect(reqwest::redirect::Policy::none())
+            .build()
+            .unwrap();
+        let res = client
+            .post("https://auth.tesla.com/oauth2/v3/token")
+            .headers(headers)
+            .body(
+                [
+                    "grant_type=authorization_code&client_id=",
+                    &secrets.client_id,
+                    "&client_secret=",
+                    &secrets.client_secret,
+                    "&code=",
+                    &code,
+                    "&audience=",
+                    &secrets.audience,
+                    "&redirect_uri=",
+                    &secrets.redirect_uri,
+                ]
+                .concat(),
+            )
+            .send()
+            .await?
+            .text()
+            .await?;
+        let keys = match serde_json::from_str(&res) {
+            Ok(AuthKeyResponse::Ok(v)) => Ok(v),
+            Ok(AuthKeyResponse::Error(e)) => Err(AuthKeyError::TeslaError(e)),
+            Err(e) => Err(AuthKeyError::SerdeError(e)),
+        }?;
+
+        Ok(keys.get_keys())
+    } else {
+        Err(AuthKeyError::Secrets)
+    }
+}

src/server/mod.rs

@@ -1,5 +1,6 @@
 use std::sync::{Arc, RwLock};
 
+use rand::Rng;
 use rocket::{
     fairing::{Fairing, Info, Kind},
     http::Header,
@@ -15,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,
 };
 
@@ -29,7 +30,27 @@ 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)>>,
+}
+
+impl ServerState {
+    pub fn new(
+        car_state: Arc<RwLock<CarState>>,
+        tcrc_state: Arc<RwLock<TcrcState>>,
+        pl_state: Option<Arc<RwLock<PlState>>>,
+        api_requests: UnboundedSender<InterfaceRequest>,
+        pli_requests: UnboundedSender<PliRequest>,
+    ) -> Self {
+        Self {
+            car_state,
+            tcrc_state,
+            pl_state,
+            api_requests,
+            pli_requests,
+            // tcrc_requests,
+            waiting_for_auth: std::sync::Mutex::new(None),
+        }
+    }
 }
 
 pub async fn launch_server(state: ServerState) {
@@ -77,11 +98,89 @@ fn rocket(state: ServerState) -> rocket::Rocket<rocket::Build> {
                 sync_time,
                 read_ram,
                 read_eeprom,
-                set_regulator_state
+                set_regulator_state,
+                reauthenticate,
+                auth
             ],
         )
 }
 
+#[derive(Debug)]
+struct Secrets {
+    client_id: String,
+    client_secret: String,
+    audience: String,
+    redirect_uri: String,
+}
+
+impl Secrets {
+    fn load() -> Option<Self> {
+        Some(Self {
+            client_id: std::env::var("CLIENT_ID").ok()?,
+            client_secret: std::env::var("CLIENT_SECRET").ok()?,
+            audience: std::env::var("AUDIENCE").ok()?,
+            redirect_uri: std::env::var("REDIRECT_URI").ok()?,
+        })
+    }
+}
+
+lazy_static::lazy_static! {
+    static ref SECRETS: Option<Secrets> = Secrets::load();
+}
+
+mod authentication;
+
+#[get("/reauthenticate")]
+async fn reauthenticate(
+    state: &State<ServerState>,
+) -> Result<rocket::response::Redirect, ServerError> {
+    let state_random: String = rand::thread_rng()
+        .sample_iter(&rand::distributions::Alphanumeric)
+        .take(16)
+        .map(char::from)
+        .collect();
+
+    let mut auth = state.waiting_for_auth.lock()?;
+
+    let secrets = SECRETS.as_ref().ok_or(ServerError::Lock)?;
+
+    let uri = rocket::http::uri::Absolute::parse_owned(format!(
+        "https://auth.tesla.com/oauth2/v3/authorize?&client_id={}&locale=en-US&prompt=login&redirect_uri={}&response_type=code&scope=openid%20vehicle_device_data%20offline_access%20user_data%20vehicle_cmds%20vehicle_charging_cmds&state={state_random}",
+        secrets.client_id,
+        secrets.redirect_uri
+    ))?;
+
+    *auth = Some((state_random.clone(), std::time::Instant::now()));
+    log::info!("set auth {auth:?}");
+    Ok(rocket::response::Redirect::to(uri))
+}
+
+#[post("/auth", format = "application/json", data = "<data>")]
+async fn auth(
+    data: rocket::serde::json::Json<tesla_common::SuccessfulAuth>,
+    state: &State<ServerState>,
+) -> Result<(), ServerError> {
+    log::info!("got reauthentication data");
+    let data = data.0;
+    let code = {
+        let mut guard = state.waiting_for_auth.lock()?;
+        if guard.as_ref().is_some_and(|(state_random, time)| {
+            *state_random == data.state
+                && std::time::Instant::now().duration_since(*time)
+                    < std::time::Duration::from_secs(30 * 60)
+        }) {
+            guard.take();
+            log::info!("matched reauthentication link");
+            Ok(data.code)
+        } else {
+            Err(ServerError::NoData)
+        }?
+    };
+    let new = authentication::register_auth_key(code).await?;
+    state.api_requests.send(InterfaceRequest::NewAuth(new))?;
+    Ok(())
+}
+
 #[get("/home")]
 async fn home(state: &State<ServerState>) -> Result<Json<bool>, ServerError> {
     let location_data = &state
@@ -177,7 +276,7 @@ async fn shutoff_status() -> Json<ShutoffStatus> {
 #[post("/set-max/<limit>")]
 async fn set_max(limit: i64, remote_addr: std::net::IpAddr) {
     log::warn!("setting max: {remote_addr:?}");
-    let limit = limit.clamp(access_config().min_rate, 15);
+    let limit = limit.clamp(access_config().min_rate, 32);
     write_to_config().max_rate = limit;
 }
 

webapp/index.html

@@ -23,10 +23,10 @@
 
 		<div id="rate-control">
 			<h3>Charge rate:</h3>
-			<input type="number" id="max-rate" max="15" min="0" onchange="change_max()" autocomplete="off" />
+			<input type="number" id="max-rate" max="32" min="0" onchange="change_max()" autocomplete="off" />
 			<button id="set-maximum" onclick="set_maximum()" disabled>Set maximum</button>
 			<br><br>
-			<input type="number" id="min-rate" max="15" min="0" onchange="change_min()" autocomplete="off" />
+			<input type="number" id="min-rate" max="32" min="0" onchange="change_min()" autocomplete="off" />
 			<button id="set-minimum" onclick="set_minimum()" disabled>Set minimum</button>
 		</div>
 		<div id="info"></div>
@@ -37,6 +37,7 @@
 			<a class="outlink" href="/regulator">Regulator control→</a>
 			<a class="outlink" href="/pid">PID control variables→</a>
 			<a class="outlink" href="/shutoff">Shutoff voltage control→</a>
+			<a class="outlink" href="/reauthenticate">Reauthenticate→</a>
 		</p>
 	</div>
 </body>