monsterrobotparty
8deb65b4fe
Everything — your hand, the room, a satellite, the markets, the sky — becomes an OSC control signal fused through a modulation matrix. Includes: - hub + normalize (One-Euro) + matrix (curves/gates/quantize) + transform (tweaks/groups) - 20+ workers: sim sensors, 8 keyless world feeds, ephemeris/almanac/clock (computed), time-warp replay (quakes/weather/db), and the dealgod market warehouse feed - planetary orbital LFOs, SQLite recorder, city targeting - live browser console: mute, group macros, drag-to-patch, inspector, Web MIDI Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Live global seismic activity from the USGS earthquake feed.
|
|
Every real earthquake on Earth in the last hour fires a percussive "event" whose
|
|
magnitude, depth, and location become musical: the planet playing itself as a drum.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
from pythonosc.udp_client import SimpleUDPClient
|
|
|
|
NAME = "world_quakes"
|
|
FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson"
|
|
POLL_SECONDS = 60
|
|
USER_AGENT = "Godstrument/1.0 (live instrument; contact monsterrobotparty@gmail.com)"
|
|
|
|
|
|
def fetch_feed():
|
|
"""Fetch and parse the USGS GeoJSON feed. Returns the parsed dict or None on failure."""
|
|
req = urllib.request.Request(FEED_URL, headers={"User-Agent": USER_AGENT})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Emit live USGS earthquakes as OSC.")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=9000)
|
|
args = parser.parse_args()
|
|
|
|
client = SimpleUDPClient(args.host, args.port)
|
|
print(f"[{NAME}] emitting /gs/quake/... every {POLL_SECONDS}s")
|
|
|
|
seen = set()
|
|
first_poll = True
|
|
backoff = 5
|
|
|
|
while True:
|
|
try:
|
|
data = fetch_feed()
|
|
backoff = 5 # reset backoff after a good fetch
|
|
except (urllib.error.URLError, urllib.error.HTTPError, OSError, ValueError) as e:
|
|
print(f"[{NAME}] warning: fetch failed ({e}); retrying in {backoff}s")
|
|
time.sleep(backoff)
|
|
backoff = min(backoff * 2, POLL_SECONDS)
|
|
continue
|
|
|
|
features = data.get("features") or []
|
|
|
|
if first_poll:
|
|
# Record existing quakes but do NOT fire them (no startup burst).
|
|
for feat in features:
|
|
fid = feat.get("id")
|
|
if fid is not None:
|
|
seen.add(fid)
|
|
first_poll = False
|
|
else:
|
|
for feat in features:
|
|
fid = feat.get("id")
|
|
if fid is None or fid in seen:
|
|
continue
|
|
seen.add(fid)
|
|
|
|
props = feat.get("properties") or {}
|
|
mag = props.get("mag")
|
|
if mag is None:
|
|
continue # skip quakes with no magnitude
|
|
|
|
place = props.get("place") or "unknown location"
|
|
geom = feat.get("geometry") or {}
|
|
coords = geom.get("coordinates") or [None, None, None]
|
|
lon = coords[0] if len(coords) > 0 else None
|
|
lat = coords[1] if len(coords) > 1 else None
|
|
depth = coords[2] if len(coords) > 2 else None
|
|
|
|
try:
|
|
client.send_message("/gs/quake/event", float(mag))
|
|
if depth is not None:
|
|
client.send_message("/gs/quake/depth", float(depth))
|
|
if lat is not None:
|
|
client.send_message("/gs/quake/lat", float(lat))
|
|
if lon is not None:
|
|
client.send_message("/gs/quake/lon", float(lon))
|
|
except (OSError, ValueError, TypeError) as e:
|
|
print(f"[{NAME}] warning: OSC send failed ({e})")
|
|
continue
|
|
|
|
depth_str = f"{depth:.0f}km" if depth is not None else "?km"
|
|
print(f" quake M{float(mag):.1f} {place} (depth {depth_str})")
|
|
|
|
time.sleep(POLL_SECONDS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|