"""Behavior the wx (NOAA) and gwx (Open-Meteo) commands share. ``WeatherCommandMixin`` goes before ``BaseCommand`` in a weather command's bases and expects the BaseCommand API (``bot``, ``logger``, ``translate``, ``send_response``, ``get_max_message_length``, ``response_translator``). Each command sets ``translation_ns`` to its own catalog namespace. """ from __future__ import annotations import asyncio import re from collections.abc import Iterable from datetime import datetime from typing import Any, Optional, Union from .clients.mqtt_weather import ( get_mqtt_weather_topic, load_mqtt_weather_format_config, mqtt_weather_display_for_topic, ) from .utils import format_temperature_high_low Number = Union[int, float] def load_open_meteo_model(config: Any, logger: Any) -> Optional[str]: """[Weather] weather_model for Open-Meteo's ``models`` parameter. Unset means ``best_match``; explicitly blank means None (omit the parameter and let Open-Meteo choose). Anything outside ``[a-z0-9_,.-]`` falls back to ``best_match`` with a warning. """ if config.has_option('Weather', 'weather_model'): model = config.get('Weather', 'weather_model', fallback='').strip().lower() if not model: return None else: model = 'best_match' if not re.fullmatch(r'[a-z0-9_,.-]+', model): logger.warning(f"Invalid weather_model '{model}', using 'best_match'") return 'best_match' return model def load_open_meteo_units(config: Any, logger: Any) -> tuple[str, str, str]: """[Weather] temperature_unit, wind_speed_unit and precipitation_unit for Open-Meteo. Lowercased; a value Open-Meteo does not accept falls back to fahrenheit, mph or inch with a warning. The result doubles as a translation key for the unit's display label. """ temperature_unit = config.get('Weather', 'temperature_unit', fallback='fahrenheit').lower() wind_speed_unit = config.get('Weather', 'wind_speed_unit', fallback='mph').lower() precipitation_unit = config.get('Weather', 'precipitation_unit', fallback='inch').lower() if temperature_unit not in ('fahrenheit', 'celsius'): logger.warning(f"Invalid temperature_unit '{temperature_unit}', using 'fahrenheit'") temperature_unit = 'fahrenheit' if wind_speed_unit not in ('mph', 'kmh', 'ms', 'kn'): logger.warning(f"Invalid wind_speed_unit '{wind_speed_unit}', using 'mph'") wind_speed_unit = 'mph' if precipitation_unit not in ('inch', 'mm'): logger.warning(f"Invalid precipitation_unit '{precipitation_unit}', using 'inch'") precipitation_unit = 'inch' return temperature_unit, wind_speed_unit, precipitation_unit # 16-point compass labels, clockwise from north, and the arrow for each 45° sector. _COMPASS_16 = ( "N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", "S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW", ) _ARROWS_8 = ("⬆️", "↗️", "➡️", "↘️", "⬇️", "↙️", "⬅️", "↖️") class WeatherCommandMixin: # Catalog namespace for this command's strings, e.g. "commands.wx". translation_ns: str = "" # Provided by BaseCommand. bot: Any logger: Any response_translator: Any translate: Any send_response: Any get_max_message_length: Any get_config_value: Any record_execution: Any _get_custom_wxsim_source: Any _get_companion_location: Any _get_bot_location: Any default_city: Any default_state: Any default_country: Any @staticmethod def _coordinates_query(lat: float, lon: float) -> str: """A "lat,lon" location for these coordinates, in fixed decimals. Plain str() of a float can give "1e-05", which the coordinate pattern does not match, so the point would be geocoded as a place name. """ return f"{lat:.5f},{lon:.5f}" @staticmethod def _parse_forecast_suffix( location_parts: list[str], max_days: int, *, allow_hourly: bool ) -> tuple[list[str], str, int]: """Strip a trailing forecast option from the location words. "tomorrow", "hourly" (when allowed), "7day"/"7-day", "Nd" or a bare "N" with 2 <= N <= max_days. Returns (remaining words, forecast type, days); anything else stays part of the location and the type is "default". """ forecast_type = "default" num_days = 7 # Default for multi-day forecast if len(location_parts) > 0: last_part = location_parts[-1].lower() if last_part == "tomorrow": forecast_type = "tomorrow" location_parts = location_parts[:-1] elif allow_hourly and last_part == "hourly": forecast_type = "hourly" location_parts = location_parts[:-1] elif last_part in ["7day", "7-day"]: forecast_type = "multiday" num_days = 7 location_parts = location_parts[:-1] else: nd_match = re.fullmatch(r"(\d+)d", last_part) if nd_match: days = int(nd_match.group(1)) if 2 <= days <= max_days: forecast_type = "multiday" num_days = days location_parts = location_parts[:-1] elif last_part.isdecimal(): days = int(last_part) if 2 <= days <= max_days: forecast_type = "multiday" num_days = days location_parts = location_parts[:-1] return location_parts, forecast_type, num_days def _is_custom_source_name(self, word: str) -> bool: """Whether *word* names a configured custom.wxsim.* or custom.mqtt_weather.* source. Such a name is a place even when it looks like an option ("custom.wxsim.hourly"). """ return bool(self._get_custom_mqtt_weather_topic(word) or self._get_custom_wxsim_source(word)) def _split_option_only(self, parts: list[str], max_days: int) -> tuple[list[str], Optional[str], str, int]: """Pull a forecast option given without a location ("wx hourly", "gwx 5d", "wx alerts"). Returns (parts without it, the option word or None, forecast type, days), so the no-location fallbacks (custom default source, the sender's position, default_city, the bot's position) can apply the option instead of showing usage. """ if len(parts) == 2: word = parts[1] if self._is_custom_source_name(word): return parts, None, "default", 7 if word.lower() == "alerts": return parts[:1], word, "alerts", 7 rest, forecast_type, num_days = self._parse_forecast_suffix(parts[1:], max_days, allow_hourly=True) if not rest and forecast_type != "default": return parts[:1], parts[1], forecast_type, num_days return parts, None, "default", 7 def _format_high_low(self, high: Optional[Number], low: Optional[Number], temp_symbol: str) -> str: """Format high/low using [Weather] temperature_*_format templates.""" return format_temperature_high_low(self.bot.config, high, low, temp_symbol, self.logger, translator=self.response_translator) def _get_custom_mqtt_weather_topic(self, location: Optional[str] = None) -> Optional[str]: """MQTT topic for custom.mqtt_weather. (see get_mqtt_weather_topic).""" return get_mqtt_weather_topic(self.bot.config, location) def _mqtt_weather_line(self, topic: str, forecast_type: str, location_name: Optional[str]) -> str: """Format the cached MQTT weather payload for ``topic``, or the matching error text.""" ns = self.translation_ns if forecast_type != "default": return self.translate(f"{ns}.mqtt_forecast_not_supported") fmt = load_mqtt_weather_format_config(self.bot.config) cache = getattr(self.bot, "mqtt_weather_cache", None) text, err = mqtt_weather_display_for_topic(topic, cache, fmt) if text is not None: return f"{location_name}: {text}" if location_name else text return self._mqtt_weather_error_key(err) def _mqtt_weather_error_key(self, err: Optional[str]) -> str: """The translated reply for an MQTT weather lookup error (overridable).""" ns = self.translation_ns if err == "no_cache": return self.translate(f"{ns}.mqtt_weather_no_subscriber") if err in ("no_data", "empty_payload", "empty_after_sanitize"): return self.translate(f"{ns}.mqtt_weather_no_data") if err == "stale": return self.translate(f"{ns}.mqtt_weather_stale") detail = (err or "unknown").replace("_", " ") return self.translate(f"{ns}.mqtt_weather_payload_error", detail=detail) def _count_display_width(self, text: str) -> int: """Count UTF-8 byte length of text. Matches RF packet byte limit from get_max_message_length().""" return len(text.encode('utf-8')) @staticmethod def _hour_label(start_time_str: str) -> str: """12-hour label for an ISO start time; empty when missing or unparseable.""" if not start_time_str: return "" try: hour = datetime.fromisoformat(start_time_str.replace('Z', '+00:00')).hour except (ValueError, TypeError): return "" return f"{hour % 12 or 12}{'AM' if hour < 12 else 'PM'}" @staticmethod def _without_arrow(direction: str) -> str: """A wind direction label without its leading arrow emoji ("↖️WNW" -> "WNW"). Hourly lines drop the arrow to keep an extra hour in the reply; the letters (translated or not) stay. """ return re.sub(r"^\W+", "", direction) @staticmethod def _short_hourly_description(description: str) -> str: """Long hourly descriptions keep their first three words, or 18 characters.""" if len(description) > 18: words = description.split() return ' '.join(words[:3]) if len(words) > 3 else description[:18] return description def _pack_hourly_lines(self, hourly_lines: Iterable[str], max_length: int) -> str: """Pack consecutive whole hours into one UTF-8 byte budget.""" lines: list[str] = [] for line in hourly_lines: if self._count_display_width("\n".join(lines + [line])) > max_length: break lines.append(line) return "\n".join(lines) or self.translate(f'{self.translation_ns}.hourly_not_available') @staticmethod async def _pace_reply(message: Any, seconds: float) -> None: """Wait between the parts of a reply so the TX limiter lets the next through. A scheduled {cmd:...} render collects the parts without transmitting, so it does not wait: the pause would only eat into its render timeout. """ if getattr(message, 'capture_sink', None) is None: await asyncio.sleep(seconds) async def _send_multiday_forecast(self, message: Any, forecast_text: str) -> bool: """Send a multi-day forecast, packing whole lines into as few messages as fit. A line too long for one message goes out on its own. The first message goes through the per-user rate limit; the ones after it skip it and are spaced 2 s apart. Returns whether every message was sent. """ max_length = self.get_max_message_length(message) lines = [line.strip() for line in forecast_text.split('\n') if line.strip()] if not lines: return False if self._count_display_width(forecast_text) <= max_length: return bool(await self.send_response(message, forecast_text)) parts: list[str] = [] current_message = "" for line in lines: test_message = current_message + "\n" + line if current_message else line if self._count_display_width(test_message) <= max_length: current_message = test_message continue if current_message: parts.append(current_message) # A single line too long for one message goes out anyway (the bot splits it). current_message = line if current_message: parts.append(current_message) sent_all = True for i, part in enumerate(parts): if i > 0: await self._pace_reply(message, 2.0) sent = await self.send_response(message, part, skip_user_rate_limit=(i > 0)) if not sent: sent_all = False break return sent_all async def _reply_from_mqtt( self, message: Any, topic: str, forecast_type: str, location: Optional[str], *, split_multiday: bool = True, ) -> bool: """Answer from a custom.mqtt_weather source; errors become the error reply.""" try: self.record_execution(message.sender_id) weather_data = self._mqtt_weather_line(topic, forecast_type, location) if split_multiday and forecast_type == "multiday": return await self._send_multiday_forecast(message, weather_data) return bool(await self.send_response(message, weather_data)) except Exception as e: self.logger.error(f"Error reading MQTT weather: {e}") await self.send_response(message, self.translate(f"{self.translation_ns}.error", error=str(e))) return True async def _reply_from_wxsim( self, message: Any, source: str, forecast_type: str, num_days: int, location: Optional[str] = None, ) -> bool: """Answer from a custom.wxsim source; errors become the error reply.""" try: self.record_execution(message.sender_id) weather_data = await self._fetch_wxsim(source, forecast_type, num_days, message, location) if forecast_type == "multiday": return await self._send_multiday_forecast(message, weather_data) return bool(await self.send_response(message, weather_data)) except Exception as e: self.logger.error(f"Error fetching WXSIM weather: {e}") await self.send_response(message, self.translate(f"{self.translation_ns}.error", error=str(e))) return True async def _fetch_wxsim( self, source: str, forecast_type: str, num_days: int, message: Any, location: Optional[str] ) -> str: """Fetch and format a WXSIM source off the event loop (each command's own way).""" raise NotImplementedError async def _fallback_place_name(self, lat: float, lon: float) -> Optional[str]: """A place name for logging a fallback position, or None to log the coordinates.""" return None def _default_city_query(self) -> Optional[str]: """``default_city[, default_state][, default_country]``, or None when unset.""" if not self.default_city: return None location_parts = [self.default_city] if self.default_state: location_parts.append(self.default_state) if self.default_country: location_parts.append(self.default_country) return ", ".join(location_parts) async def _no_location_fallback(self, message: Any) -> tuple[Optional[str], bool]: """Where to forecast when the request names no place. Tries the sender's position, then default_city, then (when use_bot_location_when_no_location is set) the bot's position. Returns (location query or None, whether it is the sender's position). """ companion_location = self._get_companion_location(message) if companion_location: # Use coordinates directly to avoid re-geocoding issues location_str = self._coordinates_query(*companion_location) display_name = await self._fallback_place_name(companion_location[0], companion_location[1]) if display_name: self.logger.info(f"Using companion location: {display_name} ({companion_location[0]}, {companion_location[1]})") else: self.logger.info(f"Using companion coordinates: {location_str}") return location_str, True default_city = self._default_city_query() if default_city: self.logger.info(f"Using default city (no args): {default_city}") return default_city, False use_bot = self.get_config_value( 'Wx_Command', 'use_bot_location_when_no_location', fallback=False, value_type='bool', ) bot_loc = self._get_bot_location() if use_bot else None if bot_loc: location_str = self._coordinates_query(*bot_loc) display_name = await self._fallback_place_name(bot_loc[0], bot_loc[1]) if display_name: self.logger.info( f"Using bot location (no args): {display_name} ({bot_loc[0]}, {bot_loc[1]})" ) else: self.logger.info(f"Using bot coordinates (no args): {location_str}") return location_str, False if use_bot: self.logger.debug( "use_bot_location_when_no_location enabled but bot_latitude/bot_longitude " "not set; showing usage" ) else: self.logger.debug("No companion/default city location found, showing usage") return None, False async def _send_weather_reply(self, message: Any, weather_data: Any, forecast_type: str) -> bool: """Send a provider's answer: two parts with alerts, split by day, or one message.""" if isinstance(weather_data, tuple) and weather_data[0] == "multi_message": # Send weather data first; if it was refused (rate limit, send failure), # do not send its second part on its own. if not await self.send_response(message, weather_data[1]): return False # Wait for bot TX rate limiter to allow next message rate_limit = self.bot.config.getfloat('Bot', 'bot_tx_rate_limit_seconds', fallback=1.0) # Use a conservative sleep time to avoid rate limiting sleep_time = max(rate_limit + 1.0, 2.0) # At least 2 seconds, or rate_limit + 1 second await self._pace_reply(message, sleep_time) # Second part of the same reply: the reply limiter already let the first through. return bool(await self.send_response(message, weather_data[2], skip_user_rate_limit=True)) if forecast_type == "multiday": # Use message splitting for multi-day forecasts return await self._send_multiday_forecast(message, weather_data) return bool(await self.send_response(message, weather_data))