first commit

src/detectors.rs

@@ -0,0 +1,55 @@
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct DetectorFile {
+    #[serde(rename = "type")]
+    pub file_type: String, // "FeatureCollection"
+    pub features: Vec<Feature>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Feature {
+    #[serde(rename = "type")]
+    pub feature_type: String, // "Feature"
+    pub geometry: Geometry,
+    pub properties: Properties,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Geometry {
+    #[serde(rename = "type")]
+    pub geometry_type: String, // "Point"
+    pub coordinates: [f64; 2],
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Properties {
+    pub station: i32,
+    pub user: i32,
+    pub generation: String,
+    pub status: String,
+
+    // Sometimes string timestamp, sometimes number 0
+    pub signal_last_ctime: serde_json::Value,
+
+    pub region_mask: i32,
+    pub comment: Option<String>,
+    pub controller_board: Option<String>,
+    pub firmware: Option<String>,
+    pub city: Option<String>,
+    pub country: Option<String>,
+    pub website: Option<String>,
+    pub counters: Vec<Counter>,
+    pub user_stations: Vec<i32>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Counter {
+    pub region: i32,
+    pub all: i32,
+    pub valid: i32,
+    pub member: i32,
+    pub used: i32,
+    pub min_distance: i64,
+    pub max_distance: i64,
+}

src/main.rs

@@ -0,0 +1,279 @@
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use http::header::{HeaderName, HeaderValue};
+
+use crate::message::{TungMessage, decode_lzw};
+use clap::Parser;
+use rand::prelude::IndexedRandom;
+use rustls::crypto::CryptoProvider;
+use std::io::{Write, stdout};
+use tokio_tungstenite::connect_async;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tokio_tungstenite::tungstenite::protocol::Message;
+
+pub mod detectors;
+pub mod message;
+
+#[derive(Parser, Debug)]
+#[command(version, about = "This project takes the information given by Blitzortung, and sends notifications to your phone/devices at the moment that there's lightning strikes near you.", long_about = None)]
+struct Args {
+    /// Latitude of the center of your view
+    #[arg(short = 'a', long)]
+    center_lat: f64,
+
+    /// Longitude of the center of your view
+    #[arg(short = 'o', long)]
+    center_lon: f64,
+
+    /// Delta, also known as radius from the center of your view
+    #[arg(short = 'd', long, default_value_t = 1.5)]
+    delta: f64,
+
+    /// The topic notifications will be sent to
+    #[arg(short = 'n', long = "ntfy")]
+    ntfy_topic: String,
+
+    /// Views detectors
+    #[arg(short = 'f', long, default_value_t = false)]
+    detector_fetch: bool,
+}
+
+fn in_bounds(lat: f64, lon: f64, center_lat: f64, center_lon: f64, delta: f64) -> bool {
+    let lat_min = center_lat - delta;
+    let lat_max = center_lat + delta;
+    let lon_min = center_lon - delta;
+    let lon_max = center_lon + delta;
+
+    (lat >= lat_min && lat <= lat_max) && (lon >= lon_min && lon <= lon_max)
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+
+    CryptoProvider::install_default(rustls_rustcrypto::provider())
+        .expect("install rustcrypto provider");
+    let mut detector_features: Option<detectors::DetectorFile> = None;
+
+    if args.detector_fetch {
+        let detector_file_contents = std::fs::read_to_string("./detectors.json").unwrap();
+        detector_features = Some(serde_json::from_str(&detector_file_contents).unwrap());
+    }
+
+    let ws_hosts = ["ws1", "ws2", "ws7", "ws8"];
+
+    loop {
+        let mut rng = rand::rng();
+        let host = ws_hosts.choose(&mut rng).unwrap();
+        let ws_url = format!("wss://{}.blitzortung.org/", host);
+
+        let mut request = ws_url
+            .clone()
+            .into_client_request()
+            .expect("failed to create client request");
+
+        let headers = request.headers_mut();
+        headers.insert(
+            HeaderName::from_static("user-agent"),
+            HeaderValue::from_static(
+                "Mozilla/5.0 (X11; Linux x86_64; rv:135.0) Gecko/20100101 Firefox/135.0",
+            ),
+        );
+        headers.insert(
+            HeaderName::from_static("accept"),
+            HeaderValue::from_static("*/*"),
+        );
+        headers.insert(
+            HeaderName::from_static("accept-language"),
+            HeaderValue::from_static("en-US,en;q=0.5"),
+        );
+        headers.insert(
+            HeaderName::from_static("accept-encoding"),
+            HeaderValue::from_static("gzip, deflate, br, zstd"),
+        );
+        headers.insert(
+            HeaderName::from_static("origin"),
+            HeaderValue::from_static("map.blitzortung.org"),
+        );
+        headers.insert(
+            HeaderName::from_static("connection"),
+            HeaderValue::from_static("keep-alive, Upgrade"),
+        );
+        headers.insert(
+            HeaderName::from_static("pragma"),
+            HeaderValue::from_static("no-cache"),
+        );
+        headers.insert(
+            HeaderName::from_static("cache-control"),
+            HeaderValue::from_static("no-cache"),
+        );
+
+        let ws_result = connect_async(request).await;
+        let (mut ws_stream, response) = match ws_result {
+            Ok((ws_stream, response)) => (ws_stream, response),
+            Err(e) => {
+                println!("Failed to connect, {}. Retrying in 5 seconds...", e);
+                tokio::time::sleep(Duration::from_secs(5)).await;
+                continue;
+            }
+        };
+
+        println!(
+            "Connected to {}! HTTP status: {}",
+            ws_url,
+            response.status()
+        );
+
+        if let Err(e) = ws_stream.send(Message::Text("{\"a\":111}".into())).await {
+            println!("Failed to send initial message: {}. Reconnecting...", e);
+            tokio::time::sleep(Duration::from_secs(5)).await;
+            continue;
+        }
+
+        let mut should_reconnect = false;
+        while let Some(msg) = ws_stream.next().await {
+            match msg {
+                Ok(Message::Text(txt)) => {
+                    let no_lzw = decode_lzw(txt.as_str());
+                    let all: TungMessage = match serde_json::from_str(&no_lzw) {
+                        Ok(val) => val,
+                        Err(e) => {
+                            println!("Failed to parse message: {}. Skipping message.", e);
+                            continue;
+                        }
+                    };
+
+                    let strike_ms = all.time / 1_000_000;
+                    let time_now_epoch = std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .expect("Time went backwards")
+                        .as_millis() as i64;
+
+                    let offset_ms = time_now_epoch - strike_ms;
+
+                    let lon = all.lon + all.lonc as f64;
+                    let lat = all.lat + all.latc as f64;
+                    if in_bounds(lat, lon, args.center_lat, args.center_lon, args.delta) {
+                        println!(
+                            "Delay: {}ms, offset to now: {}ms, status: {}, region: {} | \n Strike at: {}, {}",
+                            all.delay, offset_ms, all.status, all.region, lon, lat
+                        );
+
+                        let mut detector_details: Vec<String> = Vec::new();
+
+                        if args.detector_fetch {
+                            println!("Used detectors:");
+                            for sig in all.sig.into_iter() {
+                                for feature in detector_features.clone().unwrap().features.iter() {
+                                    if sig.sta == feature.properties.station as i64 {
+                                        let detail = format!(
+                                            "City: {}, Status: {}, Station type: {}, Country: {}, Comment: {}",
+                                            feature
+                                                .properties
+                                                .city
+                                                .clone()
+                                                .unwrap_or("".to_string()),
+                                            feature.properties.status,
+                                            feature.properties.generation,
+                                            feature
+                                                .properties
+                                                .country
+                                                .clone()
+                                                .unwrap_or("".to_string()),
+                                            feature
+                                                .properties
+                                                .comment
+                                                .clone()
+                                                .unwrap_or("".to_string())
+                                        );
+                                        println!(" {}", detail);
+                                        detector_details.push(detail);
+                                    }
+                                }
+                            }
+                        }
+
+                        let message = format!(
+                            "• Delay: {}ms\n\
+                            • Status: {}\n\
+                            • Region: {}\n\
+                            • Location: {:.5}, {:.5}\n\
+                            • Used detectors:\n{}\n\
+                            Stay safe! 🌩️",
+                            all.delay,
+                            all.status,
+                            all.region,
+                            lon,
+                            lat,
+                            {
+                                if args.detector_fetch {
+                                    let shown = 2;
+                                    let total = detector_details.len();
+                                    let mut lines = Vec::new();
+                                    for d in detector_details.iter().take(shown) {
+                                        lines.push(format!("    - {}", d));
+                                    }
+                                    if total > shown {
+                                        lines.push(format!("    (+{} others..)", total - shown));
+                                    }
+                                    lines.join("\n")
+                                } else {
+                                    "".to_string()
+                                }
+                            }
+                        );
+
+                        let json_payload = serde_json::json!({
+                            "topic": args.ntfy_topic,
+                            "message": message,
+                            "title": "⚡ Lightning Strike Detected",
+                            "priority": 4,
+                            "actions": [
+                                {
+                                    "action": "view",
+                                    "label": "View in Google Maps",
+                                    "url": format!("geo:{:.5},{:.5}", lon, lat),
+                                    "clear": false
+                                }
+                            ]
+                        });
+
+                        let client = reqwest::Client::new();
+                        let res = client
+                            .post("https://ntfy.sh/")
+                            .body(json_payload.to_string())
+                            .send()
+                            .await;
+
+                        match res {
+                            Ok(_) => println!("Notification sent to ntfy.sh!"),
+                            Err(e) => println!("Failed to send notification: {}", e),
+                        }
+                    } else {
+                        print!(
+                            "\rDelay: {}ms, offset to now: {}ms      ",
+                            all.delay, offset_ms
+                        );
+                        stdout().flush().unwrap();
+                    }
+                }
+                Ok(Message::Binary(bin)) => println!("Received binary: {:x?}", &bin[..10]),
+                Ok(_) => println!("Received other message type"),
+                Err(err) => {
+                    println!("WebSocket error: {}. Reconnecting in 5 seconds...", err);
+                    should_reconnect = true;
+                    break;
+                }
+            }
+        }
+        if should_reconnect {
+            tokio::time::sleep(Duration::from_secs(5)).await;
+            continue;
+        } else {
+            println!("WebSocket closed. Reconnecting in 5 seconds...");
+            tokio::time::sleep(Duration::from_secs(5)).await;
+            continue;
+        }
+    }
+}

src/message.rs

@@ -0,0 +1,66 @@
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Default, Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TungMessage {
+    pub time: i64,
+    pub lat: f64,
+    pub lon: f64,
+    pub alt: i64,
+    pub pol: i64,
+    pub mds: i64,
+    pub mcg: i64,
+    pub status: i64,
+    pub region: i64,
+    pub sig: Vec<Sig>,
+    pub delay: f64,
+    pub lonc: i64,
+    pub latc: i64,
+}
+
+#[derive(Default, Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Sig {
+    pub sta: i64,
+    pub time: i64,
+    pub lat: f64,
+    pub lon: f64,
+    pub alt: i64,
+    pub status: i64,
+}
+
+pub fn decode_lzw(input: &str) -> String {
+    let chars: Vec<char> = input.chars().collect();
+    if chars.is_empty() {
+        return String::new();
+    }
+
+    let mut dict: HashMap<u32, String> = HashMap::new();
+    let mut prev_sym = chars[0];
+    let mut word = prev_sym.to_string();
+    let mut out = vec![word.clone()];
+    let mut code: u32 = 256;
+
+    for &c in &chars[1..] {
+        let k = c as u32;
+        let entry = if k < 256 {
+            c.to_string()
+        } else if let Some(e) = dict.get(&k) {
+            e.clone()
+        } else {
+            format!("{}{}", word, prev_sym)
+        };
+
+        out.push(entry.clone());
+
+        prev_sym = entry.chars().next().unwrap();
+        dict.insert(code, format!("{}{}", word, prev_sym));
+        code += 1;
+
+        word = entry;
+    }
+
+    out.concat()
+}