#!/usr/bin/env python3 """Top recently-heard birds near a ZIP, from BirdWeather, for the frame's --bird-weather mode. Returns the shape the collage already expects, [{"sci","com","n"}], so the existing renderer draws it unchanged. Standalone (Python stdlib only). Filtered to species the collage can actually draw, read from the slug list bundled in apt.js, so no network call and it tracks whatever illustrations the repo ships. BirdWeather's public GraphQL needs no key. """ from __future__ import annotations import json import math import os import re import urllib.parse import urllib.request BIRDWEATHER = "https://app.birdweather.com/graphql" GEOCODER = "https://api.zippopotam.us" EBIRD = "https://api.ebird.org/v2/data/obs/geo/recent" APT_JS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "avian", "frontend", "apt.js") _drawable = None def _graphql(query, timeout): req = urllib.request.Request( BIRDWEATHER, data=json.dumps({"query": query}).encode(), headers={"Content-Type": "application/json", "User-Agent": "AvianVisitors-frame/1.0"}, ) try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read(2_000_000)) except Exception: return {} # a transient upstream failure degrades to the next radius or fallback def geocode(zip_code, country="us", timeout=20): """ZIP / postal code to (lat, lon) via the keyless zippopotam.us gazetteer.""" url = f"{GEOCODER}/{country}/{urllib.parse.quote(zip_code.strip())}" req = urllib.request.Request(url, headers={"User-Agent": "AvianVisitors-frame/1.0"}) with urllib.request.urlopen(req, timeout=timeout) as r: place = json.loads(r.read(200_000))["places"][0] return float(place["latitude"]), float(place["longitude"]) def bbox(lat, lon, miles): """Square box `miles` to each side of the point. Returns (ne, sw) corners.""" dlat = miles / 69.0 dlon = miles / (69.0 * max(0.2, math.cos(math.radians(lat)))) return (lat + dlat, lon + dlon), (lat - dlat, lon - dlon) def top_species(lat, lon, miles, days=7, limit=60, timeout=20): """BirdWeather's most-detected species in the box, as [{sci,com,n}].""" (ne_lat, ne_lon), (sw_lat, sw_lon) = bbox(lat, lon, miles) query = ( '{ topSpecies(period: {count: %d, unit: "day"}, ' 'ne: {lat: %f, lon: %f}, sw: {lat: %f, lon: %f}, limit: %d) ' '{ count species { commonName scientificName } } }' % (days, ne_lat, ne_lon, sw_lat, sw_lon, limit) ) rows = (_graphql(query, timeout).get("data") or {}).get("topSpecies") or [] out = [] for row in rows: sp = row.get("species") or {} sci = sp.get("scientificName") if sci: out.append({"sci": sci, "com": sp.get("commonName") or sci, "n": row.get("count") or 1}) return out def _haversine(lat1, lon1, lat2, lon2): """Great-circle distance between two points, in miles.""" r = 3958.8 p1, p2 = math.radians(lat1), math.radians(lat2) dp, dl = math.radians(lat2 - lat1), math.radians(lon2 - lon1) a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 return 2 * r * math.asin(math.sqrt(a)) def nearest_stations(lat, lon, n=3, boxes=(50, 150, 400), timeout=20): """The n closest BirdWeather stations to the point, as [(miles, id)], found by growing the search box until at least n are in view.""" nodes = [] for miles in boxes: (ne_lat, ne_lon), (sw_lat, sw_lon) = bbox(lat, lon, miles) query = ('{ stations(ne: {lat: %f, lon: %f}, sw: {lat: %f, lon: %f}, first: 200) ' '{ nodes { id coords { lat lon } } } }' % (ne_lat, ne_lon, sw_lat, sw_lon)) nodes = (((_graphql(query, timeout).get("data") or {}).get("stations") or {}).get("nodes")) or [] if len(nodes) >= n: break out = [] for node in nodes: c = node.get("coords") or {} if c.get("lat") is not None and c.get("lon") is not None: out.append((_haversine(lat, lon, c["lat"], c["lon"]), node["id"])) out.sort() return out[:n] def triangulate(lat, lon, n=3, days=7, timeout=20): """Estimate the birds near a point from the n closest stations, inverse- distance weighted so the nearest station counts most. Used where the local box has too few stations to fill a collage, so remote ZIPs still get birds.""" weighted = {} for miles, sid in nearest_stations(lat, lon, n, timeout=timeout): weight = 1.0 / max(miles, 1.0) query = ('{ topSpecies(stationIds: [%s], period: {count: %d, unit: "day"}, limit: 40) ' '{ count species { commonName scientificName } } }' % (sid, days)) for row in (((_graphql(query, timeout).get("data") or {}).get("topSpecies")) or []): sp = row.get("species") or {} sci = sp.get("scientificName") if not sci: continue entry = weighted.setdefault(sci, {"com": sp.get("commonName") or sci, "score": 0.0}) entry["score"] += (row.get("count") or 0) * weight out = [{"sci": sci, "com": e["com"], "n": max(1, round(e["score"]))} for sci, e in weighted.items()] out.sort(key=lambda s: -s["n"]) return out def ebird_nearby(lat, lon, days=14, key=None, timeout=20): """Recent eBird observations near the point, as [{sci,com,n}]. The deepest fallback, for spots with no station in range. Needs a free eBird API key in EBIRD_API_KEY and returns [] without one, so keyless installs just skip it.""" key = key or os.environ.get("EBIRD_API_KEY") if not key: return [] url = f"{EBIRD}?lat={lat:.4f}&lng={lon:.4f}&dist=50&back={min(days, 30)}" req = urllib.request.Request(url, headers={"X-eBirdApiToken": key, "User-Agent": "AvianVisitors-frame/1.0"}) try: with urllib.request.urlopen(req, timeout=timeout) as r: observations = json.loads(r.read(5_000_000)) except Exception: return [] tally = {} for o in observations: sci = o.get("sciName") if not sci: continue entry = tally.setdefault(sci, {"com": o.get("comName") or sci, "n": 0}) entry["n"] += o.get("howMany") or 1 out = [{"sci": sci, "com": e["com"], "n": e["n"]} for sci, e in tally.items()] out.sort(key=lambda s: -s["n"]) return out def slugify(sci): return re.sub(r"[^a-z0-9]+", "-", sci.lower()).strip("-") def drawable_slugs(apt_js=APT_JS): """Base slugs we have a cutout for, parsed once from the collage's bundled DIMS table (perched and flight entries collapse to the base slug).""" global _drawable if _drawable is None: with open(apt_js, encoding="utf-8") as f: block = re.search(r"var DIMS = (\{.*?\});", f.read(), re.S) keys = re.findall(r'"([a-z0-9-]+)"\s*:', block.group(1)) if block else [] _drawable = {re.sub(r"-2$", "", k) for k in keys} return _drawable def species_for_zip(zip_code, country="us", target=10, days=7, radii=(15, 30, 50), apt_js=APT_JS, timeout=20): """Geocode the ZIP, pull BirdWeather top species, and grow the search radius only until `target` drawable species are found, so it stays as local as the data allows. Returns the top `target` by detection count, or fewer where birds or stations are sparse. """ lat, lon = geocode(zip_code, country, timeout) have = drawable_slugs(apt_js) found = [] for miles in radii: found = [s for s in top_species(lat, lon, miles, days, 60, timeout) if slugify(s["sci"]) in have] if len(found) >= target: break if len(found) < target: # box too sparse: estimate from the nearest stations so remote ZIPs still fill tri = [s for s in triangulate(lat, lon, 3, days, timeout) if slugify(s["sci"]) in have] if len(tri) > len(found): found = tri if len(found) < target: # no station in range at all: fall back to eBird sightings, if a key is set eb = [s for s in ebird_nearby(lat, lon, days * 2, timeout=timeout) if slugify(s["sci"]) in have] if len(eb) > len(found): found = eb found.sort(key=lambda s: -(s["n"] or 0)) return found[:target] if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="Print BirdWeather top drawable species for a ZIP.") ap.add_argument("zip") ap.add_argument("--country", default="us") ap.add_argument("--target", type=int, default=10) ap.add_argument("--days", type=int, default=7) a = ap.parse_args() for s in species_for_zip(a.zip, a.country, a.target, a.days): print(f'{s["n"]:>8} {s["com"]} ({s["sci"]})')