Weather/src/main.rs

mod weather_data;
use chrono::{Local};
use crate::weather_data::{DwdWeather, MucWeather};
use rumqttc::{AsyncClient, EventLoop, MqttOptions, QoS};
use scraper::{Html as HtmlScraper, Selector};
use std::time::Duration;
use tokio::{task, time};
use dotenv::dotenv;
use std::env;

const METEO_MUNICH_API_URL: &str =
    "https://www.meteorologie.lmu.de/~quicklooks/aktuelle_messwerte/messwerte_stadt.html";

const LATITUDE_MUC: f64 = 48.163142;

const LONGITUDE_MUC: f64 = 11.542922;

#[tokio::main(flavor = "current_thread")]
async fn main() {

    dotenv().ok();

    let (client, mut eventloop) = create_conn();

    match meteo_weather().await {
        Ok(weather) => {
            task::spawn(async move {
                publish(client.clone(), &weather).await;
                time::sleep(Duration::from_secs(3)).await;
            });

            let mut count = 0;
            loop {
                let event = eventloop.poll().await;
                match &event {
                    Ok(_) => {
                        if count == 10 {
                            break;
                        }
                        count = count + 1;
                    }
                    Err(e) => {
                        println!("Error = {e:?}");
                    }
                }
            }
        }
        Err(e) => {
            println!("{}", e);
        }
    }
}

pub async fn meteo_weather() -> Result<MucWeather, reqwest::Error> {

    let response = trpl::get(METEO_MUNICH_API_URL).await;
    let response_text = response.text().await;

    let fragment = HtmlScraper::parse_fragment(&response_text);
    let mut temperature = 999.0;
    let mut humidity = 0.0;
    let mut pressure = 0.0;
    let mut precipitation = 0.0;
    
    let profile_selector = Selector::parse("#messwerte-profilwerte").unwrap();
    let pressure_selector = Selector::parse("#messwerte-winddruck").unwrap();
    let precipitation_selector = Selector::parse("#messwerte-niederschlag").unwrap();
    let row_selector = Selector::parse("tr").unwrap();
    let cell_selector = Selector::parse("td").unwrap();

    if let Some(table) = fragment.select(&profile_selector).next() {
        for row in table.select(&row_selector) {
            let mut row_data = Vec::new();
            for cell in row.select(&cell_selector) {
                let text = cell.text().collect::<Vec<_>>().join(" ");
                row_data.push(text.trim().to_string());
            }
            if row_data[0] == "Lufttemperatur" {
                temperature = row_data[1].replace(" °C", "").trim().parse::<f64>().unwrap();
            }
            if row_data[0] == "Relative Feuchte" {
                humidity = row_data[1].replace(" %", "").trim().parse::<f64>().unwrap();
            }
        }
    }

    if let Some(table) = fragment.select(&pressure_selector).next() {
        for row in table.select(&row_selector) {
            let mut row_data = Vec::new();
            for cell in row.select(&cell_selector) {
                let text = cell.text().collect::<Vec<_>>().join(" ");
                row_data.push(text.trim().to_string());
            }
            if row_data[0] == "Luftdruck NN" {
                pressure = row_data[1].replace(" hPa", "").trim().parse::<f64>().unwrap();
            }
        }
    }

    if let Some(table) = fragment.select(&precipitation_selector).next() {
        for row in table.select(&row_selector) {
            let mut row_data = Vec::new();
            for cell in row.select(&cell_selector) {
                let text = cell.text().collect::<Vec<_>>().join(" ");
                row_data.push(text.trim().to_string());
            }
            if row_data.len() > 1 && row_data[1].contains(" mm")  {
                precipitation = row_data[1].replace(" mm", "").trim().parse::<f64>().unwrap();
            }
        }
    }

    if temperature == 999.0 {
        let url = format!(
            "https://api.brightsky.dev/current_weather?lat={}&lon={}",
            LATITUDE_MUC, LONGITUDE_MUC
        );
        println!("Fetching DWD {}", url);
        let result = trpl::get(&url).await.text().await;
        let dwd_weather: DwdWeather = serde_json::from_str(&result).unwrap();
        temperature = dwd_weather.weather.temperature.unwrap();
        pressure = dwd_weather.weather.pressure_msl;
        humidity = dwd_weather.weather.relative_humidity as f64;
        precipitation = dwd_weather.weather.precipitation;
    }

    let precipitation = (precipitation * 100.0).round() / 100.00;
    let cur_date = Local::now();
    let cur_date_formatted = format!("{}", cur_date.format("%Y-%m-%d %H:%M"));
    let weather = MucWeather {
        temperature,
        pressure,
        humidity,
        precipitation,
        date: cur_date_formatted
    };
    println!("Zeile: {:?}", weather);
    Ok(weather)
}

async fn publish(client: AsyncClient, muc: &MucWeather) {

    let json = serde_json::to_string(&muc);
    publish_topic(client.clone(), "munich/weather", json.unwrap()).await;
    publish_topic(client.clone(), "munich/temperature", muc.temperature.to_string()).await;
    publish_topic(client.clone(), "munich/pressure", muc.pressure.to_string()).await;
    publish_topic(client.clone(), "munich/humidity", muc.humidity.to_string()).await;
    publish_topic(client.clone(), "munich/precipitation", muc.precipitation.to_string()).await;
    time::sleep(Duration::from_secs(2)).await;
}

async fn publish_topic(client: AsyncClient, topic: &str, value: String) {
    client
        .publish(
            topic,
            QoS::AtLeastOnce,
            false,
            value,
        )
        .await
        .unwrap();
}

fn create_conn() -> (AsyncClient, EventLoop) {
    let mqtt_client = env::var("MQTT_CLIENT");
    let mqtt_broker = env::var("MQTT_BROKER");
    let mqtt_user = env::var("MQTT_USER");
    let mqtt_password = env::var("MQTT_PASSWORD");

    let mut mqttoptions = MqttOptions::new(mqtt_client.unwrap(), mqtt_broker.unwrap(), 1883);
    mqttoptions
        .set_keep_alive(Duration::from_secs(5))
        .set_manual_acks(false)
        .set_credentials(mqtt_user.unwrap(), mqtt_password.unwrap())
        .set_clean_session(false);
    AsyncClient::new(mqttoptions, 10)
}