Refactor prom client

src/config.rs

@@ -0,0 +1,39 @@
+use serde::Deserialize;
+use anyhow::Result;
+use serde_json;
+
+#[derive(Debug,Deserialize)]
+pub struct Charger {
+    pub id: String,
+    pub channel_id: String,
+    pub owners: Vec<String>,
+}
+
+#[derive(Debug,Deserialize)]
+pub struct Mattermost {
+    pub base: String,
+    pub token: String,
+}
+
+#[derive(Debug,Deserialize)]
+pub struct Prometheus {
+    pub base: String
+}
+
+#[derive(Debug,Deserialize)]
+pub struct Config {
+    pub easee_token_path: String,
+    pub prometheus: Prometheus,
+    pub mattermost: Mattermost,
+    pub chargers: Vec<Charger>,
+}
+
+#[derive(Debug,Deserialize)]
+pub struct Regulator {
+    power_bias_watts: f64,
+    monophase_volts: f64,
+}
+
+pub fn load_config(path: &str) -> Result<Config> {
+    Ok(serde_json::from_str(&std::fs::read_to_string(path)?)?)
+}

src/control.rs

@@ -0,0 +1,121 @@
+use std::collections::HashMap;
+use std::convert::Infallible;
+use std::sync::{Arc, Mutex};
+
+use anyhow::Result;
+use easee::api::{self, ChargerOpMode, Context};
+use easee::observation::{self, Event, ObservationError};
+use tracing::{error, info, warn};
+
+use crate::mattermost::{self, Channel};
+use crate::prom::PromClient;
+use crate::config::{self, Config};
+
+use observation::{Observation,PilotMode};
+
+struct Charger {
+    inner: api::Charger,
+    owners: Vec<String>,
+    current: Option<(f64, f64, f64)>
+}
+
+impl Charger {
+    pub fn from_api(inner: api::Charger, configs: &[config::Charger]) -> Self {
+        let owners = configs.iter()
+            .find(|c| c.id == inner.id)
+            .map(|c| &c.owners)
+            .cloned()
+            .unwrap_or_default();
+        Charger { inner, owners, current: None }
+    }
+
+}
+
+pub fn start(mut ctx: Context, config: Config, mut chargers: Vec<api::Charger>) -> Result<Infallible> {
+
+    let mattermost = mattermost::Context::new(config.mattermost.base, &config.mattermost.token)?;
+    let mut stream = observation::Stream::from_context(&mut ctx)?;
+
+    // TODO
+    let chargers: HashMap<String, Charger> = chargers.into_iter()
+        .filter_map(|c| {
+            stream.subscribe(&c.id).ok()?;
+            let name = c.name.clone();
+            Some((name, Charger::from_api(c, &config.chargers)))
+        })
+        .collect();
+
+    let chargers = Arc::new(Mutex::new(chargers));
+
+    // TODO get channel on a per-charger basis
+    let channel = mattermost.channel("9d9o1a5qf7fofk3wqfa493gkfe");
+
+
+
+    info!("Controller started");
+    mattermost.send_to_channel(&channel, "Easee Controller started")?;
+
+    loop {
+
+        let evt = match stream.recv() {
+            Ok(e) => e,
+            Err(ObservationError::Stream(stream_error)) => Err(stream_error)?,
+            Err(other) => { error!("Cannot process message: {}", other); continue },
+        };
+
+        let mut chargers = chargers.lock().unwrap();
+
+        let Some(charger) = chargers.get_mut(&evt.charger)
+            else { warn!("Received message for unknown charger {}", &evt.charger); continue };
+
+        let result = handle_event(evt, charger, &mattermost, &channel);
+
+        if let Err(err) = result {
+            error!("Error handling observation: {:?}", err);
+        }
+
+    }
+}
+
+fn handle_event(evt: Event, charger: &mut Charger, ctx: &mattermost::Context, channel: &Channel) -> Result<()> {
+
+    let send = |msg: &str| ctx.send_to_channel(channel, msg);
+
+    match evt.observation {
+        Observation::PilotMode(mode) => {
+            match mode {
+                PilotMode::Disconnected => send("Car Disconnected"),
+                PilotMode::Connected => send("Car Connected"),
+                PilotMode::Charging => send("Car Charging"),
+                PilotMode::NeedsVentilation => send("Car needs ventilation"),
+                PilotMode::FaultDetected => send("Fault detected"),
+                PilotMode::Unknown => send("Unknown"),
+            }
+        },
+        Observation::ChargerOpMode(mode) => {
+            match mode {
+                ChargerOpMode::Unknown => send("Unknown"),
+                ChargerOpMode::Disconnected => send("Charger disconnected"),
+                ChargerOpMode::Paused => send("Charge paused"),
+                ChargerOpMode::Charging => send("Charging"),
+                ChargerOpMode::Finished => send("Charging finished"),
+                ChargerOpMode::Error => send("Charger error"),
+                ChargerOpMode::Ready => send("Charger ready"),
+            }?;
+            ctx.set_status(mode)
+        },
+        other => Ok(info!("{}: {:?}", evt.charger, other)),
+    }
+}
+
+pub fn adjust_power(prom: PromClient) -> Result<()> {
+
+    loop {
+        let export_power = prom.current_power()?;
+
+        
+
+    }
+
+
+}

