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>
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
"""
|
|
replay_quakes.py — a month of the Earth, scratchable.
|
|
|
|
Fetches every earthquake above a magnitude over the past N days from USGS, then
|
|
replays them compressed onto a loop (default: 26 days -> 5 minutes) so the
|
|
planet's seismicity becomes a repeating rhythm. Listens on a jog port so the hub
|
|
can push the playhead forward/back like a CDJ platter.
|
|
|
|
Drop-in replacement for world_quakes.py — emits the same /gs/quake/* addresses.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
|
|
from pythonosc.udp_client import SimpleUDPClient
|
|
from pythonosc.dispatcher import Dispatcher
|
|
from pythonosc.osc_server import ThreadingOSCUDPServer
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from timewarp import ReplayBuffer
|
|
|
|
UA = "Godstrument/1.0 (live instrument)"
|
|
|
|
|
|
class Jog:
|
|
"""Live playhead control, fed by the hub over OSC."""
|
|
def __init__(self, rate_min=0.25, rate_max=4.0):
|
|
self.rate_min, self.rate_max = rate_min, rate_max
|
|
self.rate_mult = 1.0
|
|
self.nudge = 0.0
|
|
|
|
def on_rate(self, addr, *a):
|
|
if a:
|
|
v = max(0.0, min(1.0, float(a[0])))
|
|
self.rate_mult = self.rate_min + (self.rate_max - self.rate_min) * v
|
|
|
|
def on_nudge(self, addr, *a):
|
|
if a:
|
|
self.nudge = (max(0.0, min(1.0, float(a[0]))) - 0.5) * 2.0
|
|
|
|
|
|
def fetch(days: int, minmag: float) -> list[dict]:
|
|
start = (datetime.datetime.now(datetime.timezone.utc)
|
|
- datetime.timedelta(days=days)).strftime("%Y-%m-%d")
|
|
url = ("https://earthquake.usgs.gov/fdsnws/event/1/query?format=geojson"
|
|
f"&starttime={start}&minmagnitude={minmag}&orderby=time&limit=800")
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
data = json.load(r)
|
|
out = []
|
|
for f in data.get("features", []):
|
|
p, g = f["properties"], f["geometry"]
|
|
if p.get("mag") is None or not g:
|
|
continue
|
|
c = g["coordinates"]
|
|
out.append({"t": p["time"] / 1000.0, "mag": float(p["mag"]),
|
|
"depth": float(c[2]), "lat": float(c[1]), "lon": float(c[0]),
|
|
"place": p.get("place", "")})
|
|
return out
|
|
|
|
|
|
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("--jog-port", type=int, default=9101)
|
|
ap.add_argument("--days", type=int, default=26)
|
|
ap.add_argument("--into", type=float, default=300.0, help="loop length (s)")
|
|
ap.add_argument("--minmag", type=float, default=4.5)
|
|
args = ap.parse_args()
|
|
|
|
client = SimpleUDPClient(args.host, args.port)
|
|
buf = ReplayBuffer(mode="event", into_seconds=args.into)
|
|
jog = Jog()
|
|
|
|
while not buf.loaded:
|
|
try:
|
|
events = fetch(args.days, args.minmag)
|
|
buf.load_events(events)
|
|
print(f"[replay_quakes] {len(events)} quakes over {args.days}d "
|
|
f"-> looping every {args.into:g}s (jog udp/{args.jog_port})")
|
|
except Exception as e: # noqa
|
|
print(f"[replay_quakes] fetch failed ({e}); retrying in 10s")
|
|
time.sleep(10)
|
|
|
|
disp = Dispatcher()
|
|
disp.map("/jog/rate", jog.on_rate)
|
|
disp.map("/jog/nudge", jog.on_nudge)
|
|
srv = ThreadingOSCUDPServer(("127.0.0.1", args.jog_port), disp)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
|
|
period = 1.0 / 60.0
|
|
while True:
|
|
crossed = buf.tick(period, jog.rate_mult, jog.nudge)
|
|
for e in crossed or []:
|
|
client.send_message("/gs/quake/event", e["mag"])
|
|
client.send_message("/gs/quake/depth", e["depth"])
|
|
client.send_message("/gs/quake/lat", e["lat"])
|
|
client.send_message("/gs/quake/lon", e["lon"])
|
|
time.sleep(period)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|