Update README

Cargo.toml

@@ -15,10 +15,13 @@ categories = ["api-bindings"]
 
 [dependencies]
 chrono = { version = "0.4.38", features = ["serde"] }
+futures-util = { version = "0.3.30", features = ["futures-sink"] }
+reqwest = { version = "0.12.7", features = ["json"] }
 serde = { version = "1.0.204", features = ["derive"] }
 serde_json = "1.0.121"
 serde_repr = "0.1.19"
 thiserror = "1.0.63"
+tokio = "1.39.3"
+tokio-tungstenite = { version = "0.23.1", features = ["tokio-rustls", "rustls-tls-native-roots"] }
 tracing = "0.1.40"
 tungstenite = { version = "0.23.0", optional = true, features = ["rustls-tls-native-roots"] }
-ureq = { version = "2.10.0", features = ["json"] }

README.md

@@ -6,24 +6,18 @@ Work in progress.
 
  - Authn/z
    - [x] Authentication and token retrieval
-   - [ ] Persistence of tokens
+   - [x] Persistence of tokens
  - Core functionality
    - [x] Enumerate sites and chargers
    - [x] Read energy meter
    - [x] Read charger status
-   - [ ] Control charging (start/pause/resume/stop)
+   - [x] Control charging (start/pause/resume/stop)
-   - [ ] Control dynamic current limits
+   - [x] Control dynamic current limits
  - Event stream
    - [x] Websocket connection (raw SignalR messages)
-   - [ ] Event decoding
+   - [x] Event decoding
  - Ergonomics
    - [ ] Enums for protocol constants
    - [ ] Proper SignalR support with Tokio
 
- 
- 
-
-
-
-
 [1]: https://easee.com

src/api.rs

@@ -1,7 +1,5 @@
 use std::{
-    io,
+    io, ops::{Add, Mul, Sub}, time::{Duration, Instant, SystemTime, UNIX_EPOCH}
-    ops::{Add, Mul, Sub},
-    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
 use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize};
@@ -9,11 +7,14 @@ use serde_repr::Deserialize_repr;
 use thiserror::Error;
 use tracing::{debug, info, instrument};
 
