dandri/meshcore-bot
1
0 Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-11-04 03:52:34 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
885a61742fc0248f4dcdb48872576f3aebb860ec
meshcore-bot/modules/utils.py
T
agessaman 885a61742f refactor(utils): move packet framing and advert signature helpers to the decoders 
decode_path_len_byte, encode_path_len_byte, parse_trace_payload_route_hashes and calculate_packet_hash move verbatim to packet_decode.py, which already handled packet framing and imported decode_path_len_byte from utils; verify_meshcore_advert_ed25519 moves next to the advert parser in meshcore_payload_decode.py. utils re-exports all five as the same objects, and no test patched them. get_packet_hash_placeholder stays in utils with the response template it serves. utils.py drops to 1,630 lines.
2026-10-01 14:28:06 -07:00

1631 lines
65 KiB
Python
Raw Blame History

#!/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
Reference in new issue View Git Blame Copy Permalink