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.
This commit is contained in:
agessaman committed 2026-10-03 12:42:23 -07:00
1 parent 7c186cb65c
commit 752a1c24bc
2 files changed
+39 -35

No files matched your search

+3 -35
View File
@@ -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}")
+36
View File
@@ -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