Readouts are now posted to MQTT

.cargo/config.toml

@@ -1,8 +1,8 @@
 [target.'cfg(all(target_arch = "arm", target_os = "none"))']
-runner = "probe-run --chip RP2040"
+runner = "probe-rs run --chip RP2040"
 
 [build]
-target = "thumbv6m-none-eabi"        # Cortex-M0 and Cortex-M0+
+target = "thumbv6m-none-eabi" # Cortex-M0 and Cortex-M0+
 
 [env]
 DEFMT_LOG = "debug"

Cargo.toml

@@ -10,7 +10,7 @@ cortex-m = { version = "0.7", features = ["inline-asm"] }
 cortex-m-rt = "0.7"
 defmt = "0.3"
 defmt-rtt = "0.4"
-embassy-executor = { git = "https://github.com/embassy-rs/embassy", features = [
+embassy-executor = { version = "0.3", features = [
 	"arch-cortex-m",
 	"executor-thread",
 	"executor-interrupt",
@@ -18,7 +18,7 @@ embassy-executor = { git = "https://github.com/embassy-rs/embassy", features = [
 	"nightly",
 	"integrated-timers",
 ] }
-embassy-rp = { git = "https://github.com/embassy-rs/embassy", features = [
+embassy-rp = { version = "0.1", features = [
 	"defmt",
 	"unstable-traits",
 	"nightly",
@@ -26,42 +26,60 @@ embassy-rp = { git = "https://github.com/embassy-rs/embassy", features = [
 	"time-driver",
 	"critical-section-impl",
 ] }
-embassy-time = { git = "https://github.com/embassy-rs/embassy", features = [
+embassy-time = { version = "0.1", features = [
 	"defmt",
 	"unstable-traits",
 	"defmt-timestamp-uptime",
 	"nightly",
 ] }
-embassy-usb = { git = "https://github.com/embassy-rs/embassy", features = [
+embassy-usb = { version = "0.1", features = ["defmt"] }
-	"defmt",
+embassy-net = { version = "0.1", features = [
-] }
-embassy-net = { git = "https://github.com/embassy-rs/embassy", features = [
 	"tcp",
 	"dhcpv4",
 	"nightly",
 	"medium-ethernet",
 	"defmt",
 ] }
-embassy-sync = { git = "https://github.com/embassy-rs/embassy", features = [
+embassy-sync = { version = "0.2", features = ["defmt"] }
+embassy-futures = { version = "0.1", features = ["defmt"] }
+cyw43 = { git = "https://github.com/embassy-rs/embassy", features = [
 	"defmt",
+	"firmware-logs",
 ] }
-embassy-futures = { git = "https://github.com/embassy-rs/embassy", features = [
-	"defmt",
-] }
-cyw43 = { git = "https://github.com/embassy-rs/embassy", features = ["defmt"] }
 cyw43-pio = { git = "https://github.com/embassy-rs/embassy", features = [
 	"defmt",
 ] }
 panic-probe = { version = "0.3", features = ["print-defmt"] }
-log = "0.4.17"
+log = "0.4"
-static_cell = { version = "1.1.0", features = ["nightly"] }
+static_cell = { version = "1.1", features = ["nightly"] }
 heapless = { version = "0.7.16", features = ["defmt"] }
-embedded-io = { version = "0.4.0", features = ["async", "defmt"] }
+embedded-io-async = { version = "0.5", features = ["defmt-03"] }
-crc16 = "0.4.0"
+crc16 = "0.4"
-dsmr5 = "0.2.2"
+dsmr5 = "0.2"
+rust-mqtt = { version = "0.1.5", features = [
+	"defmt",
+	"no_std",
+	"tls",
+], default-features = false }
+rand = { version = "0.8.5", features = [
+	"nightly",
+	"small_rng",
+], default-features = false }
+serde-json-core = "0.5.1"
+
+[patch.crates-io]
+embassy-executor = { git = "https://github.com/embassy-rs/embassy" }
+embassy-rp = { git = "https://github.com/embassy-rs/embassy" }
+embassy-time = { git = "https://github.com/embassy-rs/embassy" }
+embassy-usb = { git = "https://github.com/embassy-rs/embassy" }
+embassy-net = { git = "https://github.com/embassy-rs/embassy" }
+embassy-sync = { git = "https://github.com/embassy-rs/embassy" }
+embassy-futures = { git = "https://github.com/embassy-rs/embassy" }
+
+rust-mqtt = { path = "../rust-mqtt" }
 
 [build-dependencies]
-dotenvy = "0.15.7"
+dotenvy = "0.15"
 
 [profile.release]
 debug = true

src/main.rs

@@ -11,11 +11,12 @@ use embassy_net::{tcp::TcpSocket, Ipv4Address, Ipv4Cidr, Stack, StackResources};
 use embassy_rp::{
     bind_interrupts, gpio,
     peripherals::{DMA_CH0, PIN_23, PIN_25, PIO0, UART0},
-    pio::Pio,
+    pio::{self, Pio},
     uart::{self, BufferedUartRx, Parity},
 };
+use embassy_time::{Duration, Ticker};
 
-use embedded_io::asynch::{Read, Write};
+use embedded_io_async::Read;
 
 use embassy_sync::{
     blocking_mutex::raw::NoopRawMutex,
@@ -23,12 +24,15 @@ use embassy_sync::{
 };
 use gpio::{Level, Output};
 use heapless::Vec;
+use rand::{rngs::SmallRng, RngCore, SeedableRng};
+use rust_mqtt::client::{client::MqttClient, client_config::ClientConfig};
 use static_cell::make_static;
 
 use {defmt_rtt as _, panic_probe as _};
 
 bind_interrupts!(struct Irqs {
     UART0_IRQ => uart::BufferedInterruptHandler<UART0>;
+    PIO0_IRQ_0 => pio::InterruptHandler<PIO0>;
 });
 
 #[derive(Format)]
@@ -142,42 +146,11 @@ async fn main(spawner: Spawner) {
 
     spawner.spawn(uart_rx_task(rx, channel.sender())).unwrap();
 
-    // === USB SERIAL ===
-    // let driver = Driver::new(p.USB, Irqs);
-    //
-    // let mut config = Config::new(0xc0de, 0xcafe);
-    // config.manufacturer = Some("huizinga.dev");
-    // config.product = Some("Raspberry Pi Pico");
-    // config.serial_number = Some("123456789");
-    // config.max_power = 0;
-    // config.max_packet_size_0 = 64;
-    //
-    // // Needed for windows compatiblilty
-    // config.device_class = 0xEF;
-    // config.device_sub_class = 0x02;
-    // config.device_protocol = 0x01;
-    // config.composite_with_iads = true;
-    //
-    // let mut builder = Builder::new(
-    //     driver,
-    //     config,
-    //     singleton!([0; 256]),
-    //     singleton!([0; 256]),
-    //     singleton!([0; 256]),
-    //     singleton!([0; 64]),
-    // );
-    //
-    // let class = CdcAcmClass::new(&mut builder, singleton!(State::new()), 64);
-    // let usb = builder.build();
-    //
-    // spawner.spawn(usb_task(usb)).unwrap();
-    // spawner.spawn(echo_task(class, channel.sender())).unwrap();
-
     // === WIFI ===
     // To make flashing faster for development, you may want to flash the firmwares independently
     // at hardcoded addresses, instead of baking them into the program with `include_bytes!`:
-    //     probe-rs-cli download 43439A0.bin --format bin --chip RP2040 --base-address 0x10100000
+    //     probe-rs download 43439A0.bin --format bin --chip RP2040 --base-address 0x10100000
-    //     probe-rs-cli download 43439A0_clm.bin --format bin --chip RP2040 --base-address 0x10140000
+    //     probe-rs download 43439A0_clm.bin --format bin --chip RP2040 --base-address 0x10140000
     let fw = unsafe { core::slice::from_raw_parts(0x10100000 as *const u8, 224190) };
     let clm = unsafe { core::slice::from_raw_parts(0x10140000 as *const u8, 4752) };
     // let fw = include_bytes!("../firmware/43439A0.bin");
@@ -186,7 +159,7 @@ async fn main(spawner: Spawner) {
     let pwr = Output::new(p.PIN_23, Level::Low);
     let cs = Output::new(p.PIN_25, Level::High);
 
-    let mut pio = Pio::new(p.PIO0);
+    let mut pio = Pio::new(p.PIO0, Irqs);
     let spi = PioSpi::new(
         &mut pio.common,
         pio.sm0,
@@ -206,16 +179,23 @@ async fn main(spawner: Spawner) {
         .set_power_management(cyw43::PowerManagementMode::PowerSave)
         .await;
 
-    let ip = Ipv4Address::new(10, 0, 0, 77);
+    // Turn LED on while trying to connect
+    control.gpio_set(0, true).await;
 
-    // let config = embassy_net::Config::Dhcp(Default::default());
+    // let config = embassy_net::Config::dhcpv4(Default::default());
-    let config = embassy_net::Config::Static(embassy_net::StaticConfig {
+    let ip = Ipv4Address::new(10, 0, 0, 77);
+    let config = embassy_net::Config::ipv4_static(embassy_net::StaticConfigV4 {
         address: Ipv4Cidr::new(ip, 24),
-        dns_servers: Vec::new(),
         gateway: Some(Ipv4Address::new(10, 0, 0, 1)),
+        dns_servers: Vec::new(),
     });
 
-    let seed = 0x51ac_3101_6468_8cdf;
+    let mut seed = [0; 8];
+    // TODO: Make the seed actually random?
+    let mut rng = SmallRng::seed_from_u64(0x51ac_3101_6468_8cdf);
+    rng.fill_bytes(&mut seed);
+    let seed = u64::from_le_bytes(seed);
+
     let stack = make_static!(Stack::new(
         net_device,
         config,
@@ -241,99 +221,63 @@ async fn main(spawner: Spawner) {
     let mut rx_buffer = [0; 4096];
     let mut tx_buffer = [0; 4096];
 
+    let mut socket = TcpSocket::new(stack, &mut rx_buffer, &mut tx_buffer);
+    // socket.set_timeout(Some(Duration::from_secs(10)));
+    let addr = (Ipv4Address::new(10, 0, 0, 2), 1883);
+
+    while let Err(e) = socket.connect(addr).await {
+        warn!("Connect error: {:?}", e);
+        continue;
+    }
+    info!("TCP Connected!");
+
+    let mut config = ClientConfig::new(rust_mqtt::client::client_config::MqttVersion::MQTTv5, rng);
+
+    config.add_username("mqtt");
+    config.add_password("$x@w6t8ErHJwolSfVdkLqFUvxex5ZgeE");
+    config.add_max_subscribe_qos(rust_mqtt::packet::v5::publish_packet::QualityOfService::QoS1);
+    config.add_client_id("pico");
+    // Leads to InsufficientBufferSize error
+    // config.add_will("pico/test", b"disconnected", false);
+
+    let mut recv_buffer = [0; 4096];
+    let mut write_buffer = [0; 4096];
+
+    let mut client =
+        MqttClient::<_, 5, _>::new(socket, &mut write_buffer, &mut recv_buffer, config);
+
+    client.connect_to_broker().await.unwrap();
+    info!("MQTT Connected!");
+
+    // Turn LED off when connected
+    control.gpio_set(0, false).await;
+
+    let mut keep_alive = Ticker::every(Duration::from_secs(30));
     let receiver = channel.receiver();
 
     loop {
-        let mut socket = TcpSocket::new(stack, &mut rx_buffer, &mut tx_buffer);
+        match select(keep_alive.next(), receiver.receive()).await {
-        // socket.set_timeout(Some(Duration::from_secs(10)));
+            Either::First(_) => client.send_ping().await.unwrap(),
+            Either::Second(readout) => {
+                control.gpio_set(0, true).await;
+                control.gpio_set(0, false).await;
 
-        control.gpio_set(0, true).await;
+                let telegram = readout.to_telegram().unwrap();
-        let port = 1234;
+                let state = dsmr5::Result::<dsmr5::state::State>::from(&telegram).unwrap();
-        info!("Listening on {}:{}...", ip, port);
 
-        if let Err(e) = socket.accept(port).await {
+                let msg: Vec<u8, 4096> = serde_json_core::to_vec(&state).unwrap();
-            warn!("Accept error: {:?}", e);
+                info!("len: {}", msg.len());
-            continue;
-        }
 
-        info!("Received connection from {:?}", socket.remote_endpoint());
+                client
-        control.gpio_set(0, false).await;
+                    .send_message(
-
+                        "pico/test",
-        socket.write_all("Connected!\n".as_bytes()).await.unwrap();
+                        &msg,
-
+                        rust_mqtt::packet::v5::publish_packet::QualityOfService::QoS0,
-        loop {
+                        false,
-            match select(socket.read(&mut [0; 64]), receiver.recv()).await {
+                    )
-                Either::First(Ok(0)) => {
+                    .await
-                    warn!("Read OEF");
+                    .unwrap();
-                    break;
-                }
-                Either::First(Ok(_)) => {
-                    // Received something over the socket, currently does nothing
-                }
-                Either::First(Err(e)) => {
-                    warn!("Read error: {:?}", e);
-                }
-                Either::Second(readout) => {
-                    socket.write_all(&readout.buffer).await.unwrap();
-
-                    // let telegram = readout.to_telegram().unwrap();
-                    //
-                    // debug!("checksum: {}", telegram.checksum);
-                    // debug!("prefix: {}", telegram.prefix);
-                    // debug!("identification: {}", telegram.identification);
-                    //
-                    // let state = dsmr5::Result::<dsmr5::state::State>::from(&telegram).unwrap();
-                    // debug!("datetime: {}", state.datetime.unwrap().year);
-                    // debug!("meterreadings[0]: {}", state.meterreadings[0].to.unwrap());
-                    // debug!("meterreadings[1]: {}", state.meterreadings[1].to.unwrap());
-                    //
-                    // debug!("slave:");
-                    // debug!("\tdevice_type: {}", state.slaves[0].device_type.unwrap());
-                    // debug!(
-                    //     "\tmeter_reading: {}",
-                    //     state.slaves[0].meter_reading.as_ref().unwrap().1
-                    // );
-                }
-                // Either::Second(message) => match socket.write(&message).await {
-                //     Ok(_) => {}
-                //     Err(e) => {
-                //         warn!("Write error: {:?}", e);
-                //         break;
-                //     }
-                // },
             }
         }
     }
 }
-
-// struct Disconnected;
-//
-// impl From<EndpointError> for Disconnected {
-//     fn from(value: EndpointError) -> Self {
-//         match value {
-//             EndpointError::BufferOverflow => panic!("Buffer overflow"),
-//             EndpointError::Disabled => Self {},
-//         }
-//     }
-// }
-//
-// /// Echo back received packets
-// async fn echo<'d>(
-//     class: &mut CdcAcmClass<'d, Driver<'d, peripherals::USB>>,
-//     sender: &Sender<'static, NoopRawMutex, Message, 1>,
-// ) -> Result<(), Disconnected> {
-//     let mut buf = [0; 64];
-//     loop {
-//         let n = class.read_packet(&mut buf).await?;
-//         let data = from_utf8_mut(&mut buf[..n]).unwrap();
-//
-//         data.make_ascii_uppercase();
-//
-//         sender
-//             .send(Message {
-//                 data: buf,
-//                 length: n,
-//             })
-//             .await;
-//     }
-// }