Formatting and cleanup

src/api.rs

@@ -8,12 +8,22 @@ use serde_repr::Deserialize_repr;
 use thiserror::Error;
 use tracing::{debug, info, instrument};
 
-/// API Authentication context
-#[derive(Debug)]
 pub struct Context {
-    auth_header: String, //TODO mark as secret to hide in tracing
+    auth_header: String,
     refresh_token: String,
     token_expiration: Instant,
+    on_refresh: Option<Box<dyn FnMut(&mut Self)>>,
+}
+
+impl std::fmt::Debug for Context {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Context")
+            .field("auth_header", &"<secret>")
+            .field("refresh_token", &"<secret>")
+            .field("token_expiration", &self.token_expiration)
+            .field("on_refresh", &"[closure]")
+            .finish()
+    }
 }
 
 const API_BASE: &str = "https://api.easee.com/api/";
@@ -60,6 +70,7 @@ pub struct Charger {
 #[derive(Clone, Copy, Debug, Deserialize_repr, Eq, Ord, PartialEq, PartialOrd)]
 #[repr(u8)]
 pub enum ChargerOpMode {
+    Unknown = 0,
     Disconnected = 1,
     Paused = 2,
     Charging = 3,
@@ -235,7 +246,7 @@ pub enum ApiError {
     FormatError(#[from] chrono::ParseError),
 
     #[error("Invalid ID: {0:?}")]
-    InvalidID(String)
+    InvalidID(String),
 }
 
 impl From<ureq::Error> for ApiError {
@@ -267,34 +278,50 @@ pub enum TokenParseError {
 }
 
 impl Context {
-
     fn from_login_response(resp: LoginResponse) -> 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))
+            token_expiration: (Instant::now() + Duration::from_secs(resp.expires_in as u64)),
+            on_refresh: None,
         }
     }
 
     pub fn from_saved(saved: &str) -> Result<Self, TokenParseError> {
         let lines: Vec<&str> = saved.lines().collect();
-        let &[token, refresh, expire] = &*lines else { return Err(TokenParseError::IncorrectLineCount) };
+        let &[token, refresh, expire] = &*lines else {
+            return Err(TokenParseError::IncorrectLineCount);
+        };
 
         let expire: u64 = expire.parse()?;
-        let token_expiration = Instant::now() + (UNIX_EPOCH + Duration::from_secs(expire)).duration_since(SystemTime::now()).unwrap_or_default();
+        let token_expiration = Instant::now()
+            + (UNIX_EPOCH + Duration::from_secs(expire))
+                .duration_since(SystemTime::now())
+                .unwrap_or_default();
 
         Ok(Self {
             auth_header: format!("Bearer {}", token),
             refresh_token: refresh.to_owned(),
             token_expiration,
+            on_refresh: None,
         })
+    }
 
+    pub fn on_refresh<F: FnMut(&mut Self) + 'static>(mut self, on_refresh: F) -> Self {
+        self.on_refresh = Some(Box::new(on_refresh));
+        self
     }
 
     pub fn save(&self) -> String {
-        let expiration = (SystemTime::now() + (self.token_expiration - Instant::now())).duration_since(UNIX_EPOCH)
+        let expiration = (SystemTime::now() + (self.token_expiration - Instant::now()))
+            .duration_since(UNIX_EPOCH)
             .unwrap();
-        format!("{}\n{}\n{}\n", self.auth_token(), self.refresh_token, expiration.as_secs())
+        format!(
+            "{}\n{}\n{}\n",
+            self.auth_token(),
+            self.refresh_token,
+            expiration.as_secs()
+        )
     }
 
     /// Retrieve access tokens online, by logging in with the provided credentials
@@ -365,7 +392,7 @@ impl Context {
 
     pub fn charger(&mut self, id: &str) -> Result<Charger, ApiError> {
         if !id.chars().all(char::is_alphanumeric) {
-            return Err(ApiError::InvalidID(id.to_owned()))
+            return Err(ApiError::InvalidID(id.to_owned()));
         }
         self.get(&format!("chargers/{}", id))
     }
@@ -495,7 +522,6 @@ impl Charger {
     }
 }
 
