From a75ea16609515ff4dcd7ca5235d005949f654edb Mon Sep 17 00:00:00 2001 From: Stacy Olivas Date: Sun, 12 Apr 2026 11:04:27 -0700 Subject: [PATCH] fix: post-rebase compatibility fixes for #138 (radio reliability) Two small incompatibilities surfaced after rebasing dev-kg7qin-changes onto upstream/dev in code introduced by #138: - message_handler.py: replace two remaining Optional[str] annotations with str | None; the Optional import was removed during typing modernization but two annotations in this module were missed - command_manager.py: restore # noqa: F401 guard on PUBLIC_CHANNEL_KEY_HEX re-export so ruff auto-fix does not silently remove it (imported by core.py) Do not merge until #138 is merged. --- modules/command_manager.py | 52 +++++++++++++++++++------------------- modules/message_handler.py | 38 ++++++++++++++-------------- 2 files changed, 45 insertions(+), 45 deletions(-) diff --git a/modules/command_manager.py b/modules/command_manager.py index 396f4db..74ecee2 100644 --- a/modules/command_manager.py +++ b/modules/command_manager.py @@ -9,7 +9,7 @@ import random import time from dataclasses import dataclass from hashlib import sha256 -from typing import Any, Optional +from typing import Any from meshcore import EventType @@ -37,7 +37,7 @@ class InternetStatusCache: """ has_internet: bool timestamp: float - _lock: Optional[asyncio.Lock] = None + _lock: asyncio.Lock | None = None def _get_lock(self) -> asyncio.Lock: """Lazily initialize the async lock. @@ -110,7 +110,7 @@ class CommandManager: # Command queue for near-expiring global cooldowns # Key: (command_name, user_id) tuple, Value: QueuedCommand self._command_queue: dict[tuple[str, str], QueuedCommand] = {} - self._queue_processor_task: Optional[asyncio.Task] = None + self._queue_processor_task: asyncio.Task | None = None # Multi-scope reply: map of normalized scope name → 16-byte HMAC key. # flood_scope_allow_global is True when '*' (or equivalent) appears in @@ -306,11 +306,11 @@ class CommandManager: self.logger.debug(f"Applying {self.bot.tx_delay_ms}ms transmission delay") await asyncio.sleep(self.bot.tx_delay_ms / 1000.0) - def get_rate_limit_key(self, message: MeshMessage) -> Optional[str]: + def get_rate_limit_key(self, message: MeshMessage) -> str | None: """Return the key used for per-user rate limiting (pubkey when available, else sender name).""" return message.sender_pubkey or message.sender_id or None - def get_rate_limit_wait_seconds(self, rate_limit_key: Optional[str] = None) -> float: + def get_rate_limit_wait_seconds(self, rate_limit_key: str | None = None) -> float: """Return seconds to wait until we could pass rate limits (for reply retry).""" wait = 0.0 if not self.bot.rate_limiter.can_send(): @@ -322,8 +322,8 @@ class CommandManager: return wait async def _check_rate_limits( - self, skip_user_rate_limit: bool = False, rate_limit_key: Optional[str] = None, - channel: Optional[str] = None, + self, skip_user_rate_limit: bool = False, rate_limit_key: str | None = None, + channel: str | None = None, ) -> tuple[bool, str]: """Check all rate limits before sending. @@ -389,7 +389,7 @@ class CommandManager: operation_name: str, target: str, used_retry_method: bool = False, - rate_limit_key: Optional[str] = None, + rate_limit_key: str | None = None, ) -> bool: """Handle result from message send operations. @@ -497,7 +497,7 @@ class CommandManager: banned = self.bot.config.get('Banned_Users', 'banned_users', fallback='') return [user.strip() for user in banned.split(',') if user.strip()] - def is_user_banned(self, sender_id: Optional[str]) -> bool: + def is_user_banned(self, sender_id: str | None) -> bool: """Check if sender is banned using prefix (starts-with) matching. A banned entry "Awful Username" matches "Awful Username" and "Awful Username 🍆". @@ -526,7 +526,7 @@ class CommandManager: return channel_list - def load_channel_keywords(self) -> Optional[list[str]]: + def load_channel_keywords(self) -> list[str] | None: """Load channel keyword whitelist from config. When set, only these triggers (command/keyword names) are answered in channels; @@ -615,7 +615,7 @@ class CommandManager: Returns: List[tuple]: List of (trigger, response) tuples for matched keywords. """ - matches: list[tuple[str, Optional[str]]] = [] + matches: list[tuple[str, str | None]] = [] content = message.content.strip() # Check for command prefix if configured @@ -778,7 +778,7 @@ class CommandManager: # case-insensitive + ignore extra spaces return " ".join(text.lower().split()) - def match_randomline(self, message: MeshMessage) -> Optional[tuple[str, str]]: + def match_randomline(self, message: MeshMessage) -> tuple[str, str] | None: """ Exact-match message content against RandomLine triggers. Returns (key, response) or None. @@ -919,9 +919,9 @@ class CommandManager: self, recipient_id: str, content: str, - command_id: Optional[str] = None, + command_id: str | None = None, skip_user_rate_limit: bool = False, - rate_limit_key: Optional[str] = None, + rate_limit_key: str | None = None, ) -> bool: """Send a direct message using meshcore-cli command. @@ -1027,10 +1027,10 @@ class CommandManager: self, channel: str, content: str, - command_id: Optional[str] = None, + command_id: str | None = None, skip_user_rate_limit: bool = False, - rate_limit_key: Optional[str] = None, - scope: Optional[str] = None, + rate_limit_key: str | None = None, + scope: str | None = None, ) -> bool: """Send a channel message using meshcore_py (optional flood scope). @@ -1147,10 +1147,10 @@ class CommandManager: channel: str, chunks: list[str], *, - command_id: Optional[str] = None, + command_id: str | None = None, skip_user_rate_limit: bool = True, - rate_limit_key: Optional[str] = None, - scope: Optional[str] = None, + rate_limit_key: str | None = None, + scope: str | None = None, ) -> bool: """Send multiple channel messages with rate-limit spacing between chunks. @@ -1195,7 +1195,7 @@ class CommandManager: return False return True - def get_help_for_command(self, command_name: str, message: Optional[MeshMessage] = None) -> str: + def get_help_for_command(self, command_name: str, message: MeshMessage | None = None) -> str: """Get help text for a specific command (LoRa-friendly compact format). Args: @@ -1228,7 +1228,7 @@ class CommandManager: return f"Help {command_name}: {help_text}" # Next, consult plugin_loader keyword mappings (if available) - mapped_name: Optional[str] = None + mapped_name: str | None = None if hasattr(self, 'plugin_loader') and hasattr(self.plugin_loader, 'keyword_mappings'): mapped_name = self.plugin_loader.keyword_mappings.get(normalized_name) if mapped_name: @@ -1285,7 +1285,7 @@ class CommandManager: _HELP_PREFIX = "Bot Help: " _HELP_SUFFIX = " | More: 'help '" - def get_general_help(self, message: Optional[MeshMessage] = None) -> str: + def get_general_help(self, message: MeshMessage | None = None) -> str: """Get general help text from config (LoRa-friendly compact format). When message is provided, only lists commands valid for the message's channel. @@ -1773,11 +1773,11 @@ class CommandManager: return has_internet - def get_plugin_by_keyword(self, keyword: str) -> Optional[BaseCommand]: + def get_plugin_by_keyword(self, keyword: str) -> BaseCommand | None: """Get a plugin by keyword""" return self.plugin_loader.get_plugin_by_keyword(keyword) - def get_plugin_by_name(self, name: str) -> Optional[BaseCommand]: + def get_plugin_by_name(self, name: str) -> BaseCommand | None: """Get a plugin by name""" return self.plugin_loader.get_plugin_by_name(name) @@ -1785,6 +1785,6 @@ class CommandManager: """Reload a specific plugin""" return self.plugin_loader.reload_plugin(plugin_name) - def get_plugin_metadata(self, plugin_name: Optional[str] = None) -> dict[str, Any]: + def get_plugin_metadata(self, plugin_name: str | None = None) -> dict[str, Any]: """Get plugin metadata""" return self.plugin_loader.get_plugin_metadata(plugin_name) diff --git a/modules/message_handler.py b/modules/message_handler.py index 60b5969..ab3b460 100644 --- a/modules/message_handler.py +++ b/modules/message_handler.py @@ -10,7 +10,7 @@ import hmac as hmac_mod import time from collections import OrderedDict from hashlib import sha256 -from typing import Any, Optional +from typing import Any from .enums import AdvertFlags, DeviceRole, PayloadType, PayloadVersion, RouteType from .graph_trace_helper import update_mesh_graph_from_trace_data @@ -70,7 +70,7 @@ class MessageHandler: @staticmethod def _match_scope(transport_code: int, payload_type: int, pkt_payload: bytes, - scope_keys: dict[str, bytes]) -> Optional[str]: + scope_keys: dict[str, bytes]) -> str | None: """Return the scope name whose HMAC matches transport_code, or None. Mirrors the firmware's TransportKey::calcTransportCode: computes @@ -341,13 +341,13 @@ class MessageHandler: timestamp = payload.get('sender_timestamp', 'unknown') # Look up contact name from pubkey prefix - sender_id = payload.get('pubkey_prefix', '') + sender_id = sanitize_name(payload.get('pubkey_prefix', '')) sender_name = sender_id # Default to sender_id if hasattr(self.bot.meshcore, 'contacts') and self.bot.meshcore.contacts: for _contact_key, contact_data in self.bot.meshcore.contacts.items(): if contact_data.get('public_key', '').startswith(sender_id): # Use the contact name if available, otherwise use adv_name - contact_name = contact_data.get('name', contact_data.get('adv_name', sender_id)) + contact_name = sanitize_name(contact_data.get('name', contact_data.get('adv_name', sender_id))) sender_name = contact_name break @@ -996,7 +996,7 @@ class MessageHandler: except Exception as e: self.logger.error(f"Error handling RF log data: {e}") - def extract_path_from_raw_hex(self, raw_hex: str, expected_hops: int) -> Optional[str]: + def extract_path_from_raw_hex(self, raw_hex: str, expected_hops: int) -> str | None: """Extract path information directly from raw hex data. Attempts to find a sequence of node IDs in the raw packet data that matches @@ -1068,7 +1068,7 @@ class MessageHandler: self.logger.debug(f"Error extracting path from raw hex: {e}") return None - def _cleanup_stale_cache_entries(self, current_time: Optional[float] = None) -> None: + def _cleanup_stale_cache_entries(self, current_time: float | None = None) -> None: """Remove stale entries from RF data caches and enforce maximum size limits. Args: @@ -1271,7 +1271,7 @@ class MessageHandler: - def decode_meshcore_packet(self, raw_hex: str, payload_hex: Optional[str] = None) -> Optional[dict]: + def decode_meshcore_packet(self, raw_hex: str, payload_hex: str | None = None) -> dict | None: """ Decode a MeshCore packet from raw hex data - matches Packet.cpp exactly @@ -1505,7 +1505,7 @@ class MessageHandler: self.logger.error(f"Error parsing ADVERT payload: {e}", exc_info=True) return {} - def _path_bytes_to_nodes(self, path_bytes: bytes, prefix_hex_chars: Optional[int] = None) -> tuple: + def _path_bytes_to_nodes(self, path_bytes: bytes, prefix_hex_chars: int | None = None) -> tuple: """Chunk path bytes into hex node IDs using configured prefix length, with legacy 2-char fallback. Args: @@ -1543,9 +1543,9 @@ class MessageHandler: def _get_path_from_rf_data( self, rf_data: dict[str, Any], - payload_hex: Optional[str] = None, - packet_info: Optional[dict[str, Any]] = None - ) -> tuple[Optional[str], Optional[list[str]], int]: + payload_hex: str | None = None, + packet_info: dict[str, Any] | None = None + ) -> tuple[str | None, list[str] | None, int]: """Get path string, path nodes, and hop count from RF data (single source for path extraction). Prefers routing_info.path_nodes when present (no re-decode; correct multi-byte). @@ -1900,7 +1900,7 @@ class MessageHandler: # Scope matching: if the RF data is a TC_FLOOD, check whether its transport # code matches any configured flood_scopes entry. If so, the reply should # use the same scope so it reaches the same scoped network segment. - reply_scope: Optional[str] = None + reply_scope: str | None = None if recent_rf_data: rt = recent_rf_data.get("route_type_int") tc_code1 = recent_rf_data.get("transport_code1") @@ -2161,7 +2161,7 @@ class MessageHandler: geographic_distance=geographic_distance ) - def _store_observed_path(self, advert_data: dict[str, Any], path_hex: str, path_length: int, packet_type: str, packet_hash: Optional[str] = None, bytes_per_hop: Optional[int] = None): + def _store_observed_path(self, advert_data: dict[str, Any], path_hex: str, path_length: int, packet_type: str, packet_hash: str | None = None, bytes_per_hop: int | None = None): """Store a complete path in the observed_paths table. Args: @@ -2255,7 +2255,7 @@ class MessageHandler: import traceback self.logger.debug(traceback.format_exc()) - def _get_bot_location_fallback(self) -> Optional[tuple[float, float]]: + def _get_bot_location_fallback(self) -> tuple[float, float] | None: """Get bot location from config to use as fallback reference for distance-based selection. Returns: @@ -2274,7 +2274,7 @@ class MessageHandler: self.logger.debug(f"Error getting bot location fallback: {e}") return None - def _get_location_by_public_key(self, public_key: str) -> Optional[tuple[float, float]]: + def _get_location_by_public_key(self, public_key: str) -> tuple[float, float] | None: """Get location for a full public key (more accurate than prefix lookup). Prefers starred repeaters if there are somehow multiple entries (shouldn't happen with full key). @@ -2404,7 +2404,7 @@ class MessageHandler: # Use bot location as fallback reference to ensure distance-based selection bot_location_ref = self._get_bot_location_fallback() first_hop_temp_result = _get_node_location_from_db(self.bot, first_hop, bot_location_ref, recency_days) - first_hop_location_temp: Optional[tuple[float, float]] + first_hop_location_temp: tuple[float, float] | None if first_hop_temp_result: first_hop_location_temp, _ = first_hop_temp_result else: @@ -2756,7 +2756,7 @@ class MessageHandler: except Exception as e: self.logger.error(f"Error in debug packet decoding: {e}") - def _format_path_string(self, hex_path: str, bytes_per_hop: Optional[int] = None) -> str: + def _format_path_string(self, hex_path: str, bytes_per_hop: int | None = None) -> str: """ Convert a hex path string to node prefix format. @@ -3046,7 +3046,7 @@ class MessageHandler: if hash_mode != -1: return - opl: Optional[int] + opl: int | None raw_opl = contact_data.get('out_path_len') try: opl = None if raw_opl is None else int(raw_opl) @@ -3107,7 +3107,7 @@ class MessageHandler: self.logger.info(f"📦 Event payload: {contact_data}") # Get contact details - contact_name = contact_data.get('name', contact_data.get('adv_name', 'Unknown')) + contact_name = sanitize_name(contact_data.get('name', contact_data.get('adv_name', 'Unknown'))) public_key = contact_data.get('public_key', '') self.logger.info(f"Processing new contact: {contact_name} (key: {public_key[:16]}...)")