diff --git a/config.ini.example b/config.ini.example index 22fb227..f13e7a8 100644 --- a/config.ini.example +++ b/config.ini.example @@ -476,6 +476,19 @@ wind_speed_unit = mph # Default: inch precipitation_unit = inch +# Custom weather sources (extensible pattern, similar to Channels_List) +# Format: custom.. = +# custom.wxsim.default = - Default source used when 'wx' is called without a location +# custom.wxsim. = - Named source accessible via 'wx ' +# +# Example: +# custom.wxsim.default = https://westlethbridgeweather.com/latest.txt +# custom.wxsim.lethbridge = https://westlethbridgeweather.com/latest.txt +# custom.wxsim.seattle = https://seattleweather.example.com/plaintext.txt +# +# Note: Only one 'default' source can be set per provider type. +# Future providers (e.g., custom.pwaweather.location) will follow the same pattern. + [Solar_Config] # URL timeout for external API calls (seconds) url_timeout = 10 diff --git a/modules/clients/wxsim_parser.py b/modules/clients/wxsim_parser.py new file mode 100644 index 0000000..0abef2a --- /dev/null +++ b/modules/clients/wxsim_parser.py @@ -0,0 +1,765 @@ +#!/usr/bin/env python3 +""" +WXSIM Plaintext Parser +Parses WXSIM plaintext.txt forecast files into structured data + +Based on the PHP parser by Ken True (Saratoga-Weather.org) +https://github.com/ktrue/WXSIM-forecast/blob/master/plaintext-parser.php + +Usage: + from modules.clients.wxsim_parser import WXSIMParser + + # Parse from URL + parser = WXSIMParser() + text = parser.fetch_from_url('https://example.com/plaintext.txt') + if text: + forecast = parser.parse(text) + current = parser.format_current_conditions(forecast, temp_unit='fahrenheit', wind_unit='mph') + summary = parser.format_forecast_summary(forecast, num_days=7, temp_unit='fahrenheit', wind_unit='mph') + + # Or parse from file/string + with open('plaintext.txt', 'r') as f: + text = f.read() + forecast = parser.parse(text) + + # Access structured data + for period in forecast.periods: + print(f"{period.day_name}: {period.conditions} {period.high_temp}°C/{period.low_temp}°C") +""" + +import re +import requests +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Tuple, Any +from dataclasses import dataclass, field +from enum import Enum + + +class PeriodType(Enum): + """Forecast period type""" + DAY = "day" + NIGHT = "night" + UNKNOWN = "unknown" + + +@dataclass +class HourlyData: + """Single hour of forecast data""" + date: str # e.g., "May 5" + time: str # e.g., "7:00 A" or "12:00 P" + hour: int # 0-23 + temperature: float + wind_speed: int + humidity: int + sky_cover: int # %SC + visibility: int # %VST + visibility_miles: float # VIS + precip_chance: int # PC/HR + rain_total: float # RN TOT + weather: str # Weather condition text + + +@dataclass +class ForecastPeriod: + """A forecast period (day or night)""" + day_name: str # e.g., "Friday", "Today" + date: str # e.g., "May 5" + period_type: PeriodType + high_temp: Optional[float] = None + low_temp: Optional[float] = None + conditions: str = "" + wind_speed: Optional[int] = None + wind_direction: Optional[str] = None + precip_chance: Optional[int] = None + precip_amount: Optional[float] = None + hourly_data: List[HourlyData] = field(default_factory=list) + + +@dataclass +class WXSIMForecast: + """Complete WXSIM forecast data""" + city: str = "" + station: str = "" + update_time: str = "" + update_date: str = "" + periods: List[ForecastPeriod] = field(default_factory=list) + hourly_data: List[HourlyData] = field(default_factory=list) + raw_text: str = "" + + +class WXSIMParser: + """Parser for WXSIM plaintext.txt files""" + + # Weather condition mappings (abbreviated to full descriptions) + WEATHER_CONDITIONS = { + 'CLEAR': 'Clear', + 'SUNNY': 'Sunny', + 'FAIR': 'Fair', + 'FAIR-P.C.': 'Fair to Partly Cloudy', + 'P.CLOUDY': 'Partly Cloudy', + 'P.-M.CLDY': 'Partly to Mostly Cloudy', + 'M.CLOUDY': 'Mostly Cloudy', + 'M.C.-CLDY': 'Mostly Cloudy', + 'CLOUDY': 'Cloudy', + 'DNS.OVCST': 'Dense Overcast', + 'OVCST': 'Overcast', + 'FOGGY': 'Foggy', + 'DRIZZLE': 'Drizzle', + 'DRZL': 'Drizzle', + 'CHNC. DRZL': 'Chance Drizzle', + 'CHNC. SHWR': 'Chance Showers', + 'SHOWERS': 'Showers', + 'RAIN': 'Rain', + 'CHNC. RAIN': 'Chance Rain', + 'SNOW': 'Snow', + 'CHNC. SNOW': 'Chance Snow', + 'T-STM': 'Thunderstorm', + 'CHNC. T-STM': 'Chance Thunderstorm', + } + + def __init__(self): + """Initialize the parser""" + self.current_year = datetime.now().year + self.current_date = datetime.now() + + def parse(self, text: str) -> WXSIMForecast: + """Parse WXSIM plaintext content. + + Args: + text: The plaintext content from WXSIM plaintext.txt file + + Returns: + WXSIMForecast: Parsed forecast data + """ + forecast = WXSIMForecast() + forecast.raw_text = text + + lines = text.split('\n') + + # Find FORECAST RUN section (skip calibration) + forecast_start = self._find_forecast_start(lines) + if forecast_start == -1: + return forecast + + # Parse header info (city, station, date) + self._parse_header(lines[:forecast_start], forecast) + + # Parse forecast data + forecast_lines = lines[forecast_start:] + hourly_data = self._parse_hourly_data(forecast_lines) + forecast.hourly_data = hourly_data + + # Extract forecast date/time from first data point + if hourly_data: + first_data = hourly_data[0] + forecast.update_date = first_data.date # e.g., "May 5" + forecast.update_time = first_data.time # e.g., "7:00 A" + + # Group into periods (days) + forecast.periods = self._group_into_periods(hourly_data, forecast_lines) + + return forecast + + def _find_forecast_start(self, lines: List[str]) -> int: + """Find the start of the FORECAST RUN section. + + Args: + lines: All lines from the file + + Returns: + int: Index of first forecast data line, or -1 if not found + """ + for i, line in enumerate(lines): + if 'FORECAST RUN:' in line.upper(): + # Find the header line "DATE TIME TEMP..." + for j in range(i, min(i + 10, len(lines))): + if 'DATE' in lines[j] and 'TIME' in lines[j] and 'TEMP' in lines[j]: + # Return line after header + return j + 2 # Skip header and blank line + return -1 + + def _parse_header(self, lines: List[str], forecast: WXSIMForecast) -> None: + """Parse header information (city, station, date). + + Args: + lines: Header lines (before FORECAST RUN) + forecast: Forecast object to populate + """ + # Look for city/station info in header + # WXSIM format may vary, so we'll extract what we can + for line in lines: + # Look for common patterns + if 'FORECAST FOR' in line.upper(): + # Extract city name + parts = line.split('FORECAST FOR', 1) + if len(parts) > 1: + forecast.city = parts[1].strip() + elif 'BY' in line.upper() and not forecast.station: + # Extract station/forecaster name + parts = line.split('BY', 1) + if len(parts) > 1: + forecast.station = parts[1].strip() + + def _parse_hourly_data(self, lines: List[str]) -> List[HourlyData]: + """Parse hourly forecast data rows. + + Args: + lines: Lines from FORECAST RUN section onwards + + Returns: + List[HourlyData]: Parsed hourly data + """ + hourly_data = [] + + for line in lines: + line = line.strip() + if not line: + continue + + # Skip non-data lines + if (line.startswith('DATE') or + line.startswith('FORECAST') or + line.startswith('CALIBRATION') or + line.startswith('SURFACE WIND') or + line.startswith('AUTO CLOUDS') or + line.startswith('Press') or + line.startswith('(Automatically') or + line.startswith('-') or + line.startswith('(')): + continue + + # Try to parse as data row + # Format: "May 5 7:00 A 9.3 0 73 95 50 42.2 3 0.0 M.C.-CLDY CHNC. DRZL" + data = self._parse_data_row(line) + if data: + hourly_data.append(data) + + return hourly_data + + def _parse_data_row(self, line: str) -> Optional[HourlyData]: + """Parse a single data row. + + Args: + line: Single line of forecast data + + Returns: + Optional[HourlyData]: Parsed data or None if invalid + """ + # Pattern: Month Day Time Temp Wind Hum %SC %VST VIS PC/HR RN TOT Weather + # Example: "May 5 7:00 A 9.3 0 73 95 50 42.2 3 0.0 M.C.-CLDY CHNC. DRZL" + + # Match month and day at start + match = re.match(r'^([A-Za-z]+)\s+(\d+)\s+(\d{1,2}):(\d{2})\s+([AP])\s+', line) + if not match: + return None + + month_name = match.group(1) + day = int(match.group(2)) + hour_12 = int(match.group(3)) + minute = int(match.group(4)) + am_pm = match.group(5) + + # Convert to 24-hour + hour_24 = hour_12 + if am_pm == 'P' and hour_12 != 12: + hour_24 = hour_12 + 12 + elif am_pm == 'A' and hour_12 == 12: + hour_24 = 0 + + # Extract remaining fields (split by whitespace) + rest = line[match.end():].strip() + parts = rest.split() + + if len(parts) < 8: + return None + + try: + # Parse numeric fields + temp = float(parts[0]) + wind = int(parts[1]) + humidity = int(parts[2]) + sky_cover = int(parts[3]) + visibility_pct = int(parts[4]) + visibility_miles = float(parts[5]) + precip_chance = int(parts[6]) + rain_total = float(parts[7]) + + # Weather condition is everything after the numeric fields + weather = ' '.join(parts[8:]) if len(parts) > 8 else '' + + # Format date string + date_str = f"{month_name} {day}" + time_str = f"{hour_12}:{minute:02d} {am_pm}" + + return HourlyData( + date=date_str, + time=time_str, + hour=hour_24, + temperature=temp, + wind_speed=wind, + humidity=humidity, + sky_cover=sky_cover, + visibility=visibility_pct, + visibility_miles=visibility_miles, + precip_chance=precip_chance, + rain_total=rain_total, + weather=weather + ) + except (ValueError, IndexError): + return None + + def _group_into_periods(self, hourly_data: List[HourlyData], lines: List[str]) -> List[ForecastPeriod]: + """Group hourly data into forecast periods (days). + + Args: + hourly_data: List of hourly data points + lines: Original lines (to find day separators) + + Returns: + List[ForecastPeriod]: Forecast periods + """ + periods = [] + + if not hourly_data: + return periods + + # Find day separators in original lines + day_separators = self._find_day_separators(lines) + + # Group hourly data by day + current_day = None + current_period_data = [] + + for data in hourly_data: + # Check if this is a new day (by date string or hour reset) + if current_day is None or data.date != current_day: + # Save previous period if exists + if current_period_data: + period = self._create_period_from_hourly(current_day, current_period_data, day_separators) + if period: + periods.append(period) + + # Start new day + current_day = data.date + current_period_data = [data] + else: + current_period_data.append(data) + + # Add final period + if current_period_data: + period = self._create_period_from_hourly(current_day, current_period_data, day_separators) + if period: + periods.append(period) + + return periods + + def _find_day_separators(self, lines: List[str]) -> Dict[str, str]: + """Find day name separators in the file. + + Args: + lines: All lines from the file + + Returns: + Dict[str, str]: Mapping of date string to day name + """ + separators = {} + + # Look for lines like " Friday" + for i, line in enumerate(lines): + line_stripped = line.strip() + # Check if line contains a day name + day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] + for day_name in day_names: + if day_name in line_stripped: + # Try to find the date from nearby lines + # Look at next few lines for date pattern + for j in range(i + 1, min(i + 5, len(lines))): + date_match = re.search(r'([A-Za-z]+)\s+(\d+)', lines[j]) + if date_match: + date_str = f"{date_match.group(1)} {date_match.group(2)}" + separators[date_str] = day_name + break + break + + return separators + + def _create_period_from_hourly(self, date: str, hourly_data: List[HourlyData], + day_separators: Dict[str, str]) -> Optional[ForecastPeriod]: + """Create a forecast period from hourly data. + + Args: + date: Date string (e.g., "May 5") + hourly_data: Hourly data for this day + day_separators: Mapping of dates to day names + + Returns: + Optional[ForecastPeriod]: Forecast period or None + """ + if not hourly_data: + return None + + # Get day name - try to determine from date or use separator + day_name = day_separators.get(date) + if not day_name: + # Try to determine day name from date + try: + # Parse date (e.g., "May 5") + month_name, day_num = date.split() + month_map = { + 'Jan': 1, 'Feb': 2, 'Mar': 3, 'Apr': 4, 'May': 5, 'Jun': 6, + 'Jul': 7, 'Aug': 8, 'Sep': 9, 'Oct': 10, 'Nov': 11, 'Dec': 12 + } + month = month_map.get(month_name[:3], 1) + day = int(day_num) + + # Create datetime and get day name + forecast_date = datetime(self.current_year, month, day) + day_name = forecast_date.strftime('%A') + + # If it's today, use "Today" instead + today = datetime.now() + if forecast_date.date() == today.date(): + day_name = "Today" + elif forecast_date.date() == (today + timedelta(days=1)).date(): + day_name = "Tomorrow" + except (ValueError, KeyError): + day_name = "Today" + + # Determine period type (day vs night) + # Day: roughly 6 AM to 6 PM, Night: 6 PM to 6 AM + day_hours = [d for d in hourly_data if 6 <= d.hour < 18] + night_hours = [d for d in hourly_data if d.hour < 6 or d.hour >= 18] + + # Use the period with more data, or default to day + if len(day_hours) >= len(night_hours): + period_type = PeriodType.DAY + primary_data = day_hours if day_hours else hourly_data + else: + period_type = PeriodType.NIGHT + primary_data = night_hours if night_hours else hourly_data + + # Calculate high/low temps + temps = [d.temperature for d in hourly_data] + high_temp = max(temps) if temps else None + low_temp = min(temps) if temps else None + + # Get most common weather condition + conditions = self._get_primary_condition(hourly_data) + + # Get average wind speed + wind_speeds = [d.wind_speed for d in hourly_data if d.wind_speed > 0] + avg_wind = int(sum(wind_speeds) / len(wind_speeds)) if wind_speeds else None + + # Get max precip chance + precip_chances = [d.precip_chance for d in hourly_data] + max_precip_chance = max(precip_chances) if precip_chances else None + + # Get total precipitation + total_precip = sum(d.rain_total for d in hourly_data) + + period = ForecastPeriod( + day_name=day_name, + date=date, + period_type=period_type, + high_temp=high_temp, + low_temp=low_temp, + conditions=conditions, + wind_speed=avg_wind, + precip_chance=max_precip_chance, + precip_amount=total_precip if total_precip > 0 else None, + hourly_data=hourly_data + ) + + return period + + def _get_primary_condition(self, hourly_data: List[HourlyData]) -> str: + """Get the primary weather condition from hourly data. + + Args: + hourly_data: List of hourly data points + + Returns: + str: Primary weather condition description + """ + if not hourly_data: + return "Unknown" + + # Count condition occurrences + condition_counts = {} + for data in hourly_data: + # Normalize condition text + condition = data.weather.strip().upper() + if condition: + # Expand abbreviations + for abbrev, full in self.WEATHER_CONDITIONS.items(): + if abbrev in condition: + condition = full + break + condition_counts[condition] = condition_counts.get(condition, 0) + 1 + + if not condition_counts: + return "Unknown" + + # Return most common condition + return max(condition_counts.items(), key=lambda x: x[1])[0] + + def format_current_conditions(self, forecast: WXSIMForecast, + temp_unit: str = 'celsius', + wind_unit: str = 'kph') -> str: + """Format current conditions for display. + + Args: + forecast: Parsed forecast data + temp_unit: Temperature unit ('celsius' or 'fahrenheit') + wind_unit: Wind speed unit ('kph', 'mph', or 'ms') + + Returns: + str: Formatted current conditions string + """ + if not forecast.hourly_data: + return "No current data available" + + # Get the first hour from FORECAST RUN (this is the current/starting conditions) + # The last hour would be in the future, so we use the first one + current = forecast.hourly_data[0] + + # Convert temperature + temp = self._convert_temp(current.temperature, temp_unit) + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + # Convert wind speed + wind = self._convert_wind(current.wind_speed, wind_unit) + wind_unit_str = self._get_wind_unit_str(wind_unit) + + # Format condition + condition = self._normalize_condition(current.weather) + + # Build string + result = f"{condition} {temp}{temp_symbol}" + + if current.wind_speed > 0: + result += f" Wind {wind}{wind_unit_str}" + + if current.humidity > 0: + result += f" {current.humidity}%RH" + + if current.precip_chance > 0: + result += f" {current.precip_chance}% PoP" + + return result + + def format_forecast_summary(self, forecast: WXSIMForecast, num_days: int = 7, + temp_unit: str = 'celsius', + wind_unit: str = 'kph') -> str: + """Format forecast summary for display. + + Args: + forecast: Parsed forecast data + num_days: Number of days to include + temp_unit: Temperature unit ('celsius' or 'fahrenheit') + wind_unit: Wind speed unit ('kph', 'mph', or 'ms') + + Returns: + str: Formatted forecast summary + """ + if not forecast.periods: + return "No forecast data available" + + parts = [] + for period in forecast.periods[:num_days]: + # Convert temps + high = self._convert_temp(period.high_temp, temp_unit) if period.high_temp else None + low = self._convert_temp(period.low_temp, temp_unit) if period.low_temp else None + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + # Format day + day_abbrev = period.day_name[:3] if len(period.day_name) > 3 else period.day_name + + # Build period string + period_str = f"{day_abbrev}: {period.conditions}" + if high is not None and low is not None: + period_str += f" {high}{temp_symbol}/{low}{temp_symbol}" + elif high is not None: + period_str += f" {high}{temp_symbol}" + elif low is not None: + period_str += f" {low}{temp_symbol}" + + if period.precip_chance and period.precip_chance > 30: + period_str += f" {period.precip_chance}% PoP" + + parts.append(period_str) + + return "\n".join(parts) + + def _convert_temp(self, temp_c: float, unit: str) -> float: + """Convert temperature from Celsius to requested unit. + + Args: + temp_c: Temperature in Celsius + unit: Target unit ('celsius' or 'fahrenheit') + + Returns: + float: Converted temperature + """ + if unit == 'fahrenheit': + return round((temp_c * 9/5) + 32, 1) + return round(temp_c, 1) + + def _convert_wind(self, wind_kph: int, unit: str) -> float: + """Convert wind speed from km/h to requested unit. + + Args: + wind_kph: Wind speed in km/h (WXSIM default unit) + unit: Target unit ('kph', 'mph', or 'ms') + + Returns: + float: Converted wind speed + """ + # WXSIM outputs wind in km/h, but the values might be in different units + # depending on configuration. We'll assume km/h as default. + if unit == 'mph': + return round(wind_kph * 0.621371, 1) + elif unit == 'ms': + return round(wind_kph / 3.6, 1) + return float(wind_kph) + + def _get_wind_unit_str(self, unit: str) -> str: + """Get wind speed unit string. + + Args: + unit: Wind unit ('kph', 'mph', or 'ms') + + Returns: + str: Unit string + """ + unit_map = { + 'kph': 'km/h', + 'mph': 'mph', + 'ms': 'm/s' + } + return unit_map.get(unit, 'km/h') + + def _normalize_condition(self, condition: str) -> str: + """Normalize weather condition text. + + Args: + condition: Raw condition text from WXSIM + + Returns: + str: Normalized condition description + """ + condition_upper = condition.strip().upper() + + # Try to match abbreviations + for abbrev, full in self.WEATHER_CONDITIONS.items(): + if abbrev in condition_upper: + return full + + # Return original if no match + return condition.strip() if condition else "Unknown" + + def get_forecast_date(self, forecast: WXSIMForecast) -> Optional[datetime]: + """Get the forecast date as a datetime object. + + Args: + forecast: Parsed forecast data + + Returns: + Optional[datetime]: Forecast date/time or None if unavailable + """ + if not forecast.update_date: + return None + + try: + # Parse date string like "May 5" + month_name, day_num = forecast.update_date.split() + month_map = { + 'Jan': 1, 'Feb': 2, 'Mar': 3, 'Apr': 4, 'May': 5, 'Jun': 6, + 'Jul': 7, 'Aug': 8, 'Sep': 9, 'Oct': 10, 'Nov': 11, 'Dec': 12 + } + month = month_map.get(month_name[:3], 1) + day = int(day_num) + + # Parse time if available + hour = 0 + minute = 0 + if forecast.update_time: + # Parse time like "7:00 A" or "12:30 P" + time_match = re.match(r'(\d{1,2}):(\d{2})\s+([AP])', forecast.update_time) + if time_match: + hour_12 = int(time_match.group(1)) + minute = int(time_match.group(2)) + am_pm = time_match.group(3) + + # Convert to 24-hour + hour = hour_12 + if am_pm == 'P' and hour_12 != 12: + hour = hour_12 + 12 + elif am_pm == 'A' and hour_12 == 12: + hour = 0 + + # Create datetime - try current year first, but if that's in the future, + # it's likely from last year + forecast_datetime = datetime(self.current_year, month, day, hour, minute) + + # If forecast date is in the future, assume it's from last year + # (WXSIM forecasts are typically generated for the current/upcoming period) + if forecast_datetime > self.current_date: + # Try previous year + forecast_datetime = datetime(self.current_year - 1, month, day, hour, minute) + # If that's also in the future (shouldn't happen), keep original + if forecast_datetime > self.current_date: + forecast_datetime = datetime(self.current_year, month, day, hour, minute) + + return forecast_datetime + except (ValueError, KeyError, AttributeError): + return None + + def is_forecast_stale(self, forecast: WXSIMForecast, max_age_hours: int = 48) -> Tuple[bool, Optional[str]]: + """Check if forecast is stale (too old). + + Args: + forecast: Parsed forecast data + max_age_hours: Maximum age in hours before considered stale (default: 48) + + Returns: + Tuple[bool, Optional[str]]: (is_stale, reason_message) + """ + forecast_date = self.get_forecast_date(forecast) + if not forecast_date: + return True, "Could not determine forecast date" + + now = datetime.now() + age = now - forecast_date + age_hours = age.total_seconds() / 3600 + + if age_hours > max_age_hours: + return True, f"Forecast is {age_hours:.1f} hours old (max: {max_age_hours}h)" + + if age_hours < 0: + # Forecast is in the future (shouldn't happen, but handle gracefully) + return True, f"Forecast date is in the future: {forecast_date}" + + return False, None + + @staticmethod + def fetch_from_url(url: str, timeout: int = 10) -> Optional[str]: + """Fetch WXSIM plaintext data from a URL. + + Args: + url: URL to fetch plaintext.txt from + timeout: Request timeout in seconds + + Returns: + Optional[str]: Plaintext content or None on error + """ + try: + response = requests.get(url, timeout=timeout) + response.raise_for_status() + text = response.text + # Verify it looks like WXSIM data + if 'FORECAST RUN' in text.upper() or 'DATE' in text: + return text + return None + except (requests.RequestException, ValueError, AttributeError): + return None diff --git a/modules/commands/alternatives/wx_international.py b/modules/commands/alternatives/wx_international.py index 026c211..509fc37 100644 --- a/modules/commands/alternatives/wx_international.py +++ b/modules/commands/alternatives/wx_international.py @@ -13,6 +13,14 @@ from ..base_command import BaseCommand from ...models import MeshMessage from typing import Any, List, Optional, Tuple, Union +# Import WXSIM parser for custom weather sources +try: + from ...clients.wxsim_parser import WXSIMParser + WXSIM_PARSER_AVAILABLE = True +except ImportError: + WXSIM_PARSER_AVAILABLE = False + WXSIMParser = None + class GlobalWxCommand(BaseCommand): """Handles global weather commands with city/location support""" @@ -47,6 +55,12 @@ class GlobalWxCommand(BaseCommand): super().__init__(bot) self.url_timeout = 10 # seconds + # Initialize WXSIM parser if available + if WXSIM_PARSER_AVAILABLE: + self.wxsim_parser = WXSIMParser() + else: + self.wxsim_parser = None + # Get default state and country from config for city disambiguation self.default_state = self.bot.config.get('Weather', 'default_state', fallback='WA') self.default_country = self.bot.config.get('Weather', 'default_country', fallback='US') @@ -159,6 +173,140 @@ class GlobalWxCommand(BaseCommand): self.logger.debug(f"Error getting bot location: {e}") return None + def _get_custom_wxsim_source(self, location: Optional[str] = None) -> Optional[str]: + """Get custom WXSIM source URL from config. + + Looks for keys in [Weather] section with pattern: custom.wxsim. = + Similar to how Channels_List handles dotted keys. + + Args: + location: Location name or None for default source + + Returns: + Optional[str]: Source URL or None if not found + """ + if not self.wxsim_parser: + return None + + section = 'Weather' + if not self.bot.config.has_section(section): + return None + + if location: + # Strip whitespace and normalize + location = location.strip() + location_lower = location.lower() + + # Look for keys matching custom.wxsim. pattern + prefix = 'custom.wxsim.' + for key, value in self.bot.config.items(section): + if key.startswith(prefix): + # Extract the location name from the key (e.g., "custom.wxsim.lethbridge" -> "lethbridge") + key_location = key[len(prefix):].strip() + if key_location.lower() == location_lower: + return value + else: + # Check for default source: custom.wxsim.default + default_key = 'custom.wxsim.default' + if self.bot.config.has_option(section, default_key): + return self.bot.config.get(section, default_key) + + return None + + def _get_wxsim_weather(self, source_url: str, forecast_type: str = "default", + num_days: int = 7, message: MeshMessage = None, + location_name: Optional[str] = None) -> str: + """Get and format weather from WXSIM source. + + Args: + source_url: URL to WXSIM plaintext.txt file + forecast_type: "default", "tomorrow", or "multiday" + num_days: Number of days for multiday forecast + message: The MeshMessage for dynamic length calculation + location_name: Optional location name for display + + Returns: + str: Formatted weather string + """ + if not self.wxsim_parser: + return self.translate('commands.gwx.error', error="WXSIM parser not available") + + # Fetch WXSIM data + text = self.wxsim_parser.fetch_from_url(source_url, timeout=self.url_timeout) + if not text: + return self.translate('commands.gwx.error_fetching') + + # Parse the data + forecast = self.wxsim_parser.parse(text) + + # Get unit preferences from config + temp_unit = self.bot.config.get('Weather', 'temperature_unit', fallback='fahrenheit').lower() + wind_unit = self.bot.config.get('Weather', 'wind_speed_unit', fallback='mph').lower() + + # Format based on forecast type + if forecast_type == "tomorrow": + # Get tomorrow's forecast + if len(forecast.periods) > 1: + tomorrow = forecast.periods[1] + high = self.wxsim_parser._convert_temp(tomorrow.high_temp, temp_unit) if tomorrow.high_temp else None + low = self.wxsim_parser._convert_temp(tomorrow.low_temp, temp_unit) if tomorrow.low_temp else None + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + result = f"Tomorrow: {tomorrow.conditions}" + if high is not None and low is not None: + result += f" {high}{temp_symbol}/{low}{temp_symbol}" + elif high is not None: + result += f" {high}{temp_symbol}" + elif low is not None: + result += f" {low}{temp_symbol}" + + if tomorrow.precip_chance and tomorrow.precip_chance > 30: + result += f" {tomorrow.precip_chance}% PoP" + + if location_name: + return f"{location_name}: {result}" + return result + else: + return self.translate('commands.gwx.tomorrow_not_available') + + elif forecast_type == "multiday": + # Format multiday forecast + summary = self.wxsim_parser.format_forecast_summary(forecast, num_days, temp_unit, wind_unit) + if location_name: + return f"{location_name}:\n{summary}" + return summary + + else: + # Default: current conditions + today's forecast + current = self.wxsim_parser.format_current_conditions(forecast, temp_unit, wind_unit) + + # Add today's high/low if available + if forecast.periods: + today = forecast.periods[0] + high = self.wxsim_parser._convert_temp(today.high_temp, temp_unit) if today.high_temp else None + low = self.wxsim_parser._convert_temp(today.low_temp, temp_unit) if today.low_temp else None + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + if high is not None and low is not None: + current += f" | H:{high}{temp_symbol} L:{low}{temp_symbol}" + elif high is not None: + current += f" | H:{high}{temp_symbol}" + elif low is not None: + current += f" | L:{low}{temp_symbol}" + + # Add tomorrow if available + if len(forecast.periods) > 1: + tomorrow = forecast.periods[1] + tomorrow_high = self.wxsim_parser._convert_temp(tomorrow.high_temp, temp_unit) if tomorrow.high_temp else None + tomorrow_low = self.wxsim_parser._convert_temp(tomorrow.low_temp, temp_unit) if tomorrow.low_temp else None + + if tomorrow_high is not None and tomorrow_low is not None: + current += f" | Tomorrow: {tomorrow_high}{temp_symbol}/{tomorrow_low}{temp_symbol}" + + if location_name: + return f"{location_name}: {current}" + return current + def _coordinates_to_location_string(self, lat: float, lon: float) -> Optional[str]: """Convert coordinates to a location string (city name) using reverse geocoding. @@ -210,8 +358,22 @@ class GlobalWxCommand(BaseCommand): # Parse the command to extract location and forecast type parts = content.split() - # If no location specified, try to use companion location + # If no location specified, check for custom WXSIM default source first if len(parts) < 2: + wxsim_source = self._get_custom_wxsim_source(None) # Check for default + if wxsim_source: + # Use custom WXSIM default source + try: + self.record_execution(message.sender_id) + weather_data = self._get_wxsim_weather(wxsim_source, "default", 7, message) + await self.send_response(message, weather_data) + return True + except Exception as e: + self.logger.error(f"Error fetching WXSIM weather: {e}") + await self.send_response(message, self.translate('commands.gwx.error', error=str(e))) + return True + + # No custom source, try companion location companion_location = self._get_companion_location(message) if companion_location: # Convert coordinates to location string @@ -261,6 +423,23 @@ class GlobalWxCommand(BaseCommand): await self.send_response(message, self.translate('commands.gwx.usage')) return True + # Check for custom WXSIM source first (before normal geocoding) + wxsim_source = self._get_custom_wxsim_source(location) + if wxsim_source: + # Use custom WXSIM source + try: + self.record_execution(message.sender_id) + weather_data = self._get_wxsim_weather(wxsim_source, forecast_type, num_days, message, location_name=location) + if forecast_type == "multiday": + await self._send_multiday_forecast(message, weather_data) + else: + await self.send_response(message, weather_data) + return True + except Exception as e: + self.logger.error(f"Error fetching WXSIM weather: {e}") + await self.send_response(message, self.translate('commands.gwx.error', error=str(e))) + return True + try: # Record execution for this user self.record_execution(message.sender_id) diff --git a/modules/commands/wx_command.py b/modules/commands/wx_command.py index eb59668..47d1204 100644 --- a/modules/commands/wx_command.py +++ b/modules/commands/wx_command.py @@ -26,6 +26,14 @@ except ImportError: WX_INTERNATIONAL_AVAILABLE = False GlobalWxCommand = None +# Import WXSIM parser for custom weather sources +try: + from ..clients.wxsim_parser import WXSIMParser + WXSIM_PARSER_AVAILABLE = True +except ImportError: + WXSIM_PARSER_AVAILABLE = False + WXSIMParser = None + class WxCommand(BaseCommand): """Handles weather commands with zipcode support""" @@ -56,6 +64,12 @@ class WxCommand(BaseCommand): super().__init__(bot) self.wx_enabled = self.get_config_value('Wx_Command', 'enabled', fallback=True, value_type='bool') + # Initialize WXSIM parser if available + if WXSIM_PARSER_AVAILABLE: + self.wxsim_parser = WXSIMParser() + else: + self.wxsim_parser = None + # Check weather provider setting - delegate to international command if using Open-Meteo weather_provider = bot.config.get('Weather', 'weather_provider', fallback='noaa').lower() if weather_provider == 'openmeteo' and WX_INTERNATIONAL_AVAILABLE: @@ -218,6 +232,156 @@ class WxCommand(BaseCommand): self.logger.debug(f"Error getting bot location: {e}") return None + def _get_custom_wxsim_source(self, location: Optional[str] = None) -> Optional[str]: + """Get custom WXSIM source URL from config. + + Looks for keys in [Weather] section with pattern: custom.wxsim. = + Similar to how Channels_List handles dotted keys. + + Args: + location: Location name or None for default source + + Returns: + Optional[str]: Source URL or None if not found + """ + if not self.wxsim_parser: + self.logger.debug("WXSIM parser not available") + return None + + section = 'Weather' + if not self.bot.config.has_section(section): + self.logger.debug(f"Config section '{section}' does not exist") + return None + + if location: + # Strip whitespace and normalize + location = location.strip() + location_lower = location.lower() + self.logger.debug(f"Checking for WXSIM source for location: '{location}' (normalized: '{location_lower}')") + + # Look for keys matching custom.wxsim. pattern + prefix = 'custom.wxsim.' + for key, value in self.bot.config.items(section): + if key.startswith(prefix): + # Extract the location name from the key (e.g., "custom.wxsim.lethbridge" -> "lethbridge") + key_location = key[len(prefix):].strip() + if key_location.lower() == location_lower: + self.logger.debug(f"Found WXSIM source: {key} = {value}") + return value + + self.logger.debug(f"No WXSIM source found for location '{location}'") + else: + # Check for default source: custom.wxsim.default + default_key = 'custom.wxsim.default' + if self.bot.config.has_option(section, default_key): + url = self.bot.config.get(section, default_key) + self.logger.debug(f"Found default WXSIM source: {url}") + return url + self.logger.debug("No default WXSIM source configured") + + return None + + def _get_wxsim_weather(self, source_url: str, forecast_type: str = "default", + num_days: int = 7, message: MeshMessage = None, + location_name: Optional[str] = None) -> str: + """Get and format weather from WXSIM source. + + Args: + source_url: URL to WXSIM plaintext.txt file + forecast_type: "default", "tomorrow", or "multiday" + num_days: Number of days for multiday forecast + message: The MeshMessage for dynamic length calculation + location_name: Optional location name for display + + Returns: + str: Formatted weather string + """ + if not self.wxsim_parser: + return self.translate('commands.wx.error', error="WXSIM parser not available") + + # Fetch WXSIM data + text = self.wxsim_parser.fetch_from_url(source_url, timeout=self.url_timeout) + if not text: + return self.translate('commands.wx.error', error="Failed to fetch WXSIM data") + + # Parse the data + forecast = self.wxsim_parser.parse(text) + + # Validate forecast is not stale + is_stale, stale_reason = self.wxsim_parser.is_forecast_stale(forecast, max_age_hours=48) + if is_stale: + self.logger.warning(f"WXSIM forecast appears stale: {stale_reason}") + # Still return the forecast, but log the warning + # Optionally, we could return an error message here instead + + # Get unit preferences from config + temp_unit = self.bot.config.get('Weather', 'temperature_unit', fallback='fahrenheit').lower() + wind_unit = self.bot.config.get('Weather', 'wind_speed_unit', fallback='mph').lower() + + # Format based on forecast type + if forecast_type == "tomorrow": + # Get tomorrow's forecast + if len(forecast.periods) > 1: + tomorrow = forecast.periods[1] + high = self.wxsim_parser._convert_temp(tomorrow.high_temp, temp_unit) if tomorrow.high_temp else None + low = self.wxsim_parser._convert_temp(tomorrow.low_temp, temp_unit) if tomorrow.low_temp else None + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + result = f"Tomorrow: {tomorrow.conditions}" + if high is not None and low is not None: + result += f" {high}{temp_symbol}/{low}{temp_symbol}" + elif high is not None: + result += f" {high}{temp_symbol}" + elif low is not None: + result += f" {low}{temp_symbol}" + + if tomorrow.precip_chance and tomorrow.precip_chance > 30: + result += f" {tomorrow.precip_chance}% PoP" + + if location_name: + return f"{location_name}: {result}" + return result + else: + return self.translate('commands.wx.error', error="Tomorrow forecast not available") + + elif forecast_type == "multiday": + # Format multiday forecast + summary = self.wxsim_parser.format_forecast_summary(forecast, num_days, temp_unit, wind_unit) + if location_name: + return f"{location_name}:\n{summary}" + return summary + + else: + # Default: current conditions + today's forecast + current = self.wxsim_parser.format_current_conditions(forecast, temp_unit, wind_unit) + + # Add today's high/low if available (use first period as "today") + if forecast.periods: + today = forecast.periods[0] + high = self.wxsim_parser._convert_temp(today.high_temp, temp_unit) if today.high_temp else None + low = self.wxsim_parser._convert_temp(today.low_temp, temp_unit) if today.low_temp else None + temp_symbol = "°F" if temp_unit == 'fahrenheit' else "°C" + + if high is not None and low is not None: + current += f" | H:{high}{temp_symbol} L:{low}{temp_symbol}" + elif high is not None: + current += f" | H:{high}{temp_symbol}" + elif low is not None: + current += f" | L:{low}{temp_symbol}" + + # Add tomorrow if available (second period) + if len(forecast.periods) > 1: + tomorrow = forecast.periods[1] + tomorrow_high = self.wxsim_parser._convert_temp(tomorrow.high_temp, temp_unit) if tomorrow.high_temp else None + tomorrow_low = self.wxsim_parser._convert_temp(tomorrow.low_temp, temp_unit) if tomorrow.low_temp else None + + if tomorrow_high is not None and tomorrow_low is not None: + current += f" | Tomorrow: {tomorrow_high}{temp_symbol}/{tomorrow_low}{temp_symbol}" + + if location_name: + return f"{location_name}: {current}" + return current + def _coordinates_to_location_string(self, lat: float, lon: float) -> Optional[str]: """Convert coordinates to a location string (city name) using reverse geocoding. @@ -272,8 +436,22 @@ class WxCommand(BaseCommand): # Track if we're using companion location (so we always show location in response) using_companion_location = False - # If no location specified, try to use companion location + # If no location specified, check for custom WXSIM default source first if len(parts) < 2: + wxsim_source = self._get_custom_wxsim_source(None) # Check for default + if wxsim_source: + # Use custom WXSIM default source + try: + self.record_execution(message.sender_id) + weather_data = self._get_wxsim_weather(wxsim_source, "default", 7, message) + await self.send_response(message, weather_data) + return True + except Exception as e: + self.logger.error(f"Error fetching WXSIM weather: {e}") + await self.send_response(message, self.translate('commands.wx.error', error=str(e))) + return True + + # No custom source, try companion location companion_location = self._get_companion_location(message) if companion_location: # Use coordinates directly to avoid re-geocoding issues @@ -332,6 +510,26 @@ class WxCommand(BaseCommand): await self.send_response(message, self.translate('commands.wx.usage')) return True + # Check for custom WXSIM source first (before checking location type) + wxsim_source = self._get_custom_wxsim_source(location) + if wxsim_source: + self.logger.info(f"Using custom WXSIM source for location '{location}': {wxsim_source}") + # Use custom WXSIM source + try: + self.record_execution(message.sender_id) + weather_data = self._get_wxsim_weather(wxsim_source, forecast_type, num_days, message, location_name=location) + if forecast_type == "multiday": + await self._send_multiday_forecast(message, weather_data) + else: + await self.send_response(message, weather_data) + return True + except Exception as e: + self.logger.error(f"Error fetching WXSIM weather: {e}") + await self.send_response(message, self.translate('commands.wx.error', error=str(e))) + return True + else: + self.logger.debug(f"No custom WXSIM source found for location '{location}', using normal weather API") + # Check if it's coordinates, zipcode, or city name if re.match(r'^\s*-?\d+\.?\d*\s*,\s*-?\d+\.?\d*\s*$', location): # It's coordinates (lat,lon format)