mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-08-27 21:10:13 +00:00
Merge pull request #224 from agessaman/refactor/consolidate-location-tools
Enhance feed management and location resolution features
This commit is contained in:
Adam Gessaman
@@ -124,11 +124,35 @@ for i, chunk in enumerate(chunks):
|
||||
|
||||
So you are allowed to send multiple messages in sequence; you do **not** need 10 seconds between chunks. Use `skip_user_rate_limit=True` and about 1–1.5 seconds (or your configured `bot_tx_rate_limit_seconds` + buffer) between chunks.
|
||||
|
||||
## Resolving locations (shared API)
|
||||
|
||||
For place lookup (coords, ZIP, city, neighborhoods, optional repeater names), use **`modules.location`** rather than calling Nominatim directly:
|
||||
|
||||
```python
|
||||
from modules.location import OPTIONS_AQI, classify_location, resolve_location
|
||||
|
||||
# Pure classify (no network):
|
||||
normalized, location_type = classify_location("mexico city") # -> ("mexico city, mexico", "city")
|
||||
|
||||
# Best-effort resolve (uses bot Nominatim rate limiter + geocode cache):
|
||||
resolved = resolve_location(self.bot, "seattle", options=OPTIONS_AQI)
|
||||
if resolved.error:
|
||||
...
|
||||
else:
|
||||
lat, lon = resolved.lat, resolved.lon
|
||||
label = resolved.display_name
|
||||
```
|
||||
|
||||
**Presets** (`OPTIONS_AQI`, `OPTIONS_WX`, `OPTIONS_AURORA`, `OPTIONS_RAIN`, `OPTIONS_PREFIX`, `OPTIONS_SOLARFORECAST`) encode command-specific semantics via `ResolveOptions` flags (intl cities, neighborhoods, structured ZIP, region capitals, Zippopotam labels, repeater names, label style). Low-level helpers remain in **`modules.utils`** (`geocode_city_sync`, `geocode_zipcode_sync`, rate-limited Nominatim). Alert-style agency/street/county parsing stays command-local; only the geocode subset should call `resolve_location`.
|
||||
|
||||
Built-in commands other than AQI will migrate onto this API over time; new plugins should prefer it now.
|
||||
|
||||
## References
|
||||
|
||||
- [Service plugins](service-plugins.md) — built-in services and how they are enabled.
|
||||
- [Check-in API](checkin-api.md) — contract for the optional check-in submission API (local check-in service).
|
||||
- Built-in command plugins live in **modules/commands/** and **modules/commands/alternatives/**; you can use them as examples for `BaseCommand`, `get_config_value`, `handle_keyword_match`, etc.
|
||||
- Location helpers: **modules/location.py** (high-level), **modules/utils.py** (Nominatim / geocode_*).
|
||||
- Base classes: **modules/commands/base_command.py** (`BaseCommand`), **modules/service_plugins/base_service.py** (`BaseServicePlugin`).
|
||||
|
||||
## Check-in service (local)
|
||||
|
||||
+90
-587
@@ -4,21 +4,27 @@ AQI command for the MeshCore Bot
|
||||
Provides Air Quality Index information using OpenMeteo API
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
import openmeteo_requests
|
||||
import requests_cache
|
||||
from retry_requests import retry
|
||||
|
||||
from ..location import (
|
||||
OPTIONS_AQI,
|
||||
ResolveOptions,
|
||||
geocode_city_best_effort,
|
||||
resolve_location,
|
||||
)
|
||||
from ..location import (
|
||||
get_neighborhood_queries as location_neighborhood_queries,
|
||||
)
|
||||
from ..models import MeshMessage
|
||||
from ..utils import (
|
||||
abbreviate_location,
|
||||
geocode_city_sync,
|
||||
geocode_zipcode_sync,
|
||||
get_nominatim_geocoder,
|
||||
is_valid_timezone,
|
||||
rate_limited_nominatim_geocode_sync,
|
||||
rate_limited_nominatim_reverse_sync,
|
||||
normalize_us_state,
|
||||
)
|
||||
from .base_command import BaseCommand
|
||||
|
||||
@@ -177,156 +183,12 @@ class AqiCommand(BaseCommand):
|
||||
await self.send_response(message, self.astronomical_responses[location_lower])
|
||||
return True
|
||||
|
||||
# Check if it's lat,lon coordinates (decimal numbers separated by comma, with optional spaces)
|
||||
# Handle formats like: "47.6,-122.3", "47.6, -122.3", "47.980525, -122.150649", " -47.6 , 122.3 "
|
||||
if re.match(r'^\s*-?\d+\.?\d*\s*,\s*-?\d+\.?\d*\s*$', location):
|
||||
location_type = "coordinates"
|
||||
# Check if it's a US ZIP code (5 digits)
|
||||
elif re.match(r'^\s*\d{5}\s*$', location):
|
||||
location_type = "zipcode"
|
||||
# Keep the original ZIP code for structured queries
|
||||
# Don't modify the location string here - let the geocoding logic handle it
|
||||
else:
|
||||
# It's a city name (possibly with state/country)
|
||||
# Check if it might be "city country" format (space-separated)
|
||||
location_parts = location.split()
|
||||
if len(location_parts) >= 2:
|
||||
potential_city = location_parts[0]
|
||||
potential_country = location_parts[1]
|
||||
country_indicators = ['canada', 'mexico', 'uk', 'united', 'kingdom', 'france', 'germany', 'italy', 'spain', 'australia', 'japan', 'china', 'india', 'brazil']
|
||||
|
||||
# Check if second word is a country indicator
|
||||
if potential_country.lower() in country_indicators:
|
||||
# Convert space-separated to comma-separated format
|
||||
if potential_country.lower() in ['united', 'kingdom']:
|
||||
# Handle "united kingdom" case
|
||||
if len(location_parts) >= 3 and location_parts[2].lower() == 'kingdom':
|
||||
location = f"{potential_city}, uk"
|
||||
else:
|
||||
location = f"{potential_city}, {potential_country}"
|
||||
else:
|
||||
location = f"{potential_city}, {potential_country}"
|
||||
else:
|
||||
# Single word city - check if it's a well-known international city
|
||||
international_cities = {
|
||||
'beijing': 'beijing, china',
|
||||
'shanghai': 'shanghai, china',
|
||||
'tokyo': 'tokyo, japan',
|
||||
'london': 'london, uk',
|
||||
'paris': 'paris, france',
|
||||
'berlin': 'berlin, germany',
|
||||
'rome': 'rome, italy',
|
||||
'madrid': 'madrid, spain',
|
||||
'moscow': 'moscow, russia',
|
||||
'sydney': 'sydney, australia',
|
||||
'melbourne': 'melbourne, australia',
|
||||
'toronto': 'toronto, canada',
|
||||
'vancouver': 'vancouver, canada',
|
||||
'mumbai': 'mumbai, india',
|
||||
'delhi': 'delhi, india',
|
||||
'bangalore': 'bangalore, india',
|
||||
'sao paulo': 'sao paulo, brazil',
|
||||
'rio de janeiro': 'rio de janeiro, brazil',
|
||||
'mexico city': 'mexico city, mexico',
|
||||
'cairo': 'cairo, egypt',
|
||||
'istanbul': 'istanbul, turkey',
|
||||
'seoul': 'seoul, south korea',
|
||||
'bangkok': 'bangkok, thailand',
|
||||
'singapore': 'singapore, singapore',
|
||||
'hong kong': 'hong kong, china',
|
||||
'dubai': 'dubai, uae',
|
||||
'tel aviv': 'tel aviv, israel',
|
||||
'johannesburg': 'johannesburg, south africa',
|
||||
'nairobi': 'nairobi, kenya',
|
||||
'lagos': 'lagos, nigeria',
|
||||
'buenos aires': 'buenos aires, argentina',
|
||||
'lima': 'lima, peru',
|
||||
'santiago': 'santiago, chile',
|
||||
'bogota': 'bogota, colombia',
|
||||
'caracas': 'caracas, venezuela',
|
||||
'havana': 'havana, cuba',
|
||||
'kingston': 'kingston, jamaica',
|
||||
'san juan': 'san juan, puerto rico',
|
||||
'reykjavik': 'reykjavik, iceland',
|
||||
'oslo': 'oslo, norway',
|
||||
'stockholm': 'stockholm, sweden',
|
||||
'copenhagen': 'copenhagen, denmark',
|
||||
'helsinki': 'helsinki, finland',
|
||||
'warsaw': 'warsaw, poland',
|
||||
'prague': 'prague, czech republic',
|
||||
'budapest': 'budapest, hungary',
|
||||
'bucharest': 'bucharest, romania',
|
||||
'sofia': 'sofia, bulgaria',
|
||||
'zagreb': 'zagreb, croatia',
|
||||
'belgrade': 'belgrade, serbia',
|
||||
'athens': 'athens, greece',
|
||||
'lisbon': 'lisbon, portugal',
|
||||
'dublin': 'dublin, ireland',
|
||||
'brussels': 'brussels, belgium',
|
||||
'amsterdam': 'amsterdam, netherlands',
|
||||
'zurich': 'zurich, switzerland',
|
||||
'vienna': 'vienna, austria',
|
||||
'lucerne': 'lucerne, switzerland',
|
||||
'geneva': 'geneva, switzerland',
|
||||
'monaco': 'monaco, monaco',
|
||||
'andorra': 'andorra, andorra',
|
||||
'san marino': 'san marino, san marino',
|
||||
'vatican': 'vatican city, vatican',
|
||||
'luxembourg': 'luxembourg, luxembourg',
|
||||
'malta': 'valletta, malta',
|
||||
'cyprus': 'nicosia, cyprus',
|
||||
'albania': 'tirana, albania',
|
||||
'macedonia': 'skopje, macedonia',
|
||||
'montenegro': 'podgorica, montenegro',
|
||||
'bosnia': 'sarajevo, bosnia',
|
||||
'slovenia': 'ljubljana, slovenia',
|
||||
'slovakia': 'bratislava, slovakia',
|
||||
'lithuania': 'vilnius, lithuania',
|
||||
'latvia': 'riga, latvia',
|
||||
'estonia': 'tallinn, estonia',
|
||||
'belarus': 'minsk, belarus',
|
||||
'ukraine': 'kiev, ukraine',
|
||||
'moldova': 'chisinau, moldova',
|
||||
'georgia': 'tbilisi, georgia',
|
||||
'armenia': 'yerevan, armenia',
|
||||
'azerbaijan': 'baku, azerbaijan',
|
||||
'kazakhstan': 'almaty, kazakhstan',
|
||||
'uzbekistan': 'tashkent, uzbekistan',
|
||||
'kyrgyzstan': 'bishkek, kyrgyzstan',
|
||||
'tajikistan': 'dushanbe, tajikistan',
|
||||
'turkmenistan': 'ashgabat, turkmenistan',
|
||||
'afghanistan': 'kabul, afghanistan',
|
||||
'pakistan': 'islamabad, pakistan',
|
||||
'bangladesh': 'dhaka, bangladesh',
|
||||
'sri lanka': 'colombo, sri lanka',
|
||||
'nepal': 'kathmandu, nepal',
|
||||
'bhutan': 'thimphu, bhutan',
|
||||
'myanmar': 'yangon, myanmar',
|
||||
'laos': 'vientiane, laos',
|
||||
'cambodia': 'phnom penh, cambodia',
|
||||
'vietnam': 'hanoi, vietnam',
|
||||
'malaysia': 'kuala lumpur, malaysia',
|
||||
'indonesia': 'jakarta, indonesia',
|
||||
'philippines': 'manila, philippines',
|
||||
'taiwan': 'taipei, taiwan',
|
||||
'north korea': 'pyongyang, north korea',
|
||||
'mongolia': 'ulaanbaatar, mongolia',
|
||||
'kazakhstan': 'nur-sultan, kazakhstan'
|
||||
}
|
||||
|
||||
# Check if it's a known international city
|
||||
city_lower = location.lower()
|
||||
if city_lower in international_cities:
|
||||
location = international_cities[city_lower]
|
||||
|
||||
location_type = "city"
|
||||
|
||||
try:
|
||||
# Record execution for this user
|
||||
self.record_execution(message.sender_id)
|
||||
|
||||
# Get AQI data for the location
|
||||
aqi_data = await self.get_aqi_for_location(location, location_type)
|
||||
# Get AQI data for the location (single resolve with intl rewrite)
|
||||
aqi_data = await self.get_aqi_for_location(location)
|
||||
|
||||
# Send the response
|
||||
await self.send_response(message, aqi_data)
|
||||
@@ -337,304 +199,88 @@ class AqiCommand(BaseCommand):
|
||||
await self.send_response(message, f"Error getting AQI data: {e}")
|
||||
return True
|
||||
|
||||
async def get_aqi_for_location(self, location: str, location_type: str) -> str:
|
||||
"""Get AQI data for a location (city or coordinates).
|
||||
def _resolved_state_differs_from_default(self, address_info: Optional[dict]) -> bool:
|
||||
"""True when reverse-geocode state/country differs from bot default_state."""
|
||||
if not address_info or not self.default_state:
|
||||
return False
|
||||
country = address_info.get("country", "")
|
||||
state = address_info.get("state", "")
|
||||
default_abbr, default_full = normalize_us_state(self.default_state)
|
||||
defaults = {d for d in (self.default_state, default_abbr, default_full) if d}
|
||||
if country in ("United States", "US", "United States of America"):
|
||||
abbr, full = normalize_us_state(state) if state else (None, None)
|
||||
actuals = {a for a in (abbr, full, state) if a}
|
||||
return bool(actuals) and actuals.isdisjoint(defaults)
|
||||
actual_state = country or address_info.get("province") or ""
|
||||
return bool(actual_state) and actual_state not in defaults
|
||||
|
||||
async def get_aqi_for_location(
|
||||
self, location: str, location_type: Optional[str] = None
|
||||
) -> str:
|
||||
"""Get AQI data for a location (city, ZIP, or coordinates).
|
||||
|
||||
Args:
|
||||
location: Location string (city name, ZIP, or "lat,lon").
|
||||
location_type: Type of location ("city", "zipcode", "coordinates").
|
||||
location: Raw location string (city name, ZIP, or "lat,lon").
|
||||
location_type: Unused; kept for call-site/test compatibility.
|
||||
|
||||
Returns:
|
||||
str: Formatted AQI string or error message.
|
||||
"""
|
||||
try:
|
||||
# Define state abbreviation map for US states (needed for all location types)
|
||||
state_abbrev_map = {
|
||||
'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'
|
||||
}
|
||||
# Convert location to lat/lon
|
||||
if location_type == "coordinates":
|
||||
# Parse lat,lon coordinates
|
||||
try:
|
||||
lat_str, lon_str = location.split(',')
|
||||
lat = float(lat_str.strip())
|
||||
lon = float(lon_str.strip())
|
||||
opts = ResolveOptions(
|
||||
default_state=self.default_state,
|
||||
default_country=self.default_country,
|
||||
use_international_cities=OPTIONS_AQI.use_international_cities,
|
||||
use_neighborhoods=OPTIONS_AQI.use_neighborhoods,
|
||||
use_structured_zip=OPTIONS_AQI.use_structured_zip,
|
||||
label_style=OPTIONS_AQI.label_style,
|
||||
include_address_info=True,
|
||||
timeout=10,
|
||||
)
|
||||
resolved = resolve_location(self.bot, location, options=opts)
|
||||
|
||||
# Validate coordinate ranges
|
||||
if not (-90 <= lat <= 90):
|
||||
return f"Invalid latitude: {lat}. Must be between -90 and 90."
|
||||
if not (-180 <= lon <= 180):
|
||||
return f"Invalid longitude: {lon}. Must be between -180 and 180."
|
||||
if resolved.error == "invalid_latitude":
|
||||
detail = resolved.error_detail or location
|
||||
return f"Invalid latitude: {detail}. Must be between -90 and 90."
|
||||
if resolved.error == "invalid_longitude":
|
||||
detail = resolved.error_detail or location
|
||||
return f"Invalid longitude: {detail}. Must be between -180 and 180."
|
||||
if resolved.error == "invalid_coordinates":
|
||||
return f"Invalid coordinates format: {location}. Use format: lat,lon (e.g., 47.6,-122.3)"
|
||||
if resolved.error == "no_location_zipcode":
|
||||
return f"Could not find ZIP code '{location.strip()}'"
|
||||
if resolved.error == "no_location_city":
|
||||
if "," in (resolved.query or location):
|
||||
return f"Could not find city '{resolved.query or location}'"
|
||||
region = self.default_state or self.default_country
|
||||
return f"Could not find city '{resolved.query or location}' in {region}"
|
||||
if resolved.lat is None or resolved.lon is None:
|
||||
return f"Could not find location '{location}'"
|
||||
|
||||
address_info = None
|
||||
except ValueError:
|
||||
return f"Invalid coordinates format: {location}. Use format: lat,lon (e.g., 47.6,-122.3)"
|
||||
elif location_type == "zipcode":
|
||||
# Handle ZIP code geocoding with AQI-specific structured queries
|
||||
try:
|
||||
zip_code = location.strip()
|
||||
lat, lon = resolved.lat, resolved.lon
|
||||
address_info = resolved.address_info
|
||||
location_type = resolved.location_type or location_type
|
||||
|
||||
# Check for known problematic ZIP codes that need specific mapping
|
||||
zip_code_mappings = {
|
||||
'98013': 'Vashon, WA, USA',
|
||||
'98014': 'Vashon Island, WA, USA',
|
||||
# Add other problematic ZIP codes here as needed
|
||||
}
|
||||
|
||||
location_result = None
|
||||
|
||||
# First check if we have a specific mapping for this ZIP code
|
||||
if zip_code in zip_code_mappings:
|
||||
mapped_location = zip_code_mappings[zip_code]
|
||||
self.logger.debug(f"Using specific mapping for ZIP {zip_code}: {mapped_location}")
|
||||
try:
|
||||
result = rate_limited_nominatim_geocode_sync(self.bot, mapped_location, timeout=10)
|
||||
if result and result.address:
|
||||
location_result = result
|
||||
self.logger.debug(f"Found mapped location for ZIP {zip_code}: {result.address}")
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Mapping failed for ZIP {zip_code}: {e}")
|
||||
|
||||
# If no mapping, try structured queries (AQI-specific feature for better ZIP handling)
|
||||
if not location_result:
|
||||
structured_queries = [
|
||||
# Direct postalcode search
|
||||
{"postalcode": zip_code, "country": "US"},
|
||||
# Postalcode with state
|
||||
{"postalcode": zip_code, "state": self.default_state, "country": "US"},
|
||||
# Postalcode with country code
|
||||
{"postalcode": zip_code, "countrycode": "US"},
|
||||
]
|
||||
|
||||
for query in structured_queries:
|
||||
try:
|
||||
result = rate_limited_nominatim_geocode_sync(self.bot, query, timeout=10)
|
||||
if result and result.address:
|
||||
# Check if it's a US location
|
||||
if 'united states' in result.address.lower() or 'usa' in result.address.lower():
|
||||
# Additional validation: check if it's in the expected state
|
||||
if self.default_state in result.address or 'washington' in result.address.lower():
|
||||
location_result = result
|
||||
self.logger.debug(f"Found US location in {self.default_state} for ZIP {zip_code}: {result.address}")
|
||||
break
|
||||
else:
|
||||
# Found US location but not in expected state - log warning but continue searching
|
||||
self.logger.warning(f"ZIP {zip_code} found in wrong state: {result.address}")
|
||||
if not location_result: # Keep as fallback if no better result found
|
||||
location_result = result
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Structured query failed for {query}: {e}")
|
||||
continue
|
||||
|
||||
# If structured queries didn't work, use shared function as fallback
|
||||
if not location_result:
|
||||
lat, lon = geocode_zipcode_sync(self.bot, zip_code, timeout=10)
|
||||
if lat and lon:
|
||||
# Use the shared function result directly
|
||||
pass
|
||||
else:
|
||||
lat, lon = None, None
|
||||
else:
|
||||
lat = location_result.latitude
|
||||
lon = location_result.longitude
|
||||
|
||||
if lat and lon:
|
||||
|
||||
# Get detailed address info via reverse geocoding (check cache first)
|
||||
reverse_cache_key = f"reverse_{lat}_{lon}"
|
||||
cached_address = self.db_manager.get_cached_json(reverse_cache_key, "geolocation")
|
||||
if cached_address:
|
||||
address_info = cached_address
|
||||
else:
|
||||
try:
|
||||
reverse_location = rate_limited_nominatim_reverse_sync(self.bot, f"{lat}, {lon}", timeout=10)
|
||||
if reverse_location:
|
||||
address_info = reverse_location.raw.get('address', {})
|
||||
# Cache the reverse geocoding result
|
||||
self.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720)
|
||||
else:
|
||||
address_info = {}
|
||||
except:
|
||||
address_info = {}
|
||||
|
||||
# Validate that the found location makes sense for the ZIP code
|
||||
if address_info:
|
||||
found_state = address_info.get('state', '')
|
||||
found_country = address_info.get('country', '')
|
||||
|
||||
# If we found a location but it's not in the expected state, warn the user
|
||||
if found_country == 'United States' and found_state != self.default_state:
|
||||
self.logger.warning(f"ZIP code {zip_code} found in {found_state} instead of {self.default_state}")
|
||||
else:
|
||||
lat, lon = None, None
|
||||
address_info = None
|
||||
|
||||
if lat is None or lon is None:
|
||||
return f"Could not find ZIP code '{zip_code}'"
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error geocoding ZIP code {location}: {e}")
|
||||
return f"Error geocoding ZIP code: {e}"
|
||||
else: # city
|
||||
|
||||
result = self.city_to_lat_lon(location)
|
||||
if len(result) == 3:
|
||||
lat, lon, address_info = result
|
||||
else:
|
||||
lat, lon = result
|
||||
address_info = None
|
||||
|
||||
if lat is None or lon is None:
|
||||
# Check if it's an international city to provide better error message
|
||||
if ',' in location and any(country in location.lower() for country in ['canada', 'mexico', 'uk', 'france', 'germany', 'italy', 'spain', 'australia', 'japan', 'china', 'india', 'brazil', 'uae', 'russia', 'korea', 'thailand', 'singapore', 'egypt', 'turkey']):
|
||||
return f"Could not find city '{location}'"
|
||||
else:
|
||||
region = self.default_state or self.default_country
|
||||
return f"Could not find city '{location}' in {region}"
|
||||
|
||||
# Check if the found city is in a different state than default
|
||||
actual_city = location
|
||||
actual_state = self.default_state
|
||||
|
||||
if address_info:
|
||||
# Try to get the best city name from various address fields
|
||||
actual_city = (address_info.get('city') or
|
||||
address_info.get('town') or
|
||||
address_info.get('village') or
|
||||
address_info.get('hamlet') or
|
||||
address_info.get('municipality') or
|
||||
location)
|
||||
|
||||
# Get state/province/country info - handle US vs international addresses
|
||||
country = address_info.get('country', '')
|
||||
state = address_info.get('state', '')
|
||||
|
||||
# For US cities, use the state; for international cities, use the country
|
||||
if country == "United States" or country == "US" or country == "United States of America":
|
||||
# US city - use the state
|
||||
actual_state = state or self.default_state
|
||||
# Convert full state name to abbreviation if needed
|
||||
if len(actual_state) > 2 and actual_state in state_abbrev_map:
|
||||
actual_state = state_abbrev_map.get(actual_state, actual_state)
|
||||
else:
|
||||
# International city - use the country
|
||||
actual_state = (country or
|
||||
address_info.get('province') or
|
||||
self.default_state)
|
||||
# Normalize "United States" variants to "USA" to save characters
|
||||
if actual_state == "United States" or actual_state == "United States of America":
|
||||
actual_state = "USA"
|
||||
|
||||
# Also check if the default state needs to be converted for comparison
|
||||
default_state_full = self.default_state
|
||||
if len(self.default_state) == 2:
|
||||
# Convert abbreviation to full name for comparison
|
||||
abbrev_to_full_map = {v: k for k, v in state_abbrev_map.items()}
|
||||
default_state_full = abbrev_to_full_map.get(self.default_state, self.default_state)
|
||||
|
||||
# Get AQI data from OpenMeteo
|
||||
aqi_data = self.get_openmeteo_aqi(lat, lon)
|
||||
|
||||
if aqi_data == self.ERROR_FETCHING_DATA:
|
||||
return "Error fetching AQI data from OpenMeteo"
|
||||
|
||||
# Add location info for better user confirmation
|
||||
location_prefix = ""
|
||||
if location_type == "city" and address_info:
|
||||
# Always try to include city name if there's space
|
||||
# Use abbreviate_location to shorten long location strings (e.g., "United States of America" -> "USA")
|
||||
full_location = f"{actual_city}, {actual_state}" if actual_state else actual_city
|
||||
city_display = abbreviate_location(full_location, max_length=30)
|
||||
|
||||
# Check if we have space for the city name
|
||||
if location_type == "coordinates":
|
||||
location_prefix = f"{lat:.3f},{lon:.3f}: "
|
||||
elif resolved.display_name:
|
||||
city_display = resolved.display_name
|
||||
test_output = f"{city_display}: {aqi_data}"
|
||||
if len(test_output) <= 130:
|
||||
location_prefix = f"{city_display}: "
|
||||
else:
|
||||
# If no space, only show if it's a different state than default
|
||||
states_different = (actual_state != self.default_state and
|
||||
actual_state != default_state_full)
|
||||
if states_different:
|
||||
# Use abbreviated version for shorter display
|
||||
city_display_short = abbreviate_location(full_location, max_length=20)
|
||||
location_prefix = f"{city_display_short}: "
|
||||
elif location_type == "zipcode":
|
||||
location_prefix = f"{location.strip()}: "
|
||||
elif self._resolved_state_differs_from_default(address_info):
|
||||
# Over budget: only keep a short prefix when outside default region
|
||||
short = abbreviate_location(city_display, max_length=20)
|
||||
location_prefix = f"{short}: "
|
||||
elif location_type == "zipcode":
|
||||
# Add location info for ZIP codes to confirm geocoding accuracy
|
||||
if address_info:
|
||||
# Try to get city from address_info first
|
||||
actual_city = (address_info.get('city') or
|
||||
address_info.get('town') or
|
||||
address_info.get('village') or
|
||||
address_info.get('hamlet') or
|
||||
address_info.get('municipality'))
|
||||
|
||||
# If no city found in address_info, try to extract from the original geocoding result
|
||||
if not actual_city and location_result and location_result.address:
|
||||
# Extract city name from the geocoding result address
|
||||
address_parts = location_result.address.split(',')
|
||||
if len(address_parts) > 0:
|
||||
# The first part usually contains the city name
|
||||
potential_city = address_parts[0].strip()
|
||||
# Remove any house numbers or road names
|
||||
if not any(char.isdigit() for char in potential_city):
|
||||
actual_city = potential_city
|
||||
|
||||
# Fallback to 'Unknown' if still no city found
|
||||
if not actual_city:
|
||||
actual_city = 'Unknown'
|
||||
|
||||
# Get state info
|
||||
country = address_info.get('country', '')
|
||||
state = address_info.get('state', '')
|
||||
|
||||
if country == "United States" or country == "US" or country == "United States of America":
|
||||
# US city - use the state
|
||||
actual_state = state or self.default_state
|
||||
# Convert full state name to abbreviation if needed
|
||||
if len(actual_state) > 2 and actual_state in state_abbrev_map:
|
||||
actual_state = state_abbrev_map.get(actual_state, actual_state)
|
||||
else:
|
||||
# International city - use the country
|
||||
actual_state = (country or
|
||||
address_info.get('province') or
|
||||
self.default_state)
|
||||
# Normalize "United States" variants to "USA" to save characters
|
||||
if actual_state == "United States" or actual_state == "United States of America":
|
||||
actual_state = "USA"
|
||||
|
||||
# Use abbreviate_location to shorten long location strings (e.g., "United States of America" -> "USA")
|
||||
full_location = f"{actual_city}, {actual_state}" if actual_state else actual_city
|
||||
city_display = abbreviate_location(full_location, max_length=30)
|
||||
|
||||
# Check if we have space for the city name
|
||||
test_output = f"{city_display}: {aqi_data}"
|
||||
if len(test_output) <= 130:
|
||||
location_prefix = f"{city_display}: "
|
||||
else:
|
||||
# If no space, only show if it's a different state than default
|
||||
states_different = (actual_state != self.default_state and
|
||||
actual_state != default_state_full)
|
||||
if states_different:
|
||||
# Use abbreviated version for shorter display
|
||||
city_display_short = abbreviate_location(full_location, max_length=20)
|
||||
location_prefix = f"{city_display_short}: "
|
||||
else:
|
||||
# No address info available
|
||||
location_prefix = f"{zip_code}: "
|
||||
elif location_type == "coordinates":
|
||||
# Add coordinate info for clarity
|
||||
location_prefix = f"{lat:.3f},{lon:.3f}: "
|
||||
location_prefix = f"{location.strip()}: "
|
||||
|
||||
return f"{location_prefix}{aqi_data}"
|
||||
|
||||
@@ -645,167 +291,24 @@ class AqiCommand(BaseCommand):
|
||||
def city_to_lat_lon(self, city: str) -> tuple:
|
||||
"""Convert city name to latitude and longitude using default state.
|
||||
|
||||
Args:
|
||||
city: City name (can include state/country).
|
||||
|
||||
Returns:
|
||||
tuple: (latitude, longitude, address_info) or (None, None, None).
|
||||
Thin wrapper over shared ``geocode_city_best_effort`` (kept for tests/compat).
|
||||
"""
|
||||
try:
|
||||
# Check if the input contains a comma (city, state/country format)
|
||||
if ',' in city:
|
||||
# Parse city, state/country format
|
||||
city_parts = [part.strip() for part in city.split(',')]
|
||||
if len(city_parts) >= 2:
|
||||
city_name = city_parts[0]
|
||||
state_or_country = city_parts[1]
|
||||
|
||||
# AQI-specific: Check if it's a country (not a US state)
|
||||
country_indicators = ['canada', 'mexico', 'uk', 'united kingdom', 'france', 'germany', 'italy', 'spain', 'australia', 'japan', 'china', 'india', 'brazil', 'uae', 'russia', 'korea', 'thailand', 'singapore', 'egypt', 'turkey', 'israel', 'south africa', 'kenya', 'nigeria', 'argentina', 'peru', 'chile', 'colombia', 'venezuela', 'cuba', 'jamaica', 'puerto rico', 'iceland', 'norway', 'sweden', 'denmark', 'finland', 'poland', 'czech republic', 'hungary', 'romania', 'bulgaria', 'croatia', 'serbia', 'greece', 'portugal', 'ireland', 'belgium', 'netherlands', 'switzerland', 'austria', 'monaco', 'andorra', 'san marino', 'vatican', 'luxembourg', 'malta', 'cyprus', 'albania', 'macedonia', 'montenegro', 'bosnia', 'slovenia', 'slovakia', 'lithuania', 'latvia', 'estonia', 'belarus', 'ukraine', 'moldova', 'georgia', 'armenia', 'azerbaijan', 'kazakhstan', 'uzbekistan', 'kyrgyzstan', 'tajikistan', 'turkmenistan', 'afghanistan', 'pakistan', 'bangladesh', 'sri lanka', 'nepal', 'bhutan', 'myanmar', 'laos', 'cambodia', 'vietnam', 'malaysia', 'indonesia', 'philippines', 'taiwan', 'north korea', 'mongolia']
|
||||
is_country = state_or_country.lower() in country_indicators
|
||||
|
||||
if is_country:
|
||||
# Handle international cities explicitly (AQI-specific feature)
|
||||
location = rate_limited_nominatim_geocode_sync(self.bot, f"{city_name}, {state_or_country}", timeout=10)
|
||||
if location:
|
||||
# Cache the result
|
||||
self.db_manager.cache_geocoding(f"{city_name}, {state_or_country}", location.latitude, location.longitude)
|
||||
|
||||
# Use reverse geocoding to get detailed address info (check cache first)
|
||||
reverse_cache_key = f"reverse_{location.latitude}_{location.longitude}"
|
||||
cached_address = self.db_manager.get_cached_json(reverse_cache_key, "geolocation")
|
||||
if cached_address:
|
||||
return location.latitude, location.longitude, cached_address
|
||||
else:
|
||||
try:
|
||||
reverse_location = rate_limited_nominatim_reverse_sync(self.bot, f"{location.latitude}, {location.longitude}", timeout=10)
|
||||
if reverse_location:
|
||||
address_info = reverse_location.raw.get('address', {})
|
||||
# Cache the reverse geocoding result
|
||||
self.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720)
|
||||
return location.latitude, location.longitude, address_info
|
||||
except:
|
||||
pass
|
||||
return location.latitude, location.longitude, location.raw.get('address', {})
|
||||
|
||||
# Use shared geocode_city_sync function with address info
|
||||
default_country = self.bot.config.get('Weather', 'default_country', fallback='US')
|
||||
lat, lon, address_info = geocode_city_sync(
|
||||
self.bot, city, default_state=self.default_state,
|
||||
default_country=default_country,
|
||||
include_address_info=True, timeout=10
|
||||
)
|
||||
|
||||
if lat and lon:
|
||||
return lat, lon, address_info or {}
|
||||
|
||||
# AQI-specific fallback: Try neighborhood-specific queries for major cities
|
||||
neighborhood_queries = self.get_neighborhood_queries(city)
|
||||
if neighborhood_queries:
|
||||
for query in neighborhood_queries:
|
||||
location = rate_limited_nominatim_geocode_sync(self.bot, query, timeout=10)
|
||||
if location:
|
||||
# Cache the result
|
||||
self.db_manager.cache_geocoding(query, location.latitude, location.longitude)
|
||||
|
||||
# Use reverse geocoding to get detailed address info (check cache first)
|
||||
reverse_cache_key = f"reverse_{location.latitude}_{location.longitude}"
|
||||
cached_address = self.db_manager.get_cached_json(reverse_cache_key, "geolocation")
|
||||
if cached_address:
|
||||
return location.latitude, location.longitude, cached_address
|
||||
else:
|
||||
try:
|
||||
reverse_location = rate_limited_nominatim_reverse_sync(self.bot, f"{location.latitude}, {location.longitude}", timeout=10)
|
||||
if reverse_location:
|
||||
address_info = reverse_location.raw.get('address', {})
|
||||
# Cache the reverse geocoding result
|
||||
self.db_manager.cache_json(reverse_cache_key, address_info, "geolocation", cache_hours=720)
|
||||
return location.latitude, location.longitude, address_info
|
||||
except:
|
||||
pass
|
||||
return location.latitude, location.longitude, location.raw.get('address', {})
|
||||
|
||||
return (None, None, None)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error geocoding city {city}: {e}")
|
||||
lat, lon, address_info = geocode_city_best_effort(
|
||||
self.bot,
|
||||
city,
|
||||
default_state=self.default_state,
|
||||
default_country=self.default_country,
|
||||
use_neighborhoods=True,
|
||||
include_address_info=True,
|
||||
timeout=10,
|
||||
)
|
||||
if lat is None or lon is None:
|
||||
return (None, None, None)
|
||||
return lat, lon, address_info or {}
|
||||
|
||||
def get_neighborhood_queries(self, city: str) -> list:
|
||||
"""Generate neighborhood-specific search queries for major cities.
|
||||
|
||||
Args:
|
||||
city: City name.
|
||||
|
||||
Returns:
|
||||
list: List of neighborhood-specific query strings.
|
||||
"""
|
||||
city_lower = city.lower()
|
||||
|
||||
# Seattle neighborhoods
|
||||
if city_lower in ['greenwood', 'ballard', 'capitol hill', 'fremont', 'queen anne',
|
||||
'wallingford', 'university district', 'pike place', 'pioneer square',
|
||||
'belltown', 'first hill', 'central district', 'beacon hill', 'columbia city',
|
||||
'west seattle', 'magnolia', 'phinney ridge', 'crown hill', 'loyal heights']:
|
||||
return [
|
||||
f"{city}, Seattle, WA, USA",
|
||||
f"{city}, Seattle, USA"
|
||||
]
|
||||
|
||||
# New York neighborhoods
|
||||
elif city_lower in ['greenwich village', 'soho', 'tribeca', 'chinatown', 'little italy',
|
||||
'east village', 'west village', 'chelsea', 'hells kitchen', 'upper east side',
|
||||
'upper west side', 'harlem', 'brooklyn heights', 'dumbo', 'williamsburg',
|
||||
'park slope', 'red hook', 'coney island']:
|
||||
return [
|
||||
f"{city}, New York, NY, USA",
|
||||
f"{city}, New York, USA"
|
||||
]
|
||||
|
||||
# San Francisco neighborhoods
|
||||
elif city_lower in ['mission district', 'haight-ashbury', 'castro', 'soma', 'financial district',
|
||||
'north beach', 'chinatown', 'russian hill', 'pacific heights', 'marina district',
|
||||
'sunset district', 'richmond district', 'bernal heights', 'noe valley']:
|
||||
return [
|
||||
f"{city}, San Francisco, CA, USA",
|
||||
f"{city}, San Francisco, USA"
|
||||
]
|
||||
|
||||
# Los Angeles neighborhoods
|
||||
elif city_lower in ['hollywood', 'beverly hills', 'santa monica', 'venice', 'manhattan beach',
|
||||
'hermosa beach', 'redondo beach', 'pasadena', 'glendale', 'burbank',
|
||||
'west hollywood', 'culver city', 'marina del rey', 'playa del rey']:
|
||||
return [
|
||||
f"{city}, Los Angeles, CA, USA",
|
||||
f"{city}, Los Angeles, USA"
|
||||
]
|
||||
|
||||
# Chicago neighborhoods
|
||||
elif city_lower in ['loop', 'magnificent mile', 'gold coast', 'lincoln park', 'wrigleyville',
|
||||
'lakeview', 'wicker park', 'bucktown', 'logan square', 'pilsen', 'hyde park']:
|
||||
return [
|
||||
f"{city}, Chicago, IL, USA",
|
||||
f"{city}, Chicago, USA"
|
||||
]
|
||||
|
||||
# Boston neighborhoods
|
||||
elif city_lower in ['back bay', 'beacon hill', 'north end', 'south end', 'charlestown',
|
||||
'east boston', 'dorchester', 'roxbury', 'jamaica plain', 'allston',
|
||||
'brighton', 'cambridge', 'somerville']:
|
||||
return [
|
||||
f"{city}, Boston, MA, USA",
|
||||
f"{city}, Boston, USA"
|
||||
]
|
||||
|
||||
# Portland neighborhoods
|
||||
elif city_lower in ['pearl district', 'alphabet district', 'nob hill', 'mississippi district',
|
||||
'hawthorne', 'belmont', 'sellwood', 'st. johns', 'kenton', 'overlook']:
|
||||
return [
|
||||
f"{city}, Portland, OR, USA",
|
||||
f"{city}, Portland, USA"
|
||||
]
|
||||
|
||||
# No neighborhood-specific queries for this city
|
||||
return []
|
||||
"""Generate neighborhood-specific search queries for major cities."""
|
||||
return location_neighborhood_queries(city)
|
||||
|
||||
def get_openmeteo_aqi(self, lat: float, lon: float) -> str:
|
||||
"""Get AQI data from OpenMeteo API.
|
||||
|
||||
@@ -17,9 +17,26 @@ from urllib3.util.retry import Retry
|
||||
|
||||
from ..models import MeshMessage
|
||||
from ..region_capitals import REGION_DEFAULT_NOTE, region_capital_query
|
||||
from ..utils import geocode_city_sync, geocode_zipcode_sync, normalize_us_state
|
||||
from ..location import (
|
||||
US_STATE_ABBRS,
|
||||
city_display_name,
|
||||
join_location,
|
||||
reverse_geocode_region,
|
||||
titlecase_location,
|
||||
zip_to_city_string as location_zip_to_city_string,
|
||||
)
|
||||
from ..utils import geocode_city_sync, geocode_zipcode_sync
|
||||
from .base_command import BaseCommand
|
||||
|
||||
# Re-exports for weather_service / tests that import display helpers from rain_command.
|
||||
__all_location_reexports__ = (
|
||||
"US_STATE_ABBRS",
|
||||
"city_display_name",
|
||||
"join_location",
|
||||
"reverse_geocode_region",
|
||||
"titlecase_location",
|
||||
)
|
||||
|
||||
# WMO weather code -> precipitation "bucket". Buckets map to an emoji and a
|
||||
# translatable label (commands.rain.precip_types.<bucket>). Codes not listed
|
||||
# here are non-precipitating and never trigger a nowcast.
|
||||
@@ -89,123 +106,6 @@ def precip_bucket_for_code(code: Optional[int]) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def titlecase_location(text: str) -> str:
|
||||
"""Tidy a user-typed location for display.
|
||||
|
||||
'middlesboro, ky' -> 'Middlesboro, KY'; 'paris, france' -> 'Paris, France';
|
||||
'memphis' -> 'Memphis'. A 2-letter token after a comma is treated as a
|
||||
state/country code and upper-cased; everything else is title-cased.
|
||||
"""
|
||||
parts = [p.strip() for p in text.split(",") if p.strip()]
|
||||
if not parts:
|
||||
return text.strip()
|
||||
out = []
|
||||
for i, p in enumerate(parts):
|
||||
if i > 0 and len(p) == 2 and p.isalpha():
|
||||
out.append(p.upper())
|
||||
else:
|
||||
out.append(p.title())
|
||||
return ", ".join(out)
|
||||
|
||||
|
||||
# US state / territory 2-letter codes — used to drop a trailing state from a
|
||||
# typed location like "london ky" (no comma) so it doesn't become "London Ky".
|
||||
US_STATE_ABBRS = frozenset({
|
||||
"AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL",
|
||||
"IN", "IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT",
|
||||
"NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI",
|
||||
"SC", "SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY",
|
||||
"DC", "AS", "GU", "MP", "PR", "VI",
|
||||
})
|
||||
|
||||
|
||||
def city_display_name(typed_location: str, suffix: Optional[str] = None) -> str:
|
||||
"""City part of a typed location for display, dropping a trailing region the
|
||||
user appended without a comma.
|
||||
|
||||
'london ky' -> 'London'; 'paris france' -> 'Paris'; 'london, ky' -> 'London';
|
||||
'oklahoma city' -> 'Oklahoma City'. `suffix` is the geocoder's authoritative
|
||||
state/country (e.g. 'KY' or 'France'); when the typed text ends with it, it's
|
||||
stripped so it isn't doubled into the city name. The state/country is added
|
||||
back separately by the caller.
|
||||
"""
|
||||
head = typed_location.split(",")[0].strip()
|
||||
# Drop a trailing region matching the geocoder's suffix — handles country
|
||||
# names and multi-word regions ("paris france", "london united kingdom").
|
||||
if suffix and head.lower().endswith(" " + suffix.lower()):
|
||||
head = head[: -len(suffix)].strip()
|
||||
# Drop a trailing US state abbreviation ("london ky" -> "london").
|
||||
tokens = head.split()
|
||||
if len(tokens) >= 2 and tokens[-1].upper() in US_STATE_ABBRS:
|
||||
head = " ".join(tokens[:-1])
|
||||
return titlecase_location(head)
|
||||
|
||||
|
||||
def join_location(city: Optional[str], suffix: Optional[str]) -> str:
|
||||
"""Join a city and its state/country suffix as 'City, Suffix'.
|
||||
|
||||
Collapses to a single name when one side is missing or the two name the same
|
||||
place (case-insensitive) — so a country typed as the city ('spain' -> 'Spain',
|
||||
not 'Spain, Spain') or a city-state ('Singapore', not 'Singapore, Singapore')
|
||||
renders once.
|
||||
"""
|
||||
city = (city or "").strip()
|
||||
suffix = (suffix or "").strip()
|
||||
if not suffix:
|
||||
return city
|
||||
if not city or city.lower() == suffix.lower():
|
||||
return suffix
|
||||
return f"{city}, {suffix}"
|
||||
|
||||
|
||||
def reverse_geocode_region(
|
||||
bot: Any, lat: float, lon: float, *, timeout: int = 10, logger: Any = None
|
||||
) -> tuple[Optional[str], Optional[str]]:
|
||||
"""Reverse-geocode to (city, suffix), respecting the bot's Nominatim rate limiter.
|
||||
|
||||
suffix is the US state abbreviation ('TN') for US points, else the English
|
||||
country name ('Japan'). Requests language='en' so country names aren't
|
||||
localized. No caching (callers cache as needed). Shared by the rain command
|
||||
and the Weather_Service proactive push so both label locations identically.
|
||||
"""
|
||||
city: Optional[str] = None
|
||||
suffix: Optional[str] = None
|
||||
try:
|
||||
from ..utils import get_nominatim_geocoder
|
||||
limiter = getattr(bot, "nominatim_rate_limiter", None)
|
||||
if limiter is not None:
|
||||
limiter.wait_for_request_sync()
|
||||
geolocator = get_nominatim_geocoder(timeout=timeout)
|
||||
# language="en" so country names come back in English ("Japan", not "日本").
|
||||
result = geolocator.reverse(f"{lat}, {lon}", timeout=timeout, language="en")
|
||||
if limiter is not None:
|
||||
limiter.record_request()
|
||||
if result is not None and hasattr(result, "raw"):
|
||||
address = result.raw.get("address", {})
|
||||
city = (
|
||||
address.get("city")
|
||||
or address.get("town")
|
||||
or address.get("village")
|
||||
or address.get("municipality")
|
||||
or address.get("county")
|
||||
or None
|
||||
)
|
||||
country_code = (address.get("country_code") or "").lower()
|
||||
if country_code == "us":
|
||||
iso = address.get("ISO3166-2-lvl4") or address.get("ISO3166-2-lvl6") or ""
|
||||
if "-" in iso:
|
||||
suffix = iso.rsplit("-", 1)[-1]
|
||||
else:
|
||||
state_abbr, _ = normalize_us_state(address.get("state", ""))
|
||||
suffix = state_abbr or address.get("state") or None
|
||||
else:
|
||||
suffix = address.get("country") or None
|
||||
except Exception as e:
|
||||
if logger:
|
||||
logger.debug(f"Error reverse geocoding {lat},{lon}: {e}")
|
||||
return city, suffix
|
||||
|
||||
|
||||
def precip_descriptor(bucket: Optional[str]) -> tuple[str, str]:
|
||||
"""Return (emoji, English label) for a precip bucket; defaults to rain.
|
||||
|
||||
@@ -972,29 +872,10 @@ class RainCommand(BaseCommand):
|
||||
return self._reverse_geocode(lat, lon)[1]
|
||||
|
||||
def _zip_to_city_string(self, zipcode: str) -> Optional[str]:
|
||||
"""US ZIP -> 'City, ST' via Zippopotam.us (free, no key, cached).
|
||||
|
||||
OSM/Nominatim often lacks the USPS city for a ZIP centroid (returns the
|
||||
county instead), so for 5-digit US ZIPs this gives a far better name.
|
||||
Returns None on failure (caller falls back to reverse geocoding).
|
||||
"""
|
||||
z = zipcode.strip()
|
||||
if z in self._zip_cache:
|
||||
return self._zip_cache[z]
|
||||
name: Optional[str] = None
|
||||
try:
|
||||
resp = requests.get(f"https://api.zippopotam.us/us/{z}", timeout=self.url_timeout)
|
||||
if resp.ok:
|
||||
places = resp.json().get("places") or []
|
||||
if places:
|
||||
city = (places[0].get("place name") or "").strip()
|
||||
st = (places[0].get("state abbreviation") or "").strip()
|
||||
if city:
|
||||
name = join_location(city, st)
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Zippopotam ZIP lookup failed for {z}: {e}")
|
||||
if name:
|
||||
_cache_put(self._zip_cache, z, name)
|
||||
"""US ZIP -> 'City, ST' via Zippopotam.us (free, no key, cached)."""
|
||||
name = location_zip_to_city_string(
|
||||
zipcode, timeout=self.url_timeout, cache=self._zip_cache, logger=self.logger
|
||||
)
|
||||
return name
|
||||
|
||||
def _resolve_location(
|
||||
|
||||
@@ -704,8 +704,8 @@ class WxCommand(BaseCommand):
|
||||
if re.match(r'^\s*-?\d+\.?\d*\s*,\s*-?\d+\.?\d*\s*$', location):
|
||||
# It's coordinates (lat,lon format)
|
||||
location_type = "coordinates"
|
||||
elif re.match(r'^\d{5}$', location):
|
||||
# It's a zipcode
|
||||
elif re.match(r'^\s*\d{5}\s*$', location):
|
||||
# It's a zipcode (allow surrounding whitespace; strip later in geocode)
|
||||
location_type = "zipcode"
|
||||
else:
|
||||
# It's a city name (possibly with state)
|
||||
|
||||
@@ -0,0 +1,882 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Shared location classification and best-effort geocoding for command plugins.
|
||||
|
||||
Low-level Nominatim / geocode_city / geocode_zipcode helpers remain in
|
||||
``modules.utils``. This module is the high-level front door (coords | ZIP |
|
||||
city | optional repeater / region capitals / neighborhoods) with opt-in
|
||||
``ResolveOptions`` so commands can migrate without changing semantics.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from dataclasses import dataclass, replace
|
||||
from typing import Any, Literal, Mapping, Optional, Union
|
||||
|
||||
import requests
|
||||
|
||||
from .region_capitals import REGION_DEFAULT_NOTE, region_capital_query
|
||||
from .utils import (
|
||||
abbreviate_location,
|
||||
geocode_city_sync,
|
||||
geocode_zipcode_sync,
|
||||
normalize_us_state,
|
||||
rate_limited_nominatim_geocode_sync,
|
||||
rate_limited_nominatim_reverse_sync,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
COORD_RE = re.compile(r"^\s*-?\d+\.?\d*\s*,\s*-?\d+\.?\d*\s*$")
|
||||
ZIP_RE = re.compile(r"^\s*\d{5}\s*$")
|
||||
|
||||
_COUNTRY_WORD_INDICATORS = frozenset({
|
||||
"canada", "mexico", "uk", "united", "kingdom", "france", "germany", "italy",
|
||||
"spain", "australia", "japan", "china", "india", "brazil",
|
||||
})
|
||||
|
||||
# Explicit allowlist for "city, country" intl geocode shortcut (legacy AQI list).
|
||||
# Do not use utils.is_country_name here — its len>2 heuristic is too broad.
|
||||
_COUNTRY_TOKENS = frozenset({
|
||||
"canada", "mexico", "uk", "united kingdom", "france", "germany", "italy",
|
||||
"spain", "australia", "japan", "china", "india", "brazil", "uae", "russia",
|
||||
"korea", "thailand", "singapore", "egypt", "turkey", "israel", "south africa",
|
||||
"kenya", "nigeria", "argentina", "peru", "chile", "colombia", "venezuela",
|
||||
"cuba", "jamaica", "puerto rico", "iceland", "norway", "sweden", "denmark",
|
||||
"finland", "poland", "czech republic", "hungary", "romania", "bulgaria",
|
||||
"croatia", "serbia", "greece", "portugal", "ireland", "belgium",
|
||||
"netherlands", "switzerland", "austria", "monaco", "andorra", "san marino",
|
||||
"vatican", "luxembourg", "malta", "cyprus", "albania", "macedonia",
|
||||
"montenegro", "bosnia", "slovenia", "slovakia", "lithuania", "latvia",
|
||||
"estonia", "belarus", "ukraine", "moldova", "georgia", "armenia",
|
||||
"azerbaijan", "kazakhstan", "uzbekistan", "kyrgyzstan", "tajikistan",
|
||||
"turkmenistan", "afghanistan", "pakistan", "bangladesh", "sri lanka",
|
||||
"nepal", "bhutan", "myanmar", "laos", "cambodia", "vietnam", "malaysia",
|
||||
"indonesia", "philippines", "taiwan", "north korea", "south korea",
|
||||
"mongolia",
|
||||
})
|
||||
|
||||
GEOCODE_CACHE_CAP = 256
|
||||
|
||||
# Bare place → "city, country" (full-string match, including multi-word keys).
|
||||
INTERNATIONAL_CITIES: dict[str, str] = {
|
||||
"beijing": "beijing, china",
|
||||
"shanghai": "shanghai, china",
|
||||
"tokyo": "tokyo, japan",
|
||||
"london": "london, uk",
|
||||
"paris": "paris, france",
|
||||
"berlin": "berlin, germany",
|
||||
"rome": "rome, italy",
|
||||
"madrid": "madrid, spain",
|
||||
"moscow": "moscow, russia",
|
||||
"sydney": "sydney, australia",
|
||||
"melbourne": "melbourne, australia",
|
||||
"toronto": "toronto, canada",
|
||||
"vancouver": "vancouver, canada",
|
||||
"mumbai": "mumbai, india",
|
||||
"delhi": "delhi, india",
|
||||
"bangalore": "bangalore, india",
|
||||
"sao paulo": "sao paulo, brazil",
|
||||
"rio de janeiro": "rio de janeiro, brazil",
|
||||
"mexico city": "mexico city, mexico",
|
||||
"cairo": "cairo, egypt",
|
||||
"istanbul": "istanbul, turkey",
|
||||
"seoul": "seoul, south korea",
|
||||
"bangkok": "bangkok, thailand",
|
||||
"singapore": "singapore, singapore",
|
||||
"hong kong": "hong kong, china",
|
||||
"dubai": "dubai, uae",
|
||||
"tel aviv": "tel aviv, israel",
|
||||
"johannesburg": "johannesburg, south africa",
|
||||
"nairobi": "nairobi, kenya",
|
||||
"lagos": "lagos, nigeria",
|
||||
"buenos aires": "buenos aires, argentina",
|
||||
"lima": "lima, peru",
|
||||
"santiago": "santiago, chile",
|
||||
"bogota": "bogota, colombia",
|
||||
"caracas": "caracas, venezuela",
|
||||
"havana": "havana, cuba",
|
||||
"kingston": "kingston, jamaica",
|
||||
"san juan": "san juan, puerto rico",
|
||||
"reykjavik": "reykjavik, iceland",
|
||||
"oslo": "oslo, norway",
|
||||
"stockholm": "stockholm, sweden",
|
||||
"copenhagen": "copenhagen, denmark",
|
||||
"helsinki": "helsinki, finland",
|
||||
"warsaw": "warsaw, poland",
|
||||
"prague": "prague, czech republic",
|
||||
"budapest": "budapest, hungary",
|
||||
"bucharest": "bucharest, romania",
|
||||
"sofia": "sofia, bulgaria",
|
||||
"zagreb": "zagreb, croatia",
|
||||
"belgrade": "belgrade, serbia",
|
||||
"athens": "athens, greece",
|
||||
"lisbon": "lisbon, portugal",
|
||||
"dublin": "dublin, ireland",
|
||||
"brussels": "brussels, belgium",
|
||||
"amsterdam": "amsterdam, netherlands",
|
||||
"zurich": "zurich, switzerland",
|
||||
"vienna": "vienna, austria",
|
||||
"lucerne": "lucerne, switzerland",
|
||||
"geneva": "geneva, switzerland",
|
||||
"monaco": "monaco, monaco",
|
||||
"andorra": "andorra, andorra",
|
||||
"san marino": "san marino, san marino",
|
||||
"vatican": "vatican city, vatican",
|
||||
"luxembourg": "luxembourg, luxembourg",
|
||||
"malta": "valletta, malta",
|
||||
"cyprus": "nicosia, cyprus",
|
||||
"albania": "tirana, albania",
|
||||
"macedonia": "skopje, macedonia",
|
||||
"montenegro": "podgorica, montenegro",
|
||||
"bosnia": "sarajevo, bosnia",
|
||||
"slovenia": "ljubljana, slovenia",
|
||||
"slovakia": "bratislava, slovakia",
|
||||
"lithuania": "vilnius, lithuania",
|
||||
"latvia": "riga, latvia",
|
||||
"estonia": "tallinn, estonia",
|
||||
"belarus": "minsk, belarus",
|
||||
"ukraine": "kiev, ukraine",
|
||||
"moldova": "chisinau, moldova",
|
||||
"georgia": "tbilisi, georgia",
|
||||
"armenia": "yerevan, armenia",
|
||||
"azerbaijan": "baku, azerbaijan",
|
||||
"kazakhstan": "nur-sultan, kazakhstan",
|
||||
"uzbekistan": "tashkent, uzbekistan",
|
||||
"kyrgyzstan": "bishkek, kyrgyzstan",
|
||||
"tajikistan": "dushanbe, tajikistan",
|
||||
"turkmenistan": "ashgabat, turkmenistan",
|
||||
"afghanistan": "kabul, afghanistan",
|
||||
"pakistan": "islamabad, pakistan",
|
||||
"bangladesh": "dhaka, bangladesh",
|
||||
"sri lanka": "colombo, sri lanka",
|
||||
"nepal": "kathmandu, nepal",
|
||||
"bhutan": "thimphu, bhutan",
|
||||
"myanmar": "yangon, myanmar",
|
||||
"laos": "vientiane, laos",
|
||||
"cambodia": "phnom penh, cambodia",
|
||||
"vietnam": "hanoi, vietnam",
|
||||
"malaysia": "kuala lumpur, malaysia",
|
||||
"indonesia": "jakarta, indonesia",
|
||||
"philippines": "manila, philippines",
|
||||
"taiwan": "taipei, taiwan",
|
||||
"north korea": "pyongyang, north korea",
|
||||
"mongolia": "ulaanbaatar, mongolia",
|
||||
}
|
||||
|
||||
ZIP_CODE_OVERRIDES: dict[str, str] = {
|
||||
"98013": "Vashon, WA, USA",
|
||||
"98014": "Vashon Island, WA, USA",
|
||||
}
|
||||
|
||||
US_STATE_ABBRS = frozenset({
|
||||
"AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL",
|
||||
"IN", "IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT",
|
||||
"NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI",
|
||||
"SC", "SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY",
|
||||
"DC", "AS", "GU", "MP", "PR", "VI",
|
||||
})
|
||||
|
||||
_NEIGHBORHOODS: dict[str, tuple[str, ...]] = {
|
||||
"seattle": (
|
||||
"greenwood", "ballard", "capitol hill", "fremont", "queen anne",
|
||||
"wallingford", "university district", "pike place", "pioneer square",
|
||||
"belltown", "first hill", "central district", "beacon hill", "columbia city",
|
||||
"west seattle", "magnolia", "phinney ridge", "crown hill", "loyal heights",
|
||||
),
|
||||
"new york": (
|
||||
"greenwich village", "soho", "tribeca", "chinatown", "little italy",
|
||||
"east village", "west village", "chelsea", "hells kitchen", "upper east side",
|
||||
"upper west side", "harlem", "brooklyn heights", "dumbo", "williamsburg",
|
||||
"park slope", "red hook", "coney island",
|
||||
),
|
||||
"san francisco": (
|
||||
"mission district", "haight-ashbury", "castro", "soma", "financial district",
|
||||
"north beach", "chinatown", "russian hill", "pacific heights", "marina district",
|
||||
"sunset district", "richmond district", "bernal heights", "noe valley",
|
||||
),
|
||||
"los angeles": (
|
||||
"hollywood", "beverly hills", "santa monica", "venice", "manhattan beach",
|
||||
"hermosa beach", "redondo beach", "pasadena", "glendale", "burbank",
|
||||
"west hollywood", "culver city", "marina del rey", "playa del rey",
|
||||
),
|
||||
"chicago": (
|
||||
"loop", "magnificent mile", "gold coast", "lincoln park", "wrigleyville",
|
||||
"lakeview", "wicker park", "bucktown", "logan square", "pilsen", "hyde park",
|
||||
),
|
||||
"boston": (
|
||||
"back bay", "beacon hill", "north end", "south end", "charlestown",
|
||||
"east boston", "dorchester", "roxbury", "jamaica plain", "allston",
|
||||
"brighton", "cambridge", "somerville",
|
||||
),
|
||||
"portland": (
|
||||
"pearl district", "alphabet district", "nob hill", "mississippi district",
|
||||
"hawthorne", "belmont", "sellwood", "st. johns", "kenton", "overlook",
|
||||
),
|
||||
}
|
||||
|
||||
_NEIGHBORHOOD_PARENT: dict[str, tuple[str, str]] = {
|
||||
"seattle": ("Seattle", "WA"),
|
||||
"new york": ("New York", "NY"),
|
||||
"san francisco": ("San Francisco", "CA"),
|
||||
"los angeles": ("Los Angeles", "CA"),
|
||||
"chicago": ("Chicago", "IL"),
|
||||
"boston": ("Boston", "MA"),
|
||||
"portland": ("Portland", "OR"),
|
||||
}
|
||||
|
||||
LabelStyle = Literal["numeric", "query", "abbreviated", "city_region"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ResolveOptions:
|
||||
"""Opt-in semantics for ``resolve_location``."""
|
||||
|
||||
default_state: Optional[str] = None
|
||||
default_country: Optional[str] = None
|
||||
timeout: int = 10
|
||||
use_international_cities: bool = True
|
||||
use_neighborhoods: bool = True
|
||||
use_structured_zip: bool = True
|
||||
use_region_capitals: bool = False
|
||||
use_zippopotam_labels: bool = False
|
||||
allow_repeater_names: bool = False
|
||||
fallback_coords: Optional[tuple[float, float]] = None
|
||||
fallback_label: Optional[str] = None
|
||||
include_address_info: bool = True
|
||||
label_style: LabelStyle = "abbreviated"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ResolvedLocation:
|
||||
lat: Optional[float]
|
||||
lon: Optional[float]
|
||||
location_type: Optional[str]
|
||||
query: str
|
||||
display_name: Optional[str]
|
||||
address_info: Optional[dict]
|
||||
error: Optional[str] = None
|
||||
error_detail: Optional[str] = None
|
||||
region_note: Optional[str] = None
|
||||
|
||||
|
||||
OPTIONS_AQI = ResolveOptions()
|
||||
OPTIONS_WX = ResolveOptions(
|
||||
use_neighborhoods=False, use_structured_zip=False, label_style="city_region",
|
||||
)
|
||||
OPTIONS_AURORA = ResolveOptions(
|
||||
use_international_cities=False, use_neighborhoods=False, use_structured_zip=False,
|
||||
label_style="numeric", include_address_info=False,
|
||||
)
|
||||
OPTIONS_RAIN = ResolveOptions(
|
||||
use_region_capitals=True, use_zippopotam_labels=True,
|
||||
use_international_cities=False, use_neighborhoods=False, use_structured_zip=False,
|
||||
label_style="city_region", include_address_info=False,
|
||||
)
|
||||
OPTIONS_PREFIX = ResolveOptions(
|
||||
allow_repeater_names=True, use_international_cities=False,
|
||||
use_neighborhoods=False, use_structured_zip=False,
|
||||
label_style="query", include_address_info=False,
|
||||
)
|
||||
OPTIONS_SOLARFORECAST = OPTIONS_PREFIX
|
||||
|
||||
|
||||
def titlecase_location(text: str) -> str:
|
||||
parts = [p.strip() for p in text.split(",") if p.strip()]
|
||||
if not parts:
|
||||
return text.strip()
|
||||
out = []
|
||||
for i, p in enumerate(parts):
|
||||
if i > 0 and len(p) == 2 and p.isalpha():
|
||||
out.append(p.upper())
|
||||
else:
|
||||
out.append(p.title())
|
||||
return ", ".join(out)
|
||||
|
||||
|
||||
def city_display_name(typed_location: str, suffix: Optional[str] = None) -> str:
|
||||
head = typed_location.split(",")[0].strip()
|
||||
if suffix and head.lower().endswith(" " + suffix.lower()):
|
||||
head = head[: -len(suffix)].strip()
|
||||
tokens = head.split()
|
||||
if len(tokens) >= 2 and tokens[-1].upper() in US_STATE_ABBRS:
|
||||
head = " ".join(tokens[:-1])
|
||||
return titlecase_location(head)
|
||||
|
||||
|
||||
def join_location(city: Optional[str], suffix: Optional[str]) -> str:
|
||||
city = (city or "").strip()
|
||||
suffix = (suffix or "").strip()
|
||||
if not suffix:
|
||||
return city
|
||||
if not city or city.lower() == suffix.lower():
|
||||
return suffix
|
||||
return f"{city}, {suffix}"
|
||||
|
||||
|
||||
def reverse_geocode_region(
|
||||
bot: Any, lat: float, lon: float, *, timeout: int = 10, logger: Any = None
|
||||
) -> tuple[Optional[str], Optional[str]]:
|
||||
city: Optional[str] = None
|
||||
suffix: Optional[str] = None
|
||||
try:
|
||||
from .utils import get_nominatim_geocoder
|
||||
limiter = getattr(bot, "nominatim_rate_limiter", None)
|
||||
if limiter is not None:
|
||||
limiter.wait_for_request_sync()
|
||||
geolocator = get_nominatim_geocoder(timeout=timeout)
|
||||
result = geolocator.reverse(f"{lat}, {lon}", timeout=timeout, language="en")
|
||||
if limiter is not None:
|
||||
limiter.record_request()
|
||||
if result is not None and hasattr(result, "raw"):
|
||||
address = result.raw.get("address", {})
|
||||
city = (
|
||||
address.get("city") or address.get("town") or address.get("village")
|
||||
or address.get("municipality") or address.get("county") or None
|
||||
)
|
||||
country_code = (address.get("country_code") or "").lower()
|
||||
if country_code == "us":
|
||||
iso = address.get("ISO3166-2-lvl4") or address.get("ISO3166-2-lvl6") or ""
|
||||
if "-" in iso:
|
||||
suffix = iso.rsplit("-", 1)[-1]
|
||||
else:
|
||||
state_abbr, _ = normalize_us_state(address.get("state", ""))
|
||||
suffix = state_abbr or address.get("state") or None
|
||||
else:
|
||||
suffix = address.get("country") or None
|
||||
except Exception as e:
|
||||
if logger:
|
||||
logger.debug(f"Error reverse geocoding {lat},{lon}: {e}")
|
||||
return city, suffix
|
||||
|
||||
|
||||
def cache_put(
|
||||
cache: dict, key: Any, value: Any, *, cap: int = GEOCODE_CACHE_CAP
|
||||
) -> None:
|
||||
"""Insert into a size-capped cache, evicting the oldest entry when full."""
|
||||
if key not in cache and len(cache) >= cap:
|
||||
cache.pop(next(iter(cache)))
|
||||
cache[key] = value
|
||||
|
||||
|
||||
def zip_to_city_string(
|
||||
zipcode: str, *, timeout: int = 10, cache: Optional[dict[str, str]] = None, logger: Any = None
|
||||
) -> Optional[str]:
|
||||
z = zipcode.strip()
|
||||
if cache is not None and z in cache:
|
||||
return cache[z]
|
||||
name: Optional[str] = None
|
||||
try:
|
||||
resp = requests.get(f"https://api.zippopotam.us/us/{z}", timeout=timeout)
|
||||
if resp.ok:
|
||||
places = resp.json().get("places") or []
|
||||
if places:
|
||||
city = (places[0].get("place name") or "").strip()
|
||||
st = (places[0].get("state abbreviation") or "").strip()
|
||||
if city:
|
||||
name = join_location(city, st)
|
||||
except Exception as e:
|
||||
if logger:
|
||||
logger.debug(f"Zippopotam ZIP lookup failed for {z}: {e}")
|
||||
if name and cache is not None:
|
||||
cache_put(cache, z, name)
|
||||
return name
|
||||
|
||||
|
||||
def parse_coordinates(raw: str) -> Optional[tuple[float, float]]:
|
||||
"""Return (lat, lon) when valid; None for bad format or out-of-range."""
|
||||
coords, _error, _detail = parse_coordinates_detailed(raw)
|
||||
return coords
|
||||
|
||||
|
||||
def parse_coordinates_detailed(
|
||||
raw: str,
|
||||
) -> tuple[Optional[tuple[float, float]], Optional[str], Optional[str]]:
|
||||
"""Return ((lat, lon)|None, error_code|None, error_detail|None)."""
|
||||
if not COORD_RE.match(raw or ""):
|
||||
return None, "invalid_coordinates", None
|
||||
try:
|
||||
a, b = raw.split(",", 1)
|
||||
lat, lon = float(a.strip()), float(b.strip())
|
||||
except ValueError:
|
||||
return None, "invalid_coordinates", None
|
||||
if not (-90 <= lat <= 90):
|
||||
return None, "invalid_latitude", str(lat)
|
||||
if not (-180 <= lon <= 180):
|
||||
return None, "invalid_longitude", str(lon)
|
||||
return (lat, lon), None, None
|
||||
|
||||
|
||||
def _rewrite_space_country(location: str) -> str:
|
||||
parts = location.split()
|
||||
if len(parts) < 2:
|
||||
return location
|
||||
potential_country = parts[1].lower()
|
||||
if potential_country not in _COUNTRY_WORD_INDICATORS:
|
||||
return location
|
||||
city = parts[0]
|
||||
if potential_country in ("united", "kingdom"):
|
||||
if len(parts) >= 3 and parts[2].lower() == "kingdom":
|
||||
return f"{city}, uk"
|
||||
return f"{city}, {parts[1]}"
|
||||
return f"{city}, {parts[1]}"
|
||||
|
||||
|
||||
def classify_location(
|
||||
raw: str,
|
||||
*,
|
||||
use_international_cities: bool = True,
|
||||
) -> tuple[str, str]:
|
||||
"""Return (normalized_location, location_type)."""
|
||||
location = (raw or "").strip()
|
||||
if COORD_RE.match(location):
|
||||
return location, "coordinates"
|
||||
if ZIP_RE.match(location):
|
||||
return location.strip(), "zipcode"
|
||||
|
||||
if use_international_cities:
|
||||
city_lower = location.lower()
|
||||
if city_lower in INTERNATIONAL_CITIES:
|
||||
return INTERNATIONAL_CITIES[city_lower], "city"
|
||||
|
||||
location = _rewrite_space_country(location)
|
||||
return location, "city"
|
||||
|
||||
|
||||
def get_neighborhood_queries(city: str) -> list[str]:
|
||||
city_lower = city.lower().strip()
|
||||
for parent_key, names in _NEIGHBORHOODS.items():
|
||||
if city_lower in names:
|
||||
parent, st = _NEIGHBORHOOD_PARENT[parent_key]
|
||||
return [f"{city}, {parent}, {st}, USA", f"{city}, {parent}, USA"]
|
||||
return []
|
||||
|
||||
|
||||
def _is_country_token(text: str) -> bool:
|
||||
return text.strip().lower() in _COUNTRY_TOKENS
|
||||
|
||||
|
||||
def _address_from_result(bot: Any, lat: float, lon: float, timeout: int) -> dict:
|
||||
db = getattr(bot, "db_manager", None)
|
||||
reverse_cache_key = f"reverse_{lat}_{lon}"
|
||||
if db is not None:
|
||||
cached = db.get_cached_json(reverse_cache_key, "geolocation")
|
||||
if cached:
|
||||
return cached
|
||||
try:
|
||||
rev = rate_limited_nominatim_reverse_sync(bot, f"{lat}, {lon}", timeout=timeout)
|
||||
if rev:
|
||||
info = rev.raw.get("address", {}) or {}
|
||||
if db is not None:
|
||||
db.cache_json(reverse_cache_key, info, "geolocation", cache_hours=720)
|
||||
return info
|
||||
except Exception:
|
||||
pass
|
||||
return {}
|
||||
|
||||
|
||||
def _cache_geocode(bot: Any, query: str, lat: float, lon: float) -> None:
|
||||
db = getattr(bot, "db_manager", None)
|
||||
if db is not None:
|
||||
try:
|
||||
db.cache_geocoding(query, lat, lon)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def geocode_city_best_effort(
|
||||
bot: Any,
|
||||
city: str,
|
||||
*,
|
||||
default_state: Optional[str] = None,
|
||||
default_country: Optional[str] = None,
|
||||
use_neighborhoods: bool = True,
|
||||
include_address_info: bool = True,
|
||||
timeout: int = 10,
|
||||
) -> tuple[Optional[float], Optional[float], Optional[dict]]:
|
||||
try:
|
||||
if "," in city:
|
||||
parts = [p.strip() for p in city.split(",")]
|
||||
if len(parts) >= 2 and _is_country_token(parts[1]):
|
||||
query = f"{parts[0]}, {parts[1]}"
|
||||
loc = rate_limited_nominatim_geocode_sync(bot, query, timeout=timeout)
|
||||
if loc:
|
||||
_cache_geocode(bot, query, loc.latitude, loc.longitude)
|
||||
addr = _address_from_result(bot, loc.latitude, loc.longitude, timeout)
|
||||
if not addr:
|
||||
addr = loc.raw.get("address", {}) or {}
|
||||
return loc.latitude, loc.longitude, addr if include_address_info else None
|
||||
|
||||
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")
|
||||
|
||||
lat, lon, address_info = geocode_city_sync(
|
||||
bot, city, default_state=default_state, default_country=default_country,
|
||||
include_address_info=include_address_info, timeout=timeout,
|
||||
)
|
||||
if lat is not None and lon is not None:
|
||||
return lat, lon, address_info or ({} if include_address_info else None)
|
||||
|
||||
if use_neighborhoods:
|
||||
for query in get_neighborhood_queries(city):
|
||||
loc = rate_limited_nominatim_geocode_sync(bot, query, timeout=timeout)
|
||||
if loc:
|
||||
_cache_geocode(bot, query, loc.latitude, loc.longitude)
|
||||
addr = _address_from_result(bot, loc.latitude, loc.longitude, timeout)
|
||||
if not addr:
|
||||
addr = loc.raw.get("address", {}) or {}
|
||||
return loc.latitude, loc.longitude, addr if include_address_info else None
|
||||
return None, None, None
|
||||
except Exception:
|
||||
return None, None, None
|
||||
|
||||
|
||||
def geocode_zipcode_best_effort(
|
||||
bot: Any,
|
||||
zipcode: str,
|
||||
*,
|
||||
default_state: Optional[str] = None,
|
||||
use_structured_zip: bool = True,
|
||||
include_address_info: bool = True,
|
||||
timeout: int = 10,
|
||||
) -> tuple[Optional[float], Optional[float], Optional[dict]]:
|
||||
zip_code = zipcode.strip()
|
||||
if default_state is None:
|
||||
default_state = bot.config.get("Weather", "default_state", fallback="")
|
||||
location_result = None
|
||||
try:
|
||||
if zip_code in ZIP_CODE_OVERRIDES:
|
||||
mapped = ZIP_CODE_OVERRIDES[zip_code]
|
||||
try:
|
||||
result = rate_limited_nominatim_geocode_sync(bot, mapped, timeout=timeout)
|
||||
if result and getattr(result, "address", None):
|
||||
location_result = result
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if use_structured_zip and not location_result:
|
||||
structured_queries: list[Union[str, Mapping[str, str]]] = [
|
||||
{"postalcode": zip_code, "country": "US"},
|
||||
{"postalcode": zip_code, "state": default_state or "", "country": "US"},
|
||||
{"postalcode": zip_code, "countrycode": "US"},
|
||||
]
|
||||
for query in structured_queries:
|
||||
try:
|
||||
result = rate_limited_nominatim_geocode_sync(bot, query, timeout=timeout)
|
||||
if result and getattr(result, "address", None):
|
||||
addr_l = result.address.lower()
|
||||
if "united states" in addr_l or "usa" in addr_l:
|
||||
if default_state and (
|
||||
default_state in result.address or "washington" in addr_l
|
||||
):
|
||||
location_result = result
|
||||
break
|
||||
if not location_result:
|
||||
location_result = result
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if location_result:
|
||||
lat, lon = location_result.latitude, location_result.longitude
|
||||
else:
|
||||
lat, lon = geocode_zipcode_sync(bot, zip_code, timeout=timeout)
|
||||
|
||||
if lat is None or lon is None:
|
||||
return None, None, None
|
||||
address_info = _address_from_result(bot, lat, lon, timeout) if include_address_info else None
|
||||
return lat, lon, address_info
|
||||
except Exception:
|
||||
return None, None, None
|
||||
|
||||
|
||||
def get_bot_lat_lon(bot: Any) -> Optional[tuple[float, float]]:
|
||||
try:
|
||||
lat = bot.config.getfloat("Bot", "bot_latitude", fallback=None)
|
||||
lon = bot.config.getfloat("Bot", "bot_longitude", fallback=None)
|
||||
if lat is not None and lon is not None and -90 <= lat <= 90 and -180 <= lon <= 180:
|
||||
return (lat, lon)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def get_config_default_lat_lon(bot: Any, section: str) -> Optional[tuple[float, float]]:
|
||||
try:
|
||||
if not bot.config.has_section(section):
|
||||
return None
|
||||
lat = bot.config.getfloat(section, "default_lat", fallback=None)
|
||||
lon = bot.config.getfloat(section, "default_lon", fallback=None)
|
||||
if lat is not None and lon is not None and -90 <= lat <= 90 and -180 <= lon <= 180:
|
||||
return (lat, lon)
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def get_companion_lat_lon(bot: Any, message: Any) -> Optional[tuple[float, float]]:
|
||||
try:
|
||||
sender_pubkey = getattr(message, "sender_pubkey", None)
|
||||
if not sender_pubkey or not hasattr(bot, "db_manager"):
|
||||
return None
|
||||
query = """
|
||||
SELECT latitude, longitude
|
||||
FROM complete_contact_tracking
|
||||
WHERE public_key = ?
|
||||
AND latitude IS NOT NULL AND longitude IS NOT NULL
|
||||
AND latitude != 0 AND longitude != 0
|
||||
ORDER BY COALESCE(last_advert_timestamp, last_heard) DESC
|
||||
LIMIT 1
|
||||
"""
|
||||
results = bot.db_manager.execute_query(query, (sender_pubkey,))
|
||||
if results:
|
||||
row = results[0]
|
||||
return (float(row["latitude"]), float(row["longitude"]))
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def lookup_repeater_lat_lon(
|
||||
bot: Any, repeater_name: str
|
||||
) -> Optional[tuple[float, float, str]]:
|
||||
try:
|
||||
if not hasattr(bot, "db_manager"):
|
||||
return None
|
||||
query = """
|
||||
SELECT latitude, longitude, name
|
||||
FROM complete_contact_tracking
|
||||
WHERE role IN ('repeater', 'roomserver')
|
||||
AND latitude IS NOT NULL AND longitude IS NOT NULL
|
||||
AND latitude != 0 AND longitude != 0
|
||||
AND LOWER(name) LIKE LOWER(?)
|
||||
ORDER BY
|
||||
CASE
|
||||
WHEN LOWER(name) = LOWER(?) THEN 1
|
||||
WHEN LOWER(name) LIKE LOWER(?) THEN 2
|
||||
ELSE 3
|
||||
END,
|
||||
COALESCE(last_advert_timestamp, last_heard) DESC
|
||||
LIMIT 1
|
||||
"""
|
||||
exact = repeater_name.strip()
|
||||
results = bot.db_manager.execute_query(
|
||||
query, (f"%{exact}%", exact, f"{exact}%")
|
||||
)
|
||||
if results:
|
||||
row = results[0]
|
||||
lat, lon = row.get("latitude"), row.get("longitude")
|
||||
if lat is not None and lon is not None:
|
||||
return float(lat), float(lon), row.get("name") or exact
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _display_from_address(
|
||||
address_info: Optional[dict],
|
||||
*,
|
||||
query: str,
|
||||
default_state: Optional[str],
|
||||
label_style: LabelStyle,
|
||||
lat: Optional[float] = None,
|
||||
lon: Optional[float] = None,
|
||||
bot: Any = None,
|
||||
timeout: int = 10,
|
||||
zippopotam: bool = False,
|
||||
location_type: Optional[str] = None,
|
||||
) -> Optional[str]:
|
||||
if label_style == "numeric" and lat is not None and lon is not None:
|
||||
return f"{lat:.1f},{lon:.1f}"
|
||||
if label_style == "query":
|
||||
return query
|
||||
if label_style == "city_region":
|
||||
if location_type == "zipcode" and zippopotam:
|
||||
zname = zip_to_city_string(query, timeout=timeout)
|
||||
if zname:
|
||||
return f"{zname} ({query.strip()})"
|
||||
if lat is not None and lon is not None and bot is not None:
|
||||
city, suffix = reverse_geocode_region(bot, lat, lon, timeout=timeout)
|
||||
if location_type == "city":
|
||||
typed = city_display_name(query, suffix)
|
||||
return join_location(typed, suffix) or query
|
||||
if city:
|
||||
return join_location(city, suffix)
|
||||
return query
|
||||
if address_info:
|
||||
city = (
|
||||
address_info.get("city") or address_info.get("town")
|
||||
or address_info.get("village") or address_info.get("hamlet")
|
||||
or address_info.get("municipality") or query
|
||||
)
|
||||
country = address_info.get("country", "")
|
||||
state = address_info.get("state", "")
|
||||
if country in ("United States", "US", "United States of America"):
|
||||
abbr, _ = normalize_us_state(state) if state else (None, None)
|
||||
suffix = abbr or state or default_state or ""
|
||||
else:
|
||||
suffix = country or address_info.get("province") or default_state or ""
|
||||
if suffix in ("United States", "United States of America"):
|
||||
suffix = "USA"
|
||||
full = f"{city}, {suffix}" if suffix else str(city)
|
||||
return abbreviate_location(full, max_length=30)
|
||||
if label_style == "abbreviated" and location_type == "coordinates" and lat is not None and lon is not None:
|
||||
return f"{lat:.3f},{lon:.3f}"
|
||||
if location_type == "zipcode":
|
||||
return query.strip()
|
||||
return query
|
||||
|
||||
|
||||
def resolve_location(
|
||||
bot: Any,
|
||||
raw: Optional[str],
|
||||
*,
|
||||
options: Optional[ResolveOptions] = None,
|
||||
) -> ResolvedLocation:
|
||||
opts = options or OPTIONS_AQI
|
||||
default_state = opts.default_state
|
||||
default_country = opts.default_country
|
||||
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")
|
||||
|
||||
if raw is None or not str(raw).strip():
|
||||
if opts.fallback_coords:
|
||||
lat, lon = opts.fallback_coords
|
||||
label = opts.fallback_label or _display_from_address(
|
||||
None, query=f"{lat},{lon}", default_state=default_state,
|
||||
label_style=opts.label_style, lat=lat, lon=lon, bot=bot,
|
||||
timeout=opts.timeout,
|
||||
)
|
||||
return ResolvedLocation(
|
||||
lat=lat, lon=lon, location_type="fallback", query="",
|
||||
display_name=label, address_info=None,
|
||||
)
|
||||
return ResolvedLocation(
|
||||
lat=None, lon=None, location_type=None, query="",
|
||||
display_name=None, address_info=None, error="no_location",
|
||||
)
|
||||
|
||||
text = str(raw).strip()
|
||||
if opts.allow_repeater_names:
|
||||
hit = lookup_repeater_lat_lon(bot, text)
|
||||
if hit:
|
||||
lat, lon, name = hit
|
||||
return ResolvedLocation(
|
||||
lat=lat, lon=lon, location_type="repeater", query=name,
|
||||
display_name=name, address_info=None,
|
||||
)
|
||||
|
||||
region_note: Optional[str] = None
|
||||
query, location_type = classify_location(text, use_international_cities=False)
|
||||
|
||||
if location_type == "city":
|
||||
if opts.use_region_capitals:
|
||||
cap = region_capital_query(query)
|
||||
if cap:
|
||||
query = cap
|
||||
region_note = REGION_DEFAULT_NOTE
|
||||
if opts.use_international_cities:
|
||||
query, _ = classify_location(query, use_international_cities=True)
|
||||
|
||||
if location_type == "coordinates":
|
||||
parsed, err, detail = parse_coordinates_detailed(query)
|
||||
if err or not parsed:
|
||||
return ResolvedLocation(
|
||||
lat=None, lon=None, location_type="coordinates", query=query,
|
||||
display_name=None, address_info=None,
|
||||
error=err or "invalid_coordinates", error_detail=detail,
|
||||
)
|
||||
lat, lon = parsed
|
||||
display = _display_from_address(
|
||||
None, query=query, default_state=default_state,
|
||||
label_style=opts.label_style, lat=lat, lon=lon, bot=bot,
|
||||
timeout=opts.timeout, location_type="coordinates",
|
||||
)
|
||||
return ResolvedLocation(
|
||||
lat=lat, lon=lon, location_type="coordinates", query=query,
|
||||
display_name=display, address_info=None, region_note=region_note,
|
||||
)
|
||||
|
||||
if location_type == "zipcode":
|
||||
lat, lon, address_info = geocode_zipcode_best_effort(
|
||||
bot, query, default_state=default_state,
|
||||
use_structured_zip=opts.use_structured_zip,
|
||||
include_address_info=opts.include_address_info, timeout=opts.timeout,
|
||||
)
|
||||
if lat is None or lon is None:
|
||||
return ResolvedLocation(
|
||||
lat=None, lon=None, location_type="zipcode", query=query,
|
||||
display_name=None, address_info=None, error="no_location_zipcode",
|
||||
)
|
||||
display = _display_from_address(
|
||||
address_info, query=query, default_state=default_state,
|
||||
label_style=opts.label_style, lat=lat, lon=lon, bot=bot,
|
||||
timeout=opts.timeout, zippopotam=opts.use_zippopotam_labels,
|
||||
location_type="zipcode",
|
||||
)
|
||||
return ResolvedLocation(
|
||||
lat=lat, lon=lon, location_type="zipcode", query=query.strip(),
|
||||
display_name=display, address_info=address_info, region_note=region_note,
|
||||
)
|
||||
|
||||
lat, lon, address_info = geocode_city_best_effort(
|
||||
bot, query, default_state=default_state, default_country=default_country,
|
||||
use_neighborhoods=opts.use_neighborhoods,
|
||||
include_address_info=opts.include_address_info, timeout=opts.timeout,
|
||||
)
|
||||
if lat is None or lon is None:
|
||||
return ResolvedLocation(
|
||||
lat=None, lon=None, location_type="city", query=query,
|
||||
display_name=None, address_info=None, error="no_location_city",
|
||||
region_note=region_note,
|
||||
)
|
||||
display = _display_from_address(
|
||||
address_info, query=query, default_state=default_state,
|
||||
label_style=opts.label_style, lat=lat, lon=lon, bot=bot,
|
||||
timeout=opts.timeout, location_type="city",
|
||||
)
|
||||
return ResolvedLocation(
|
||||
lat=lat, lon=lon, location_type="city", query=query,
|
||||
display_name=display, address_info=address_info, region_note=region_note,
|
||||
)
|
||||
|
||||
|
||||
async def resolve_location_async(
|
||||
bot: Any,
|
||||
raw: Optional[str],
|
||||
*,
|
||||
options: Optional[ResolveOptions] = None,
|
||||
) -> ResolvedLocation:
|
||||
"""Async twin of ``resolve_location`` (same behavior via thread offload)."""
|
||||
opts = options or OPTIONS_PREFIX
|
||||
return await asyncio.to_thread(resolve_location, bot, raw, options=opts)
|
||||
|
||||
|
||||
def resolve_with_message_fallbacks(
|
||||
bot: Any,
|
||||
raw: Optional[str],
|
||||
message: Any,
|
||||
*,
|
||||
config_section: Optional[str] = None,
|
||||
options: Optional[ResolveOptions] = None,
|
||||
) -> ResolvedLocation:
|
||||
opts = options or OPTIONS_AQI
|
||||
if raw is None or not str(raw).strip():
|
||||
for coords in (
|
||||
get_companion_lat_lon(bot, message),
|
||||
get_config_default_lat_lon(bot, config_section) if config_section else None,
|
||||
get_bot_lat_lon(bot),
|
||||
):
|
||||
if coords:
|
||||
return resolve_location(bot, None, options=replace(opts, fallback_coords=coords))
|
||||
return ResolvedLocation(
|
||||
lat=None, lon=None, location_type=None, query="",
|
||||
display_name=None, address_info=None, error="no_location",
|
||||
)
|
||||
return resolve_location(bot, raw, options=opts)
|
||||
+9
-5
@@ -12,7 +12,7 @@ import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional, Union
|
||||
from typing import Any, Mapping, Optional, Union
|
||||
|
||||
try:
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
@@ -863,12 +863,14 @@ def get_nominatim_geocoder(user_agent: str = "meshcore-bot", timeout: int = 10)
|
||||
return Nominatim(user_agent=user_agent, timeout=timeout)
|
||||
|
||||
|
||||
async def rate_limited_nominatim_geocode(bot: Any, query: str, timeout: int = 10) -> Optional[Any]:
|
||||
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.
|
||||
query: Location query string or structured Nominatim dict (e.g. postalcode).
|
||||
timeout: Request timeout in seconds.
|
||||
|
||||
Returns:
|
||||
@@ -921,12 +923,14 @@ async def rate_limited_nominatim_reverse(bot: Any, coordinates: str, timeout: in
|
||||
return result
|
||||
|
||||
|
||||
def rate_limited_nominatim_geocode_sync(bot: Any, query: str, timeout: int = 10) -> Optional[Any]:
|
||||
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.
|
||||
query: Location query string or structured Nominatim dict (e.g. postalcode).
|
||||
timeout: Request timeout in seconds.
|
||||
|
||||
Returns:
|
||||
|
||||
@@ -0,0 +1,317 @@
|
||||
"""Unit tests for modules.location (shared location service)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import configparser
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from modules.location import (
|
||||
GEOCODE_CACHE_CAP,
|
||||
INTERNATIONAL_CITIES,
|
||||
OPTIONS_AQI,
|
||||
OPTIONS_PREFIX,
|
||||
OPTIONS_RAIN,
|
||||
ResolveOptions,
|
||||
cache_put,
|
||||
classify_location,
|
||||
geocode_city_best_effort,
|
||||
get_neighborhood_queries,
|
||||
parse_coordinates,
|
||||
parse_coordinates_detailed,
|
||||
resolve_location,
|
||||
resolve_location_async,
|
||||
zip_to_city_string,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bot():
|
||||
b = MagicMock()
|
||||
cfg = configparser.ConfigParser()
|
||||
cfg.add_section("Weather")
|
||||
cfg.set("Weather", "default_state", "WA")
|
||||
cfg.set("Weather", "default_country", "US")
|
||||
cfg.add_section("Bot")
|
||||
b.config = cfg
|
||||
b.db_manager = Mock()
|
||||
b.db_manager.get_cached_geocoding = Mock(return_value=(None, None))
|
||||
b.db_manager.cache_geocoding = Mock()
|
||||
b.db_manager.get_cached_json = Mock(return_value=None)
|
||||
b.db_manager.cache_json = Mock()
|
||||
b.db_manager.execute_query = Mock(return_value=[])
|
||||
b.logger = Mock()
|
||||
rl = Mock()
|
||||
rl.wait_for_request_sync = Mock()
|
||||
rl.record_request = Mock()
|
||||
b.nominatim_rate_limiter = rl
|
||||
return b
|
||||
|
||||
|
||||
def _loc(lat=47.6, lon=-122.3, address="Seattle, WA, USA", info=None):
|
||||
loc = Mock()
|
||||
loc.latitude = lat
|
||||
loc.longitude = lon
|
||||
loc.address = address
|
||||
loc.raw = {"address": info or {"city": "Seattle", "state": "Washington", "country": "United States", "country_code": "us"}}
|
||||
return loc
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestClassifyLocation:
|
||||
def test_coordinates(self):
|
||||
q, t = classify_location("47.6,-122.3")
|
||||
assert t == "coordinates"
|
||||
|
||||
def test_zip_with_whitespace(self):
|
||||
q, t = classify_location(" 98101 ")
|
||||
assert t == "zipcode"
|
||||
assert q == "98101"
|
||||
|
||||
def test_intl_single(self):
|
||||
q, t = classify_location("london")
|
||||
assert t == "city"
|
||||
assert q == "london, uk"
|
||||
|
||||
def test_intl_multiword(self):
|
||||
q, t = classify_location("mexico city")
|
||||
assert t == "city"
|
||||
assert q == "mexico city, mexico"
|
||||
|
||||
def test_space_country(self):
|
||||
q, t = classify_location("vancouver canada")
|
||||
assert t == "city"
|
||||
assert q == "vancouver, canada"
|
||||
|
||||
def test_intl_disabled(self):
|
||||
q, t = classify_location("london", use_international_cities=False)
|
||||
assert q == "london"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestParseCoordinates:
|
||||
def test_valid(self):
|
||||
assert parse_coordinates("47.6, -122.3") == pytest.approx((47.6, -122.3))
|
||||
|
||||
def test_invalid_range(self):
|
||||
assert parse_coordinates("200,0") is None
|
||||
|
||||
def test_invalid_format(self):
|
||||
assert parse_coordinates("seattle") is None
|
||||
|
||||
def test_detailed_invalid_latitude(self):
|
||||
coords, err, detail = parse_coordinates_detailed("91,0")
|
||||
assert coords is None
|
||||
assert err == "invalid_latitude"
|
||||
assert detail == "91.0"
|
||||
|
||||
def test_detailed_invalid_longitude(self):
|
||||
coords, err, detail = parse_coordinates_detailed("0,200")
|
||||
assert coords is None
|
||||
assert err == "invalid_longitude"
|
||||
assert detail == "200.0"
|
||||
|
||||
def test_detailed_valid(self):
|
||||
coords, err, detail = parse_coordinates_detailed("47.6,-122.3")
|
||||
assert coords == pytest.approx((47.6, -122.3))
|
||||
assert err is None
|
||||
assert detail is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestNeighborhoods:
|
||||
def test_greenwood(self):
|
||||
qs = get_neighborhood_queries("greenwood")
|
||||
assert qs[0] == "greenwood, Seattle, WA, USA"
|
||||
|
||||
def test_unknown(self):
|
||||
assert get_neighborhood_queries("nowhere") == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestKazakhstanDedup:
|
||||
def test_single_canonical(self):
|
||||
assert INTERNATIONAL_CITIES["kazakhstan"] == "nur-sultan, kazakhstan"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestCountryTokenPath:
|
||||
def test_france_uses_direct_nominatim(self, bot):
|
||||
loc = _loc(48.85, 2.35, "Paris, France", {"city": "Paris", "country": "France"})
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync", return_value=loc
|
||||
) as geo, patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
) as shared, patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync", return_value=loc
|
||||
):
|
||||
lat, lon, _ = geocode_city_best_effort(bot, "paris, france")
|
||||
geo.assert_called()
|
||||
shared.assert_not_called()
|
||||
assert lat == pytest.approx(48.85)
|
||||
|
||||
def test_texas_uses_shared_geocode(self, bot):
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(33.66, -95.55, {"city": "Paris", "state": "Texas", "country": "United States"}),
|
||||
) as shared, patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
) as geo:
|
||||
lat, lon, _ = geocode_city_best_effort(bot, "paris, texas")
|
||||
shared.assert_called_once()
|
||||
for call in geo.call_args_list:
|
||||
assert call[0][1] != "paris, texas"
|
||||
assert lat == pytest.approx(33.66)
|
||||
|
||||
def test_arbitrary_second_token_not_country(self, bot):
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(1.0, 2.0, {"city": "Foo"}),
|
||||
) as shared, patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
) as geo:
|
||||
geocode_city_best_effort(bot, "foo, bar")
|
||||
geocode_city_best_effort(bot, "springfield, greene")
|
||||
assert shared.call_count == 2
|
||||
for call in geo.call_args_list:
|
||||
q = call[0][1]
|
||||
assert q not in ("foo, bar", "springfield, greene")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestResolveLocation:
|
||||
def test_coords(self, bot):
|
||||
r = resolve_location(bot, "47.6,-122.3", options=OPTIONS_AQI)
|
||||
assert r.location_type == "coordinates"
|
||||
assert r.lat == pytest.approx(47.6)
|
||||
assert r.error is None
|
||||
|
||||
def test_invalid_latitude_error(self, bot):
|
||||
r = resolve_location(bot, "91,0", options=OPTIONS_AQI)
|
||||
assert r.error == "invalid_latitude"
|
||||
assert r.error_detail == "91.0"
|
||||
|
||||
def test_invalid_longitude_error(self, bot):
|
||||
r = resolve_location(bot, "0,200", options=OPTIONS_AQI)
|
||||
assert r.error == "invalid_longitude"
|
||||
assert r.error_detail == "200.0"
|
||||
|
||||
def test_empty_no_fallback(self, bot):
|
||||
r = resolve_location(bot, None, options=OPTIONS_AQI)
|
||||
assert r.error == "no_location"
|
||||
|
||||
def test_empty_with_fallback(self, bot):
|
||||
opts = ResolveOptions(fallback_coords=(48.0, -122.0), label_style="numeric")
|
||||
r = resolve_location(bot, "", options=opts)
|
||||
assert r.lat == pytest.approx(48.0)
|
||||
assert r.location_type == "fallback"
|
||||
|
||||
def test_city_via_geocode(self, bot):
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(47.6, -122.3, {"city": "Seattle", "state": "Washington", "country": "United States"}),
|
||||
):
|
||||
r = resolve_location(bot, "seattle", options=OPTIONS_AQI)
|
||||
assert r.lat == pytest.approx(47.6)
|
||||
assert r.location_type == "city"
|
||||
|
||||
def test_region_capitals_flag(self, bot):
|
||||
loc = _loc(48.85, 2.35, "Paris, France", {"city": "Paris", "country": "France"})
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
return_value=loc,
|
||||
), patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync",
|
||||
return_value=loc,
|
||||
):
|
||||
r = resolve_location(bot, "france", options=OPTIONS_RAIN)
|
||||
assert r.region_note is not None
|
||||
assert r.lat == pytest.approx(48.85)
|
||||
assert "Paris" in r.query or "paris" in r.query.lower()
|
||||
|
||||
def test_repeater_lookup(self, bot):
|
||||
bot.db_manager.execute_query.return_value = [
|
||||
{"latitude": 47.1, "longitude": -122.1, "name": "KR7ABC"}
|
||||
]
|
||||
r = resolve_location(bot, "KR7ABC", options=OPTIONS_PREFIX)
|
||||
assert r.location_type == "repeater"
|
||||
assert r.lat == pytest.approx(47.1)
|
||||
|
||||
def test_zip_override(self, bot):
|
||||
loc = _loc(47.45, -122.46, "Vashon, WA, USA")
|
||||
with patch("modules.location.rate_limited_nominatim_geocode_sync", return_value=loc), patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync", return_value=loc
|
||||
):
|
||||
r = resolve_location(bot, "98013", options=OPTIONS_AQI)
|
||||
assert r.lat == pytest.approx(47.45)
|
||||
assert r.location_type == "zipcode"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestZipCacheCap:
|
||||
def test_cache_put_evicts_oldest(self):
|
||||
cache: dict[str, str] = {}
|
||||
for i in range(GEOCODE_CACHE_CAP + 5):
|
||||
cache_put(cache, f"{i:05d}", f"City{i}")
|
||||
assert len(cache) == GEOCODE_CACHE_CAP
|
||||
assert "00000" not in cache
|
||||
assert f"{GEOCODE_CACHE_CAP + 4:05d}" in cache
|
||||
|
||||
def test_zip_to_city_string_uses_capped_cache(self):
|
||||
cache: dict[str, str] = {}
|
||||
for i in range(GEOCODE_CACHE_CAP):
|
||||
cache[f"{i:05d}"] = f"Old{i}"
|
||||
mock_resp = Mock()
|
||||
mock_resp.ok = True
|
||||
mock_resp.json.return_value = {
|
||||
"places": [{"place name": "Seattle", "state abbreviation": "WA"}]
|
||||
}
|
||||
with patch("modules.location.requests.get", return_value=mock_resp):
|
||||
name = zip_to_city_string("99999", cache=cache)
|
||||
assert name == "Seattle, WA"
|
||||
assert len(cache) == GEOCODE_CACHE_CAP
|
||||
assert "99999" in cache
|
||||
assert "00000" not in cache
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAsyncParity:
|
||||
def test_coords_match_sync(self, bot):
|
||||
sync = resolve_location(bot, "47.6,-122.3", options=OPTIONS_AQI)
|
||||
async_r = asyncio.run(resolve_location_async(bot, "47.6,-122.3", options=OPTIONS_AQI))
|
||||
assert async_r.lat == sync.lat
|
||||
assert async_r.lon == sync.lon
|
||||
assert async_r.error == sync.error
|
||||
assert async_r.location_type == sync.location_type
|
||||
|
||||
def test_empty_match_sync(self, bot):
|
||||
sync = resolve_location(bot, None, options=OPTIONS_PREFIX)
|
||||
async_r = asyncio.run(resolve_location_async(bot, None, options=OPTIONS_PREFIX))
|
||||
assert async_r.error == sync.error
|
||||
|
||||
def test_city_match_sync(self, bot):
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(47.6, -122.3, {"city": "Seattle", "state": "Washington", "country": "United States"}),
|
||||
):
|
||||
sync = resolve_location(bot, "seattle", options=OPTIONS_AQI)
|
||||
async_r = asyncio.run(resolve_location_async(bot, "seattle", options=OPTIONS_AQI))
|
||||
assert async_r.lat == sync.lat
|
||||
assert async_r.display_name == sync.display_name
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRainReexports:
|
||||
def test_helpers_importable_from_rain(self):
|
||||
from modules.commands.rain_command import (
|
||||
city_display_name,
|
||||
join_location,
|
||||
reverse_geocode_region,
|
||||
titlecase_location,
|
||||
)
|
||||
assert titlecase_location("memphis") == "Memphis"
|
||||
assert join_location("Paris", "France") == "Paris, France"
|
||||
assert city_display_name("london ky") == "London"
|
||||
assert callable(reverse_geocode_region)
|
||||
@@ -0,0 +1,700 @@
|
||||
"""Characterization tests for current command location-resolution behavior.
|
||||
|
||||
These lock today's semantics (before modules/location.py) so regressions are
|
||||
visible after the shared location service lands. All Nominatim / geocode /
|
||||
OpenMeteo calls are mocked — no live network.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import configparser
|
||||
from typing import Optional
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.conftest import mock_message
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared fixtures / helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _base_config() -> configparser.ConfigParser:
|
||||
cfg = configparser.ConfigParser()
|
||||
cfg.add_section("Bot")
|
||||
cfg.set("Bot", "bot_name", "TestBot")
|
||||
cfg.set("Bot", "timezone", "America/Los_Angeles")
|
||||
cfg.add_section("Channels")
|
||||
cfg.set("Channels", "monitor_channels", "general")
|
||||
cfg.set("Channels", "respond_to_dms", "true")
|
||||
cfg.add_section("Keywords")
|
||||
cfg.add_section("Weather")
|
||||
cfg.set("Weather", "default_state", "WA")
|
||||
cfg.set("Weather", "default_country", "US")
|
||||
cfg.set("Weather", "weather_provider", "noaa")
|
||||
return cfg
|
||||
|
||||
|
||||
def _make_bot(**extra_sections: dict[str, dict[str, str]]) -> MagicMock:
|
||||
bot = MagicMock()
|
||||
bot.logger = Mock()
|
||||
cfg = _base_config()
|
||||
for section, values in extra_sections.items():
|
||||
if not cfg.has_section(section):
|
||||
cfg.add_section(section)
|
||||
for k, v in values.items():
|
||||
cfg.set(section, k, v)
|
||||
bot.config = cfg
|
||||
bot.translator = MagicMock()
|
||||
bot.translator.translate = Mock(side_effect=lambda key, **kw: key)
|
||||
bot.command_manager = MagicMock()
|
||||
bot.command_manager.monitor_channels = ["general"]
|
||||
bot.command_manager.send_response = AsyncMock(return_value=True)
|
||||
bot.db_manager = MagicMock()
|
||||
bot.db_manager.get_cached_geocoding = Mock(return_value=(None, None))
|
||||
bot.db_manager.cache_geocoding = Mock()
|
||||
bot.db_manager.get_cached_json = Mock(return_value=None)
|
||||
bot.db_manager.cache_json = Mock()
|
||||
bot.db_manager.execute_query = Mock(return_value=[])
|
||||
bot.nominatim_rate_limiter = Mock()
|
||||
bot.nominatim_rate_limiter.wait_for_request_sync = Mock()
|
||||
bot.nominatim_rate_limiter.record_request = Mock()
|
||||
return bot
|
||||
|
||||
|
||||
def _make_geopy_location(
|
||||
lat: float = 47.6062,
|
||||
lon: float = -122.3321,
|
||||
address: str = "Seattle, Washington, United States",
|
||||
address_info: Optional[dict] = None,
|
||||
) -> Mock:
|
||||
loc = Mock()
|
||||
loc.latitude = lat
|
||||
loc.longitude = lon
|
||||
loc.address = address
|
||||
loc.raw = {"address": address_info or {"city": "Seattle", "state": "Washington", "country": "United States", "country_code": "us"}}
|
||||
return loc
|
||||
|
||||
|
||||
def _run(coro):
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AQI — classification via execute() (capture get_aqi_for_location args)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def aqi_cmd():
|
||||
bot = _make_bot(Aqi_Command={"enabled": "true"})
|
||||
with patch("modules.commands.aqi_command.get_nominatim_geocoder", return_value=Mock()), \
|
||||
patch("modules.commands.aqi_command.requests_cache.CachedSession"), \
|
||||
patch("modules.commands.aqi_command.retry", side_effect=lambda s, **kw: s), \
|
||||
patch("modules.commands.aqi_command.openmeteo_requests.Client", return_value=Mock()):
|
||||
from modules.commands.aqi_command import AqiCommand
|
||||
cmd = AqiCommand(bot)
|
||||
cmd.send_response = AsyncMock(return_value=True)
|
||||
cmd.record_execution = Mock()
|
||||
return cmd
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiLocationClassification:
|
||||
"""Lock AQI's front-door location typing and rewrites.
|
||||
|
||||
execute() passes the raw user location into get_aqi_for_location; typing and
|
||||
intl rewrites happen inside resolve_location / classify_location.
|
||||
"""
|
||||
|
||||
def _capture(self, cmd, content: str) -> tuple[str, str]:
|
||||
from modules.location import classify_location
|
||||
|
||||
with patch.object(cmd, "get_aqi_for_location", new_callable=AsyncMock) as m:
|
||||
m.return_value = "ok"
|
||||
_run(cmd.execute(mock_message(content=content)))
|
||||
assert m.called, f"get_aqi_for_location not called for {content!r}"
|
||||
raw = m.call_args[0][0]
|
||||
return classify_location(raw, use_international_cities=True)
|
||||
|
||||
def test_coordinates_basic(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi 47.6,-122.3")
|
||||
assert typ == "coordinates"
|
||||
assert loc == "47.6,-122.3"
|
||||
|
||||
def test_coordinates_with_spaces(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi 47.6, -122.3")
|
||||
assert typ == "coordinates"
|
||||
assert loc == "47.6, -122.3"
|
||||
|
||||
def test_coordinates_negative_lat(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi -33.9, 151.2")
|
||||
assert typ == "coordinates"
|
||||
|
||||
def test_zipcode_five_digits(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi 98101")
|
||||
assert typ == "zipcode"
|
||||
assert loc == "98101"
|
||||
|
||||
def test_plain_city(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi seattle")
|
||||
assert typ == "city"
|
||||
assert loc == "seattle"
|
||||
|
||||
def test_city_state_comma_kept(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi paris, tx")
|
||||
assert typ == "city"
|
||||
assert loc == "paris, tx"
|
||||
|
||||
def test_space_separated_city_country_rewritten(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi vancouver canada")
|
||||
assert typ == "city"
|
||||
assert loc == "vancouver, canada"
|
||||
|
||||
def test_united_kingdom_rewritten_to_uk(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi london united kingdom")
|
||||
assert typ == "city"
|
||||
assert loc == "london, uk"
|
||||
|
||||
def test_international_city_single_token_london(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi london")
|
||||
assert typ == "city"
|
||||
assert loc == "london, uk"
|
||||
|
||||
def test_international_city_single_token_tokyo(self, aqi_cmd):
|
||||
loc, typ = self._capture(aqi_cmd, "aqi tokyo")
|
||||
assert typ == "city"
|
||||
assert loc == "tokyo, japan"
|
||||
|
||||
def test_multiword_intl_key_rewritten(self, aqi_cmd):
|
||||
"""Multi-word intl keys rewrite like single-token ones."""
|
||||
loc, typ = self._capture(aqi_cmd, "aqi mexico city")
|
||||
assert typ == "city"
|
||||
assert loc == "mexico city, mexico"
|
||||
|
||||
def test_execute_passes_raw_location(self, aqi_cmd):
|
||||
with patch.object(aqi_cmd, "get_aqi_for_location", new_callable=AsyncMock) as m:
|
||||
m.return_value = "ok"
|
||||
_run(aqi_cmd.execute(mock_message(content="aqi mexico city")))
|
||||
assert m.call_args[0][0] == "mexico city"
|
||||
|
||||
def test_astronomical_early_out_skips_geocode(self, aqi_cmd):
|
||||
with patch.object(aqi_cmd, "get_aqi_for_location", new_callable=AsyncMock) as m:
|
||||
_run(aqi_cmd.execute(mock_message(content="aqi mars")))
|
||||
m.assert_not_called()
|
||||
aqi_cmd.send_response.assert_called_once()
|
||||
body = aqi_cmd.send_response.call_args[0][1]
|
||||
assert "Mars" in body or "mars" in body.lower() or "CO2" in body
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiNeighborhoodQueries:
|
||||
def test_seattle_neighborhood(self, aqi_cmd):
|
||||
queries = aqi_cmd.get_neighborhood_queries("greenwood")
|
||||
assert queries == [
|
||||
"greenwood, Seattle, WA, USA",
|
||||
"greenwood, Seattle, USA",
|
||||
]
|
||||
|
||||
def test_nyc_neighborhood(self, aqi_cmd):
|
||||
queries = aqi_cmd.get_neighborhood_queries("williamsburg")
|
||||
assert "New York" in queries[0]
|
||||
|
||||
def test_unknown_returns_empty(self, aqi_cmd):
|
||||
assert aqi_cmd.get_neighborhood_queries("notaneighborhood") == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiCityToLatLon:
|
||||
def test_uses_geocode_city_sync_for_plain_city(self, aqi_cmd):
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(47.6, -122.3, {"city": "Seattle"}),
|
||||
) as geo:
|
||||
lat, lon, addr = aqi_cmd.city_to_lat_lon("seattle")
|
||||
geo.assert_called_once()
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert lon == pytest.approx(-122.3)
|
||||
assert addr == {"city": "Seattle"}
|
||||
|
||||
def test_country_comma_path_uses_nominatim_directly(self, aqi_cmd):
|
||||
loc = _make_geopy_location(48.8566, 2.3522, "Paris, France", {"city": "Paris", "country": "France"})
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
return_value=loc,
|
||||
) as geo, patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
) as shared:
|
||||
lat, lon, addr = aqi_cmd.city_to_lat_lon("paris, france")
|
||||
geo.assert_called()
|
||||
shared.assert_not_called()
|
||||
assert lat == pytest.approx(48.8566)
|
||||
assert lon == pytest.approx(2.3522)
|
||||
|
||||
def test_neighborhood_fallback_when_geocode_city_fails(self, aqi_cmd):
|
||||
loc = _make_geopy_location(47.69, -122.35, "Greenwood, Seattle")
|
||||
with patch(
|
||||
"modules.location.geocode_city_sync",
|
||||
return_value=(None, None, None),
|
||||
), patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
return_value=loc,
|
||||
) as geo:
|
||||
lat, lon, addr = aqi_cmd.city_to_lat_lon("greenwood")
|
||||
assert lat == pytest.approx(47.69)
|
||||
assert any("Seattle" in str(c) for c in geo.call_args_list)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiZipcodePath:
|
||||
def test_zip_override_98013_queries_mapped_place(self, aqi_cmd):
|
||||
loc = _make_geopy_location(47.45, -122.46, "Vashon, Washington, United States")
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
return_value=loc,
|
||||
) as geo, patch.object(
|
||||
aqi_cmd, "get_openmeteo_aqi", return_value="🟢 20 (Good)"
|
||||
), patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync",
|
||||
return_value=loc,
|
||||
):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("98013", "zipcode"))
|
||||
assert "🟢" in result or "20" in result
|
||||
first_query = geo.call_args_list[0][0][1]
|
||||
assert "Vashon" in first_query
|
||||
|
||||
def test_zip_falls_back_to_geocode_zipcode_sync(self, aqi_cmd):
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync",
|
||||
return_value=None,
|
||||
), patch(
|
||||
"modules.location.geocode_zipcode_sync",
|
||||
return_value=(47.6, -122.3),
|
||||
) as zip_geo, patch.object(
|
||||
aqi_cmd, "get_openmeteo_aqi", return_value="ok"
|
||||
), patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync",
|
||||
return_value=None,
|
||||
):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("98101", "zipcode"))
|
||||
zip_geo.assert_called()
|
||||
assert result.startswith("98101:") or "ok" in result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiCoordinateErrors:
|
||||
def test_invalid_latitude_message(self, aqi_cmd):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("91,0"))
|
||||
assert result == "Invalid latitude: 91.0. Must be between -90 and 90."
|
||||
|
||||
def test_invalid_longitude_message(self, aqi_cmd):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("0,200"))
|
||||
assert result == "Invalid longitude: 200.0. Must be between -180 and 180."
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiPrefixBudget:
|
||||
def test_under_budget_includes_display_name(self, aqi_cmd):
|
||||
from modules.location import ResolvedLocation
|
||||
|
||||
resolved = ResolvedLocation(
|
||||
lat=47.6, lon=-122.3, location_type="city", query="seattle",
|
||||
display_name="Seattle, WA",
|
||||
address_info={"city": "Seattle", "state": "Washington", "country": "United States"},
|
||||
)
|
||||
with patch("modules.commands.aqi_command.resolve_location", return_value=resolved), \
|
||||
patch.object(aqi_cmd, "get_openmeteo_aqi", return_value="🟢 20"):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("seattle"))
|
||||
assert result.startswith("Seattle, WA:")
|
||||
|
||||
def test_over_budget_same_state_omits_prefix(self, aqi_cmd):
|
||||
from modules.location import ResolvedLocation
|
||||
|
||||
aqi_cmd.default_state = "WA"
|
||||
long_aqi = "X" * 120
|
||||
resolved = ResolvedLocation(
|
||||
lat=47.6, lon=-122.3, location_type="city", query="seattle",
|
||||
display_name="Seattle, WA",
|
||||
address_info={"city": "Seattle", "state": "Washington", "country": "United States"},
|
||||
)
|
||||
with patch("modules.commands.aqi_command.resolve_location", return_value=resolved), \
|
||||
patch.object(aqi_cmd, "get_openmeteo_aqi", return_value=long_aqi):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("seattle"))
|
||||
assert result == long_aqi
|
||||
assert not result.startswith("Seattle")
|
||||
|
||||
def test_over_budget_different_state_keeps_short_prefix(self, aqi_cmd):
|
||||
from modules.location import ResolvedLocation
|
||||
|
||||
aqi_cmd.default_state = "WA"
|
||||
long_aqi = "X" * 120
|
||||
resolved = ResolvedLocation(
|
||||
lat=30.27, lon=-97.74, location_type="city", query="austin",
|
||||
display_name="Austin, TX",
|
||||
address_info={"city": "Austin", "state": "Texas", "country": "United States"},
|
||||
)
|
||||
with patch("modules.commands.aqi_command.resolve_location", return_value=resolved), \
|
||||
patch.object(aqi_cmd, "get_openmeteo_aqi", return_value=long_aqi):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("austin"))
|
||||
assert result.endswith(long_aqi)
|
||||
assert result.startswith("Austin")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAqiIntlResolveWiring:
|
||||
def test_london_rewrites_via_resolve(self, aqi_cmd):
|
||||
loc = _make_geopy_location(
|
||||
51.5, -0.12, "London, UK", {"city": "London", "country": "United Kingdom"}
|
||||
)
|
||||
with patch(
|
||||
"modules.location.rate_limited_nominatim_geocode_sync", return_value=loc
|
||||
) as geo, patch(
|
||||
"modules.location.geocode_city_sync", return_value=(None, None, None)
|
||||
), patch(
|
||||
"modules.location.rate_limited_nominatim_reverse_sync", return_value=loc
|
||||
), patch.object(aqi_cmd, "get_openmeteo_aqi", return_value="ok"):
|
||||
result = _run(aqi_cmd.get_aqi_for_location("london"))
|
||||
assert "ok" in result
|
||||
# Intl rewrite should query "london, uk" on the country path
|
||||
assert any(
|
||||
isinstance(c[0][1], str) and "london" in c[0][1].lower()
|
||||
for c in geo.call_args_list
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WX — type detection + thin geocode wrappers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wx_cmd():
|
||||
bot = _make_bot(Wx_Command={"enabled": "true"})
|
||||
from modules.commands.wx_command import WxCommand
|
||||
return WxCommand(bot)
|
||||
|
||||
|
||||
def _wx_classify(location: str) -> str:
|
||||
"""Desired shared classify contract (AQI / location.classify_location)."""
|
||||
import re
|
||||
if re.match(r'^\s*-?\d+\.?\d*\s*,\s*-?\d+\.?\d*\s*$', location):
|
||||
return "coordinates"
|
||||
if re.match(r'^\s*\d{5}\s*$', location):
|
||||
return "zipcode"
|
||||
return "city"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestWxLocationClassification:
|
||||
def test_coordinates(self):
|
||||
assert _wx_classify("47.6,-122.3") == "coordinates"
|
||||
assert _wx_classify("47.6, -122.3") == "coordinates"
|
||||
|
||||
def test_zipcode_no_whitespace(self):
|
||||
assert _wx_classify("98101") == "zipcode"
|
||||
|
||||
def test_zipcode_with_whitespace_is_zipcode(self):
|
||||
"""Desired: surrounding whitespace does not demote ZIP to city."""
|
||||
assert _wx_classify(" 98101 ") == "zipcode"
|
||||
|
||||
def test_wx_execute_accepts_zip_with_whitespace(self, wx_cmd):
|
||||
"""Live WxCommand.execute type-detect must treat ' 98101 ' as zipcode."""
|
||||
import inspect
|
||||
import re
|
||||
|
||||
from modules.commands.wx_command import WxCommand
|
||||
|
||||
source = inspect.getsource(WxCommand.execute)
|
||||
assert r"^\s*\d{5}\s*$" in source
|
||||
assert re.match(r"^\s*\d{5}\s*$", " 98101 ")
|
||||
|
||||
def test_city(self):
|
||||
assert _wx_classify("seattle") == "city"
|
||||
assert _wx_classify("paris, tx") == "city"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestWxGeocodeWrappers:
|
||||
def test_zipcode_delegates_to_geocode_zipcode_sync(self, wx_cmd):
|
||||
with patch(
|
||||
"modules.commands.wx_command.geocode_zipcode_sync",
|
||||
return_value=(47.6, -122.3),
|
||||
) as geo:
|
||||
lat, lon = wx_cmd.zipcode_to_lat_lon("98101")
|
||||
geo.assert_called_once()
|
||||
assert lat == pytest.approx(47.6)
|
||||
|
||||
def test_city_delegates_to_geocode_city_sync(self, wx_cmd):
|
||||
with patch(
|
||||
"modules.commands.wx_command.geocode_city_sync",
|
||||
return_value=(47.6, -122.3, {"city": "Seattle"}),
|
||||
) as geo:
|
||||
lat, lon, addr = wx_cmd.city_to_lat_lon("seattle")
|
||||
geo.assert_called_once()
|
||||
kwargs = geo.call_args.kwargs
|
||||
assert kwargs.get("include_address_info") is True
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert addr == {"city": "Seattle"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rain — _resolve_location contracts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rain_cmd():
|
||||
bot = _make_bot(
|
||||
Rain_Command={"enabled": "true", "zip_city_lookup": "false"},
|
||||
)
|
||||
bot.config.set("Bot", "bot_latitude", "47.6")
|
||||
bot.config.set("Bot", "bot_longitude", "-122.3")
|
||||
from modules.commands.rain_command import RainCommand
|
||||
cmd = RainCommand(bot)
|
||||
return cmd
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRainResolveLocation:
|
||||
def test_coordinates(self, rain_cmd):
|
||||
with patch.object(rain_cmd, "_coordinates_to_location_string", return_value="Seattle, WA"):
|
||||
lat, lon, label, err = rain_cmd._resolve_location(
|
||||
mock_message(content="rain 47.6,-122.3"), "47.6,-122.3"
|
||||
)
|
||||
assert err is None
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert lon == pytest.approx(-122.3)
|
||||
assert label == "Seattle, WA"
|
||||
|
||||
def test_invalid_coordinates(self, rain_cmd):
|
||||
lat, lon, label, err = rain_cmd._resolve_location(
|
||||
mock_message(content="rain 200,0"), "200,0"
|
||||
)
|
||||
assert err == "commands.rain.error"
|
||||
assert lat is None
|
||||
|
||||
def test_zipcode_uses_geocode_zipcode_sync(self, rain_cmd):
|
||||
with patch(
|
||||
"modules.commands.rain_command.geocode_zipcode_sync",
|
||||
return_value=(47.6, -122.3),
|
||||
) as geo, patch.object(
|
||||
rain_cmd, "_coordinates_to_location_string", return_value="Seattle, WA"
|
||||
):
|
||||
lat, lon, label, err = rain_cmd._resolve_location(
|
||||
mock_message(content="rain 98101"), "98101"
|
||||
)
|
||||
geo.assert_called_once()
|
||||
assert err is None
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert "98101" in label or "Seattle" in label
|
||||
|
||||
def test_city_uses_geocode_city_sync(self, rain_cmd):
|
||||
with patch(
|
||||
"modules.commands.rain_command.geocode_city_sync",
|
||||
return_value=(36.16, -86.78, None),
|
||||
) as geo, patch.object(
|
||||
rain_cmd, "_suffix_for_coords", return_value="TN"
|
||||
):
|
||||
lat, lon, label, err = rain_cmd._resolve_location(
|
||||
mock_message(content="rain nashville"), "nashville"
|
||||
)
|
||||
geo.assert_called_once()
|
||||
assert err is None
|
||||
assert "Nashville" in label or "nashville" in label.lower()
|
||||
assert "TN" in label
|
||||
|
||||
def test_empty_uses_bot_location(self, rain_cmd):
|
||||
rain_cmd.bot.db_manager.execute_query.return_value = []
|
||||
with patch.object(
|
||||
rain_cmd, "_coordinates_to_location_string", return_value="Botville, WA"
|
||||
):
|
||||
lat, lon, label, err = rain_cmd._resolve_location(
|
||||
mock_message(content="rain"), None
|
||||
)
|
||||
assert err is None
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert lon == pytest.approx(-122.3)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prefix / solarforecast — repeater-first parse order
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def prefix_cmd():
|
||||
bot = _make_bot(
|
||||
Prefix_Command={"enabled": "true"},
|
||||
External_Data={"repeater_prefix_api_url": ""},
|
||||
)
|
||||
from modules.commands.prefix_command import PrefixCommand
|
||||
return PrefixCommand(bot)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def solar_cmd():
|
||||
bot = _make_bot(Solarforecast_Command={"enabled": "true"})
|
||||
with patch("modules.commands.solarforecast_command.get_nominatim_geocoder", return_value=Mock()):
|
||||
from modules.commands.solarforecast_command import SolarforecastCommand
|
||||
return SolarforecastCommand(bot)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestPrefixParseLocationOrder:
|
||||
def test_repeater_wins_before_city(self, prefix_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
prefix_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(47.1, -122.1)
|
||||
) as rep, patch(
|
||||
"modules.commands.prefix_command.geocode_city", new_callable=AsyncMock
|
||||
) as city:
|
||||
lat, lon, typ = await prefix_cmd._parse_location_to_lat_lon("KR7ABC")
|
||||
return lat, lon, typ, rep, city
|
||||
|
||||
lat, lon, typ, rep, city = _run(_run_parse())
|
||||
assert typ == "repeater"
|
||||
assert lat == pytest.approx(47.1)
|
||||
rep.assert_awaited_once()
|
||||
city.assert_not_called()
|
||||
|
||||
def test_coordinates(self, prefix_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
prefix_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(None, None)
|
||||
):
|
||||
return await prefix_cmd._parse_location_to_lat_lon("47.6,-122.3")
|
||||
|
||||
lat, lon, typ = _run(_run_parse())
|
||||
assert typ == "coordinates"
|
||||
assert lat == pytest.approx(47.6)
|
||||
|
||||
def test_zipcode(self, prefix_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
prefix_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(None, None)
|
||||
), patch(
|
||||
"modules.commands.prefix_command.geocode_zipcode",
|
||||
new_callable=AsyncMock,
|
||||
return_value=(47.6, -122.3),
|
||||
):
|
||||
return await prefix_cmd._parse_location_to_lat_lon("98101")
|
||||
|
||||
lat, lon, typ = _run(_run_parse())
|
||||
assert typ == "zipcode"
|
||||
assert lat == pytest.approx(47.6)
|
||||
|
||||
def test_city_fallback(self, prefix_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
prefix_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(None, None)
|
||||
), patch(
|
||||
"modules.commands.prefix_command.geocode_city",
|
||||
new_callable=AsyncMock,
|
||||
return_value=(47.6, -122.3, None),
|
||||
):
|
||||
return await prefix_cmd._parse_location_to_lat_lon("seattle")
|
||||
|
||||
lat, lon, typ = _run(_run_parse())
|
||||
assert typ == "city"
|
||||
assert lat == pytest.approx(47.6)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSolarforecastParseLocationOrder:
|
||||
def test_repeater_wins(self, solar_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
solar_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(48.0, -122.0)
|
||||
), patch.object(
|
||||
solar_cmd, "_city_to_lat_lon", new_callable=AsyncMock
|
||||
) as city:
|
||||
lat, lon, typ = await solar_cmd._parse_location("SomeRepeater")
|
||||
return lat, lon, typ, city
|
||||
|
||||
lat, lon, typ, city = _run(_run_parse())
|
||||
assert typ == "repeater"
|
||||
city.assert_not_called()
|
||||
|
||||
def test_coordinates(self, solar_cmd):
|
||||
async def _run_parse():
|
||||
with patch.object(
|
||||
solar_cmd, "_repeater_name_to_lat_lon", new_callable=AsyncMock, return_value=(None, None)
|
||||
):
|
||||
return await solar_cmd._parse_location("47.6, -122.3")
|
||||
|
||||
lat, lon, typ = _run(_run_parse())
|
||||
assert typ == "coordinates"
|
||||
assert lat == pytest.approx(47.6)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Alert — _parse_query geocode-related typing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def alert_cmd():
|
||||
bot = _make_bot(
|
||||
Alert_Command={
|
||||
"enabled": "true",
|
||||
"agency.city.seattle": "1234",
|
||||
"agency.county.king": "5678",
|
||||
# Legacy agency.* keys are treated as counties today
|
||||
"agency.everett": "9999",
|
||||
}
|
||||
)
|
||||
from modules.commands.alert_command import AlertCommand
|
||||
return AlertCommand(bot)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAlertParseQuery:
|
||||
def test_coordinates(self, alert_cmd):
|
||||
qtype, location, lat, lon = alert_cmd._parse_query("47.6,-122.3")
|
||||
assert qtype == "coordinates"
|
||||
assert lat == pytest.approx(47.6)
|
||||
assert lon == pytest.approx(-122.3)
|
||||
|
||||
def test_zipcode(self, alert_cmd):
|
||||
qtype, location, lat, lon = alert_cmd._parse_query("98101")
|
||||
assert qtype == "zipcode"
|
||||
assert location == "98101"
|
||||
assert lat is None and lon is None
|
||||
|
||||
def test_county_alias_sno(self, alert_cmd):
|
||||
qtype, location, lat, lon = alert_cmd._parse_query("sno")
|
||||
assert qtype == "county"
|
||||
assert location == "sno"
|
||||
|
||||
def test_county_alias_sea(self, alert_cmd):
|
||||
qtype, location, _, _ = alert_cmd._parse_query("sea")
|
||||
assert qtype == "county"
|
||||
|
||||
def test_street_city_heuristic(self, alert_cmd):
|
||||
qtype, location, lat, lon = alert_cmd._parse_query("178th seattle")
|
||||
assert qtype == "street_city"
|
||||
assert "178th" in location
|
||||
assert "seattle" in location.lower()
|
||||
|
||||
def test_configured_city(self, alert_cmd):
|
||||
qtype, location, _, _ = alert_cmd._parse_query("seattle")
|
||||
assert qtype == "city"
|
||||
assert location == "seattle"
|
||||
|
||||
def test_configured_county(self, alert_cmd):
|
||||
qtype, location, _, _ = alert_cmd._parse_query("king")
|
||||
assert qtype == "county"
|
||||
assert location == "king"
|
||||
|
||||
def test_legacy_agency_key_treated_as_county(self, alert_cmd):
|
||||
qtype, location, _, _ = alert_cmd._parse_query("everett")
|
||||
assert qtype == "county"
|
||||
assert location == "everett"
|
||||
|
||||
def test_unknown_defaults_to_city(self, alert_cmd):
|
||||
qtype, location, _, _ = alert_cmd._parse_query("bothell")
|
||||
assert qtype == "city"
|
||||
assert location == "bothell"
|
||||
Reference in New Issue
Block a user