""" Weather plugin – primary: OpenWeatherMap, fallback: Open‑Meteo. Outputs a formatted code block with emojis and perfectly aligned columns. """ import logging import os import aiohttp import simplematrixbotlib as botlib from dotenv import load_dotenv from urllib.parse import quote from plugins.common import html_escape, collapsible_summary, code_block OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY", "") # --------------------------------------------------------------------------- # OpenWeatherMap helpers # --------------------------------------------------------------------------- async def openweathermap_get(session: aiohttp.ClientSession, location: str) -> dict | None: if not OPENWEATHER_API_KEY: logging.info("OpenWeatherMap key missing, skipping primary") return None url = "https://api.openweathermap.org/data/2.5/weather" params = { "q": location, "appid": OPENWEATHER_API_KEY, "units": "metric", } try: async with session.get(url, params=params, timeout=10) as resp: if resp.status == 200: return await resp.json() logging.info(f"OpenWeatherMap HTTP {resp.status}, falling back") except Exception as e: logging.warning(f"OpenWeatherMap request error: {e}") return None # --------------------------------------------------------------------------- # Open‑Meteo helpers (fallback) # --------------------------------------------------------------------------- async def meteo_geocode(session: aiohttp.ClientSession, location: str) -> dict | None: url = "https://geocoding-api.open-meteo.com/v1/search" params = {"name": location, "count": 1, "language": "en"} try: async with session.get(url, params=params, timeout=10) as resp: if resp.status == 200: data = await resp.json() results = data.get("results", []) if results: r = results[0] return { "name": r["name"], "latitude": r["latitude"], "longitude": r["longitude"], "country": r.get("country", ""), "state": r.get("admin1", ""), "timezone": r.get("timezone", "UTC"), } except Exception as e: logging.warning(f"Open‑Meteo geocode error: {e}") return None async def meteo_weather(session: aiohttp.ClientSession, lat: float, lon: float, timezone: str = "auto") -> dict | None: url = "https://api.open-meteo.com/v1/forecast" params = { "latitude": lat, "longitude": lon, "current_weather": "true", "temperature_unit": "fahrenheit", "windspeed_unit": "mph", "timezone": timezone, } try: async with session.get(url, params=params, timeout=10) as resp: if resp.status == 200: return await resp.json() except Exception as e: logging.warning(f"Open‑Meteo weather error: {e}") return None # --------------------------------------------------------------------------- # Formatting # --------------------------------------------------------------------------- def format_openweathermap(data: dict) -> str: """Build a code block from OpenWeatherMap response.""" city = data.get("name", "Unknown") sys_data = data.get("sys", {}) country = sys_data.get("country", "") main = data.get("main", {}) temp_c = main.get("temp", 0) temp_f = round(temp_c * 9 / 5 + 32, 1) humidity = main.get("humidity", 0) wind_speed = data.get("wind", {}).get("speed", 0) weather_list = data.get("weather", []) description = weather_list[0]["description"].capitalize() if weather_list else "Unknown" emoji_map = { "Clear": "☀️", "Clouds": "☁️", "Rain": "🌧️", "Drizzle": "🌦️", "Thunderstorm": "⛈️", "Snow": "❄️", "Mist": "🌫️", "Fog": "🌫️", "Haze": "🌫️", "Smoke": "🌫️", "Dust": "🌫️", "Sand": "🌫️", "Ash": "🌫️", "Squall": "💨", "Tornado": "🌪️", } main_weather = weather_list[0].get("main", "") if weather_list else "" weather_emoji = emoji_map.get(main_weather, "🌡️") location = f"{city}, {country}" if country else city rows = [ ("🌍", "Location", location), (weather_emoji, "Condition", description), ("🌡️", "Temperature", f"{temp_c:.1f}°C / {temp_f:.1f}°F"), ("💧", "Humidity", f"{humidity}%"), ("💨", "Wind Speed", f"{wind_speed} m/s"), ] sections = [{"title": "", "rows": rows}] return code_block(f"🌤️ Weather for {location}", sections) def format_meteo(loc_info: dict, weather_data: dict) -> str: """Build a code block from Open‑Meteo response.""" c = weather_data["current_weather"] code = c["weathercode"] wmo_emoji = { 0: ("Clear sky", "☀️"), 1: ("Mainly clear", "🌤️"), 2: ("Partly cloudy", "⛅"), 3: ("Overcast", "☁️"), 45: ("Fog", "🌫️"), 51: ("Light drizzle", "🌦️"), 61: ("Slight rain", "🌧️"), 63: ("Moderate rain", "🌧️"), 71: ("Slight snow", "❄️"), 95: ("Thunderstorm", "⛈️"), } desc, emoji = wmo_emoji.get(code, ("Unknown", "🌡️")) location_parts = [loc_info["name"]] if loc_info.get("state") and loc_info["state"] != loc_info["name"]: location_parts.append(loc_info["state"]) if loc_info.get("country"): location_parts.append(loc_info["country"]) location = ", ".join(location_parts) temp_f = c["temperature"] temp_c = round((temp_f - 32) * 5 / 9, 1) wind = c["windspeed"] # mph rows = [ ("🌍", "Location", location), (emoji, "Condition", desc), ("🌡️", "Temperature", f"{temp_c}°C / {temp_f}°F"), ("💨", "Wind Speed", f"{wind} mph"), ] sections = [{"title": "", "rows": rows}] return code_block(f"🌤️ Weather for {location}", sections) # --------------------------------------------------------------------------- # Plugin entry point # --------------------------------------------------------------------------- async def handle_command(room, message, bot, prefix, config): match = botlib.MessageMatch(room, message, bot, prefix) if not (match.is_not_from_this_bot() and match.prefix() and match.command("weather")): return args = match.args() if not args: await bot.api.send_text_message( room.room_id, "Usage: !weather \nExample: !weather London or !weather New York,US" ) return location = " ".join(args) logging.info("Received !weather command for '%s'", location) async with aiohttp.ClientSession() as session: # 1. Try OpenWeatherMap owm_data = await openweathermap_get(session, location) if owm_data and owm_data.get("cod") == 200: block = format_openweathermap(owm_data) output = collapsible_summary(f"🌤️ Weather: {html_escape(location)}", block) await bot.api.send_markdown_message(room.room_id, output) return # 2. Fallback: Open‑Meteo logging.info("Falling back to Open‑Meteo") loc_info = await meteo_geocode(session, location) if not loc_info: await bot.api.send_text_message( room.room_id, f"Location '{html_escape(location)}' not found." ) return wdata = await meteo_weather(session, loc_info["latitude"], loc_info["longitude"], loc_info.get("timezone", "auto")) if not wdata: await bot.api.send_text_message( room.room_id, "Could not fetch weather data from any provider." ) return block = format_meteo(loc_info, wdata) output = collapsible_summary(f"🌤️ Weather: {html_escape(location)}", block) await bot.api.send_markdown_message(room.room_id, output) logging.info("Sent weather via Open‑Meteo (fallback)") def setup(bot): logging.info("Weather plugin loaded (OpenWeatherMap + Open‑Meteo fallback)") # --------------------------------------------------------------------------- # Plugin Metadata # --------------------------------------------------------------------------- __version__ = "1.1.1" __author__ = "Funguy Bot" __description__ = "Weather data plugin" __help__ = """
!weather – Current weather

!weather <location> – Shows temperature, conditions, humidity, wind in a clean, aligned table. Uses OpenWeatherMap primary, Open‑Meteo fallback.

"""