-
 #[cfg(test)]
 mod test {
     use std::time::{Duration, Instant};
@@ -503,10 +529,12 @@ mod test {
     use super::Context;
     #[test]
     fn token_save() {
-
-        let ctx = Context { auth_header: "Bearer aaaaaaa0".to_owned()
-                                   , refresh_token: "abcdef".to_owned()
-                                   , token_expiration: Instant::now() + Duration::from_secs(1234) };
+        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,
+        };
 
         let saved = ctx.save();
         let ctx2 = Context::from_saved(&saved).unwrap();
@@ -514,6 +542,5 @@ mod test {
         assert_eq!(&ctx.auth_header, &ctx2.auth_header);
         assert_eq!(&ctx.refresh_token, &ctx2.refresh_token);
         assert!((ctx.token_expiration - ctx2.token_expiration) < Duration::from_secs(5))
-
     }
 }

src/observation.rs

@@ -1,5 +1,17 @@
-use crate::api::ChargerOpMode;
+use serde::{de::IntoDeserializer, Deserialize};
+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},
+    signalr::{self, StreamError},
+    stream::NegotiateError,
+};
+
+#[derive(Clone, Copy, Debug, Deserialize_repr)]
 #[repr(u8)]
 pub enum PilotMode {
     Disconnected = b'A',
@@ -7,8 +19,24 @@ pub enum PilotMode {
     Charging = b'C',
     NeedsVentilation = b'D',
     FaultDetected = b'F',
+    Unknown = b'\x00',
 }
 
+impl From<&str> for PilotMode {
+    fn from(value: &str) -> Self {
+        use PilotMode::*;
+        match value {
+            "A" => Disconnected,
+            "B" => Connected,
+            "C" => Charging,
+            "D" => NeedsVentilation,
+            "F" => FaultDetected,
+            _ => Unknown,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize_repr)]
 #[repr(u8)]
 pub enum PhaseMode {
     Ignore = 0,
@@ -17,19 +45,123 @@ pub enum PhaseMode {
     Phase2 = 3,
 }
 
-pub enum ReasonForNoCurrent {}
+#[derive(Clone, Copy, Debug)]
+pub enum InputPin {
+    T1,
+    T2,
+    T3,
+    T4,
+    T5,
+}
 
+#[derive(Clone, Copy, Debug, Deserialize_repr)]
+#[repr(u8)]
+enum DataType {
+    Boolean = 2,
+    Double = 3,
+    Integer = 4,
+    String = 6,
+}
+
+#[derive(Clone, Debug)]
+pub enum ObservationData {
+    Boolean(bool),
+    Double(f64),
+    Integer(i64),
+    String(String),
+}
+
+#[derive(Error, Debug)]
+pub enum ParseError {
+    #[error("integer: {0}")]
+    Integer(#[from] ParseIntError),
+
+    #[error("double: {0}")]
+    Double(#[from] ParseFloatError),
+}
+
+impl ObservationData {
+    fn from_dynamic(value: String, data_type: DataType) -> Result<ObservationData, ParseError> {
+        Ok(match data_type {
+            DataType::Boolean => ObservationData::Boolean(value.parse::<i64>()? != 0),
+            DataType::Double => ObservationData::Double(value.parse()?),
+            DataType::Integer => ObservationData::Integer(value.parse()?),
+            DataType::String => ObservationData::String(value),
+        })
+    }
+
+    fn dynamic_type(&self) -> DataType {
+        match self {
+            ObservationData::Boolean(_) => DataType::Boolean,
+            ObservationData::Double(_) => DataType::Double,
+            ObservationData::Integer(_) => DataType::Integer,
+            ObservationData::String(_) => DataType::String,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct ReasonForNoCurrent(u16);
+
+impl std::fmt::Display for ReasonForNoCurrent {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self.0 {
+                0 => "OK",
+                1 => "LoadBalance: circuit too low",
+                2 => "LoadBalance: dynamic circuit too low",
+                3 => "LoadBalance: max dynamic offline",
+                4 => "LoadBalance: circuit fuse too low",
+                5 => "LoadBalance: waiting in queue",
+                6 => "LoadBalance: waiting in charged queue",
+                7 => "Error: illegal grid type",
+                8 => "Error: not received request from car",
+                9 => "Error: master communication lost",
+                10 => "Error: no current from equalizer",
+                11 => "Error: no current, phase disconnected",
+                25 => "Error: limited by circuit fuse",
+                26 => "Error: limited by circuit max current",
+                27 => "Error: limited by dynamic circuit current",
+                28 => "Error: limited by equalizer",
+                29 => "Error: limited by circuit load balancing",
+                30 => "Error: limited by offline settings",
+                53 => "Info: charger disabled",
+                54 => "Waiting: pending schedule",
+                55 => "Waiting: pending authorization",
+                56 => "Error: charger in error state",
+                57 => "Error: Erratic EV",
+                75 => "Cable: limited by cable rating",
+                76 => "Schedule: limited by schedule",
+                77 => "Charger limit: limited by charger max current",
+                78 => "Charger Limit: limited by dynamic charger current",
+                79 => "Car limit: limited by car not charging",
+                80 => "Local: limited by local adjustment",
+                81 => "Car limit: limited by car",
+                100 => "Error: undefined",
+                other => return write!(f, "Code {other}"),
+            }
+        )
+    }
+}
+
+#[derive(Debug)]
 pub enum Observation {
     SelfTestResult(String),
-    SelfTestDetails(serde_json::Value),
-    WifiEvent(u64),
-    ChargerOfflineReason(u64),
-    CircuitMaxCurrent { phase: u8, amperes: u64 },
+    SelfTestDetails(String),
+    WifiEvent(i64),
+    ChargerOfflineReason(i64),
+    CircuitMaxCurrent { phase: u8, amperes: i64 },
     SiteID(String),
     IsEnabled(bool),
-    Temperature(u64),
+    Temperature(i64),
     TriplePhase(bool),
     DynamicChargerCurrent(f64),
+
+    ICCID(String),
+    MobileNetworkOperator(String),
+
     ReasonForNoCurrent(ReasonForNoCurrent),
     PilotMode(PilotMode),
     SmartCharging(bool),
@@ -37,4 +169,147 @@ pub enum Observation {
     CableRating(f64),
     UserId(String),
     ChargerOpMode(ChargerOpMode),
+    IntCurrent { pin: InputPin, current: f64 },
+
+    TotalPower(f64),
+    EnergyPerHour(f64),
+    LifetimeEnergy(f64),
+
+    Unknown { code: u16, value: ObservationData },
+}
+
+fn op_mode_from_int(mode: i64) -> ChargerOpMode {
+    use ChargerOpMode::*;
+    match mode {
+        1 => Disconnected,
+        2 => Paused,
+        3 => Charging,
+        4 => Finished,
+        5 => Error,
+        6 => Ready,
+        _ => Unknown,
+    }
+}
+
+impl Observation {
+    fn try_from_data(code: u16, data: ObservationData) -> Observation {
+        use InputPin::*;
+        use Observation::*;
+        use ObservationData::*;
+        match (code, data) {
+            (1, String(result)) => SelfTestResult(result),
+            (2, String(details)) => SelfTestDetails(details),
+            (10, Integer(wifi)) => WifiEvent(wifi),
+            (11, Integer(reason)) => ChargerOfflineReason(reason),
+            (22, Integer(amperes)) => CircuitMaxCurrent { phase: 1, amperes },
+            (23, Integer(amperes)) => CircuitMaxCurrent { phase: 2, amperes },
+            (24, Integer(amperes)) => CircuitMaxCurrent { phase: 3, amperes },
+            (26, String(site)) => SiteID(site),
+            (31, Boolean(enabled)) => IsEnabled(enabled),
+            (32, Integer(temperature)) => Temperature(temperature),
+            (38, Integer(1)) => TriplePhase(false),
+            (38, Integer(3)) => TriplePhase(true),
+            (48, Double(current)) => DynamicChargerCurrent(current),
+            (81, String(iccid)) => ICCID(iccid),
+            (84, String(operator)) => MobileNetworkOperator(operator),
+            (96, Integer(reason)) => ReasonForNoCurrent(self::ReasonForNoCurrent(reason as u16)),
+            (100, String(l)) => PilotMode(super::observation::PilotMode::from(&*l)),
+            (102, Boolean(enabled)) => SmartCharging(enabled),
+            (103, Boolean(locked)) => CableLocked(locked),
+            (104, Double(amps)) => CableRating(amps),
+            (107, String(tok_rev)) => UserId(tok_rev.chars().rev().collect()),
+            (109, Integer(mode)) => ChargerOpMode(op_mode_from_int(mode)),
+            (120, Double(power)) => TotalPower(power),
+            (182, Double(current)) => IntCurrent { pin: T2, current },
+            (183, Double(current)) => IntCurrent { pin: T3, current },
+            (184, Double(current)) => IntCurrent { pin: T4, current },
+            (185, Double(current)) => IntCurrent { pin: T5, current },
+
+            (code, value) => Unknown { code, value },
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Event {
+    pub charger: String,
+    pub observation: Observation,
+}
+
+pub struct Stream {
+    inner: signalr::Stream,
+}
+
+#[derive(Debug, Error)]
+pub enum ObservationError {
+    #[error("stream: {0}")]
+    Stream(#[from] StreamError),
+
+    #[error("Protocol error")]
+    Protocol(signalr::Message),
+
+    #[error("JSON: {0}")]
+    Deserialize(#[from] serde_json::Error),
+
+    #[error("Parsing: {0}")]
+    Parsing(#[from] ParseError),
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(rename_all = "camelCase")]
+struct ProductUpdate {
+    data_type: DataType,
+    id: u16,
+    mid: String,
+    timestamp: UtcDateTime,
+    value: String,
+}
+
+impl Stream {
+    pub fn from_context(ctx: &mut Context) -> Result<Self, NegotiateError> {
+        Ok(Self {
+            inner: signalr::Stream::from_ws(crate::stream::Stream::open(ctx)?),
+        })
+    }
+
+    pub 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()?;
+            match &msg {
+                Empty | Ping | InvocationResult { .. } => info!("Skipped message: {msg:?}"),
+                Invocation { target, arguments } if target == "ProductUpdate" => {
+                    if arguments.len() != 1 {
+                        return de(msg);
+                    };
+                    let evt = ProductUpdate::deserialize(&arguments[0])?;
+                    return decode_update(evt);
+                }
+                Invocation { .. } => continue,
+                _other => return de(msg),
+            }
+        }
+    }
+    pub fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
+        self.inner
+            .invoke("SubscribeWithCurrentState", json!([id, true]))
+    }
+}
+
+fn decode_update(update: ProductUpdate) -> Result<Event, ObservationError> {
+    let ProductUpdate {
+        data_type,
+        id,
+        mid,
+        timestamp,
+        value,
+    } = update;
+    let data = ObservationData::from_dynamic(value, data_type)?;
+    let obs = Observation::try_from_data(id, data);
+    let _ = timestamp;
+    Ok(Event {
+        charger: mid,
+        observation: obs,
+    })
 }

src/signalr.rs

@@ -1,4 +1,4 @@
-use serde_json::Value;
+use serde_json::{json, Value};
 use thiserror::Error;
 
 use crate::stream::RecvError;
@@ -122,4 +122,15 @@ impl Stream {
         let json = self.buffer.pop().unwrap();
         Ok(Message::from_json(json)?)
     }
+
+    pub fn invoke(
+        &mut self,
+        target: &str,
+        args: serde_json::Value,
+    ) -> Result<(), tungstenite::Error> {
+        self.ws.send(json!( { "arguments": args,
+                                  "invocationId": "0",
+                                  "target": target,
+                                  "type": 1} ))
+    }
 }

src/stream.rs

@@ -1,8 +1,7 @@
-use std::net::TcpStream;
-
 use super::api::{ApiError, Context};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
+use std::net::TcpStream;
 use thiserror::Error;
 use tungstenite::{stream::MaybeTlsStream, Message, WebSocket};
 
@@ -46,31 +45,12 @@ pub struct Stream {
 impl Stream {
     pub fn open(ctx: &mut Context) -> Result<Stream, NegotiateError> {
         let r: NegotiateResponse = ctx.post_raw(STREAM_API_NEGOTIATION_URL, &())?;
-        dbg!(&r);
 
         let token = ctx.auth_token();
         let wss_url = format!(
             "{}?id={}&access_token={}",
             WSS_URL, r.connection_token, token
         );
-        dbg!(&wss_url);
-
-        /*
-        let req = tungstenite::http::Request::builder()
-            .uri(WSS_URL)
-            .header("Accept", "* / *")
-            .header("Host", "streams.easee.com")
-            .header("Origin", "https://portal.easee.com")
-            .header("Connection", "keep-alive, Upgrade")
-            .header("Upgrade", "websocket")
-            .header("Sec-WebSocket-Version", "13")
-            .header("Sec-WebSocket-Key", tungstenite::handshake::client::generate_key())
-            .header("Sec-Fetch-Dest", "websocket")
-            .header("Sec-Fetch-Mode", "websocket")
-            .header("Sec-Fetch-Site", "same-site")
-            .header("Cookie", format!("easee_skaat=\"{}\"", token))
-            .body(()).unwrap();
-        */
 
         let resp = tungstenite::client::connect(&wss_url);
 
@@ -87,7 +67,7 @@ impl Stream {
         Ok(stream)
     }
 
-    fn send<T: Serialize>(&mut self, msg: T) -> Result<(), tungstenite::Error> {
+    pub 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))
@@ -106,11 +86,4 @@ impl Stream {
 
         Ok(msgs)
     }
-
-    pub fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
-        self.send(json!( { "arguments": [id, true],
-                               "invocationId": "0",
-                               "target": "SubscribeWithCurrentState",
-                               "type": 1} ))
-    }
 }