Implement InfluxDB support and reading the configuration from a file.

base-station/software/Cargo.toml

@@ -29,6 +29,10 @@ nb = "0.1.2"
 gpio-cdev = { git = "https://github.com/mgottschlag/gpio-cdev", branch = "tokio", features = [ "async-tokio" ], optional = true }
 chrono = "0.4.11"
 nrf24l01-stick-driver = { git = "https://github.com/mgottschlag/nrf24l01-stick", optional = true }
+influxdb = { version = "0.3", features = [ "derive" ] }
+serde = { version = "1.0.119", features = ["derive"] }
+toml = "0.5.8"
+structopt = "0.3"
 
 [patch.crates-io]
 sysfs_gpio = { git = "https://github.com/mgottschlag/rust-sysfs-gpio", branch = "new-futures" }

base-station/software/default_config.toml

@@ -0,0 +1,5 @@
+[influxdb]
+enabled = false
+# address = "http://localhost:8086"
+# user = ""
+# password = ""

base-station/software/src/config.rs

@@ -0,0 +1,72 @@
+use std::fs;
+use std::path::Path;
+
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+pub struct Config {
+    pub influxdb: InfluxDbConfig,
+}
+
+impl Config {
+    pub fn load<P: AsRef<Path>>(path: P) -> Result<Config, ConfigError> {
+        let file = fs::read(path)?;
+        let config = toml::from_slice::<Config>(&file)?;
+        config.validate()?;
+        Ok(config)
+    }
+
+    fn validate(&self) -> Result<(), ConfigError> {
+        self.influxdb.validate()?;
+        Ok(())
+    }
+}
+
+#[derive(Deserialize)]
+pub struct InfluxDbConfig {
+    #[serde(default)]
+    pub enabled: bool,
+    #[serde(default)]
+    pub address: String,
+    #[serde(default)]
+    pub user: String,
+    #[serde(default)]
+    pub password: String,
+}
+
+impl InfluxDbConfig {
+    fn validate(&self) -> Result<(), ConfigError> {
+        if (self.user != "") != (self.password != "") {
+            return Err(ConfigError::Settings(
+                "need to specify both InfluxDB user and password".to_owned(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl Default for InfluxDbConfig {
+    fn default() -> InfluxDbConfig {
+        InfluxDbConfig {
+            enabled: false,
+            address: "http://localhost:8086".to_owned(),
+            user: "".to_owned(),
+            password: "".to_owned(),
+        }
+    }
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum ConfigError {
+    #[error("I/O error")]
+    Io(#[from] std::io::Error),
+    #[error("Parser error")]
+    Parser(#[from] toml::de::Error),
+    #[error("Invalid settings")]
+    Settings(String),
+}
+
+fn default_disabled() -> bool {
+    // https://github.com/serde-rs/serde/issues/1030
+    true
+}

base-station/software/src/main.rs

@@ -6,21 +6,26 @@
 compile_error!("Select an SPI implementation via the corresponding cargo feature!");
 
 use std::env;
+use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use chrono::offset::Local;
+use chrono::Utc;
 use log::{error, info};
-use protocol::{GetValues, Packet, Report, Value, ValueType, Values};
 use rand::{rngs::ThreadRng, Rng};
+use structopt::StructOpt;
 use tokio::sync::Mutex;
 use tokio::time::delay_for;
 
+use config::Config;
 use current::CurrentValues;
+use protocol::{GetValues, Packet, Report, Value, ValueType, Values};
 use publish::Publish;
-use radio::{Radio, RadioConfig};
+use radio::Radio;
 use tsdb::TimeSeriesDatabase;
 
+mod config;
 mod current;
 mod publish;
 mod radio;
@@ -28,6 +33,19 @@ mod radio;
 mod spi;
 mod tsdb;
 
+#[derive(Debug, StructOpt)]
+#[structopt(name = "base-station", about = "Smart home base station.")]
+struct Options {
+    /// Config file
+    #[structopt(
+        short,
+        long,
+        default_value = "/etc/smart-home.toml",
+        parse(from_os_str)
+    )]
+    config: PathBuf,
+}
+
 #[tokio::main]
 async fn main() {
     env::set_var(
@@ -36,14 +54,18 @@ async fn main() {
     );
     env_logger::init();
 
+    // Load the configuration.
+    let options = Options::from_args();
+    let config = Config::load(options.config)
+        .expect("could not open config file - did you forget to pass \"-c\"?");
+
     // Initialize radio, MQTT and InfluxDB.
-    let radio_config = RadioConfig{
-        // TODO: Configure pins.
+    let (radio, mut incoming_packets) = Radio::init().await.expect("radio initialization failed");
+    let tsdb = if config.influxdb.enabled {
+        Some(TimeSeriesDatabase::init(&config.influxdb))
+    } else {
+        None
     };
-    let (radio, mut incoming_packets) = Radio::init(radio_config)
-        .await
-        .expect("radio initialization failed");
-    let tsdb = TimeSeriesDatabase::init();
     let publish = Publish::init();
 
     let mut app = Application {
@@ -68,7 +90,7 @@ async fn main() {
 struct Application {
     radio: Arc<Mutex<Radio>>,
     publish: Arc<Mutex<Publish>>,
-    tsdb: Arc<Mutex<TimeSeriesDatabase>>,
+    tsdb: Arc<Mutex<Option<TimeSeriesDatabase>>>,
     current: CurrentValues,
     rng: ThreadRng,
 }
@@ -110,6 +132,7 @@ impl Application {
     async fn handle_report(&mut self, device_id: u8, report: Report) {
         let location = radio::get_device_location(device_id);
         let count = report.count as usize;
+        let time = Utc::now();
         // Update local copy of current values.
         for i in 0..count {
             match report.values[i] {
@@ -118,15 +141,15 @@ impl Application {
                     error!("device tried to report a time");
                 }
                 Value::Temperature(temperature) => {
-                    let temperature = temperature as f32 / 10.0;
+                    let temperature = temperature as f32 / report.values[i].decimal_factor() as f32;
                     info!("{:?} temperature: {} °C", location, temperature);
                 }
                 Value::Pressure(pressure) => {
-                    let pressure = pressure as f32 / 100.0;
+                    let pressure = pressure as f32 / report.values[i].decimal_factor() as f32;
                     info!("{:?} pressure: {} HPa", location, pressure);
                 }
                 Value::Humidity(humidity) => {
-                    let humidity = humidity as f32 / 100.0;
+                    let humidity = humidity as f32 / report.values[i].decimal_factor() as f32;
                     info!("{:?} humidity: {}%", location, humidity);
                 }
             }
@@ -145,8 +168,9 @@ impl Application {
         let tsdb = self.tsdb.clone();
         tokio::spawn(async move {
             let mut tsdb = tsdb.lock().await;
-            tsdb.insert(Instant::now(), location, &report.values[0..count])
-                .await;
+            if let Some(tsdb) = tsdb.as_mut() {
+                tsdb.insert(time, location, &report.values[0..count]).await;
+            }
         });
     }
 

base-station/software/src/radio/gpio.rs

@@ -18,10 +18,9 @@ use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::time::delay_for;
 
 use super::super::spi::EmbeddedHalSpidev;
-use super::{get_key, RadioConfig};
+use super::get_key;
 
 pub struct RadioTask {
-    config: super::RadioConfig,
     nrf24: Option<RxMode<NRF24L01<GpioError, Pin, Pin, EmbeddedHalSpidev>>>,
     irq: Option<AsyncLineEventHandle>,
     tx: UnboundedReceiver<(u8, Packet)>,
@@ -30,12 +29,10 @@ pub struct RadioTask {
 
 impl RadioTask {
     pub fn new(
-        config: RadioConfig,
         tx: UnboundedReceiver<(u8, Packet)>,
         rx: UnboundedSender<(u8, Packet)>,
     ) -> RadioTask {
         RadioTask {
-            config,
             nrf24: None,
             irq: None,
             tx,

base-station/software/src/radio/mod.rs

@@ -25,11 +25,6 @@ const WEATHER_STATION_0_KEY: [u8; 16] = include!("../../../../common/weather_sta
 const WEATHER_STATION_1_ID: u8 = 0x31;
 const WEATHER_STATION_1_KEY: [u8; 16] = include!("../../../../common/weather_station_1_key.txt");
 
-/// Hardware configuration.
-pub struct RadioConfig {
-    // TODO
-}
-
 /// Wrapper around the NRF24 radio driver.
 pub struct Radio {
     tx: UnboundedSender<(u8, Packet)>,
@@ -37,13 +32,11 @@ pub struct Radio {
 
 impl Radio {
     /// Initializes the radio and returns both radio and the stream of incoming packets.
-    pub async fn init(
-        config: RadioConfig,
-    ) -> Result<(Radio, UnboundedReceiver<(u8, Packet)>), RadioError> {
+    pub async fn init() -> Result<(Radio, UnboundedReceiver<(u8, Packet)>), RadioError> {
         info!("Initializing radio...");
         let (tx_send, tx_recv) = mpsc::unbounded_channel();
         let (rx_send, rx_recv) = mpsc::unbounded_channel();
-        let mut task = RadioTask::new(config, tx_recv, rx_send);
+        let mut task = RadioTask::new(tx_recv, rx_send);
         task.init().await?;
         info!("radio initialized.");
 

base-station/software/src/radio/nrfstick.rs

@@ -4,19 +4,16 @@ use std::time::Duration;
 
 use futures_util::stream::StreamExt;
 use log::{error, info};
-use nrf24l01_stick_driver::{
-    Configuration, CrcMode, DataRate, Receiver, MAX_PAYLOAD_LEN, NRF24L01,
-};
+use nrf24l01_stick_driver::{Configuration, CrcMode, DataRate, Receiver, NRF24L01};
 use protocol::Packet;
 use rand::Rng;
 use tokio::select;
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::time::delay_for;
 
-use super::{get_key, RadioConfig};
+use super::get_key;
 
 pub struct RadioTask {
-    config: super::RadioConfig,
     nrf: Option<Receiver>,
 
     tx: UnboundedReceiver<(u8, Packet)>,
@@ -25,16 +22,10 @@ pub struct RadioTask {
 
 impl RadioTask {
     pub fn new(
-        config: RadioConfig,
         tx: UnboundedReceiver<(u8, Packet)>,
         rx: UnboundedSender<(u8, Packet)>,
     ) -> RadioTask {
-        RadioTask {
-            config,
-            nrf: None,
-            tx,
-            rx,
-        }
+        RadioTask { nrf: None, tx, rx }
     }
 
     pub async fn init(&mut self) -> Result<(), RadioError> {

base-station/software/src/tsdb.rs

@@ -1,13 +1,18 @@
 //! Interface to fill/query a timeseries database (InfluxDB).
 
-use std::time::Instant;
+use chrono::{DateTime, Utc};
+use influxdb::InfluxDbWriteable;
+use influxdb::{Client, Timestamp};
+use log::error;
 
+use super::config::InfluxDbConfig;
 use protocol::{Location, Value};
 
 // TODO: Configuration mechanism.
 
 /// Interface to InfluxDB.
 pub struct TimeSeriesDatabase {
+    client: Client,
     // TODO
 }
 
@@ -17,9 +22,16 @@ impl TimeSeriesDatabase {
     /// The function does not return an error. Instead, the code will automatically retry
     /// connection in case the connection cannot be established or the server closes the
     /// connection.
-    pub fn init() -> TimeSeriesDatabase {
-        // TODO
-        TimeSeriesDatabase {}
+    pub fn init(config: &InfluxDbConfig) -> TimeSeriesDatabase {
+        // TODO: Configurable address
+        let client = Client::new(&config.address, "weather");
+        let client = if config.user != "" && config.password != "" {
+            client.with_auth(&config.user, &config.password)
+        } else {
+            client
+        };
+        // TODO: Ping the server.
+        TimeSeriesDatabase { client }
     }
 
     /// Inserts values into the database.
@@ -34,7 +46,84 @@ impl TimeSeriesDatabase {
     /// * `time` - time at which the values were received
     /// * `location` - location of the device which sent the values
     /// * `values` - list of values to be inserted into the database
-    pub async fn insert(&mut self, _time: Instant, _location: Location, _values: &[Value]) {
-        // TODO
+    pub async fn insert(&mut self, time: DateTime<Utc>, location: Location, values: &[Value]) {
+        let location = location.to_str();
+
+        for value in values {
+            if let Err(e) = self.try_insert(time, location.to_owned(), value).await {
+                error!("Could not insert value into TSDB: {:?}", e);
+                // TODO: Ping the server and do not try to write anything until ping succeds to
+                // reduce debug spam.
+            }
+        }
+    }
+
+    async fn try_insert(
+        &mut self,
+        time: DateTime<Utc>,
+        location: String,
+        value: &Value,
+    ) -> Result<(), TsdbError> {
+        match *value {
+            Value::Temperature(temperature) => {
+                let temperature = temperature as f32 / value.decimal_factor() as f32;
+                let reading = TemperatureReading {
+                    time: time,
+                    temperature,
+                    location,
+                };
+                self.client.query(&reading.into_query("weather")).await?;
+            }
+            Value::Pressure(pressure) => {
+                let pressure = pressure as f32 / value.decimal_factor() as f32;
+                let reading = PressureReading {
+                    time: time,
+                    pressure,
+                    location,
+                };
+                self.client.query(&reading.into_query("weather")).await?;
+            }
+            Value::Humidity(humidity) => {
+                let humidity = humidity as f32 / value.decimal_factor() as f32;
+                let reading = HumidityReading {
+                    time: time,
+                    humidity,
+                    location,
+                };
+                self.client.query(&reading.into_query("weather")).await?;
+            }
+            _ => {}
+        }
+        Ok(())
     }
 }
+
+#[derive(InfluxDbWriteable)]
+struct TemperatureReading {
+    time: DateTime<Utc>,
+    temperature: f32,
+    #[tag]
+    location: String,
+}
+
+#[derive(InfluxDbWriteable)]
+struct PressureReading {
+    time: DateTime<Utc>,
+    pressure: f32,
+    #[tag]
+    location: String,
+}
+
+#[derive(InfluxDbWriteable)]
+struct HumidityReading {
+    time: DateTime<Utc>,
+    humidity: f32,
+    #[tag]
+    location: String,
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum TsdbError {
+    #[error("Database error")]
+    Database(#[from] influxdb::Error),
+}

common/rust-protocol/src/lib.rs

@@ -236,6 +236,16 @@ impl Location {
             _ => Err(Error::InvalidLocation),
         }
     }
+
+    pub fn to_str(self) -> &'static str {
+        match self {
+            Self::Livingroom => "Livingroom",
+            Self::Bathroom => "Bathroom",
+            Self::Bedroom => "Bedroom",
+            Self::Kitchen => "Kitchen",
+            Self::Balcony => "Balcony",
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -322,6 +332,16 @@ impl Value {
             ValueType::Humidity => Self::Humidity(value as u16),
         }
     }
+
+    pub fn decimal_factor(&self) -> u32 {
+        match self {
+            Self::Invalid => 1,
+            Self::Time(_) => 1,
+            Self::Temperature(_) => 10,
+            Self::Pressure(_) => 100,
+            Self::Humidity(_) => 100,
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]