#!/usr/bin/env python3 """ Utility functions for the MeshCore Bot Shared helper functions used across multiple modules """ import asyncio import socket import urllib.error import urllib.request from collections.abc import Generator from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Mapping, Optional, Union from .mesh_paths import ( # noqa: F401 re-exported: callers import these from utils _get_node_location_from_db, _normalized_message_path_string, bytes_per_hop_from_routing_and_nodes, calculate_path_distances, extract_path_node_ids_from_message, message_hop_count, message_path_bytes_per_hop, node_ids_from_path_string, parse_path_string, public_key_has_prefix, ) from .meshcore_payload_decode import verify_meshcore_advert_ed25519 # noqa: F401 re-exported from .packet_decode import ( # noqa: F401 re-exported: callers import these from utils calculate_packet_hash, decode_path_len_byte, encode_path_len_byte, parse_trace_payload_route_hashes, ) try: from zoneinfo import ZoneInfo, ZoneInfoNotFoundError except ImportError: ZoneInfo = None # type: ignore[misc, assignment] ZoneInfoNotFoundError = Exception # type: ignore[misc, assignment] def is_valid_timezone(tz_str: str) -> bool: """Return True if the string is a valid IANA timezone name.""" if not (tz_str and tz_str.strip()): return False if ZoneInfo is not None: try: ZoneInfo(tz_str.strip()) return True except ZoneInfoNotFoundError: return False try: pytz = __import__("pytz") pytz.timezone(tz_str.strip()) return True except Exception: return False def get_config_timezone(config: Any, logger: Optional[Any] = None) -> tuple[Any, str]: """Resolve [Bot] timezone from config; fall back to system timezone if invalid or empty. Returns: (tz, iana_str): tz is a timezone object for datetime; iana_str is an IANA string for APIs (e.g. OpenMeteo). When falling back to system, iana_str is "UTC". """ timezone_str = (config.get('Bot', 'timezone', fallback='') or '').strip() if timezone_str and is_valid_timezone(timezone_str): pytz = __import__("pytz") return (pytz.timezone(timezone_str), timezone_str) if timezone_str and logger: logger.warning("Invalid timezone '%s', using system timezone", timezone_str) # System timezone for datetime; use "UTC" for API when we don't have an IANA name tz = datetime.now().astimezone().tzinfo return (tz, "UTC") def espn_dates_for_local_day( local_tz: Any, now: Optional[datetime] = None, ) -> tuple[str, str, float, float]: """Return ESPN scoreboard date range and local-day bounds for filtering events. ESPN buckets scoreboard events by UTC calendar date. A single local calendar day can span two UTC dates (e.g. 9pm PT is the next UTC day). This returns the min/max YYYYMMDD strings to query, plus local midnight timestamps for filtering. Returns: (start_yyyymmdd, end_yyyymmdd, local_start_ts, local_end_ts) """ if now is None: now = datetime.now(local_tz) elif now.tzinfo is None: if hasattr(local_tz, 'localize'): now = local_tz.localize(now) else: now = now.replace(tzinfo=local_tz) else: now = now.astimezone(local_tz) if hasattr(local_tz, 'localize'): local_start = local_tz.localize(datetime(now.year, now.month, now.day)) else: local_start = now.replace(hour=0, minute=0, second=0, microsecond=0) local_end = local_start + timedelta(days=1) utc_dates = { local_start.astimezone(timezone.utc).strftime("%Y%m%d"), (local_end - timedelta(seconds=1)).astimezone(timezone.utc).strftime("%Y%m%d"), } return min(utc_dates), max(utc_dates), local_start.timestamp(), local_end.timestamp() def filter_events_local_day( events: list[dict], local_start_ts: float, local_end_ts: float, ts_key: str = "event_timestamp", ) -> list[dict]: """Keep events whose kickoff falls in [local_start, local_end); keep missing timestamps.""" filtered: list[dict] = [] for event in events: ts = event.get(ts_key) if ts is None or (local_start_ts <= ts < local_end_ts): filtered.append(event) return filtered def format_temperature_high_low( config: Any, high: Optional[Union[int, float]], low: Optional[Union[int, float]], units_str: str, logger: Optional[Any] = None, translator: Any = None, ) -> str: """Format a daily high/low pair (or single value) using [Weather] templates. Config keys (optional; defaults match prior bot behavior): temperature_high_low_format — both values: {high}, {low}, {units}, {high_label}, {low_label} temperature_high_only_format — {high}, {units}, {high_label} temperature_low_only_format — {low}, {units}, {low_label} """ if translator is not None: high_label = translator.translate("common.temp_high_label") low_label = translator.translate("common.temp_low_label") else: high_label = "H" low_label = "L" section = "Weather" default_pair = f"{high_label}:{{high}}{{units}} {low_label}:{{low}}{{units}}" default_high_only = f"{high_label}:{{high}}{{units}}" default_low_only = f"{low_label}:{{low}}{{units}}" def _norm(v: Optional[Union[int, float]]) -> Optional[int]: if v is None: return None try: if isinstance(v, float): return int(round(v)) return int(v) except (TypeError, ValueError): return None hi = _norm(high) lo = _norm(low) if hi is None and lo is None: return "" if config.has_section(section): pair_fmt = config.get(section, "temperature_high_low_format", fallback=default_pair) high_only_fmt = config.get(section, "temperature_high_only_format", fallback=default_high_only) low_only_fmt = config.get(section, "temperature_low_only_format", fallback=default_low_only) else: pair_fmt, high_only_fmt, low_only_fmt = default_pair, default_high_only, default_low_only def _try_format(fmt: str, **kwargs: Any) -> Optional[str]: try: return fmt.format(**kwargs) except (KeyError, ValueError, IndexError) as e: if logger is not None and hasattr(logger, "warning"): logger.warning("Invalid temperature format template %r: %s", fmt, e) return None labels = {"high_label": high_label, "low_label": low_label} if hi is not None and lo is not None: out = _try_format(pair_fmt, high=hi, low=lo, units=units_str, **labels) if out is not None: return out return _try_format(default_pair, high=hi, low=lo, units=units_str, **labels) or f"{high_label}:{hi}{units_str} {low_label}:{lo}{units_str}" if hi is not None: out = _try_format(high_only_fmt, high=hi, low=lo, units=units_str, **labels) if out is not None: return out return _try_format(default_high_only, high=hi, low=lo, units=units_str, **labels) or f"{high_label}:{hi}{units_str}" out = _try_format(low_only_fmt, high=hi, low=lo, units=units_str, **labels) if out is not None: return out return _try_format(default_low_only, high=hi, low=lo, units=units_str, **labels) or f"{low_label}:{lo}{units_str}" def abbreviate_location(location: str, max_length: int = 20) -> str: """Abbreviate a location string to fit within character limits. Args: location: The location string to abbreviate. max_length: Maximum length for the abbreviated string (default: 20). Returns: str: Abbreviated location string. """ if not location: return location # Apply common abbreviations first abbreviated = location abbreviations = [ ('Central Business District', 'CBD'), ('United States of America', 'USA'), ('Business District', 'BD'), ('British Columbia', 'BC'), ('United States', 'USA'), ('United Kingdom', 'UK'), ('Washington', 'WA'), ('California', 'CA'), ('New York', 'NY'), ('Texas', 'TX'), ('Florida', 'FL'), ('Illinois', 'IL'), ('Pennsylvania', 'PA'), ('Ohio', 'OH'), ('Georgia', 'GA'), ('North Carolina', 'NC'), ('Michigan', 'MI'), ('New Jersey', 'NJ'), ('Virginia', 'VA'), ('Tennessee', 'TN'), ('Indiana', 'IN'), ('Arizona', 'AZ'), ('Massachusetts', 'MA'), ('Missouri', 'MO'), ('Maryland', 'MD'), ('Wisconsin', 'WI'), ('Colorado', 'CO'), ('Minnesota', 'MN'), ('South Carolina', 'SC'), ('Alabama', 'AL'), ('Louisiana', 'LA'), ('Kentucky', 'KY'), ('Oregon', 'OR'), ('Oklahoma', 'OK'), ('Connecticut', 'CT'), ('Utah', 'UT'), ('Iowa', 'IA'), ('Nevada', 'NV'), ('Arkansas', 'AR'), ('Mississippi', 'MS'), ('Kansas', 'KS'), ('New Mexico', 'NM'), ('Nebraska', 'NE'), ('West Virginia', 'WV'), ('Idaho', 'ID'), ('Hawaii', 'HI'), ('New Hampshire', 'NH'), ('Maine', 'ME'), ('Montana', 'MT'), ('Rhode Island', 'RI'), ('Delaware', 'DE'), ('South Dakota', 'SD'), ('North Dakota', 'ND'), ('Alaska', 'AK'), ('Vermont', 'VT'), ('Wyoming', 'WY') ] # Sort by length (longest first) to ensure longer matches are checked before shorter ones # This prevents "United States" from matching before "United States of America" abbreviations.sort(key=lambda x: len(x[0]), reverse=True) # Apply abbreviations in order for full_term, abbrev in abbreviations: if full_term in abbreviated: abbreviated = abbreviated.replace(full_term, abbrev) # If still too long after abbreviations, try to truncate intelligently if len(abbreviated) > max_length: # Try to keep the most important part (usually the city name) parts = abbreviated.split(', ') if len(parts) > 1: # Keep the first part (usually city) and truncate if needed first_part = parts[0] abbreviated = first_part if len(first_part) <= max_length else first_part[:max_length - 3] + '...' else: # Just truncate with ellipsis abbreviated = abbreviated[:max_length-3] + '...' return abbreviated def truncate_string(text: str, max_length: int, ellipsis: str = '...') -> str: """Truncate a string to a maximum length with ellipsis. Args: text: The string to truncate. max_length: Maximum length including ellipsis. ellipsis: String to append when truncating (default: '...'). Returns: str: Truncated string. """ if not text or len(text) <= max_length: return text return text[:max_length - len(ellipsis)] + ellipsis def decode_escape_sequences(text: str) -> str: """Decode escape sequences in config strings (e.g. Keywords, Scheduled_Messages). Processes \\n (newline), \\t (tab), \\r (carriage return), \\\\ (literal backslash). Use a single backslash in config: \\n for newline; \\\\n for literal backslash + n. Args: text: The text string to process. Returns: str: The text with escape sequences decoded. """ if not text: return text text = text.replace('\\\\', '\x00') # Temporary placeholder for backslash text = text.replace('\\n', '\n') # Newline text = text.replace('\\t', '\t') # Tab text = text.replace('\\r', '\r') # Carriage return text = text.replace('\x00', '\\') # Restore backslash return text def format_location_for_display(city: Optional[str], state: Optional[str] = None, country: Optional[str] = None, max_length: int = 20) -> Optional[str]: """Format location data for display with intelligent abbreviation. Args: city: City name (may include neighborhood/district). state: State/province name (optional). country: Country name (optional). max_length: Maximum length for the formatted location (default: 20). Returns: Optional[str]: Formatted location string or None if no city provided. """ if not city: return None # Start with city (which may include neighborhood) location_parts = [city] # Add state if available and different from city if state and state not in location_parts: location_parts.append(state) # Join parts and abbreviate if needed full_location = ', '.join(location_parts) return abbreviate_location(full_location, max_length) def get_major_city_queries(city: str, state_abbr: Optional[str] = None) -> list[str]: """Get prioritized geocoding queries for major cities that have multiple locations. This helps ensure that common city names resolve to the most likely major city rather than a small town with the same name. Args: city: City name (normalized, lowercase). state_abbr: Optional state abbreviation (e.g., "CA", "NY"). Returns: List[str]: List of geocoding query strings in priority order. """ city_lower = city.lower().strip() # Comprehensive mapping of major cities with multiple locations # Format: 'city_name': [list of queries in priority order] major_city_mappings = { 'new york': ['New York, NY, USA', 'New York City, NY, USA'], 'los angeles': ['Los Angeles, CA, USA'], 'chicago': ['Chicago, IL, USA'], 'houston': ['Houston, TX, USA'], 'phoenix': ['Phoenix, AZ, USA'], 'philadelphia': ['Philadelphia, PA, USA'], 'san antonio': ['San Antonio, TX, USA'], 'san diego': ['San Diego, CA, USA'], 'dallas': ['Dallas, TX, USA'], 'san jose': ['San Jose, CA, USA'], 'austin': ['Austin, TX, USA'], 'jacksonville': ['Jacksonville, FL, USA'], 'san francisco': ['San Francisco, CA, USA'], 'columbus': ['Columbus, OH, USA'], 'fort worth': ['Fort Worth, TX, USA'], 'charlotte': ['Charlotte, NC, USA'], 'seattle': ['Seattle, WA, USA'], 'denver': ['Denver, CO, USA'], 'washington': ['Washington, DC, USA'], 'boston': ['Boston, MA, USA'], 'el paso': ['El Paso, TX, USA'], 'detroit': ['Detroit, MI, USA'], 'nashville': ['Nashville, TN, USA'], 'portland': ['Portland, OR, USA', 'Portland, ME, USA'], 'oklahoma city': ['Oklahoma City, OK, USA'], 'las vegas': ['Las Vegas, NV, USA'], 'memphis': ['Memphis, TN, USA'], 'louisville': ['Louisville, KY, USA'], 'baltimore': ['Baltimore, MD, USA'], 'milwaukee': ['Milwaukee, WI, USA'], 'albuquerque': ['Albuquerque, NM, USA'], 'tucson': ['Tucson, AZ, USA'], 'fresno': ['Fresno, CA, USA'], 'sacramento': ['Sacramento, CA, USA'], 'kansas city': ['Kansas City, MO, USA', 'Kansas City, KS, USA'], 'mesa': ['Mesa, AZ, USA'], 'atlanta': ['Atlanta, GA, USA'], 'omaha': ['Omaha, NE, USA'], 'colorado springs': ['Colorado Springs, CO, USA'], 'raleigh': ['Raleigh, NC, USA'], 'virginia beach': ['Virginia Beach, VA, USA'], 'miami': ['Miami, FL, USA'], 'oakland': ['Oakland, CA, USA'], 'minneapolis': ['Minneapolis, MN, USA'], 'tulsa': ['Tulsa, OK, USA'], 'cleveland': ['Cleveland, OH, USA'], 'wichita': ['Wichita, KS, USA'], 'arlington': ['Arlington, TX, USA', 'Arlington, VA, USA'], 'new orleans': ['New Orleans, LA, USA'], 'honolulu': ['Honolulu, HI, USA'], # Cities with multiple locations that need disambiguation 'albany': ['Albany, NY, USA', 'Albany, OR, USA', 'Albany, CA, USA'], 'springfield': ['Springfield, IL, USA', 'Springfield, MO, USA', 'Springfield, MA, USA'], 'franklin': ['Franklin, TN, USA', 'Franklin, MA, USA'], 'georgetown': ['Georgetown, TX, USA', 'Georgetown, SC, USA'], 'madison': ['Madison, WI, USA', 'Madison, AL, USA'], 'auburn': ['Auburn, AL, USA', 'Auburn, WA, USA'], 'troy': ['Troy, NY, USA', 'Troy, MI, USA'], 'clinton': ['Clinton, IA, USA', 'Clinton, MS, USA'], 'paris': ['Paris, TX, USA', 'Paris, IL, USA', 'Paris, TN, USA'], } # Check if this is a major city if city_lower in major_city_mappings: queries = major_city_mappings[city_lower].copy() # If state abbreviation was provided, prioritize queries with that state if state_abbr: state_upper = state_abbr.upper() # Move matching state queries to the front matching = [q for q in queries if f', {state_upper},' in q or q.endswith(f', {state_upper}')] non_matching = [q for q in queries if q not in matching] if matching: return matching + non_matching return queries # Not a major city - return empty list (caller should use standard geocoding) return [] def truncate_to_bytes(text: str, limit: int) -> str: """Trim ``text`` to ``limit`` UTF-8 bytes without splitting a character. Mesh budgets are byte budgets. Trimming with ``len(text)`` instead overshoots by however much the emoji and accents in the string cost beyond one byte each, so a reply that measured as a fit spills into a second message. """ if limit <= 0: return "" encoded = text.encode("utf-8") if len(encoded) <= limit: return text return encoded[:limit].decode("utf-8", errors="ignore") def get_packet_hash_placeholder(message: Any) -> str: """Return the correlated MeshCore packet hash for template placeholders. Reads only ``message.routing_info["packet_hash"]`` so an uncorrelated RF cache entry cannot leak another packet's identity. Missing, empty, and the error sentinel ``0000000000000000`` return ``""`` so the placeholder renders empty (and piped templates can hide a label with ``prefix_if_nonempty``). """ if message is None: return "" routing_info = getattr(message, "routing_info", None) if not isinstance(routing_info, dict): return "" packet_hash = routing_info.get("packet_hash") if not packet_hash or packet_hash == "0000000000000000": return "" return str(packet_hash) def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Calculate haversine distance between two points in kilometers. Args: lat1: Latitude of first point in degrees. lon1: Longitude of first point in degrees. lat2: Latitude of second point in degrees. lon2: Longitude of second point in degrees. Returns: float: Distance in kilometers. """ import math # Convert latitude and longitude from degrees to radians lat1_rad = math.radians(lat1) lon1_rad = math.radians(lon1) lat2_rad = math.radians(lat2) lon2_rad = math.radians(lon2) # Haversine formula dlat = lat2_rad - lat1_rad dlon = lon2_rad - lon1_rad a = math.sin(dlat/2)**2 + math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(dlon/2)**2 c = 2 * math.asin(math.sqrt(a)) # Earth's radius in kilometers earth_radius = 6371.0 return earth_radius * c # Optional geocoding helper libraries try: import pycountry PYCOUNTRY_AVAILABLE = True except ImportError: PYCOUNTRY_AVAILABLE = False try: import us US_AVAILABLE = True except ImportError: US_AVAILABLE = False def normalize_country_name(country_input: str) -> tuple[Optional[str], Optional[str]]: """Normalize country name to ISO code and standard name. Args: country_input: Country name or code (e.g., "Sweden", "SE", "United States", "USA", "US") Returns: tuple: (iso_code, standard_name) or (None, None) if not found Example: ("SE", "Sweden") or ("US", "United States") """ if not PYCOUNTRY_AVAILABLE: return None, None if not country_input: return None, None country_input = country_input.strip() # Try to find by alpha_2 code (e.g., "US", "SE") if len(country_input) == 2: try: country = pycountry.countries.get(alpha_2=country_input.upper()) if country: return country.alpha_2, country.name except (KeyError, AttributeError): pass # Try to find by alpha_3 code (e.g., "USA", "SWE") if len(country_input) == 3: try: country = pycountry.countries.get(alpha_3=country_input.upper()) if country: return country.alpha_2, country.name except (KeyError, AttributeError): pass # Try to find by name (case-insensitive, handles common variants) country_input_lower = country_input.lower() # Handle common variants country_variants = { 'usa': 'United States', 'u.s.a.': 'United States', 'u.s.': 'United States', 'uk': 'United Kingdom', 'u.k.': 'United Kingdom', 'great britain': 'United Kingdom', } search_name = country_variants.get(country_input_lower, country_input) try: # Try exact match first country = pycountry.countries.get(name=search_name) if country: return country.alpha_2, country.name # Try fuzzy search for country in pycountry.countries: if country.name.lower() == search_name.lower(): return country.alpha_2, country.name except (KeyError, AttributeError): pass return None, None def normalize_us_state(state_input: str) -> tuple[Optional[str], Optional[str]]: """Normalize US state name to abbreviation and full name. Args: state_input: State name or abbreviation (e.g., "Washington", "WA", "California", "CA") Returns: tuple: (abbreviation, full_name) or (None, None) if not found Example: ("WA", "Washington") or ("CA", "California") """ if not US_AVAILABLE: return None, None if not state_input: return None, None state_input = state_input.strip() # Try to find by abbreviation if len(state_input) == 2: try: state = us.states.lookup(state_input.upper()) if state: return state.abbr, state.name except (AttributeError, KeyError): pass # Try to find by name try: state = us.states.lookup(state_input) if state: return state.abbr, state.name except (AttributeError, KeyError): pass return None, None def is_country_name(text: str) -> bool: """Check if text is likely a country name. Args: text: Text to check Returns: bool: True if text appears to be a country name """ if not text: return False if PYCOUNTRY_AVAILABLE: iso_code, _ = normalize_country_name(text) if iso_code is not None: return True if US_AVAILABLE: state_abbr, _ = normalize_us_state(text) if state_abbr: return False # It's a US state, not a country if len(text) <= 2: return False # Unknown 2-char (not a known country or US state) return len(text) > 2 # Longer text, assume country def is_us_state(text: str) -> bool: """Check if text is likely a US state name or abbreviation. Args: text: Text to check Returns: bool: True if text appears to be a US state """ if not text: return False if US_AVAILABLE: state_abbr, _ = normalize_us_state(text) return state_abbr is not None return False def parse_location_string(location: str) -> tuple[str, Optional[str], Optional[str]]: """Parse a location string into city, state/country parts. Args: location: Location string (e.g., "Stockholm, Sweden" or "Seattle, WA") Returns: tuple: (city, state_or_country, type) where type is "state", "country", or None Example: ("Stockholm", "Sweden", "country") or ("Seattle", "WA", "state") """ if ',' not in location: return location.strip(), None, None parts = [p.strip() for p in location.rsplit(',', 1)] if len(parts) != 2: return location.strip(), None, None city, second_part = parts # Check if it's a US state if is_us_state(second_part): state_abbr, _ = normalize_us_state(second_part) return city, state_abbr, "state" # Check if it's a country if is_country_name(second_part): iso_code, country_name = normalize_country_name(second_part) if iso_code: return city, country_name, "country" # If 2 chars or less, assume state abbreviation if len(second_part) <= 2: return city, second_part.upper(), "state" # Otherwise, assume country return city, second_part, "country" def get_nominatim_geocoder(user_agent: str = "meshcore-bot", timeout: int = 10) -> Any: """Get a Nominatim geocoder instance with proper User-Agent. Args: user_agent: User-Agent string for Nominatim (required by their policy). timeout: Request timeout in seconds. Returns: Any: Nominatim geocoder instance (from geopy). """ from geopy.geocoders import Nominatim return Nominatim(user_agent=user_agent, timeout=timeout) async def rate_limited_nominatim_geocode( bot: Any, query: Union[str, Mapping[str, str]], timeout: int = 10 ) -> Optional[Any]: """Perform rate-limited Nominatim geocoding (forward geocoding). Args: bot: Bot instance (must have nominatim_rate_limiter attribute). query: Location query string or structured Nominatim dict (e.g. postalcode). timeout: Request timeout in seconds. Returns: Optional[Any]: Geocoding result or None if failed/timed out. """ if not hasattr(bot, 'nominatim_rate_limiter'): # Fallback if rate limiter not initialized geolocator = get_nominatim_geocoder(timeout=timeout) return await asyncio.to_thread(geolocator.geocode, query, timeout=timeout) # Atomically reserve a slot shared with synchronous worker-thread callers. await bot.nominatim_rate_limiter.wait_and_request() geolocator = get_nominatim_geocoder(timeout=timeout) return await asyncio.to_thread(geolocator.geocode, query, timeout=timeout) async def rate_limited_nominatim_reverse(bot: Any, coordinates: str, timeout: int = 10) -> Optional[Any]: """Perform rate-limited Nominatim reverse geocoding. Args: bot: Bot instance (must have nominatim_rate_limiter attribute). coordinates: Coordinates string in format "lat, lon". timeout: Request timeout in seconds. Returns: Optional[Any]: Reverse geocoding result or None if failed/timed out. """ if not hasattr(bot, 'nominatim_rate_limiter'): # Fallback if rate limiter not initialized geolocator = get_nominatim_geocoder(timeout=timeout) return await asyncio.to_thread(geolocator.reverse, coordinates, timeout=timeout) # Atomically reserve a slot shared with synchronous worker-thread callers. await bot.nominatim_rate_limiter.wait_and_request() geolocator = get_nominatim_geocoder(timeout=timeout) return await asyncio.to_thread(geolocator.reverse, coordinates, timeout=timeout) def rate_limited_nominatim_geocode_sync( bot: Any, query: Union[str, Mapping[str, str]], timeout: int = 10 ) -> Optional[Any]: """Perform rate-limited Nominatim geocoding (synchronous version). Args: bot: Bot instance (must have nominatim_rate_limiter attribute). query: Location query string or structured Nominatim dict (e.g. postalcode). timeout: Request timeout in seconds. Returns: Optional[Any]: Geocoding result or None if failed/timed out. """ if not hasattr(bot, 'nominatim_rate_limiter'): # Fallback if rate limiter not initialized geolocator = get_nominatim_geocoder(timeout=timeout) return geolocator.geocode(query, timeout=timeout) # Reserve the slot before the request, not after: this runs on worker threads. bot.nominatim_rate_limiter.wait_and_request_sync() geolocator = get_nominatim_geocoder(timeout=timeout) return geolocator.geocode(query, timeout=timeout) def rate_limited_nominatim_reverse_sync(bot: Any, coordinates: str, timeout: int = 10) -> Optional[Any]: """Perform rate-limited Nominatim reverse geocoding (synchronous version). Args: bot: Bot instance (must have nominatim_rate_limiter attribute). coordinates: Coordinates string in format "lat, lon". timeout: Request timeout in seconds. Returns: Optional[Any]: Reverse geocoding result or None if failed/timed out. """ if not hasattr(bot, 'nominatim_rate_limiter'): # Fallback if rate limiter not initialized geolocator = get_nominatim_geocoder(timeout=timeout) return geolocator.reverse(coordinates, timeout=timeout) # Reserve the slot before the request, not after: this runs on worker threads. bot.nominatim_rate_limiter.wait_and_request_sync() geolocator = get_nominatim_geocoder(timeout=timeout) return geolocator.reverse(coordinates, timeout=timeout) _GeocodeSteps = Generator[tuple[str, Any], Any, Any] def _run_geocode_steps_sync(bot: Any, steps: _GeocodeSteps, timeout: int) -> Any: """Drive a geocoding step generator with the synchronous rate-limited Nominatim calls. Each ``("geocode", query)`` or ``("reverse", coordinates)`` the generator yields is performed and its result (or exception) sent back in, so the generator's own error handling sees exactly what a direct call would raise. """ result: Any = None error: Optional[BaseException] = None while True: try: request = steps.throw(error) if error is not None else steps.send(result) except StopIteration as stop: return stop.value kind, arg = request fetch = rate_limited_nominatim_geocode_sync if kind == "geocode" else rate_limited_nominatim_reverse_sync result, error = None, None try: result = fetch(bot, arg, timeout=timeout) except BaseException as e: error = e async def _run_geocode_steps_async(bot: Any, steps: _GeocodeSteps, timeout: int) -> Any: """Async counterpart of ``_run_geocode_steps_sync``, using the async rate-limited calls.""" result: Any = None error: Optional[BaseException] = None while True: try: request = steps.throw(error) if error is not None else steps.send(result) except StopIteration as stop: return stop.value kind, arg = request fetch = rate_limited_nominatim_geocode if kind == "geocode" else rate_limited_nominatim_reverse result, error = None, None try: result = await fetch(bot, arg, timeout=timeout) except BaseException as e: error = e def _geocode_zipcode_steps(bot: Any, zipcode: str, default_country: Optional[str]) -> _GeocodeSteps: """Shared body of ``geocode_zipcode`` and ``geocode_zipcode_sync``, as a step generator.""" try: # Get default country from config if not provided if default_country is None: default_country = bot.config.get('Weather', 'default_country', fallback='US') # Check cache first cache_query = f"{zipcode}, {default_country}" cached_lat, cached_lon = bot.db_manager.get_cached_geocoding(cache_query) if cached_lat is not None and cached_lon is not None: return cached_lat, cached_lon # Use rate-limited Nominatim to geocode the zipcode location = (yield ("geocode", cache_query)) if location: # Cache the result for future use bot.db_manager.cache_geocoding(cache_query, location.latitude, location.longitude) return location.latitude, location.longitude else: return None, None except Exception as e: bot.logger.error(f"Error geocoding zipcode {zipcode}: {e}") return None, None async def geocode_zipcode(bot: Any, zipcode: str, default_country: Optional[str] = None, timeout: int = 10) -> tuple[Optional[float], Optional[float]]: """Shared function to geocode a ZIP code to lat/lon coordinates. Checks cache first, then makes rate-limited API call if needed. Args: bot: Bot instance (must have db_manager and nominatim_rate_limiter). zipcode: ZIP code string. default_country: Default country code (e.g., "US"). If None, reads from bot.config. timeout: Request timeout in seconds. Returns: Tuple[Optional[float], Optional[float]]: Tuple of (latitude, longitude) or (None, None) if not found. """ return await _run_geocode_steps_async(bot, _geocode_zipcode_steps(bot, zipcode, default_country), timeout) def geocode_zipcode_sync(bot: Any, zipcode: str, default_country: Optional[str] = None, timeout: int = 10) -> tuple[Optional[float], Optional[float]]: """Synchronous version of geocode_zipcode. Args: bot: Bot instance (must have db_manager and nominatim_rate_limiter). zipcode: ZIP code string. default_country: Default country code (e.g., "US"). If None, reads from bot.config. timeout: Request timeout in seconds. Returns: Tuple[Optional[float], Optional[float]]: Tuple of (latitude, longitude) or (None, None) if not found. """ return _run_geocode_steps_sync(bot, _geocode_zipcode_steps(bot, zipcode, default_country), timeout) def _geocode_city_steps(bot: Any, city: str, default_state: Optional[str], default_country: Optional[str], include_address_info: bool, *, async_rules: bool) -> _GeocodeSteps: """Shared body of ``geocode_city`` and ``geocode_city_sync``, as a step generator. The two have drifted on one rule, kept here behind ``async_rules``: the async variant always reverse-geocodes an unqualified city's first hit to judge whether it is an obscure place in the default country (city name and county both count, and either name containing the other is a match), and reuses that lookup for the address info; the sync variant only does so when address info was requested, with a stricter name match. """ try: # Get defaults from config if not provided if default_state is None: default_state = bot.config.get('Weather', 'default_state', fallback='') if default_country is None: default_country = bot.config.get('Weather', 'default_country', fallback='US') city_clean = city.strip() state_abbr = None country_name = None # Parse city, state/country format if present if ',' in city_clean: parts = [p.strip() for p in city_clean.rsplit(',', 1)] if len(parts) == 2: city_clean = parts[0] second_part = parts[1] # Use geocoding helpers to determine if it's a state or country try: _, parsed_part, part_type = parse_location_string(f"{city_clean}, {second_part}") if part_type == "state": state_abbr, _ = normalize_us_state(second_part) if not state_abbr: state_abbr = second_part.upper() if len(second_part) <= 2 else None elif part_type == "country": iso_code, country_name = normalize_country_name(second_part) if iso_code: # Use the normalized country name for better geocoding country_name = country_name else: country_name = second_part else: # Fallback to original logic if len(second_part) <= 2: state_abbr = second_part.upper() else: country_name = second_part except ImportError: # Fallback if helpers not available if len(second_part) <= 2: state_abbr = second_part.upper() else: country_name = second_part # Handle major cities with multiple locations (prioritize major cities). # Skip when user specified a country (e.g. "Paris, FR") so we honor their choice. major_city_queries = get_major_city_queries(city_clean, state_abbr) if major_city_queries and not country_name: # Try major city options first for major_city_query in major_city_queries: cached_lat, cached_lon = bot.db_manager.get_cached_geocoding(major_city_query) if cached_lat and cached_lon: lat, lon = cached_lat, cached_lon else: location = (yield ("geocode", major_city_query)) if location: bot.db_manager.cache_geocoding(major_city_query, location.latitude, location.longitude) lat, lon = location.latitude, location.longitude else: continue # Get address info if requested address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info # If country name was parsed (not a state abbreviation), try geocoding with country first if country_name: # Try with country name directly (e.g., "Stockholm, Sweden") country_query = f"{city_clean}, {country_name}" cached_lat, cached_lon = bot.db_manager.get_cached_geocoding(country_query) if cached_lat and cached_lon: lat, lon = cached_lat, cached_lon else: location = (yield ("geocode", country_query)) if location: bot.db_manager.cache_geocoding(country_query, location.latitude, location.longitude) lat, lon = location.latitude, location.longitude else: lat, lon = None, None if lat and lon: address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info # If state abbreviation was parsed, use it if state_abbr: state_query = f"{city_clean}, {state_abbr}, {default_country}" cached_lat, cached_lon = bot.db_manager.get_cached_geocoding(state_query) if cached_lat and cached_lon: lat, lon = cached_lat, cached_lon else: location = (yield ("geocode", state_query)) if location: bot.db_manager.cache_geocoding(state_query, location.latitude, location.longitude) lat, lon = location.latitude, location.longitude else: lat, lon = None, None if lat and lon: address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info # If no country/state specified, try city name alone first (finds most prominent international city) # This handles cases like "Tokyo" -> Tokyo, Japan (not Tokyo, WA) if not state_abbr and not country_name: location = (yield ("geocode", city_clean)) if location: # Check if result is in default country and is a small/obscure location # If so, we'll try with default country/state as fallback result_in_default_country = False is_obscure_location = False # The async variant always checks the result; the sync one only # when address info was requested. if async_rules or include_address_info: try: reverse_location = (yield ("reverse", f"{location.latitude}, {location.longitude}")) if reverse_location: address = reverse_location.raw.get('address', {}) result_country = address.get('country', '').upper() result_country_code = address.get('country_code', '').upper() # Check if result is in default country default_country_upper = default_country.upper() if (result_country == default_country_upper or result_country_code == default_country_upper or 'United States' in result_country and default_country_upper == 'US'): result_in_default_country = True # Check if it's an obscure location (county, township, small town) place_type = address.get('type', '').lower() if async_rules: place_name = (address.get('city') or address.get('town') or address.get('village') or address.get('municipality') or address.get('county', '')).lower() # Obscure if it's a county, township, or if neither name contains the other name_mismatch = bool(place_name and city_clean.lower() not in place_name and place_name not in city_clean.lower()) else: place_name = address.get('city') or address.get('town') or address.get('village') or '' # Obscure if it's a county, township, or if city name doesn't match name_mismatch = city_clean.lower() not in place_name.lower() if ('county' in place_type or 'township' in place_type or name_mismatch): is_obscure_location = True except: pass # If result is in default country and is obscure, skip it and try with default country/state if result_in_default_country and is_obscure_location: # Fall through to try with default country/state pass else: # Use the international result (either not in default country, or is a proper city match) bot.db_manager.cache_geocoding(city_clean, location.latitude, location.longitude) lat, lon = location.latitude, location.longitude address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: if not async_rules or not reverse_location: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info # Try with default state (fallback for US cities when no country specified). # Skip when default_state is empty (e.g. non-US default_country or key unset). if default_state and default_state.strip(): cache_query = f"{city_clean}, {default_state}, {default_country}" cached_lat, cached_lon = bot.db_manager.get_cached_geocoding(cache_query) if cached_lat and cached_lon: lat, lon = cached_lat, cached_lon else: location = (yield ("geocode", cache_query)) if location: bot.db_manager.cache_geocoding(cache_query, location.latitude, location.longitude) lat, lon = location.latitude, location.longitude else: lat, lon = None, None if lat and lon: address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info # Try without state location = (yield ("geocode", f"{city_clean}, {default_country}")) if location: bot.db_manager.cache_geocoding(f"{city_clean}, {default_country}", location.latitude, location.longitude) lat, lon = location.latitude, location.longitude address_info = None if include_address_info: # Check cache for reverse geocoding result reverse_cache_key = f"reverse_{lat}_{lon}" cached_address = bot.db_manager.get_cached_json(reverse_cache_key, "geolocation") if cached_address: address_info = cached_address else: try: reverse_location = (yield ("reverse", f"{lat}, {lon}")) if reverse_location: address_info = reverse_location.raw.get('address', {}) # Cache the reverse geocoding result bot.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720) except: address_info = {} return lat, lon, address_info return None, None, None except Exception as e: bot.logger.error(f"Error geocoding city {city}: {e}") return None, None, None async def geocode_city(bot: Any, city: str, default_state: Optional[str] = None, default_country: Optional[str] = None, include_address_info: bool = False, timeout: int = 10) -> tuple[Optional[float], Optional[float], Optional[dict]]: """Shared function to geocode a city name to lat/lon coordinates. Uses intelligent fallback logic with major city prioritization. Args: bot: Bot instance (must have db_manager and nominatim_rate_limiter). city: City name (may include state/country, e.g., "Seattle, WA" or "Paris, France"). default_state: Default state abbreviation (e.g., "WA"). If None, reads from bot.config. default_country: Default country code (e.g., "US"). If None, reads from bot.config. include_address_info: If True, also return address info via reverse geocoding. timeout: Request timeout in seconds. Returns: Tuple[Optional[float], Optional[float], Optional[Dict]]: Tuple of (latitude, longitude, address_info_dict) or (None, None, None) if not found. address_info_dict is None if include_address_info is False. """ return await _run_geocode_steps_async( bot, _geocode_city_steps(bot, city, default_state, default_country, include_address_info, async_rules=True), timeout, ) def geocode_city_sync(bot: Any, city: str, default_state: Optional[str] = None, default_country: Optional[str] = None, include_address_info: bool = False, timeout: int = 10) -> tuple[Optional[float], Optional[float], Optional[dict]]: """Synchronous version of geocode_city. Args: bot: Bot instance (must have db_manager and nominatim_rate_limiter). city: City name (may include state/country, e.g., "Seattle, WA" or "Paris, France"). default_state: Default state abbreviation (e.g., "WA"). If None, reads from bot.config. default_country: Default country code (e.g., "US"). If None, reads from bot.config. include_address_info: If True, also return address info via reverse geocoding. timeout: Request timeout in seconds. Returns: Tuple[Optional[float], Optional[float], Optional[Dict]]: Tuple of (latitude, longitude, address_info_dict) or (None, None, None) if not found. address_info_dict is None if include_address_info is False. """ return _run_geocode_steps_sync( bot, _geocode_city_steps(bot, city, default_state, default_country, include_address_info, async_rules=False), timeout, ) def resolve_path(file_path: Union[str, Path], base_dir: Union[str, Path] = '.') -> str: """Resolve a file path relative to a base directory. If the path is absolute, it is returned as-is (no symlink/canonical resolution). If the path is relative, it is resolved relative to the base directory. Args: file_path: Path to resolve (can be string or Path object). base_dir: Base directory for resolving relative paths (default: current directory). Returns: str: Resolved absolute path as a string. Examples: >>> resolve_path('data.db', '/opt/bot') '/opt/bot/data.db' >>> resolve_path('/var/lib/bot/data.db', '/opt/bot') '/var/lib/bot/data.db' """ file_path = Path(file_path) if not isinstance(file_path, Path) else file_path base_dir = Path(base_dir) if not isinstance(base_dir, Path) else base_dir if file_path.is_absolute(): # Important on macOS: `/var/...` may be a symlink to `/private/var/...`. # Tests (and callers) expect the absolute path string to stay stable. return str(file_path) else: return str((base_dir.resolve() / file_path).resolve()) def check_internet_connectivity(host: str = "8.8.8.8", port: int = 53, timeout: float = 3.0) -> bool: """Check if internet connectivity is available by attempting to connect to a reliable host. First tries a lightweight DNS port check (faster, doesn't require DNS resolution). If that fails (e.g., DNS port is blocked), falls back to an HTTP request check. Args: host: Host to connect to (default: 8.8.8.8, Google's public DNS). port: Port to connect to (default: 53, DNS port). timeout: Connection timeout in seconds (default: 3.0). Returns: bool: True if connection successful, False otherwise. """ # First try: DNS port check (fastest, works if DNS port is open) try: socket.setdefaulttimeout(timeout) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, port)) sock.close() socket.setdefaulttimeout(None) # Reset to default return True except (OSError, socket.timeout): socket.setdefaulttimeout(None) # Reset to default # DNS check failed, try HTTP fallback pass # Fallback: HTTP request check (works even if DNS port is blocked) try: # Use a reliable HTTP endpoint that's likely to be accessible # Using IP address to avoid DNS resolution issues http_url = "http://1.1.1.1" # Cloudflare DNS urllib.request.urlopen(http_url, timeout=timeout).close() return True except (urllib.error.URLError, OSError, socket.timeout): # If IP-based check fails, try a hostname-based check try: http_url = "http://www.google.com" urllib.request.urlopen(http_url, timeout=timeout).close() return True except (urllib.error.URLError, OSError, socket.timeout): return False async def check_internet_connectivity_async(host: str = "8.8.8.8", port: int = 53, timeout: float = 3.0) -> bool: """Async version of check_internet_connectivity. First tries a lightweight DNS port check (faster, doesn't require DNS resolution). If that fails (e.g., DNS port is blocked), falls back to an HTTP request check. Args: host: Host to connect to (default: 8.8.8.8, Google's public DNS). port: Port to connect to (default: 53, DNS port). timeout: Connection timeout in seconds (default: 3.0). Returns: bool: True if connection successful, False otherwise. """ # First try: DNS port check (fastest, works if DNS port is open) try: reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port), timeout=timeout ) writer.close() await writer.wait_closed() return True except (asyncio.TimeoutError, OSError, ConnectionError): # DNS check failed, try HTTP fallback pass except Exception: # Unexpected error, try HTTP fallback pass # Fallback: HTTP request check (works even if DNS port is blocked) # Run urllib in executor to avoid blocking loop = asyncio.get_event_loop() try: # Use a reliable HTTP endpoint that's likely to be accessible # Using IP address to avoid DNS resolution issues http_url = "http://1.1.1.1" # Cloudflare DNS await asyncio.wait_for( loop.run_in_executor( None, lambda: urllib.request.urlopen(http_url, timeout=timeout).close() ), timeout=timeout ) return True except (asyncio.TimeoutError, urllib.error.URLError, OSError, socket.timeout): # If IP-based check fails, try a hostname-based check try: http_url = "http://www.google.com" await asyncio.wait_for( loop.run_in_executor( None, lambda: urllib.request.urlopen(http_url, timeout=timeout).close() ), timeout=timeout ) return True except (asyncio.TimeoutError, urllib.error.URLError, OSError, socket.timeout): return False except Exception: return False # Maximum plausible elapsed ms (5 minutes) for device clock validation. # Values above indicate device time is far in the past (e.g. epoch); negative = in the future. _ELAPSED_MS_MAX = 5 * 60 * 1000 # 5 minutes in milliseconds def format_elapsed_display(ts: Any, translator: Any = None) -> str: """Format elapsed time from sender timestamp for {elapsed} placeholder. Returns "Nms" when valid, or the i18n "Sync Device Clock" when the device clock is invalid (e.g. T-Deck before GPS sync: 0, future, or far in the past). Args: ts: Sender timestamp (int, float, None, or 'unknown'). translator: Bot translator for i18n; uses "Sync Device Clock" if None. Returns: str: e.g. "1234ms" or translated "Sync Device Clock". """ def _sync_str() -> str: if translator: return translator.translate('elapsed.sync_device_clock') return "Sync Device Clock" if ts is None or ts == 'unknown': return _sync_str() try: ts_f = float(ts) except (TypeError, ValueError): return _sync_str() from datetime import datetime, timezone UTC = timezone.utc elapsed_ms = (datetime.now(UTC).timestamp() - ts_f) * 1000 if elapsed_ms < 0 or elapsed_ms > _ELAPSED_MS_MAX: return _sync_str() if elapsed_ms >= 1000: return f"{round(elapsed_ms / 1000, 1)}s" return f"{round(elapsed_ms)}ms" def format_keyword_response_with_placeholders( response_format: str, message: Any, bot: Any, mesh_info: Optional[dict[str, Any]] = None ) -> str: """Format a keyword response string with all available placeholders. Supports both message-based placeholders and mesh-info-based placeholders. This is a shared function used by both Keywords and Scheduled_Messages. Args: response_format: Response format string with placeholders. message: MeshMessage instance (can be None for scheduled messages). bot: Bot instance (must have config, db_manager). mesh_info: Optional mesh network info dict (for scheduled message placeholders). Returns: str: Formatted response string. """ try: replacements = {} # Message-based placeholders (require message object) if message: # Basic message fields replacements['sender'] = message.sender_id or "Unknown" replacements['path'] = message.path or "Unknown" replacements['snr'] = message.snr or "Unknown" replacements['rssi'] = message.rssi or "Unknown" # Compute elapsed from message.timestamp (same as TestCommand) so it's available # for all keywords. Using message.elapsed would miss when it's unset on some paths. _translator = getattr(bot, 'translator', None) replacements['elapsed'] = format_elapsed_display( getattr(message, 'timestamp', None), _translator ) # Build connection_info routing_info = message.path or "Unknown routing" if "via ROUTE_TYPE_" in routing_info: parts = routing_info.split(" via ROUTE_TYPE_") if len(parts) > 0: routing_info = parts[0] snr_info = f"SNR: {message.snr or 'Unknown'} dB" rssi_info = f"RSSI: {message.rssi or 'Unknown'} dBm" connection_info = f"{routing_info} | {snr_info} | {rssi_info}" replacements['connection_info'] = connection_info # Calculate path distances path_distance, firstlast_distance = calculate_path_distances( bot, message.path or "", message=message ) replacements['path_distance'] = path_distance replacements['firstlast_distance'] = firstlast_distance # Format timestamp try: tz, _ = get_config_timezone(bot.config, getattr(bot, 'logger', None)) dt = datetime.now(tz) time_str = dt.strftime("%H:%M:%S") except Exception: time_str = "Unknown" replacements['timestamp'] = time_str replacements['packet_hash'] = get_packet_hash_placeholder(message) # Shared with BaseCommand.get_hops_display_values and the hops_min filter, so # a keyword response and a command response report the same hop count for the # same packet. "?" only when the count cannot be determined at all. hops_val = message_hop_count(message) if hops_val is None: replacements['hops'] = "?" replacements['hops_label'] = "?" else: replacements['hops'] = str(hops_val) replacements['hops_label'] = "1 hop" if hops_val == 1 else f"{hops_val} hops" else: # No message - use defaults for message-based placeholders replacements['sender'] = "Unknown" replacements['path'] = "Unknown" replacements['snr'] = "Unknown" replacements['rssi'] = "Unknown" replacements['elapsed'] = "Unknown" replacements['connection_info'] = "Unknown" replacements['path_distance'] = "" replacements['firstlast_distance'] = "" replacements['timestamp'] = "Unknown" replacements['packet_hash'] = "" replacements['hops'] = "?" replacements['hops_label'] = "?" # Mesh-info-based placeholders (from scheduled messages) if mesh_info: replacements.update({ 'total_contacts': mesh_info.get('total_contacts', 0), 'total_repeaters': mesh_info.get('total_repeaters', 0), 'total_companions': mesh_info.get('total_companions', 0), 'total_roomservers': mesh_info.get('total_roomservers', 0), 'total_sensors': mesh_info.get('total_sensors', 0), 'recent_activity_24h': mesh_info.get('recent_activity_24h', 0), 'new_companions_7d': mesh_info.get('new_companions_7d', 0), 'new_repeaters_7d': mesh_info.get('new_repeaters_7d', 0), 'new_roomservers_7d': mesh_info.get('new_roomservers_7d', 0), 'new_sensors_7d': mesh_info.get('new_sensors_7d', 0), 'total_contacts_30d': mesh_info.get('total_contacts_30d', 0), 'total_repeaters_30d': mesh_info.get('total_repeaters_30d', 0), 'total_companions_30d': mesh_info.get('total_companions_30d', 0), 'total_roomservers_30d': mesh_info.get('total_roomservers_30d', 0), 'total_sensors_30d': mesh_info.get('total_sensors_30d', 0), # Legacy placeholders 'repeaters': mesh_info.get('total_repeaters', 0), 'companions': mesh_info.get('total_companions', 0), }) else: # No mesh_info - use defaults mesh_defaults = { 'total_contacts': 0, 'total_repeaters': 0, 'total_companions': 0, 'total_roomservers': 0, 'total_sensors': 0, 'recent_activity_24h': 0, 'new_companions_7d': 0, 'new_repeaters_7d': 0, 'new_roomservers_7d': 0, 'new_sensors_7d': 0, 'total_contacts_30d': 0, 'total_repeaters_30d': 0, 'total_companions_30d': 0, 'total_roomservers_30d': 0, 'total_sensors_30d': 0, 'repeaters': 0, 'companions': 0, } replacements.update(mesh_defaults) # Format the response with all replacements return response_format.format(**replacements) except (KeyError, ValueError) as e: # If formatting fails, return as-is (might not have all placeholders) if hasattr(bot, 'logger'): bot.logger.debug(f"Error formatting response with placeholders: {e}") return response_format