+pub use reqwest::{self, StatusCode};
+
 pub struct Context {
     auth_header: String,
     refresh_token: String,
     token_expiration: Instant,
     on_refresh: Option<Box<dyn FnMut(&mut Self) + Send>>,
+    client: reqwest::Client,
 }
 
 impl std::fmt::Debug for Context {
@@ -316,7 +317,7 @@ pub enum ApiError {
 
     /// HTTP call failed (404, etc)
     #[error("ureq")]
-    Ureq(#[source] Box<ureq::Error>),
+    HTTP(#[source] Box<reqwest::Error>),
 
     /// HTTP call succeeded but the returned JSON document didn't match the expected format
     #[error("unexpected data: {1} when processing {0}")]
@@ -334,20 +335,20 @@ pub enum ApiError {
     InvalidID(String),
 }
 
-impl From<ureq::Error> for ApiError {
+impl From<reqwest::Error> for ApiError {
-    fn from(value: ureq::Error) -> Self {
+    fn from(value: reqwest::Error) -> Self {
-        ApiError::Ureq(Box::new(value))
+        ApiError::HTTP(Box::new(value))
     }
 }
 
 trait JsonExplicitError {
     /// Explicitely report the received JSON object we failed to parse
-    fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError>;
+    async fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError>;
 }
 
-impl JsonExplicitError for ureq::Response {
+impl JsonExplicitError for reqwest::Response {
-    fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError> {
+    async fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError> {
-        let resp: serde_json::Value = self.into_json()?;
+        let resp: serde_json::Value = self.json().await?;
         let parsed = T::deserialize(&resp);
         parsed.map_err(|e| ApiError::UnexpectedData(resp, e))
     }
@@ -363,12 +364,13 @@ pub enum TokenParseError {
 }
 
 impl Context {
-    fn from_login_response(resp: LoginResponse) -> Self {
+    fn from_login_response(resp: LoginResponse, client: reqwest::Client) -> Self {
         Self {
             auth_header: format!("Bearer {}", &resp.access_token),
             refresh_token: resp.refresh_token,
             token_expiration: (Instant::now() + Duration::from_secs(resp.expires_in as u64)),
             on_refresh: None,
+            client,
         }
     }
 
@@ -389,6 +391,7 @@ impl Context {
             refresh_token: refresh.to_owned(),
             token_expiration,
             on_refresh: None,
+            client: reqwest::Client::new(),
         })
     }
 
@@ -410,7 +413,7 @@ impl Context {
     }
 
     /// Retrieve access tokens online, by logging in with the provided credentials
-    pub fn from_login(user: &str, password: &str) -> Result<Self, ApiError> {
+    pub async fn from_login(user: &str, password: &str) -> Result<Self, ApiError> {
         #[derive(Serialize)]
         #[serde(rename_all = "camelCase")]
         struct Params<'t> {
@@ -418,23 +421,26 @@ impl Context {
             password: &'t str,
         }
 
+        let client = reqwest::Client::new();
+
         info!("Logging into API");
         let url: String = format!("{}accounts/login", API_BASE);
-        let resp: LoginResponse = ureq::post(&url)
+        let resp: LoginResponse = client.post(&url)
-            .send_json(Params {
+            .json(&Params {
                 user_name: user,
                 password,
-            })?
+            })
-            .into_json_with_error()?;
+            .send().await?
+            .into_json_with_error().await?;
 
-        Ok(Self::from_login_response(resp))
+        Ok(Self::from_login_response(resp, client))
     }
 
     /// Check if the token has reached its expiration date
-    fn check_expired(&mut self) -> Result<(), ApiError> {
+    async fn check_expired(&mut self) -> Result<(), ApiError> {
         if self.token_expiration < Instant::now() {
             debug!("Token has expired");
-            self.refresh_token()?;
+            self.refresh_token().await?;
         }
         Ok(())
     }
@@ -444,7 +450,7 @@ impl Context {
     }
 
     /// Use the refresh token to refresh credentials
-    pub fn refresh_token(&mut self) -> Result<(), ApiError> {
+    pub async fn refresh_token(&mut self) -> Result<(), ApiError> {
         #[derive(Serialize)]
         #[serde(rename_all = "camelCase")]
         struct Params<'t> {
@@ -456,51 +462,52 @@ impl Context {
             refresh_token: &self.refresh_token,
         };
         let url = format!("{}accounts/refresh_token", API_BASE);
-        let resp: LoginResponse = ureq::post(&url)
+        let resp: LoginResponse = self.client.post(&url)
-            .set("Content-type", "application/json")
+            .header("Content-type", "application/json")
-            .send_json(params)?
+            .json(&params)
-            .into_json_with_error()?;
+            .send().await?
+            .into_json_with_error().await?;
 
-        *self = Self::from_login_response(resp);
+        *self = Self::from_login_response(resp, self.client.clone());
         Ok(())
     }
 
     /// List all sites available to the user
-    pub fn sites(&mut self) -> Result<Vec<Site>, ApiError> {
+    pub async fn sites(&mut self) -> Result<Vec<Site>, ApiError> {
-        self.get("sites")
+        self.get("sites").await
     }
 
-    pub fn site(&mut self, id: i32) -> Result<SiteDetails, ApiError> {
+    pub async fn site(&mut self, id: i32) -> Result<SiteDetails, ApiError> {
-        self.get(&format!("sites/{id}"))
+        self.get(&format!("sites/{id}")).await
     }
 
     /// List all chargers available to the user
-    pub fn chargers(&mut self) -> Result<Vec<Charger>, ApiError> {
+    pub async fn chargers(&mut self) -> Result<Vec<Charger>, ApiError> {
-        self.get("chargers")
+        self.get("chargers").await
     }
 
-    pub fn charger(&mut self, id: &str) -> Result<Charger, ApiError> {
+    pub async fn charger(&mut self, id: &str) -> Result<Charger, ApiError> {
         if !id.chars().all(char::is_alphanumeric) {
             return Err(ApiError::InvalidID(id.to_owned()));
         }
-        self.get(&format!("chargers/{}", id))
+        self.get(&format!("chargers/{}", id)).await
     }
 
-    pub fn circuit(&mut self, site_id: u32, circuit_id: u32) -> Result<Circuit, ApiError> {
+    pub async fn circuit(&mut self, site_id: u32, circuit_id: u32) -> Result<Circuit, ApiError> {
-        self.get(&format!("site/{site_id}/circuit/{circuit_id}"))
+        self.get(&format!("site/{site_id}/circuit/{circuit_id}")).await
     }
 
-    pub fn circuit_dynamic_current(
+    pub async fn circuit_dynamic_current(
         &mut self,
         site_id: u32,
         circuit_id: u32,
     ) -> Result<Triphase, ApiError> {
         self.get(&format!(
             "sites/{site_id}/circuits/{circuit_id}/dynamicCurrent"
-        ))
+        )).await
     }
 
-    pub fn set_circuit_dynamic_current(
+    pub async fn set_circuit_dynamic_current(
         &mut self,
         site_id: u32,
         circuit_id: u32,
@@ -509,65 +516,67 @@ impl Context {
         self.post(
             &format!("sites/{site_id}/circuits/{circuit_id}/dynamicCurrent"),
             &current,
-        )
+        ).await
     }
 
     #[instrument]
-    fn get<T: DeserializeOwned>(&mut self, path: &str) -> Result<T, ApiError> {
+    async fn get<T: DeserializeOwned>(&mut self, path: &str) -> Result<T, ApiError> {
-        self.check_expired()?;
+        self.check_expired().await?;
         let url: String = format!("{}{}", API_BASE, path);
-        let req = ureq::get(&url)
-            .set("Accept", "application/json")
-            .set("Authorization", &self.auth_header);
 
-        let mut resp = req.clone().call()?;
+        let req = self.client.get(url)
+            .header("Accept", "application/json")
+            .header("Authorization", &self.auth_header)
+            .build()?;
+
+        let mut resp = self.client.execute(req.try_clone().unwrap()).await?;
 
         if resp.status() == 401 {
-            self.refresh_token()?;
+            self.refresh_token().await?;
-            resp = req.call()?
+            resp = self.client.execute(req).await?
         }
 
-        resp.into_json_with_error()
+        resp.into_json_with_error().await
     }
 
-    fn maybe_get<T: DeserializeOwned>(&mut self, path: &str) -> Result<Option<T>, ApiError> {
+    async fn maybe_get<T: DeserializeOwned>(&mut self, path: &str) -> Result<Option<T>, ApiError> {
-        match self.get(path) {
+        match self.get(path).await {
             Ok(r) => Ok(Some(r)),
-            Err(ApiError::Ureq(e)) => match &*e {
+            Err(ApiError::HTTP(e)) if e.status() == Some(StatusCode::NOT_FOUND)=> Ok(None),
-                ureq::Error::Status(404, _) => Ok(None),
-                _ => Err(ApiError::Ureq(e)),
-            },
             Err(other) => Err(other),
         }
     }
 
-    pub(crate) fn post<T: DeserializeOwned, P: Serialize>(
+    pub(crate) async fn post<T: DeserializeOwned, P: Serialize>(
         &mut self,
         path: &str,
         params: &P,
     ) -> Result<T, ApiError> {
         let url: String = format!("{}{}", API_BASE, path);
-        self.post_raw(&url, params)
+        self.post_raw(&url, params).await
     }
 
-    pub(crate) fn post_raw<T: DeserializeOwned, P: Serialize>(
+    pub(crate) async fn post_raw<T: DeserializeOwned, P: Serialize>(
         &mut self,
         url: &str,
         params: &P,
     ) -> Result<T, ApiError> {
-        self.check_expired()?;
+        self.check_expired().await?;
-        let req = ureq::post(url)
+        let req = self.client.post(url)
-            .set("Accept", "application/json")
+            .header("Accept", "application/json")
-            .set("Authorization", &self.auth_header);
+            .header("Authorization", &self.auth_header)
+            .json(params);
 
-        let mut resp = req.clone().send_json(params)?;
+        let mut resp = req
+            .try_clone().unwrap()
+            .send().await?;
 
         if resp.status() == 401 {
-            self.refresh_token()?;
+            self.refresh_token().await?;
-            resp = req.send_json(params)?
+            resp = req.send().await?
         }
 
-        resp.into_json_with_error()
+        resp.into_json_with_error().await
     }
 }
 
@@ -584,12 +593,12 @@ pub struct MeterReading {
 
 impl Site {
     /// Read all energy meters from the given site
-    pub fn lifetime_energy(&self, ctx: &mut Context) -> Result<Vec<MeterReading>, ApiError> {
+    pub async fn lifetime_energy(&self, ctx: &mut Context) -> Result<Vec<MeterReading>, ApiError> {
-        ctx.get(&format!("sites/{}/energy", self.id))
+        ctx.get(&format!("sites/{}/energy", self.id)).await
     }
 
-    pub fn details(&self, ctx: &mut Context) -> Result<SiteDetails, ApiError> {
+    pub async fn details(&self, ctx: &mut Context) -> Result<SiteDetails, ApiError> {
-        ctx.get(&format!("sites/{}", self.id))
+        ctx.get(&format!("sites/{}", self.id)).await
     }
 }
 
@@ -598,63 +607,63 @@ impl Circuit {
         format!("sites/{}/circuits/{}/dynamicCurrent", self.site_id, self.id)
     }
 
-    pub fn dynamic_current(&self, ctx: &mut Context) -> Result<Triphase, ApiError> {
+    pub async fn dynamic_current(&self, ctx: &mut Context) -> Result<Triphase, ApiError> {
-        ctx.circuit_dynamic_current(self.site_id, self.id)
+        ctx.circuit_dynamic_current(self.site_id, self.id).await
     }
 
-    pub fn set_dynamic_current(
+    pub async fn set_dynamic_current(
         &self,
         ctx: &mut Context,
         current: SetCurrent,
     ) -> Result<(), ApiError> {
-        ctx.post(&self.dynamic_current_path(), &current)
+        ctx.post(&self.dynamic_current_path(), &current).await
     }
 }
 
 impl Charger {
     /// Enable "smart charging" on the charger. This just turns the LED blue, and disables basic charging plans.
-    pub fn enable_smart_charging(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub async fn enable_smart_charging(&self, ctx: &mut Context) -> Result<(), ApiError> {
         let url = format!("chargers/{}/commands/smart_charging", &self.id);
-        ctx.post(&url, &())
+        ctx.post(&url, &()).await
     }
 
     /// Read the state of a charger
-    pub fn state(&self, ctx: &mut Context) -> Result<ChargerState, ApiError> {
+    pub async fn state(&self, ctx: &mut Context) -> Result<ChargerState, ApiError> {
         let url = format!("chargers/{}/state", self.id);
-        ctx.get(&url)
+        ctx.get(&url).await
     }
 
     /// Read info about the ongoing charging session
-    pub fn ongoing_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
+    pub async fn ongoing_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
-        ctx.maybe_get(&format!("chargers/{}/sessions/ongoing", &self.id))
+        ctx.maybe_get(&format!("chargers/{}/sessions/ongoing", &self.id)).await
     }
 
     /// Read info about the last charging session (not including ongoing one)
-    pub fn latest_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
+    pub async fn latest_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
-        ctx.maybe_get(&format!("chargers/{}/sessions/latest", &self.id))
+        ctx.maybe_get(&format!("chargers/{}/sessions/latest", &self.id)).await
     }
 
-    fn command(&self, ctx: &mut Context, command: &str) -> Result<CommandReply, ApiError> {
+    async fn command(&self, ctx: &mut Context, command: &str) -> Result<CommandReply, ApiError> {
-        ctx.post(&format!("chargers/{}/commands/{}", self.id, command), &())
+        ctx.post(&format!("chargers/{}/commands/{}", self.id, command), &()).await
     }
 
-    pub fn start(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub async fn start(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "start_charging")?;
+        self.command(ctx, "start_charging").await?;
         Ok(())
     }
 
-    pub fn pause(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub async fn pause(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "pause_charging")?;
+        self.command(ctx, "pause_charging").await?;
         Ok(())
     }
 
-    pub fn resume(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub async fn resume(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "resume_charging")?;
+        self.command(ctx, "resume_charging").await?;
         Ok(())
     }
 
-    pub fn stop(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub async fn stop(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "stop_charging")?;
+        self.command(ctx, "stop_charging").await?;
         Ok(())
     }
 }
@@ -666,11 +675,15 @@ mod test {
     use super::Context;
     #[test]
     fn token_save() {
+
+        let client = reqwest::Client::new();
+
         let ctx = Context {
             auth_header: "Bearer aaaaaaa0".to_owned(),
             refresh_token: "abcdef".to_owned(),
             token_expiration: Instant::now() + Duration::from_secs(1234),
             on_refresh: None,
+            client: client.clone(),
         };
 
         let saved = ctx.save();

src/observation.rs

@@ -1,9 +1,9 @@
 use serde::{de::{DeserializeOwned, IntoDeserializer}, Deserialize};
+use serde_json::json;
 use serde_repr::Deserialize_repr;
 use std::num::{ParseFloatError, ParseIntError};
 use thiserror::Error;
 use tracing::info;
-use ureq::json;
 
 use crate::{
     api::{ChargerOpMode, Context, OutputPhase, UtcDateTime},
@@ -325,17 +325,17 @@ struct ProductUpdate {
 }
 
 impl Stream {
-    pub fn from_context(ctx: &mut Context) -> Result<Self, NegotiateError> {
+    pub async fn from_context(ctx: &mut Context) -> Result<Self, NegotiateError> {
         Ok(Self {
-            inner: signalr::Stream::from_ws(crate::stream::Stream::open(ctx)?),
+            inner: signalr::Stream::from_ws(crate::stream::Stream::open(ctx).await?),
         })
     }
 
-    pub fn recv(&mut self) -> Result<Event, ObservationError> {
+    pub async fn recv(&mut self) -> Result<Event, ObservationError> {
         use signalr::Message::*;
         let de = |msg| -> Result<Event, ObservationError> { Err(ObservationError::Protocol(msg)) };
         loop {
-            let msg = self.inner.recv()?;
+            let msg = self.inner.recv().await?;
             match &msg {
                 Ping => continue,
                 Empty | InvocationResult { .. } => info!("Skipped message: {msg:?}"),
@@ -351,9 +351,9 @@ impl Stream {
             }
         }
     }
-    pub fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
+    pub async fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
         self.inner
-            .invoke("SubscribeWithCurrentState", json!([id, true]))
+            .invoke("SubscribeWithCurrentState", json!([id, true])).await
     }
 }
 

src/signalr.rs

@@ -113,9 +113,9 @@ impl Stream {
         Self { ws, buffer: vec![] }
     }
 
-    pub fn recv(&mut self) -> Result<Message, StreamError> {
+    pub async fn recv(&mut self) -> Result<Message, StreamError> {
         while self.buffer.is_empty() {
-            self.buffer = self.ws.recv()?;
+            self.buffer = self.ws.recv().await?;
             self.buffer.reverse();
         }
 
@@ -123,7 +123,7 @@ impl Stream {
         Ok(Message::from_json(json)?)
     }
 
-    pub fn invoke(
+    pub async fn invoke(
         &mut self,
         target: &str,
         args: serde_json::Value,
@@ -131,6 +131,6 @@ impl Stream {
         self.ws.send(json!( { "arguments": args,
                                   "invocationId": "0",
                                   "target": target,
-                                  "type": 1} ))
+                                  "type": 1} )).await
     }
 }

src/stream.rs

@@ -1,9 +1,11 @@
 use super::api::{ApiError, Context};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::net::TcpStream;
+use tokio::net::TcpStream;
 use thiserror::Error;
-use tungstenite::{stream::MaybeTlsStream, Message, WebSocket};
+//use tungstenite::{stream::MaybeTlsStream, Message, WebSocket};
+use tokio_tungstenite::{MaybeTlsStream, tungstenite::Message, WebSocketStream};
+use futures_util::{SinkExt,StreamExt};
 
 const STREAM_API_NEGOTIATION_URL: &str =
     "https://streams.easee.com/hubs/products/negotiate?negotiateVersion=1";
@@ -36,15 +38,18 @@ pub enum RecvError {
 
     #[error("WS error: {0}")]
     TungsteniteError(#[from] tungstenite::Error),
+
+    #[error("End of stream")]
+    EndOfStream,
 }
 
 pub struct Stream {
-    sock: WebSocket<MaybeTlsStream<TcpStream>>,
+    sock: WebSocketStream<MaybeTlsStream<TcpStream>>,
 }
 
 impl Stream {
-    pub fn open(ctx: &mut Context) -> Result<Stream, NegotiateError> {
+    pub async fn open(ctx: &mut Context) -> Result<Stream, NegotiateError> {
-        let r: NegotiateResponse = ctx.post_raw(STREAM_API_NEGOTIATION_URL, &())?;
+        let r: NegotiateResponse = ctx.post_raw(STREAM_API_NEGOTIATION_URL, &()).await?;
 
         let token = ctx.auth_token();
         let wss_url = format!(
@@ -52,7 +57,8 @@ impl Stream {
             WSS_URL, r.connection_token, token
         );
 
-        let resp = tungstenite::client::connect(&wss_url);
+        let resp = tokio_tungstenite::connect_async(wss_url).await;
+        //let resp = tungstenite::client::connect(&wss_url);
 
         if let Err(tungstenite::Error::Http(he)) = &resp {
             eprintln!(
@@ -62,19 +68,20 @@ impl Stream {
         }
 
         let mut stream = Stream { sock: resp?.0 };
-        stream.send(json!({ "protocol": "json", "version": 1 }))?;
+        stream.send(json!({ "protocol": "json", "version": 1 })).await?;
 
         Ok(stream)
     }
 
-    pub fn send<T: Serialize>(&mut self, msg: T) -> Result<(), tungstenite::Error> {
+    pub async fn send<T: Serialize>(&mut self, msg: T) -> Result<(), tungstenite::Error> {
         let mut msg = serde_json::to_string(&msg).unwrap();
         msg.push('\x1E');
-        self.sock.send(Message::Text(msg))
+        self.sock.send(Message::Text(msg)).await
     }
 
-    pub fn recv(&mut self) -> Result<Vec<serde_json::Value>, RecvError> {
+    pub async fn recv(&mut self) -> Result<Vec<serde_json::Value>, RecvError> {
-        let msg = self.sock.read()?;
+        let msg = self.sock.next().await
+            .ok_or(RecvError::EndOfStream)??;
         let Message::Text(txt) = msg else {
             return Err(RecvError::BadMessageType);
         };