From 752a1c24bc3037edec74f54205592ea16f146f5f Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 3 Oct 2026 12:42:23 -0700 Subject: [PATCH] refactor(rf_correlation): RfCorrelationMixin owns inserting RF rows and caching signal values handle_rf_log_data no longer updates the caches itself. _remember_signal replaces its two copies of the LRU-bounded SNR/RSSI cache update, and _insert_rf_row takes the block that files a built row: the authenticated channel cache, the TC_FLOOD debug line, recent_rf_data, the timestamp and prefix indexes, and eviction. Log text and levels are unchanged. Checked by replaying the 4,009-packet decode corpus through handle_rf_log_data on the old and new code (frozen clock, small cache limits so eviction runs, with and without a transmission tracker, varied SNR/RSSI/metadata/scope fields): the caches, indexes, every logger and tracker call and 400 lookups afterwards are byte-identical, and three mutations of the moved code each change the output. --- modules/message_handler.py | 38 +++----------------------------------- modules/rf_correlation.py | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 35 deletions(-) diff --git a/modules/message_handler.py b/modules/message_handler.py index 06db066..5378741 100644 --- a/modules/message_handler.py +++ b/modules/message_handler.py @@ -580,23 +580,13 @@ class MessageHandler(MeshGraphRecorderMixin, ContactEventsMixin, RfCorrelationMi self.logger.debug(f"Got pubkey_prefix from metadata: {pubkey_prefix[:16]}...") if packet_prefix and snr_value is not None: - # Cache the SNR value for this packet prefix (LRU-bounded) - self.snr_cache[packet_prefix] = snr_value - self.snr_cache.move_to_end(packet_prefix) - while len(self.snr_cache) > self._max_signal_cache_size: - self.snr_cache.popitem(last=False) - self.logger.debug(f"Cached SNR {snr_value} for packet prefix {packet_prefix}") + self._remember_signal(self.snr_cache, packet_prefix, snr_value, "SNR") # Extract and cache RSSI if available if "rssi" in payload: rssi_value = payload.get("rssi") if packet_prefix and rssi_value is not None: - # Cache the RSSI value for this packet prefix (LRU-bounded) - self.rssi_cache[packet_prefix] = rssi_value - self.rssi_cache.move_to_end(packet_prefix) - while len(self.rssi_cache) > self._max_signal_cache_size: - self.rssi_cache.popitem(last=False) - self.logger.debug(f"Cached RSSI {rssi_value} for packet prefix {packet_prefix}") + self._remember_signal(self.rssi_cache, packet_prefix, rssi_value, "RSSI") # Store recent RF data with timestamp for SNR/RSSI matching only if packet_prefix: @@ -903,29 +893,7 @@ class MessageHandler(MeshGraphRecorderMixin, ContactEventsMixin, RfCorrelationMi if _lib_pkt_hex else (decoded_packet.get("payload_hex") if decoded_packet else None), } - self._cache_authenticated_channel_rf_data( - rf_data, decoded_packet, current_time - ) - if rf_data.get("route_type_int") == 0: - self.logger.debug( - "TC_FLOOD scope fields: tc_code1=%s payload_type=%s payload_hex_prefix=%s", - rf_data.get("transport_code1"), - rf_data.get("payload_type_int"), - (rf_data.get("scope_payload_hex") or "")[:16], - ) - self.recent_rf_data.append(rf_data) - - # Update correlation indexes - self.rf_data_by_timestamp[current_time] = rf_data - if packet_prefix: - if packet_prefix not in self.rf_data_by_pubkey: - self.rf_data_by_pubkey[packet_prefix] = [] - self.rf_data_by_pubkey[packet_prefix].append(rf_data) - - # Clean up old data from all indexes - self._cleanup_stale_cache_entries(current_time) - - self.logger.debug(f"Stored recent RF data with routing info: {rf_data}") + self._insert_rf_row(rf_data, decoded_packet, current_time) except Exception as e: self.logger.error(f"Error handling RF log data: {e}") diff --git a/modules/rf_correlation.py b/modules/rf_correlation.py index b307e34..e7e5256 100644 --- a/modules/rf_correlation.py +++ b/modules/rf_correlation.py @@ -30,6 +30,7 @@ class RfCorrelationMixin: _is_rf_data_scope_eligible: Any _last_cache_cleanup: Any _max_rf_cache_size: Any + _max_signal_cache_size: Any bot: Any channel_rf_data: Any enhanced_correlation: Any @@ -40,6 +41,41 @@ class RfCorrelationMixin: rf_data_by_timestamp: Any rf_data_timeout: Any + def _remember_signal(self, cache: Any, packet_prefix: str, value: Any, label: str) -> None: + """Cache an SNR or RSSI value for a packet prefix (LRU-bounded).""" + cache[packet_prefix] = value + cache.move_to_end(packet_prefix) + while len(cache) > self._max_signal_cache_size: + cache.popitem(last=False) + self.logger.debug(f"Cached {label} {value} for packet prefix {packet_prefix}") + + def _insert_rf_row(self, rf_data: dict[str, Any], decoded_packet: Any, current_time: float) -> None: + """Add a received packet's row to the RF caches and indexes, then evict stale entries.""" + packet_prefix = rf_data["packet_prefix"] + self._cache_authenticated_channel_rf_data( + rf_data, decoded_packet, current_time + ) + if rf_data.get("route_type_int") == 0: + self.logger.debug( + "TC_FLOOD scope fields: tc_code1=%s payload_type=%s payload_hex_prefix=%s", + rf_data.get("transport_code1"), + rf_data.get("payload_type_int"), + (rf_data.get("scope_payload_hex") or "")[:16], + ) + self.recent_rf_data.append(rf_data) + + # Update correlation indexes + self.rf_data_by_timestamp[current_time] = rf_data + if packet_prefix: + if packet_prefix not in self.rf_data_by_pubkey: + self.rf_data_by_pubkey[packet_prefix] = [] + self.rf_data_by_pubkey[packet_prefix].append(rf_data) + + # Clean up old data from all indexes + self._cleanup_stale_cache_entries(current_time) + + self.logger.debug(f"Stored recent RF data with routing info: {rf_data}") + @staticmethod def _channel_message_identity( channel_idx: Any, sender_timestamp: Any, txt_type: Any, text: Any