Godstrument/workers/world_ephemeris.py
type-two 510ddaaf8c 🔧 review fixes — the five-angle audit's whole findings list
Correctness (four were my own regressions from the REVELATION wave):
· ☸ WHEELS polymeter permanently detuned after any skipped frame —
  seqAbsStep advanced by ++ per transition, but one frame can cross two
  16ths (bpm ≳155, or the clamped dt on every tab refocus). Proven to
  drift 12 steps out at 174bpm; now advances by the true distance, so
  seqAbsStep % 16 === seqCurStep always holds.
· Sample & hold with exactly one lit step froze on its first sample
  forever (guard keyed on the wrapped step, which never changed) — now
  keyed on the absolute step.
· PSALM double-activation left the MICROPHONE HOT after "stop": the
  re-entrancy guard sat after the getUserMedia await, orphaning the
  first stream/interval/context. Guard is now synchronous, a stop
  during the prompt is honoured, and a second tap while starting
  cancels. Mic off means off — the promise the feature makes.
· Live-recording a drum tap into a short wheel wrote steps past wlen
  that never played; the tap now folds onto the lane's own wheel.
· Switching MIDI out port left notes droning on the old port (pending
  off-timers fired at the NEW one) — midiFlushOut() releases every held
  note + all-notes-off before the swap.
· world_quakes' seen-set grew unbounded on long-running rigs.

Security: stored XSS via the commune room name — the one untrusted
string reaching innerHTML unescaped (hub only lower-cased it). Escaped
client-side at both sinks, and the hub now strips markup chars too.

Robustness: world_ephemeris/world_almanac imported swisseph unguarded,
so a machine without pyswisseph crash-looped them every 10s forever
against the supervisor's exit-0 contract — both now retire quietly
(verified by simulating the missing dep), and pyswisseph is documented
in requirements.

Mobile: the top bar's five separately-pinned chips collided below
~500px — the first thing a phone visitor saw was a broken header. Now
a clean row with the account chip on its own line and the source
counter moved left.

Housekeeping: dead measureSpaced(), dead #account .code selector, and
an 888K stale worktree removed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-26 19:43:20 +10:00

114 lines
4.5 KiB
Python

"""
world_ephemeris.py — as above, so below. The sky itself, as control signals.
Computes real planetary positions with the Swiss Ephemeris (Moshier mode — NO
data files, no downloads, good from 3000 BC to 3000 AD). Nothing is stored or
fetched; the whole sky is generated on demand, like world_sky.py does for the
sun. Astrology data is ideal for an instrument because it's inherently cyclic at
wildly different rates — the Moon wheels monthly, the Sun yearly, Saturn every 29
years — so you get a bank of natural LFOs spanning seconds to a lifetime.
Emits (all normalized-friendly):
/gs/astro/moon moon phase, 0 (new) .. 1 (full)
/gs/astro/moon_lon Moon ecliptic longitude 0..360 (fast wheel ~13deg/day)
/gs/astro/sun_lon Sun longitude 0..360 (the seasons)
/gs/astro/<planet> ecliptic longitude 0..360 for mercury..saturn (slow LFOs)
/gs/astro/tension closeness to HARD aspects (square/opposition) — dissonance
/gs/astro/harmony closeness to SOFT aspects (trine/sextile) — consonance
/gs/astro/retro fraction of planets retrograde 0..1
/gs/astro/mercury_retro 1 if Mercury is retrograde else 0
"""
from __future__ import annotations
import argparse
import datetime
import itertools
import math
import time
try: # house rule: import-safe. Without pyswisseph this
import swisseph as swe # worker retires quietly (exit 0) instead of crash-
except Exception: # looping every 10s under run.py's supervisor.
import sys
print("world_ephemeris: pyswisseph not installed — skipping (optional)")
sys.exit(0)
from pythonosc.udp_client import SimpleUDPClient
FLG = swe.FLG_MOSEPH | swe.FLG_SPEED
BODIES = {
"sun": swe.SUN, "moon": swe.MOON, "mercury": swe.MERCURY, "venus": swe.VENUS,
"mars": swe.MARS, "jupiter": swe.JUPITER, "saturn": swe.SATURN,
"uranus": swe.URANUS, "neptune": swe.NEPTUNE, "pluto": swe.PLUTO,
}
# which longitudes are worth emitting as sources (the rest still feed aspects)
EMIT_LON = ["moon", "sun", "mercury", "venus", "mars", "jupiter", "saturn"]
HARD = [(0, 8), (180, 8), (90, 7)] # conjunction/opposition/square + orbs
SOFT = [(120, 7), (60, 5)] # trine/sextile
def sky(jd):
lon, speed = {}, {}
for name, body in BODIES.items():
xx, _ = swe.calc_ut(jd, body, FLG)
lon[name], speed[name] = xx[0], xx[3]
return lon, speed
def aspects(lon):
hard = soft = 0.0
for a, b in itertools.combinations(lon, 2):
d = abs(lon[a] - lon[b]) % 360
d = min(d, 360 - d)
for ang, orb in HARD:
if abs(d - ang) < orb:
hard += 1 - abs(d - ang) / orb
for ang, orb in SOFT:
if abs(d - ang) < orb:
soft += 1 - abs(d - ang) / orb
return hard, soft
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("--interval", type=float, default=20.0)
args = ap.parse_args()
client = SimpleUDPClient(args.host, args.port)
print(f"[ephemeris] emitting /gs/astro/* (Swiss Ephemeris, no files) "
f"every {args.interval:g}s")
while True:
try:
now = datetime.datetime.now(datetime.timezone.utc)
jd = swe.julday(now.year, now.month, now.day,
now.hour + now.minute / 60 + now.second / 3600)
lon, speed = sky(jd)
phase_angle = (lon["moon"] - lon["sun"]) % 360
illum = (1 - math.cos(math.radians(phase_angle))) / 2
client.send_message("/gs/astro/moon", float(illum))
for name in EMIT_LON:
addr = "moon_lon" if name == "moon" else \
"sun_lon" if name == "sun" else name
client.send_message(f"/gs/astro/{addr}", float(lon[name]))
hard, soft = aspects(lon)
client.send_message("/gs/astro/tension", float(hard))
client.send_message("/gs/astro/harmony", float(soft))
retro = [n for n, s in speed.items() if s < 0 and n != "moon"]
client.send_message("/gs/astro/retro", len(retro) / len(BODIES))
client.send_message("/gs/astro/mercury_retro",
1.0 if speed["mercury"] < 0 else 0.0)
except Exception as e: # noqa
print(f"[ephemeris] warn: {e}")
time.sleep(args.interval)
if __name__ == "__main__":
main()