cleanups

Software/MainBoard/rust/src/hal/esp.rs

@@ -0,0 +1,994 @@
+use crate::bail;
+use crate::config::{NetworkConfig, PlantControllerConfig};
+use crate::hal::{PLANT_COUNT, TIME_ACCESS};
+use crate::log::{LogMessage, LOG_ACCESS};
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+
+use crate::fat_error::{ContextExt, FatError, FatResult};
+use crate::hal::little_fs2storage_adapter::LittleFs2Filesystem;
+use alloc::string::ToString;
+use alloc::sync::Arc;
+use alloc::{format, string::String, vec, vec::Vec};
+use core::net::{IpAddr, Ipv4Addr, SocketAddr};
+use core::str::FromStr;
+use core::sync::atomic::Ordering;
+use embassy_executor::Spawner;
+use embassy_net::udp::UdpSocket;
+use embassy_net::{DhcpConfig, Ipv4Cidr, Runner, Stack, StackResources, StaticConfigV4};
+use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
+use embassy_sync::mutex::{Mutex, MutexGuard};
+use embassy_sync::once_lock::OnceLock;
+use embassy_time::{Duration, Timer, WithTimeout};
+use embedded_storage::nor_flash::{check_erase, NorFlash, ReadNorFlash};
+use esp_bootloader_esp_idf::ota::OtaImageState::Valid;
+use esp_bootloader_esp_idf::ota::{Ota, OtaImageState, Slot};
+use esp_bootloader_esp_idf::partitions::FlashRegion;
+use esp_hal::gpio::{Input, RtcPinWithResistors};
+use esp_hal::rng::Rng;
+use esp_hal::rtc_cntl::{
+    sleep::{TimerWakeupSource, WakeupLevel},
+    Rtc,
+};
+use esp_hal::system::software_reset;
+use esp_println::println;
+use esp_storage::FlashStorage;
+use esp_wifi::wifi::{
+    AccessPointConfiguration, AccessPointInfo, AuthMethod, ClientConfiguration, Configuration,
+    ScanConfig, ScanTypeConfig, WifiController, WifiDevice, WifiState,
+};
+use littlefs2::fs::Filesystem;
+use littlefs2_core::{FileType, PathBuf, SeekFrom};
+use log::{info, warn};
+use mcutie::{
+    Error, McutieBuilder, McutieReceiver, McutieTask, MqttMessage, PublishDisplay, Publishable,
+    QoS, Topic,
+};
+use portable_atomic::AtomicBool;
+use smoltcp::socket::udp::PacketMetadata;
+use smoltcp::wire::DnsQueryType;
+use sntpc::{get_time, NtpContext, NtpTimestampGenerator};
+
+#[esp_hal::ram(rtc_fast, persistent)]
+static mut LAST_WATERING_TIMESTAMP: [i64; PLANT_COUNT] = [0; PLANT_COUNT];
+#[esp_hal::ram(rtc_fast, persistent)]
+static mut CONSECUTIVE_WATERING_PLANT: [u32; PLANT_COUNT] = [0; PLANT_COUNT];
+#[esp_hal::ram(rtc_fast, persistent)]
+static mut LOW_VOLTAGE_DETECTED: i8 = 0;
+#[esp_hal::ram(rtc_fast, persistent)]
+static mut RESTART_TO_CONF: i8 = 0;
+
+const CONFIG_FILE: &str = "config.json";
+const NTP_SERVER: &str = "pool.ntp.org";
+
+static MQTT_CONNECTED_EVENT_RECEIVED: AtomicBool = AtomicBool::new(false);
+static MQTT_ROUND_TRIP_RECEIVED: AtomicBool = AtomicBool::new(false);
+pub static MQTT_STAY_ALIVE: AtomicBool = AtomicBool::new(false);
+static MQTT_BASE_TOPIC: OnceLock<String> = OnceLock::new();
+
+#[derive(Serialize, Debug)]
+pub struct FileInfo {
+    filename: String,
+    size: usize,
+}
+
+#[derive(Serialize, Debug)]
+pub struct FileList {
+    total: usize,
+    used: usize,
+    files: Vec<FileInfo>,
+}
+
+#[derive(Copy, Clone, Default)]
+struct Timestamp {
+    stamp: DateTime<Utc>,
+}
+
+// Minimal esp-idf equivalent for gpio_hold on esp32c6 via ROM functions
+extern "C" {
+    fn gpio_pad_hold(gpio_num: u32);
+    fn gpio_pad_unhold(gpio_num: u32);
+}
+
+#[inline(always)]
+pub fn hold_enable(gpio_num: u8) {
+    unsafe { gpio_pad_hold(gpio_num as u32) }
+}
+
+#[inline(always)]
+pub fn hold_disable(gpio_num: u8) {
+    unsafe { gpio_pad_unhold(gpio_num as u32) }
+}
+
+impl NtpTimestampGenerator for Timestamp {
+    fn init(&mut self) {
+        self.stamp = DateTime::default();
+    }
+
+    fn timestamp_sec(&self) -> u64 {
+        self.stamp.timestamp() as u64
+    }
+
+    fn timestamp_subsec_micros(&self) -> u32 {
+        self.stamp.timestamp_subsec_micros()
+    }
+}
+
+pub struct Esp<'a> {
+    pub fs: Arc<Mutex<CriticalSectionRawMutex, Filesystem<'static, LittleFs2Filesystem>>>,
+    pub rng: Rng,
+    //first starter (ap or sta will take these)
+    pub interface_sta: Option<WifiDevice<'static>>,
+    pub interface_ap: Option<WifiDevice<'static>>,
+    pub controller: Arc<Mutex<CriticalSectionRawMutex, WifiController<'static>>>,
+
+    pub boot_button: Input<'a>,
+
+    // RTC-capable GPIO used as external wake source (store the raw peripheral)
+    pub wake_gpio1: esp_hal::peripherals::GPIO1<'static>,
+
+    pub ota: Ota<'static, FlashStorage>,
+    pub ota_target: &'static mut FlashRegion<'static, FlashStorage>,
+    pub current: Slot,
+    pub slot0_state: OtaImageState,
+    pub slot1_state: OtaImageState,
+}
+
+// SAFETY: On this target we never move Esp across OS threads; the firmware runs single-core
+// cooperative tasks with Embassy. All interior mutability of non-Send peripherals is gated
+// behind &mut self or embassy_sync Mutex with CriticalSectionRawMutex, which does not rely on
+// thread scheduling. Therefore it is sound to mark Esp as Send to satisfy trait object bounds
+// (e.g., Box<dyn BoardInteraction + Send>). If you add fields that are accessed from multiple
+// CPU cores/threads, reconsider this.
+unsafe impl Send for Esp<'_> {}
+
+macro_rules! mk_static {
+    ($t:ty,$val:expr) => {{
+        static STATIC_CELL: static_cell::StaticCell<$t> = static_cell::StaticCell::new();
+        #[deny(unused_attributes)]
+        let x = STATIC_CELL.uninit().write(($val));
+        x
+    }};
+}
+
+impl Esp<'_> {
+    pub(crate) async fn delete_file(&self, filename: String) -> FatResult<()> {
+        let file = PathBuf::try_from(filename.as_str())?;
+        let access = self.fs.lock().await;
+        access.remove(&*file)?;
+        Ok(())
+    }
+    pub(crate) async fn write_file(
+        &mut self,
+        filename: String,
+        offset: u32,
+        buf: &[u8],
+    ) -> Result<(), FatError> {
+        let file = PathBuf::try_from(filename.as_str())?;
+        let access = self.fs.lock().await;
+        access.open_file_with_options_and_then(
+            |options| options.read(true).write(true).create(true),
+            &*file,
+            |file| {
+                file.seek(SeekFrom::Start(offset))?;
+                file.write(buf)?;
+                Ok(())
+            },
+        )?;
+        Ok(())
+    }
+
+    pub async fn get_size(&mut self, filename: String) -> FatResult<usize> {
+        let file = PathBuf::try_from(filename.as_str())?;
+        let access = self.fs.lock().await;
+        let data = access.metadata(&*file)?;
+        Ok(data.len())
+    }
+    pub(crate) async fn get_file(
+        &mut self,
+        filename: String,
+        chunk: u32,
+    ) -> FatResult<([u8; 512], usize)> {
+        use littlefs2::io::Error as lfs2Error;
+
+        let file = PathBuf::try_from(filename.as_str())?;
+        let access = self.fs.lock().await;
+        let mut buf = [0_u8; 512];
+        let mut read = 0;
+        let offset = chunk * buf.len() as u32;
+        access.open_file_with_options_and_then(
+            |options| options.read(true),
+            &*file,
+            |file| {
+                let length = file.len()? as u32;
+                if length == 0 {
+                    Err(lfs2Error::IO)
+                } else if length > offset {
+                    file.seek(SeekFrom::Start(offset))?;
+                    read = file.read(&mut buf)?;
+                    Ok(())
+                } else {
+                    //exactly at end, do nothing
+                    Ok(())
+                }
+            },
+        )?;
+        Ok((buf, read))
+    }
+
+    pub(crate) async fn write_ota(&mut self, offset: u32, buf: &[u8]) -> Result<(), FatError> {
+        let _ = check_erase(self.ota_target, offset, offset + 4096);
+        self.ota_target.erase(offset, offset + 4096)?;
+
+        let mut temp = vec![0; buf.len()];
+        let read_back = temp.as_mut_slice();
+        //change to nor flash, align writes!
+        self.ota_target.write(offset, buf)?;
+        self.ota_target.read(offset, read_back)?;
+        if buf != read_back {
+            info!("Expected {:?} but got {:?}", buf, read_back);
+            bail!(
+                "Flash error, read back does not match write buffer at offset {:x}",
+                offset
+            )
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn finalize_ota(&mut self) -> Result<(), FatError> {
+        let current = self.ota.current_slot()?;
+        if self.ota.current_ota_state()? != OtaImageState::Valid {
+            info!(
+                "Validating current slot {:?} as it was able to ota",
+                current
+            );
+            self.ota.set_current_ota_state(Valid)?;
+        }
+
+        self.ota.set_current_slot(current.next())?;
+        info!("switched slot");
+        self.ota.set_current_ota_state(OtaImageState::New)?;
+        info!("switched state for new partition");
+        let state_new = self.ota.current_ota_state()?;
+        info!("state on new partition now {:?}", state_new);
+        //determine nextslot crc
+
+        self.set_restart_to_conf(true);
+        Ok(())
+    }
+
+    pub(crate) fn mode_override_pressed(&mut self) -> bool {
+        self.boot_button.is_low()
+    }
+
+    pub(crate) async fn sntp(
+        &mut self,
+        _max_wait_ms: u32,
+        stack: Stack<'_>,
+    ) -> FatResult<DateTime<Utc>> {
+        println!("start sntp");
+        let mut rx_meta = [PacketMetadata::EMPTY; 16];
+        let mut rx_buffer = [0; 4096];
+        let mut tx_meta = [PacketMetadata::EMPTY; 16];
+        let mut tx_buffer = [0; 4096];
+
+        let mut socket = UdpSocket::new(
+            stack,
+            &mut rx_meta,
+            &mut rx_buffer,
+            &mut tx_meta,
+            &mut tx_buffer,
+        );
+        socket.bind(123).unwrap();
+
+        let context = NtpContext::new(Timestamp::default());
+
+        let ntp_addrs = stack
+            .dns_query(NTP_SERVER, DnsQueryType::A)
+            .await
+            .expect("Failed to resolve DNS");
+        if ntp_addrs.is_empty() {
+            bail!("Failed to resolve DNS");
+        }
+        info!("NTP server: {:?}", ntp_addrs);
+
+        let mut counter = 0;
+        loop {
+            let addr: IpAddr = ntp_addrs[0].into();
+            let timeout = get_time(SocketAddr::from((addr, 123)), &socket, context)
+                .with_timeout(Duration::from_millis((_max_wait_ms / 10) as u64))
+                .await;
+
+            match timeout {
+                Ok(result) => {
+                    let time = result?;
+                    info!("Time: {:?}", time);
+                    return DateTime::from_timestamp(time.seconds as i64, 0)
+                        .context("Could not convert Sntp result");
+                }
+                Err(err) => {
+                    warn!("sntp timeout, retry: {:?}", err);
+                    counter += 1;
+                    if counter > 10 {
+                        bail!("Failed to get time from NTP server");
+                    }
+                    Timer::after(Duration::from_millis(100)).await;
+                }
+            }
+        }
+    }
+
+    pub(crate) async fn wifi_scan(&mut self) -> FatResult<Vec<AccessPointInfo>> {
+        info!("start wifi scan");
+        let mut lock = self.controller.try_lock()?;
+        info!("start wifi scan lock");
+        let scan_config = ScanConfig {
+            ssid: None,
+            bssid: None,
+            channel: None,
+            show_hidden: false,
+            scan_type: ScanTypeConfig::Active {
+                min: Default::default(),
+                max: Default::default(),
+            },
+        };
+        let rv = lock.scan_with_config_async(scan_config).await?;
+        info!("end wifi scan lock");
+        Ok(rv)
+    }
+
+    pub(crate) fn last_pump_time(&self, plant: usize) -> Option<DateTime<Utc>> {
+        let ts = unsafe { LAST_WATERING_TIMESTAMP }[plant];
+        DateTime::from_timestamp_millis(ts)
+    }
+    pub(crate) fn store_last_pump_time(&mut self, plant: usize, time: DateTime<Utc>) {
+        unsafe {
+            LAST_WATERING_TIMESTAMP[plant] = time.timestamp_millis();
+        }
+    }
+    pub(crate) fn set_low_voltage_in_cycle(&mut self) {
+        unsafe {
+            LOW_VOLTAGE_DETECTED = 1;
+        }
+    }
+    pub(crate) fn clear_low_voltage_in_cycle(&mut self) {
+        unsafe {
+            LOW_VOLTAGE_DETECTED = 0;
+        }
+    }
+
+    pub(crate) fn low_voltage_in_cycle(&mut self) -> bool {
+        unsafe { LOW_VOLTAGE_DETECTED == 1 }
+    }
+    pub(crate) fn store_consecutive_pump_count(&mut self, plant: usize, count: u32) {
+        unsafe {
+            CONSECUTIVE_WATERING_PLANT[plant] = count;
+        }
+    }
+    pub(crate) fn consecutive_pump_count(&mut self, plant: usize) -> u32 {
+        unsafe { CONSECUTIVE_WATERING_PLANT[plant] }
+    }
+    pub(crate) fn get_restart_to_conf(&mut self) -> bool {
+        unsafe { RESTART_TO_CONF == 1 }
+    }
+    pub(crate) fn set_restart_to_conf(&mut self, to_conf: bool) {
+        unsafe {
+            if to_conf {
+                RESTART_TO_CONF = 1;
+            } else {
+                RESTART_TO_CONF = 0;
+            }
+        }
+    }
+
+    pub(crate) async fn wifi_ap(&mut self) -> FatResult<Stack<'static>> {
+        let ssid = match self.load_config().await {
+            Ok(config) => config.network.ap_ssid.as_str().to_string(),
+            Err(_) => "PlantCtrl Emergency Mode".to_string(),
+        };
+
+        let spawner = Spawner::for_current_executor().await;
+
+        let device = self.interface_ap.take().unwrap();
+        let gw_ip_addr_str = "192.168.71.1";
+        let gw_ip_addr = Ipv4Addr::from_str(gw_ip_addr_str).expect("failed to parse gateway ip");
+
+        let config = embassy_net::Config::ipv4_static(StaticConfigV4 {
+            address: Ipv4Cidr::new(gw_ip_addr, 24),
+            gateway: Some(gw_ip_addr),
+            dns_servers: Default::default(),
+        });
+
+        let seed = (self.rng.random() as u64) << 32 | self.rng.random() as u64;
+
+        println!("init secondary stack");
+        // Init network stack
+        let (stack, runner) = embassy_net::new(
+            device,
+            config,
+            mk_static!(StackResources<4>, StackResources::<4>::new()),
+            seed,
+        );
+        let stack = mk_static!(Stack, stack);
+
+        let client_config = Configuration::AccessPoint(AccessPointConfiguration {
+            ssid: ssid.clone(),
+            ..Default::default()
+        });
+
+        self.controller
+            .lock()
+            .await
+            .set_configuration(&client_config)?;
+
+        println!("start new");
+        self.controller.lock().await.start()?;
+        println!("start net task");
+        spawner.spawn(net_task(runner)).ok();
+        println!("run dhcp");
+        spawner.spawn(run_dhcp(stack.clone(), gw_ip_addr_str)).ok();
+
+        loop {
+            if stack.is_link_up() {
+                break;
+            }
+            Timer::after(Duration::from_millis(500)).await;
+        }
+        while !stack.is_config_up() {
+            Timer::after(Duration::from_millis(100)).await
+        }
+        println!("Connect to the AP `${ssid}` and point your browser to http://{gw_ip_addr_str}/");
+        stack
+            .config_v4()
+            .inspect(|c| println!("ipv4 config: {c:?}"));
+
+        Ok(stack.clone())
+    }
+
+    pub(crate) async fn wifi(
+        &mut self,
+        network_config: &NetworkConfig,
+    ) -> FatResult<Stack<'static>> {
+        esp_wifi::wifi_set_log_verbose();
+        let ssid = network_config.ssid.clone();
+        match &ssid {
+            Some(ssid) => {
+                if ssid.is_empty() {
+                    bail!("Wifi ssid was empty")
+                }
+            }
+            None => {
+                bail!("Wifi ssid was empty")
+            }
+        }
+        let ssid = ssid.unwrap().to_string();
+        info!("attempting to connect wifi {ssid}");
+        let password = match network_config.password {
+            Some(ref password) => password.to_string(),
+            None => "".to_string(),
+        };
+        let max_wait = network_config.max_wait;
+
+        let spawner = Spawner::for_current_executor().await;
+
+        let device = self.interface_sta.take().unwrap();
+        let config = embassy_net::Config::dhcpv4(DhcpConfig::default());
+
+        let seed = (self.rng.random() as u64) << 32 | self.rng.random() as u64;
+
+        // Init network stack
+        let (stack, runner) = embassy_net::new(
+            device,
+            config,
+            mk_static!(StackResources<8>, StackResources::<8>::new()),
+            seed,
+        );
+        let stack = mk_static!(Stack, stack);
+
+        let client_config = Configuration::Client(ClientConfiguration {
+            ssid,
+            bssid: None,
+            auth_method: AuthMethod::WPA2Personal, //FIXME read from config, fill via scan
+            password,
+            channel: None,
+        });
+        self.controller
+            .lock()
+            .await
+            .set_configuration(&client_config)?;
+        spawner.spawn(net_task(runner)).ok();
+        self.controller.lock().await.start_async().await?;
+
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + max_wait as u64 * 1000;
+        loop {
+            let state = esp_wifi::wifi::sta_state();
+            match state {
+                WifiState::StaStarted => {
+                    self.controller.lock().await.connect()?;
+                    break;
+                }
+                _ => {}
+            }
+            if {
+                let guard = TIME_ACCESS.get().await.lock().await;
+                guard.current_time_us()
+            } > timeout
+            {
+                bail!("Timeout waiting for wifi sta ready")
+            }
+            Timer::after(Duration::from_millis(500)).await;
+        }
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + max_wait as u64 * 1000;
+        loop {
+            let state = esp_wifi::wifi::sta_state();
+            match state {
+                WifiState::StaConnected => {
+                    break;
+                }
+                _ => {}
+            }
+            if {
+                let guard = TIME_ACCESS.get().await.lock().await;
+                guard.current_time_us()
+            } > timeout
+            {
+                bail!("Timeout waiting for wifi sta connected")
+            }
+            Timer::after(Duration::from_millis(500)).await;
+        }
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + max_wait as u64 * 1000;
+        while !stack.is_link_up() {
+            if {
+                let guard = TIME_ACCESS.get().await.lock().await;
+                guard.current_time_us()
+            } > timeout
+            {
+                bail!("Timeout waiting for wifi link up")
+            }
+            Timer::after(Duration::from_millis(500)).await;
+        }
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + max_wait as u64 * 1000;
+        while !stack.is_config_up() {
+            if {
+                let guard = TIME_ACCESS.get().await.lock().await;
+                guard.current_time_us()
+            } > timeout
+            {
+                bail!("Timeout waiting for wifi config up")
+            }
+            Timer::after(Duration::from_millis(100)).await
+        }
+
+        info!("Connected WIFI, dhcp: {:?}", stack.config_v4());
+        Ok(stack.clone())
+    }
+
+    pub fn deep_sleep(
+        &mut self,
+        duration_in_ms: u64,
+        mut rtc: MutexGuard<CriticalSectionRawMutex, Rtc>,
+    ) -> ! {
+        // Configure and enter deep sleep using esp-hal. Also keep prior behavior where
+        // duration_in_ms == 0 triggers an immediate reset.
+
+        // Mark the current OTA image as valid if we reached here while in pending verify.
+        if let Ok(cur) = self.ota.current_ota_state() {
+            if cur == OtaImageState::PendingVerify {
+                self.ota
+                    .set_current_ota_state(OtaImageState::Valid)
+                    .expect("Could not set image to valid");
+            }
+        }
+
+        if duration_in_ms == 0 {
+            software_reset();
+        } else {
+            let timer = TimerWakeupSource::new(core::time::Duration::from_millis(duration_in_ms));
+            let mut wake_pins: [(&mut dyn RtcPinWithResistors, WakeupLevel); 1] =
+                [(&mut self.wake_gpio1, WakeupLevel::Low)];
+            let ext1 = esp_hal::rtc_cntl::sleep::Ext1WakeupSource::new(&mut wake_pins);
+            rtc.sleep_deep(&[&timer, &ext1]);
+        }
+    }
+
+    pub(crate) async fn load_config(&mut self) -> FatResult<PlantControllerConfig> {
+        let cfg = PathBuf::try_from(CONFIG_FILE)?;
+        let config_exist = self.fs.lock().await.exists(&cfg);
+        if !config_exist {
+            bail!("No config file stored")
+        }
+        let data = self.fs.lock().await.read::<4096>(&cfg)?;
+        let config: PlantControllerConfig = serde_json::from_slice(&data)?;
+        return Ok(config);
+    }
+    pub(crate) async fn save_config(&mut self, config: Vec<u8>) -> FatResult<()> {
+        let filesystem = self.fs.lock().await;
+        let cfg = PathBuf::try_from(CONFIG_FILE)?;
+        filesystem.write(&cfg, &*config)?;
+        Ok(())
+    }
+    pub(crate) async fn list_files(&self) -> FatResult<FileList> {
+        let path = PathBuf::new();
+
+        let fs = self.fs.lock().await;
+        let free_size = fs.available_space()?;
+        let total_size = fs.total_space();
+
+        let mut result = FileList {
+            total: total_size,
+            used: total_size - free_size,
+            files: Vec::new(),
+        };
+
+        fs.read_dir_and_then(&path, |dir| {
+            for entry in dir {
+                let e = entry?;
+                if e.file_type() == FileType::File {
+                    result.files.push(FileInfo {
+                        filename: e.path().to_string(),
+                        size: e.metadata().len(),
+                    });
+                }
+            }
+            Ok(())
+        })?;
+        Ok(result)
+    }
+
+    pub(crate) async fn init_rtc_deepsleep_memory(
+        &self,
+        init_rtc_store: bool,
+        to_config_mode: bool,
+    ) {
+        if init_rtc_store {
+            unsafe {
+                LAST_WATERING_TIMESTAMP = [0; PLANT_COUNT];
+                CONSECUTIVE_WATERING_PLANT = [0; PLANT_COUNT];
+                LOW_VOLTAGE_DETECTED = 0;
+                if to_config_mode {
+                    RESTART_TO_CONF = 1
+                } else {
+                    RESTART_TO_CONF = 0;
+                }
+            };
+        } else {
+            unsafe {
+                if to_config_mode {
+                    RESTART_TO_CONF = 1;
+                }
+                LOG_ACCESS
+                    .lock()
+                    .await
+                    .log(
+                        LogMessage::RestartToConfig,
+                        RESTART_TO_CONF as u32,
+                        0,
+                        "",
+                        "",
+                    )
+                    .await;
+                LOG_ACCESS
+                    .lock()
+                    .await
+                    .log(
+                        LogMessage::LowVoltage,
+                        LOW_VOLTAGE_DETECTED as u32,
+                        0,
+                        "",
+                        "",
+                    )
+                    .await;
+                for i in 0..PLANT_COUNT {
+                    log::info!(
+                        "LAST_WATERING_TIMESTAMP[{}] = UTC {}",
+                        i,
+                        LAST_WATERING_TIMESTAMP[i]
+                    );
+                }
+                for i in 0..PLANT_COUNT {
+                    log::info!(
+                        "CONSECUTIVE_WATERING_PLANT[{}] = {}",
+                        i,
+                        CONSECUTIVE_WATERING_PLANT[i]
+                    );
+                }
+            }
+        }
+    }
+
+    pub(crate) async fn mqtt(
+        &mut self,
+        network_config: &'static NetworkConfig,
+        stack: Stack<'static>,
+    ) -> FatResult<()> {
+        let base_topic = network_config
+            .base_topic
+            .as_ref()
+            .context("missing base topic")?;
+        if base_topic.is_empty() {
+            bail!("Mqtt base_topic was empty")
+        }
+        MQTT_BASE_TOPIC
+            .init(base_topic.to_string())
+            .map_err(|_| FatError::String {
+                error: "Error setting basetopic".to_string(),
+            })?;
+
+        let mqtt_url = network_config
+            .mqtt_url
+            .as_ref()
+            .context("missing mqtt url")?;
+        if mqtt_url.is_empty() {
+            bail!("Mqtt url was empty")
+        }
+
+        let last_will_topic = format!("{}/state", base_topic);
+        let round_trip_topic = format!("{}/internal/roundtrip", base_topic);
+        let stay_alive_topic = format!("{}/stay_alive", base_topic);
+
+        let mut builder: McutieBuilder<'_, String, PublishDisplay<String, &str>, 0> =
+            McutieBuilder::new(stack, "plant ctrl", mqtt_url);
+        if network_config.mqtt_user.is_some() && network_config.mqtt_password.is_some() {
+            builder = builder.with_authentication(
+                network_config.mqtt_user.as_ref().unwrap().as_str(),
+                network_config.mqtt_password.as_ref().unwrap().as_str(),
+            );
+            info!("With authentification");
+        }
+
+        let lwt = Topic::General(last_will_topic);
+        let lwt = mk_static!(Topic<String>, lwt);
+        let lwt = lwt.with_display("lost").retain(true).qos(QoS::AtLeastOnce);
+        builder = builder.with_last_will(lwt);
+        //TODO make configurable
+        builder = builder.with_device_id("plantctrl");
+
+        let builder: McutieBuilder<'_, String, PublishDisplay<String, &str>, 2> = builder
+            .with_subscriptions([
+                Topic::General(round_trip_topic.clone()),
+                Topic::General(stay_alive_topic.clone()),
+            ]);
+
+        let keep_alive = Duration::from_secs(60 * 60 * 2).as_secs() as u16;
+        let (receiver, task) = builder.build(keep_alive);
+
+        let spawner = Spawner::for_current_executor().await;
+        spawner.spawn(mqtt_incoming_task(
+            receiver,
+            round_trip_topic.clone(),
+            stay_alive_topic.clone(),
+        ))?;
+        spawner.spawn(mqtt_runner(task))?;
+
+        LOG_ACCESS
+            .lock()
+            .await
+            .log(LogMessage::StayAlive, 0, 0, "", &stay_alive_topic)
+            .await;
+
+        LOG_ACCESS
+            .lock()
+            .await
+            .log(LogMessage::MqttInfo, 0, 0, "", mqtt_url)
+            .await;
+
+        let mqtt_timeout = 15000;
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + mqtt_timeout as u64 * 1000;
+        while !MQTT_CONNECTED_EVENT_RECEIVED.load(Ordering::Relaxed) {
+            let cur = TIME_ACCESS.get().await.lock().await.current_time_us();
+            if cur > timeout {
+                bail!("Timeout waiting MQTT connect event")
+            }
+            Timer::after(Duration::from_millis(100)).await;
+        }
+
+        Topic::General(round_trip_topic.clone())
+            .with_display("online_text")
+            .publish()
+            .await
+            .unwrap();
+
+        let timeout = {
+            let guard = TIME_ACCESS.get().await.lock().await;
+            guard.current_time_us()
+        } + mqtt_timeout as u64 * 1000;
+        while !MQTT_ROUND_TRIP_RECEIVED.load(Ordering::Relaxed) {
+            let cur = TIME_ACCESS.get().await.lock().await.current_time_us();
+            if cur > timeout {
+                //ensure we do not further try to publish
+                MQTT_CONNECTED_EVENT_RECEIVED.store(false, Ordering::Relaxed);
+                bail!("Timeout waiting MQTT roundtrip")
+            }
+            Timer::after(Duration::from_millis(100)).await;
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn mqtt_inner(&mut self, subtopic: &str, message: &str) -> FatResult<()> {
+        if !subtopic.starts_with("/") {
+            bail!("Subtopic without / at start {}", subtopic);
+        }
+        if subtopic.len() > 192 {
+            bail!("Subtopic exceeds 192 chars {}", subtopic);
+        }
+        let base_topic = MQTT_BASE_TOPIC
+            .try_get()
+            .context("missing base topic in static!")?;
+
+        let full_topic = format!("{base_topic}{subtopic}");
+
+        loop {
+            let result = Topic::General(full_topic.as_str())
+                .with_display(message)
+                .retain(true)
+                .publish()
+                .await;
+            match result {
+                Ok(()) => return Ok(()),
+                Err(err) => {
+                    let retry = match err {
+                        Error::IOError => false,
+                        Error::TimedOut => true,
+                        Error::TooLarge => false,
+                        Error::PacketError => false,
+                        Error::Invalid => false,
+                    };
+                    if !retry {
+                        bail!(
+                            "Error during mqtt send on topic {} with message {:#?} error is {:?}",
+                            &full_topic,
+                            message,
+                            err
+                        );
+                    }
+                    info!(
+                        "Retransmit for {} with message {:#?} error is {:?} retrying {}",
+                        &full_topic, message, err, retry
+                    );
+                    Timer::after(Duration::from_millis(100)).await;
+                }
+            }
+        }
+    }
+    pub(crate) async fn mqtt_publish(&mut self, subtopic: &str, message: &str) {
+        let online = MQTT_CONNECTED_EVENT_RECEIVED.load(Ordering::Relaxed);
+        if !online {
+            return;
+        }
+        let roundtrip_ok = MQTT_ROUND_TRIP_RECEIVED.load(Ordering::Relaxed);
+        if !roundtrip_ok {
+            info!("MQTT roundtrip not received yet, dropping message");
+            return;
+        }
+        match self.mqtt_inner(subtopic, message).await {
+            Ok(()) => {}
+            Err(err) => {
+                info!(
+                    "Error during mqtt send on topic {} with message {:#?} error is {:?}",
+                    subtopic, message, err
+                );
+            }
+        };
+    }
+}
+
+#[embassy_executor::task]
+async fn mqtt_runner(
+    task: McutieTask<'static, String, PublishDisplay<'static, String, &'static str>, 2>,
+) {
+    task.run().await;
+}
+
+#[embassy_executor::task]
+async fn mqtt_incoming_task(
+    receiver: McutieReceiver,
+    round_trip_topic: String,
+    stay_alive_topic: String,
+) {
+    loop {
+        let message = receiver.receive().await;
+        match message {
+            MqttMessage::Connected => {
+                info!("Mqtt connected");
+                MQTT_CONNECTED_EVENT_RECEIVED.store(true, Ordering::Relaxed);
+            }
+            MqttMessage::Publish(topic, payload) => match topic {
+                Topic::DeviceType(_type_topic) => {}
+                Topic::Device(_device_topic) => {}
+                Topic::General(topic) => {
+                    let subtopic = topic.as_str();
+
+                    if subtopic.eq(round_trip_topic.as_str()) {
+                        MQTT_ROUND_TRIP_RECEIVED.store(true, Ordering::Relaxed);
+                    } else if subtopic.eq(stay_alive_topic.as_str()) {
+                        let value = payload.eq_ignore_ascii_case("true".as_ref())
+                            || payload.eq_ignore_ascii_case("1".as_ref());
+                        let a = match value {
+                            true => 1,
+                            false => 0,
+                        };
+                        LOG_ACCESS
+                            .lock()
+                            .await
+                            .log(LogMessage::MqttStayAliveRec, a, 0, "", "")
+                            .await;
+                        MQTT_STAY_ALIVE.store(value, Ordering::Relaxed);
+                    } else {
+                        LOG_ACCESS
+                            .lock()
+                            .await
+                            .log(LogMessage::UnknownTopic, 0, 0, "", &*topic)
+                            .await;
+                    }
+                }
+            },
+            MqttMessage::Disconnected => {
+                MQTT_CONNECTED_EVENT_RECEIVED.store(false, Ordering::Relaxed);
+                info!("Mqtt disconnected");
+            }
+            MqttMessage::HomeAssistantOnline => {
+                info!("Home assistant is online");
+            }
+        }
+    }
+}
+
+#[embassy_executor::task(pool_size = 2)]
+async fn net_task(mut runner: Runner<'static, WifiDevice<'static>>) {
+    runner.run().await;
+}
+
+#[embassy_executor::task]
+async fn run_dhcp(stack: Stack<'static>, gw_ip_addr: &'static str) {
+    use core::net::{Ipv4Addr, SocketAddrV4};
+
+    use edge_dhcp::{
+        io::{self, DEFAULT_SERVER_PORT},
+        server::{Server, ServerOptions},
+    };
+    use edge_nal::UdpBind;
+    use edge_nal_embassy::{Udp, UdpBuffers};
+
+    let ip = Ipv4Addr::from_str(gw_ip_addr).expect("dhcp task failed to parse gw ip");
+
+    let mut buf = [0u8; 1500];
+
+    let mut gw_buf = [Ipv4Addr::UNSPECIFIED];
+
+    let buffers = UdpBuffers::<3, 1024, 1024, 10>::new();
+    let unbound_socket = Udp::new(stack, &buffers);
+    let mut bound_socket = unbound_socket
+        .bind(SocketAddr::V4(SocketAddrV4::new(
+            Ipv4Addr::UNSPECIFIED,
+            DEFAULT_SERVER_PORT,
+        )))
+        .await
+        .unwrap();
+
+    loop {
+        _ = io::server::run(
+            &mut Server::<_, 64>::new_with_et(ip),
+            &ServerOptions::new(ip, Some(&mut gw_buf)),
+            &mut bound_socket,
+            &mut buf,
+        )
+        .await
+        .inspect_err(|e| log::warn!("DHCP server error: {e:?}"));
+        Timer::after(Duration::from_millis(500)).await;
+    }
+}