#!/usr/bin/env python3 """weather -- one fetcher for the whole fleet. Hits Open-Meteo (model=best_match, so NOAA blend in the US, ECMWF/ICON abroad), normalises the reading, caches the last good value, and prints it. Every app (waybar, clock-panel, an Emacs svg.el panel) shells out to this and renders the JSON -- nobody re-implements the API call, the code mapping, or the cache. Stdlib only: no pip, no venv. Runs as a bare executable. Contract (--format json): { "temp_f": 82, "feels_f": 92, "condition": "clear", # clear|partly|cloud|fog|rain|snow|storm "wmo_code": 0, "windy": false, # wind_mph >= threshold (default 15) "wind_mph": 6, "wind_dir": "WNW", "wind_deg": 293, "location": "New Orleans", "observed": "2026-07-18T01:00", "stale": false } Location resolves in order: --lat/--lon, then $WEATHER_LAT/$WEATHER_LON, then ~/.config/weather/config.json, then `whereami` (with --geo). No coordinates are baked in, so nothing personal ships in the repo -- each machine configures its own. """ import argparse import json import os import subprocess import sys import time import urllib.error import urllib.request from pathlib import Path API = "https://api.open-meteo.com/v1/forecast" CONFIG = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "weather" / "config.json" CACHE = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "weather" / "current.json" DEFAULT_TTL = 900 # 15 min -- weather changes slowly; keep API calls rare DEFAULT_WINDY = 15 # mph; Beaufort 4, where wind stops being ambient # WMO weather-interpretation code -> our semantic condition. # Ranges collapse to the seven condition glyphs; "windy" is derived from speed. _WMO = { 0: "clear", 1: "partly", 2: "partly", 3: "cloud", 45: "fog", 48: "fog", 51: "rain", 53: "rain", 55: "rain", 56: "rain", 57: "rain", 61: "rain", 63: "rain", 65: "rain", 66: "rain", 67: "rain", 80: "rain", 81: "rain", 82: "rain", 71: "snow", 73: "snow", 75: "snow", 77: "snow", 85: "snow", 86: "snow", 95: "storm", 96: "storm", 99: "storm", } _CARDINALS = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", "S", "SSW", "SW", "WSW", "W", "WNW", "NW", "NNW"] def condition_for(code): return _WMO.get(int(code), "cloud") # unknown code -> cloud, the safe neutral def cardinal(deg): return _CARDINALS[int((deg % 360) / 22.5 + 0.5) % 16] def resolve_location(args): """Return (lat, lon, name). Raise if nothing configured.""" if args.lat is not None and args.lon is not None: return args.lat, args.lon, args.location env_lat, env_lon = os.environ.get("WEATHER_LAT"), os.environ.get("WEATHER_LON") if env_lat and env_lon: return float(env_lat), float(env_lon), args.location or os.environ.get("WEATHER_NAME") if CONFIG.exists(): cfg = json.loads(CONFIG.read_text()) if args.geo or cfg.get("use_geo"): geo = _whereami() if geo: return geo[0], geo[1], args.location or cfg.get("name") if "lat" in cfg and "lon" in cfg: return cfg["lat"], cfg["lon"], args.location or cfg.get("name") if args.geo: geo = _whereami() if geo: return geo[0], geo[1], args.location raise SystemExit( "weather: no location. Pass --lat/--lon, set $WEATHER_LAT/$WEATHER_LON, " f"write {CONFIG} (\"lat\"/\"lon\"[/\"name\"]), or use --geo on a machine " "that travels." ) def _whereami(): """Best-effort coords from the `whereami` tool. Only meaningful where the machine actually moves (velox); silently gives up otherwise.""" try: out = subprocess.run(["whereami"], capture_output=True, text=True, timeout=20).stdout except (OSError, subprocess.SubprocessError): return None for line in out.splitlines(): if "coords:" in line: try: nums = line.split("coords:")[1].split("(")[0] lat, lon = (float(x) for x in nums.replace(",", " ").split()[:2]) return lat, lon except (ValueError, IndexError): return None return None def fetch(lat, lon, name, threshold): params = ( f"?latitude={lat}&longitude={lon}" "¤t=temperature_2m,apparent_temperature,weather_code," "wind_speed_10m,wind_direction_10m" "&temperature_unit=fahrenheit&wind_speed_unit=mph" "&models=best_match&timezone=auto" ) req = urllib.request.Request(API + params, headers={"User-Agent": "weather-cli/1.0"}) with urllib.request.urlopen(req, timeout=15) as resp: data = json.load(resp) cur = data["current"] mph = round(cur["wind_speed_10m"]) deg = cur["wind_direction_10m"] return { "temp_f": round(cur["temperature_2m"]), "feels_f": round(cur["apparent_temperature"]), "condition": condition_for(cur["weather_code"]), "wmo_code": int(cur["weather_code"]), "windy": mph >= threshold, "wind_mph": mph, "wind_dir": cardinal(deg), "wind_deg": round(deg), "location": name or f"{lat:.2f},{lon:.2f}", "observed": cur["time"], "stale": False, } def read_cache(): try: blob = json.loads(CACHE.read_text()) return blob["fetched_at"], blob["reading"] except (OSError, ValueError, KeyError): return None, None def write_cache(reading): CACHE.parent.mkdir(parents=True, exist_ok=True) CACHE.write_text(json.dumps({"fetched_at": int(time.time()), "reading": reading})) def get_reading(args): """Cache-first, network-second, stale-cache-on-failure. One shared cache means five widgets polling still hit the API at most once per TTL.""" fetched_at, cached = read_cache() if cached and not args.no_cache and fetched_at and (time.time() - fetched_at) < args.ttl: return cached lat, lon, name = resolve_location(args) try: reading = fetch(lat, lon, name, args.windy_threshold) write_cache(reading) return reading except (urllib.error.URLError, OSError, KeyError, ValueError) as e: if cached: # never blank the bar: serve last good, flagged cached["stale"] = True return cached raise SystemExit(f"weather: fetch failed and no cache to fall back on ({e})") def render(reading, fmt): if fmt == "json": return json.dumps(reading) if fmt == "plain": wind = "windy" if reading["windy"] else reading["wind_dir"] s = f"{reading['temp_f']}°F {reading['condition']}, wind {wind} {reading['wind_mph']}" return s + (" (stale)" if reading["stale"] else "") if fmt == "waybar": # Data only; the caller's stylesheet picks the icon by `alt`/`class`. wind = "windy" if reading["windy"] else f"{reading['wind_dir']} {reading['wind_mph']}" return json.dumps({ "text": f"{reading['temp_f']}° {reading['wind_mph']}", "alt": reading["condition"], "class": (["stale"] if reading["stale"] else []) + [reading["condition"]] + (["windy"] if reading["windy"] else []), "tooltip": f"{reading['location']}: {reading['temp_f']}°F " f"(feels {reading['feels_f']}°), {reading['condition']}, " f"wind {reading['wind_dir']} {reading['wind_mph']} mph" + (" [stale]" if reading["stale"] else ""), }) raise SystemExit(f"weather: unknown format {fmt!r}") def main(argv=None): p = argparse.ArgumentParser(description="Normalised current weather from Open-Meteo.") p.add_argument("--lat", type=float, help="latitude override") p.add_argument("--lon", type=float, help="longitude override") p.add_argument("--location", help="display name for the reading") p.add_argument("--geo", action="store_true", help="geolocate via `whereami` (traveling machines)") p.add_argument("--format", choices=["json", "plain", "waybar"], default="json") p.add_argument("--windy-threshold", type=int, default=DEFAULT_WINDY, help="mph at which windy=true") p.add_argument("--ttl", type=int, default=DEFAULT_TTL, help="cache seconds") p.add_argument("--no-cache", action="store_true", help="force a fresh fetch") args = p.parse_args(argv) reading = get_reading(args) # Derive windy from cached wind_mph so a changed threshold takes effect # without waiting for the cache to expire. reading["windy"] = reading["wind_mph"] >= args.windy_threshold print(render(reading, args.format)) if __name__ == "__main__": main()