dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-09-26 09:27:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: post-rebase compatibility fixes for #138 (radio reliability)

Browse Source 
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.
This commit is contained in:
Stacy Olivas
2026-04-14 10:02:48 -07:00
committed by agessaman
parent 973d1fc947
commit a75ea16609
2 changed files with 45 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/command_manager.py
+26 -26
View File
@@ -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 <command>'"
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)
modules/message_handler.py
+19 -19
View File
@@ -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]}...)")