src/main.rs

@@ -4,11 +4,14 @@ use anyhow::{Context as AnyhowContext, Result};
 use clap::ValueEnum;
 use clap::{Parser, Subcommand};
 use easee::api::{ApiError, Charger, ChargerState, ChargingSession, Context};
-use easee::stream;
+use easee::observation;
 
 use tracing::info;
 
 mod prom;
+mod mattermost;
+mod config;
+mod control;
 
 #[derive(Debug, Clone, Copy, ValueEnum)]
 enum Command {
@@ -37,6 +40,7 @@ enum Mode {
     },
     Stream,
     Power,
+    Control,
 }
 
 #[derive(Debug, Parser)]
@@ -47,8 +51,8 @@ struct CLI {
     #[arg(short, long)]
     charger_id: Vec<String>,
 
-    #[arg(short, long, default_value = "http://localhost:9090")]
-    prometheus: String,
+    #[arg(short, long, default_value = "./config.json")]
+    config: String,
 
     #[command(subcommand)]
     mode: Mode,
@@ -72,17 +76,23 @@ fn login() -> Result<()> {
     let username = username.trim();
     let password = password.trim();
 
-    let ctx = easee::api::Context::from_login(&username, &password)?;
-    eprintln!("Login successful.");
+    let mut ctx = easee::api::Context::from_login(&username, &password)?;
+    info!("Login successful.");
 
-    std::fs::write(SAVED_TOKEN_PATH, ctx.save().as_bytes())?;
+    save_context(&mut ctx);
     return Ok(());
 }
 
 fn load_context() -> Result<easee::api::Context> {
     let saved = std::fs::read_to_string(SAVED_TOKEN_PATH)
         .context("Cannot read saved token (did you log in ?)")?;
-    Ok(easee::api::Context::from_saved(&saved)?)
+    let ctx = easee::api::Context::from_saved(&saved)?
+        .on_refresh(save_context);
+    Ok(ctx)
+}
+
+fn save_context(ctx: &mut Context) {
+    std::fs::write(SAVED_TOKEN_PATH, ctx.save().as_bytes()).unwrap();
 }
 
 fn load_chargers(ctx: &mut Context, names: &[String]) -> Result<Vec<Charger>> {
@@ -101,13 +111,13 @@ fn load_chargers(ctx: &mut Context, names: &[String]) -> Result<Vec<Charger>> {
 
 fn stream(names: &[String]) -> Result<()> {
     let mut ctx = load_context()?;
-    let mut stream = stream::Stream::open(&mut ctx)?;
+
+    let mut stream = observation::Stream::from_context(&mut ctx)?;
     let chargers = load_chargers(&mut ctx, names)?;
     for c in &chargers {
         stream.subscribe(&c.id)?;
     }
 
-    let mut stream = easee::signalr::Stream::from_ws(stream);
     loop {
         println!("{:?}", stream.recv()?);
     }
@@ -141,7 +151,13 @@ fn main() -> Result<()> {
             .expect("Tracing subscriber failed");
     }
 
+    // We need to do this before loading the context
+    if let Mode::Login = args.mode {
+        login()?; return Ok(())
+    }
+
     let mut ctx = load_context()?;
+    let config = config::load_config(&args.config)?;
 
     match args.mode {
         Mode::Login => login()?,
@@ -166,12 +182,15 @@ fn main() -> Result<()> {
         }
         Mode::Stream => stream(&args.charger_id)?,
         Mode::Power => {
-            let pow = prom::current_power(&*args.prometheus)?;
+            let pow = prom::PromClient::new(config.prometheus.base).current_power()?;
             println!("P1:{}W P2:{}W P3:{}W", pow.0, pow.1, pow.2);
+        },
+        Mode::Control => {
+            let chargers = load_chargers(&mut ctx, &args.charger_id)?;
+            control::start(ctx, config, chargers)?;
         }
     };
 
-    std::fs::write(SAVED_TOKEN_PATH, ctx.save().as_bytes())?;
     Ok(())
 }
 

src/mattermost.rs

@@ -0,0 +1,63 @@
+use easee::api::ChargerOpMode;
+use ureq::json;
+use anyhow::Result;
+
+pub struct Context {
+    pub base: String,
+    pub auth_header: String,
+}
+
+pub struct Channel {
+    channel_id: String,
+}
+
+
+impl Context {
+
+    pub fn new(base: String, token: &str) -> Result<Self> {
+        Ok(Self {
+            base,
+            auth_header: format!("Bearer {token}"),
+        })
+    }
+
+    fn path(&self, rel: &str) -> String {
+        format!("{}/api/v4/{}", self.base, rel)
+    }
+
+    pub fn set_custom_status(&self, text: &str, emoji: &str) -> Result<()> {
+        let path = &self.path("users/me/status/custom");
+        ureq::put(path)
+            .send_json(json!( { "emoji": emoji, "text": text } ))?;
+        Ok(())
+    }
+
+    pub fn set_status(&self, mode: ChargerOpMode) -> Result<()> {
+        use ChargerOpMode::*;
+        let (text, emoji) = match mode {
+            Unknown => ("Unknown", "interrobang"),
+            Disconnected => ("Disconnected", "zzz"),
+            Paused => ("Paused", "double_vertical_bar"),
+            Charging => ("Charging", "zap"),
+            Finished => ("Finished", "white_check_mark"),
+            Error => ("Error", "no_entry_sign"),
+            Ready => ("Ready", "electric_plug"),
+        };
+        self.set_custom_status(text, emoji)
+    }
+
+    pub fn channel(&self, id: &str) -> Channel {
+        Channel { channel_id: id.to_owned() }
+    }
+
+    pub fn send_to_channel(&self, channel: &Channel, msg: &str) -> Result<()> {
+        let path = self.path("posts");
+        ureq::post(&path)
+            .set("Authorization", &self.auth_header)
+            .send_json(json!( 
+                { "channel_id": channel.channel_id, "message": msg }
+            ))?;
+        Ok(())
+    }
+
+}

src/prom.rs

@@ -41,33 +41,44 @@ struct MatrixEntry {
     values: Vec<(f64, String)>,
 }
 
-pub fn current_power(base: &str) -> Result<(f64, f64, f64)> {
-    let url = format!("{}{}", base, PROM_QUERY);
+pub struct PromClient {
+    base: String,
+    power_query_url: String
+}
 
-    let reply: PromReply = ureq::get(&url).call()?.into_json()?;
-
-    let PromReply::Success {
-        data: PromData::Vector(v),
-    } = reply
-    else {
-        bail!("Could not understand Prometheus reply: {:?}", reply)
-    };
-
-    let mut r = (None, None, None);
-
-    for entry in &v {
-        let val: f64 = entry.value.1.parse()?;
-        match entry.metric.get("phase").map(|s| &**s) {
-            Some("a") => r.0 = Some(val),
-            Some("b") => r.1 = Some(val),
-            Some("c") => r.2 = Some(val),
-            _ => warn!("Metric with unexpected phase"),
-        }
+impl PromClient {
+    pub fn new(base: String) -> Self {
+        let power_query_url = format!("{}{}", &base, PROM_QUERY);
+        PromClient { base, power_query_url }
     }
 
-    Ok((
-        r.0.ok_or_else(|| anyhow!("Missing phase a"))?,
-        r.1.ok_or_else(|| anyhow!("Missing phase b"))?,
-        r.2.ok_or_else(|| anyhow!("Missing phase c"))?,
-    ))
+    pub fn current_power(&self) -> Result<(f64, f64, f64)> {
+        let reply: PromReply = ureq::get(&self.power_query_url).call()?.into_json()?;
+
+        let PromReply::Success {
+            data: PromData::Vector(v),
+        } = reply
+        else {
+            bail!("Could not understand Prometheus reply: {:?}", reply)
+        };
+
+        let mut r = (None, None, None);
+
+        for entry in &v {
+            let val: f64 = entry.value.1.parse()?;
+            match entry.metric.get("phase").map(|s| &**s) {
+                Some("a") => r.0 = Some(val),
+                Some("b") => r.1 = Some(val),
+                Some("c") => r.2 = Some(val),
+                _ => warn!("Metric with unexpected phase"),
+            }
+        }
+
+        Ok((
+            r.0.ok_or_else(|| anyhow!("Missing phase a"))?,
+            r.1.ok_or_else(|| anyhow!("Missing phase b"))?,
+            r.2.ok_or_else(|| anyhow!("Missing phase c"))?,
+        ))
+}
+
 }