From e7493efcc182e196dfcb380e2485cdbbbbadd169 Mon Sep 17 00:00:00 2001
From: Jannik Beyerstedt <code@jbeyerstedt.de>
Date: Sat, 20 Jun 2026 20:15:11 +0200
Subject: [PATCH] v2x: Reduce SPAT rate

---
 src/main.rs | 89 ++++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 84 insertions(+), 5 deletions(-)

diff --git a/src/main.rs b/src/main.rs
index 91b7fc3..23ed774 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -10,6 +10,8 @@
 use alloc::string::ToString as _;
 use alloc::vec::Vec;
 use core::cell::RefCell;
+#[cfg(feature = "spat")]
+use core::cell::UnsafeCell;
 
 use critical_section::Mutex;
 use crossbeam_queue::ArrayQueue;
@@ -43,6 +45,8 @@ const WIFI_CHANNEL: radio::Channel = 180;
 pub const DISPLAY_WIDTH: u16 = 320;
 #[cfg(feature = "screen")]
 pub const DISPLAY_HEIGHT: u16 = 172;
+#[cfg(feature = "spat")]
+const SPAT_RATE_LIMIT: u8 = 5; // keep every n-th message
 
 // This creates a default app-descriptor required by the esp-idf bootloader.
 // For more information see: <https://docs.espressif.com/projects/esp-idf/en/stable/esp32/api-reference/system/app_image_format.html#application-description>
@@ -235,7 +239,7 @@ async fn main(spawner: Spawner) -> ! {
                     use c_its_parser::ItsMessage;
                     // use c_its_parser::standards::extensions::ItsMessageId;
 
-                    // debug!("Got new {:?} message", ItsMessageId::from(&msg));
+                    // info!("Got new {:?} message", ItsMessageId::from(&msg));
 
                     match msg {
                         #[cfg(feature = "spat")]
@@ -295,7 +299,28 @@ async fn main(spawner: Spawner) -> ! {
     clippy::needless_pass_by_value,
     reason = "adhering to callback interface"
 )]
+#[allow(clippy::too_many_lines)]
 fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) {
+    #[cfg(feature = "spat")]
+    const PRUNE_INTERVAL: chrono::Duration = chrono::Duration::seconds(60);
+    #[cfg(feature = "spat")]
+    static mut RATELIMIT_CACHE_LAST_PRUNE: UnsafeCell<chrono::NaiveDateTime> =
+        UnsafeCell::new(chrono::NaiveDateTime::MIN);
+
+    #[cfg(feature = "spat")]
+    static mut SPAT_RATELIMIT_CACHE: UnsafeCell<cache::Cache<u32, RatelimitData>> = UnsafeCell::new(
+        cache::Cache::<u32, RatelimitData>::new(chrono::Duration::seconds(20)),
+    );
+
+    #[cfg(feature = "spat")]
+    let now = {
+        let time = esp_hal::time::Instant::now().duration_since_epoch();
+        // unwrap is fine since value range is u64 microseconds
+        chrono::DateTime::from_timestamp(time.as_secs().cast_signed(), 0)
+            .unwrap()
+            .naive_utc()
+    };
+
     // Ignore frames with errors and non-data frames
     if frame.rx_cntl.rx_state != 0 {
         warn!("Received frame has RX error: {}", frame.rx_cntl.rx_state);
@@ -340,16 +365,37 @@ fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) {
             }
 
             // "parse" ITS PDU Header
-            if let Ok((payload, message_id, _)) = get_its_header(&packet) {
-                // drop all unsupported message IDs
+            if let Ok((payload, message_id, station_id)) = get_its_header(&packet) {
+                // drop all unsupported message IDs and rate-limit SPATEMs
                 if match message_id {
                     #[cfg(feature = "denm")]
                     c_its_parser::standards::extensions::ItsMessageId::Denm => false,
                     #[cfg(feature = "cam")]
                     c_its_parser::standards::extensions::ItsMessageId::Cam => false,
                     #[cfg(feature = "spat")]
-                    c_its_parser::standards::extensions::ItsMessageId::Spatem
-                    | c_its_parser::standards::extensions::ItsMessageId::Mapem => false,
+                    c_its_parser::standards::extensions::ItsMessageId::Spatem => {
+                        // reduce SPAT rate per station ID
+                        let mut drop_msg = false;
+                        unsafe {
+                            #[allow(
+                                static_mut_refs,
+                                reason = "this function will only run consecutive"
+                            )]
+                            let spat_cache = SPAT_RATELIMIT_CACHE.get();
+
+                            (*spat_cache).update_or_init(now, station_id, |i| {
+                                i.msg_count = i.msg_count.wrapping_add(1);
+
+                                if (i.msg_count % SPAT_RATE_LIMIT) > 0 {
+                                    drop_msg = true;
+                                }
+                            });
+                        }
+
+                        drop_msg
+                    }
+                    #[cfg(feature = "spat")]
+                    c_its_parser::standards::extensions::ItsMessageId::Mapem => false,
                     _ => true,
                 } {
                     return;
@@ -368,6 +414,39 @@ fn handle_frame(frame: wifi::sniffer::PromiscuousPkt<'_>) {
             }
         }
     }
+
+    #[cfg(feature = "spat")]
+    unsafe {
+        #[allow(static_mut_refs, reason = "this function will only run consecutive")]
+        let last_prune_time = RATELIMIT_CACHE_LAST_PRUNE.get();
+        if *last_prune_time + PRUNE_INTERVAL < now {
+            #[allow(static_mut_refs, reason = "this function will only run consecutive")]
+            let spat_cache = SPAT_RATELIMIT_CACHE.get();
+            (*spat_cache).prune(now);
+
+            *last_prune_time = now;
+        }
+    }
+}
+
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+struct RatelimitData {
+    station_id: u32,
+    msg_count: u8,
+}
+
+impl cache::Cachable<u32> for RatelimitData {
+    fn key(&self) -> u32 {
+        self.station_id
+    }
+}
+impl cache::Initable<Self, u32> for RatelimitData {
+    fn init(key: u32) -> Self {
+        Self {
+            station_id: key,
+            ..Default::default()
+        }
+    }
 }
 
 fn get_its_header<'p>(