base-station: Broadcast weather data via MQTT.

base-station/software/src/main.rs

@@ -2,6 +2,7 @@ use std::env;
 use std::io;
 use std::thread::sleep;
 use std::time::Duration;
+use std::sync::mpsc;
 
 use log::error;
 use ::mqtt::control::variable_header::ConnectReturnCode;
@@ -19,32 +20,62 @@ fn main() {
     );
     env_logger::init();
 
-    radio::start();
+    let (sensor_send, sensor_recv) = mpsc::channel();
+
+    radio::start(sensor_send);
 
     // Whenever the MQTT connection returns an error, we wait for a couple of seconds and then try
     // to reconnect.
     loop {
         // TODO: Pass events from ratio to mqtt connection.
-        if let Err(e) = run_mqtt_connection() {
+        if let Err(e) = run_mqtt_connection(&sensor_recv) {
             error!("MQTT error: {:?}", e);
         }
         sleep(Duration::from_secs(3));
     }
 }
 
-fn run_mqtt_connection() -> Result<(), Error> {
+fn run_mqtt_connection(updates: &mpsc::Receiver<SensorUpdate>) -> Result<(), Error> {
     let mut m = MQTT::connect("127.0.0.1:1883")?;
 
-    let mut counter = 0;
     loop {
-        // TODO: Real values
-        m.publish("gottschlag/livingroom/temperature", &format!("{}", counter))?;
-
-        sleep(Duration::from_secs(3));
-        counter += 1;
+        let update = updates.recv().unwrap();
+        let location = match update.location {
+            0 => "livingroom",
+            _ => {
+                error!("Unknown location in sensor update.");
+                continue;
+            },
+        };
+        for value in update.data.iter() {
+            match value {
+                SensorData::Temperature(value) => {
+                    m.publish(&format!("gottschlag/{}/temperature", location), &format!("{}", value))?;
+                },
+                SensorData::Humidity(value) => {
+                    m.publish(&format!("gottschlag/{}/humidity", location), &format!("{}", value))?;
+                },
+                SensorData::Pressure(value) => {
+                    m.publish(&format!("gottschlag/{}/pressure", location), &format!("{}", value))?;
+                }
+            }
+        }
     }
 }
 
+pub struct SensorUpdate {
+    location: u32,
+    data: Vec<SensorData>,
+}
+
+pub enum SensorData {
+    Temperature(f32),
+    Humidity(f32),
+    Pressure(f32),
+}
+
+
+
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
     #[error("I/O error")]

base-station/software/src/radio.rs

@@ -1,6 +1,7 @@
 use std::convert::TryInto;
 use std::thread;
 use std::time::{Duration, Instant};
+use std::sync::mpsc;
 
 use embedded_nrf24l01::{Configuration, CrcMode, DataRate, NRF24L01};
 use linux_embedded_hal::spidev::{SpiModeFlags, Spidev, SpidevOptions};
@@ -10,13 +11,14 @@ use log::{error, info};
 
 use crate::spi::EmbeddedHalSpidev;
 use crate::Error;
+use crate::{SensorUpdate, SensorData};
 
-pub fn start() {
+pub fn start(updates: mpsc::Sender<SensorUpdate>) {
     // TODO: Channels for sensor readings.
     thread::spawn(move || {
         // Whenever a error occurs, reinitialize the radio module shortly later.
         loop {
-            match radio_thread() {
+            match radio_thread(&updates) {
                 Ok(()) => {}
                 Err(e) => {
                     error!("Radio error: {:?}", e);
@@ -27,7 +29,7 @@ pub fn start() {
     });
 }
 
-fn radio_thread() -> Result<(), Error> {
+fn radio_thread(updates: &mpsc::Sender<SensorUpdate>) -> Result<(), Error> {
     info!("Initializing radio...");
 
     // The NRF module is connected as follows:
@@ -90,13 +92,20 @@ fn radio_thread() -> Result<(), Error> {
     loop {
         if let Some(pipe) = nrf24.can_read().unwrap() {
             let payload = nrf24.read().unwrap();
-            info!("packet received on pipe {}: {:?}", pipe, payload.as_ref());
+            info!("packet received on pipe {}: {:?}, {}", pipe, payload.as_ref(), payload.len());
             let pressure = u32::from_le_bytes(payload.as_ref()[0..4].try_into().unwrap());
             let temperature = u32::from_le_bytes(payload.as_ref()[4..8].try_into().unwrap());
             let humidity = u32::from_le_bytes(payload.as_ref()[8..12].try_into().unwrap());
-            info!("pressure: {}", pressure as f32 / 100.0);
-            info!("temperature: {}", temperature as f32 / 100.0);
-            info!("humidity: {}", humidity as f32 / 1024.0);
+            let pressure = pressure as f32 / 100.0;
+            let temperature = temperature as f32 / 100.0;
+            let humidity = humidity as f32 / 1024.0;
+            info!("pressure: {}", pressure);
+            info!("temperature: {}", temperature);
+            info!("humidity: {}", humidity);
+            updates.send(SensorUpdate{
+                location: 0,
+                data: vec![SensorData::Temperature(temperature), SensorData::Pressure(pressure), SensorData::Humidity(humidity)],
+            }).unwrap();
             let end = Instant::now();
             let elapsed = end.duration_since(start);
             info!("Debug: {:?}", elapsed);