C3MA/PlantCtrl
6
0
Fork 0
Code Issues 8 Pull Requests Packages Projects Releases Wiki Activity

Integrate mcutie library for MQTT functionality

Browse Source 
- Added `mcutie` as a dependency in `Cargo.toml` and updated `Cargo.lock`.
- Replaced commented-out MQTT logic with fully implemented functionality in `esp.rs`.
- Enhanced MQTT publish and subscription handling with configurable topics and error handling.
- Updated MQTT connection logic to improve reliability and logging.
This commit is contained in:
Empire Phoenix
2026-04-26 19:56:16 +02:00
parent 2e4eb283b5
commit fc0e18da56
3 changed files with 186 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
Software/MainBoard/rust/src/hal/esp.rs
View File

@@ -39,10 +39,10 @@ use esp_radio::wifi::scan::{ScanConfig, ScanTypeConfig};
use esp_radio::wifi::sta::StationConfig;
use esp_radio::wifi::{AuthenticationMethod, Config, Interface, WifiController};
use log::{error, info, warn};
// pub use mcutie::{
// Error, McutieBuilder, McutieReceiver, McutieTask, MqttMessage, PublishDisplay, Publishable,
// QoS, Topic,
// };
use mcutie::{
Error, McutieBuilder, McutieReceiver, McutieTask, MqttMessage, PublishDisplay, Publishable,
QoS, Topic,
};
use portable_atomic::AtomicBool;
use sntpc::{get_time, NtpContext, NtpTimestampGenerator, NtpUdpSocket};
@@ -678,62 +678,62 @@ impl Esp<'_> {
let round_trip_topic = format!("{base_topic}/internal/roundtrip");
let stay_alive_topic = format!("{base_topic}/stay_alive");
// let mut builder: McutieBuilder<'_, String, PublishDisplay<String, &str>, 0> =
// McutieBuilder::new(stack, "plant ctrl", mqtt_url);
// if let (Some(mqtt_user), Some(mqtt_password)) = (
// network_config.mqtt_user.as_ref(),
// network_config.mqtt_password.as_ref(),
// ) {
// builder = builder.with_authentication(mqtt_user, mqtt_password);
// 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);
//
// spawner.spawn(mqtt_incoming_task(
// receiver,
// round_trip_topic.clone(),
// stay_alive_topic.clone(),
// )?);
// spawner.spawn(mqtt_runner(task)?);
//
// log(LogMessage::StayAlive, 0, 0, "", &stay_alive_topic);
//
// log(LogMessage::MqttInfo, 0, 0, "", mqtt_url);
//
let mut builder: McutieBuilder<'_, String, PublishDisplay<String, &str>, 0> =
McutieBuilder::new(stack, "plant ctrl", mqtt_url);
if let (Some(mqtt_user), Some(mqtt_password)) = (
network_config.mqtt_user.as_ref(),
network_config.mqtt_password.as_ref(),
) {
builder = builder.with_authentication(mqtt_user, mqtt_password);
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);
spawner.spawn(mqtt_incoming_task(
receiver,
round_trip_topic.clone(),
stay_alive_topic.clone(),
)?);
spawner.spawn(mqtt_runner(task)?);
log(LogMessage::StayAlive, 0, 0, "", &stay_alive_topic);
log(LogMessage::MqttInfo, 0, 0, "", mqtt_url);
let mqtt_timeout = 15000;
// let res = async {
// while !MQTT_CONNECTED_EVENT_RECEIVED.load(Ordering::Relaxed) {
// crate::hal::PlantHal::feed_watchdog();
// Timer::after(Duration::from_millis(100)).await;
// }
// Ok::<(), FatError>(())
// }
// .with_timeout(Duration::from_millis(mqtt_timeout as u64))
// .await;
//
// if res.is_err() {
// bail!("Timeout waiting MQTT connect event")
// }
//
// let _ = Topic::General(round_trip_topic.clone())
// .with_display("online_text")
// .publish()
// .await;
let res = async {
while !MQTT_CONNECTED_EVENT_RECEIVED.load(Ordering::Relaxed) {
crate::hal::PlantHal::feed_watchdog();
Timer::after(Duration::from_millis(100)).await;
}
Ok::<(), FatError>(())
}
.with_timeout(Duration::from_millis(mqtt_timeout as u64))
.await;
if res.is_err() {
bail!("Timeout waiting MQTT connect event")
}
let _ = Topic::General(round_trip_topic.clone())
.with_display("online_text")
.publish()
.await;
let res = async {
while !MQTT_ROUND_TRIP_RECEIVED.load(Ordering::Relaxed) {
@@ -767,36 +767,37 @@ impl Esp<'_> {
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;
// }
// }
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,
Error::Rejected => 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) {
@@ -821,54 +822,52 @@ impl Esp<'_> {
}
#[embassy_executor::task]
async fn mqtt_runner(//task: McutieTask<'static, String, PublishDisplay<'static, String, &'static str>, 2>,
async fn mqtt_runner(
task: McutieTask<'static, String, PublishDisplay<'static, String, &'static str>, 2>,
) {
//task.run().await;
task.run().await;
}
#[embassy_executor::task]
async fn mqtt_incoming_task(
//receiver: McutieReceiver,
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(LogMessage::MqttStayAliveRec, a, 0, "", "");
// MQTT_STAY_ALIVE.store(value, Ordering::Relaxed);
// } else {
// log(LogMessage::UnknownTopic, 0, 0, "", &topic);
// }
// }
// },
// MqttMessage::Disconnected => {
// MQTT_CONNECTED_EVENT_RECEIVED.store(false, Ordering::Relaxed);
// info!("Mqtt disconnected");
// }
// MqttMessage::HomeAssistantOnline => {
// info!("Home assistant is online");
// }
// }
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(LogMessage::MqttStayAliveRec, a, 0, "", "");
MQTT_STAY_ALIVE.store(value, Ordering::Relaxed);
} else {
log(LogMessage::UnknownTopic, 0, 0, "", &topic);
}
}
},
MqttMessage::Disconnected => {
MQTT_CONNECTED_EVENT_RECEIVED.store(false, Ordering::Relaxed);
info!("Mqtt disconnected");
}
}
}
}