#!/usr/bin/env python3 """ Weather Agent Fetches weather for a configurable location and returns structured data + markdown. Called by Daily Briefing agents with location config. Can also run standalone. """ import os import sys from datetime import datetime from shared import MT, api_request, log_run, get_instance_config AGENT_ID = "weather" # Default location (Providence, UT) DEFAULT_LOCATION = { "name": "Providence", "state": "Utah", "country": "US", "lat": 41.7064, "lon": -111.8133, } WMO_CODES = { 0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast", 45: "Foggy", 48: "Icy fog", 51: "Light drizzle", 53: "Drizzle", 55: "Heavy drizzle", 61: "Light rain", 63: "Rain", 65: "Heavy rain", 66: "Freezing rain", 67: "Heavy freezing rain", 71: "Light snow", 73: "Snow", 75: "Heavy snow", 77: "Snow grains", 80: "Light showers", 81: "Showers", 82: "Heavy showers", 85: "Light snow showers", 86: "Heavy snow showers", 95: "Thunderstorm", 96: "Thunderstorm w/ hail", 99: "Severe thunderstorm", } DAY_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] def build_url(location): """Build the Open-Meteo API URL for a given location.""" lat = location["lat"] lon = location["lon"] return ( f"https://api.open-meteo.com/v1/forecast?" f"latitude={lat}&longitude={lon}" f"¤t=temperature_2m,apparent_temperature,weather_code,wind_speed_10m,relative_humidity_2m" f"&daily=weather_code,temperature_2m_max,temperature_2m_min,precipitation_sum,wind_speed_10m_max,sunrise,sunset" f"&temperature_unit=fahrenheit&wind_speed_unit=mph&precipitation_unit=inch" f"&timezone=America/Denver&forecast_days=7" ) def location_label(location): """Human-readable location string.""" parts = [location.get("name", "")] if location.get("state"): parts.append(location["state"]) if location.get("country") and location["country"] != "US": parts.append(location["country"]) return ", ".join(p for p in parts if p) def fetch_weather(location=None): """Fetch weather data from Open-Meteo for a given location.""" loc = location or DEFAULT_LOCATION url = build_url(loc) return api_request(url) def format_section(weather, location=None): """Format weather into a markdown section and a one-line summary.""" loc = location or DEFAULT_LOCATION label = location_label(loc) now = datetime.now(MT) current = weather["current"] daily = weather["daily"] condition = WMO_CODES.get(current["weather_code"], "Unknown") temp = round(current["temperature_2m"]) feels = round(current["apparent_temperature"]) wind = round(current["wind_speed_10m"]) humidity = current["relative_humidity_2m"] md = f"## Weather — {label}\n\n" md += "### Current Conditions\n\n" md += "| | |\n|---|---|\n" md += f"| **Condition** | {condition} |\n" md += f"| **Temperature** | {temp}°F (feels like {feels}°F) |\n" md += f"| **Wind** | {wind} mph |\n" md += f"| **Humidity** | {humidity}% |\n\n" md += "### 7-Day Forecast\n\n" md += "| Day | Condition | High | Low | Precip | Wind |\n" md += "|-----|-----------|------|-----|--------|------|\n" for i in range(len(daily["time"])): d = datetime.strptime(daily["time"][i], "%Y-%m-%d") day_name = DAY_NAMES[d.weekday()] if i == 0: day_name = "**Today**" elif i == 1: day_name = "Tomorrow" cond = WMO_CODES.get(daily["weather_code"][i], "?") hi = round(daily["temperature_2m_max"][i]) lo = round(daily["temperature_2m_min"][i]) precip = daily["precipitation_sum"][i] wind_max = round(daily["wind_speed_10m_max"][i]) precip_str = f'{precip}"' if precip > 0 else "-" md += f"| {day_name} | {cond} | {hi}°F | {lo}°F | {precip_str} | {wind_max} mph |\n" sunrise = daily["sunrise"][0].split("T")[1] if daily["sunrise"][0] else "?" sunset = daily["sunset"][0].split("T")[1] if daily["sunset"][0] else "?" md += f"\n**Sunrise:** {sunrise} | **Sunset:** {sunset}\n" summary = f"{label}: {condition}, {temp}°F (feels like {feels}°F), wind {wind} mph" return md, summary def build_result(weather, location): """Extract the structured 'result' dict. This is what API consumers (Synap, WSIT) read.""" loc = location or DEFAULT_LOCATION current = weather["current"] daily = weather["daily"] condition = WMO_CODES.get(current["weather_code"], "Unknown") forecast = [] for i in range(len(daily["time"])): forecast.append({ "date": daily["time"][i], "weekday": DAY_NAMES[datetime.strptime(daily["time"][i], "%Y-%m-%d").weekday()], "condition": WMO_CODES.get(daily["weather_code"][i], "Unknown"), "weather_code": daily["weather_code"][i], "temp_high_f": round(daily["temperature_2m_max"][i]), "temp_low_f": round(daily["temperature_2m_min"][i]), "precip_in": daily["precipitation_sum"][i], "wind_max_mph": round(daily["wind_speed_10m_max"][i]), "sunrise": daily["sunrise"][i], "sunset": daily["sunset"][i], }) return { "location": { "name": loc.get("name", ""), "state": loc.get("state", ""), "country": loc.get("country", ""), "lat": loc["lat"], "lon": loc["lon"], "label": location_label(loc), }, "current": { "condition": condition, "weather_code": current["weather_code"], "temperature_f": round(current["temperature_2m"]), "feels_like_f": round(current["apparent_temperature"]), "wind_mph": round(current["wind_speed_10m"]), "humidity_pct": current["relative_humidity_2m"], }, "forecast": forecast, "fetched_at": datetime.now(MT).isoformat(), } def run_structured(location=None): """Fetch weather and return {result, markdown, summary}. This is the new contract — structured data for API consumers, markdown for the wiki.""" loc = location or DEFAULT_LOCATION weather = fetch_weather(loc) section, summary = format_section(weather, loc) result = build_result(weather, loc) return {"result": result, "markdown": section, "summary": summary} def run(location=None): """Legacy entrypoint: returns (markdown_section, summary). Kept for backward compat with daily_briefing.py's older call sites. Prefer run_structured(). Deliberately does NOT call log_run — logging is the caller's responsibility. """ out = run_structured(location) return out["markdown"], out["summary"] def _location_from_config(cfg): """Map an instance config dict to a location dict. Config keys: name, state, country, lat, lon.""" if not cfg or "lat" not in cfg or "lon" not in cfg: return DEFAULT_LOCATION return { "name": cfg.get("name", "Custom"), "state": cfg.get("state", ""), "country": cfg.get("country", "US"), "lat": cfg["lat"], "lon": cfg["lon"], } def _main_from_api(): """Entry point when invoked via /api/instances/{id}/trigger. Reads INSTANCE_ID + RUN_ID from env, fetches config, runs, and posts structured result back to the dashboard.""" instance_id = int(os.environ["INSTANCE_ID"]) try: cfg = get_instance_config(instance_id) loc = _location_from_config(cfg) out = run_structured(loc) log_run(AGENT_ID, "success", output=out["markdown"], result=out["result"], metadata={"location": out["result"]["location"]["label"], "summary": out["summary"]}) print(out["summary"]) except Exception as e: err_msg = f"{type(e).__name__}: {e}" print(f"Error: {err_msg}", file=sys.stderr) log_run(AGENT_ID, "failed", err=err_msg) sys.exit(1) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Weather Agent") parser.add_argument("--from-api", action="store_true", help="Run from /trigger: read INSTANCE_ID/RUN_ID from env, fetch config, post result") parser.add_argument("--lat", type=float, help="Latitude") parser.add_argument("--lon", type=float, help="Longitude") parser.add_argument("--name", type=str, help="City/location name") parser.add_argument("--state", type=str, help="State") parser.add_argument("--country", type=str, default="US", help="Country code") args = parser.parse_args() if args.from_api: _main_from_api() sys.exit(0) loc = DEFAULT_LOCATION if args.lat and args.lon: loc = { "name": args.name or "Custom", "state": args.state or "", "country": args.country, "lat": args.lat, "lon": args.lon, } try: out = run_structured(loc) print(out["markdown"]) print(f"\nSummary: {out['summary']}") except Exception as e: err_msg = f"{type(e).__name__}: {e}" print(f"Error: {err_msg}", file=sys.stderr) log_run(AGENT_ID, "failed", err=err_msg) sys.exit(1)