monsterrobotparty
7f4b4bcabe
- world_zodiac.py: Chinese zodiac year/element + 2-hour animal, N/S summer (computed) - world_fire.py: live wildfires from NASA EONET (the fire element) - overlapping groups: a source can now belong to many concept groups at once (elements fire/earth/air/water/spirit, plus wealth + summer vibes) — they stack - transform.py: member_groups (1:many) replaces 1:1 grouping Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""
|
|
world_fire.py — the world on fire. Live wildfires.
|
|
|
|
Pulls open wildfire events worldwide from NASA's EONET (Earth Observatory
|
|
Natural Event Tracker — free, no key). The count is a slow burn; each newly
|
|
detected fire is a flare-up you can trigger on. The elemental fire of the set —
|
|
bushfires, forest fires, the literal heat of the planet.
|
|
|
|
Emits:
|
|
/gs/fire/count number of open wildfire events worldwide
|
|
/gs/fire/event fired when a new wildfire is detected (magnitude 1)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import argparse
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
|
|
from pythonosc.udp_client import SimpleUDPClient
|
|
|
|
URL = ("https://eonet.gsfc.nasa.gov/api/v3/events"
|
|
"?category=wildfires&status=open&limit=500")
|
|
UA = "Godstrument/1.0 (live instrument)"
|
|
|
|
|
|
def fetch():
|
|
req = urllib.request.Request(URL, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.load(r).get("events", [])
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--host", default="127.0.0.1")
|
|
ap.add_argument("--port", type=int, default=9000)
|
|
ap.add_argument("--poll", type=float, default=600.0)
|
|
args = ap.parse_args()
|
|
|
|
client = SimpleUDPClient(args.host, args.port)
|
|
print(f"[fire] emitting /gs/fire/count|event (NASA EONET wildfires) "
|
|
f"(re-query every {args.poll:g}s)")
|
|
|
|
seen: set[str] = set()
|
|
count = 0
|
|
first = True
|
|
last_poll = 0.0
|
|
while True:
|
|
try:
|
|
if time.time() - last_poll > args.poll or first:
|
|
events = fetch()
|
|
count = len(events)
|
|
for e in events:
|
|
eid = e.get("id")
|
|
if eid and eid not in seen:
|
|
seen.add(eid)
|
|
if not first: # no burst on the first load
|
|
client.send_message("/gs/fire/event", 1.0)
|
|
print(f" 🔥 {e.get('title', 'wildfire')}")
|
|
print(f" {count} wildfires burning worldwide")
|
|
first = False
|
|
last_poll = time.time()
|
|
client.send_message("/gs/fire/count", float(count))
|
|
except Exception as e: # noqa
|
|
print(f"[fire] warn: {e}")
|
|
time.sleep(10.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|