ai-agents/agents/weather_agent.py

#!/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"&current=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)