229 lines
8.8 KiB
Python
229 lines
8.8 KiB
Python
"""
|
||
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 <location>\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__ = """
|
||
<details>
|
||
<summary><strong>!weather</strong> – Current weather</summary>
|
||
<p><code>!weather <location></code> – Shows temperature, conditions, humidity, wind in a clean, aligned table. Uses OpenWeatherMap primary, Open‑Meteo fallback.</p>
|
||
</details>
|
||
"""
|