""" hub.py — the traffic cop and the brain. Everything (your hand, the room, a satellite, bitcoin) fires OSC at this hub on 127.0.0.1:9000. It doesn't care where a signal came from — a fingertip and a plane over Tokyo are the same kind of thing here. The hub: 1. receives every OSC message into a thread-safe raw state, 2. on a fixed control-rate clock, normalizes + One-Euro-filters each signal to a clean 0..1 (events become decaying impulses), 3. runs the modulation matrix (matrix.py) to turn signals into destinations, 4. emits results as OSC /out/ (for Ableton / TouchDesigner), optional MIDI CC/notes, and a live WebSocket feed for the browser visualizer. Run: python hub.py (uses config.json) python hub.py --config myset.json Ports: OSC in 9000 | OSC out 9001 | WebSocket 8765 | HTTP viz 8088 """ from __future__ import annotations import argparse import asyncio import json import math import os import queue import threading import time from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer import websockets from pythonosc.dispatcher import Dispatcher from pythonosc.osc_server import ThreadingOSCUDPServer from pythonosc.udp_client import SimpleUDPClient from normalize import AdaptiveNormalizer, OneEuroFilter, ImpulseEnvelope from matrix import ModMatrix from transform import TweakBank HERE = os.path.dirname(os.path.abspath(__file__)) START = time.monotonic() # sidereal orbital periods in Earth-years — a group assigned a planet cycles at # this rate relative to `orbit_base_seconds` (one Earth orbit = base seconds). PLANET_PERIODS = { "mercury": 0.2408, "venus": 0.6152, "earth": 1.0, "mars": 1.8808, "jupiter": 11.862, "saturn": 29.457, "uranus": 84.02, "neptune": 164.8, "pluto": 248.0, } def now() -> float: return time.monotonic() - START # --------------------------------------------------------------------------- # Per-signal processing chain # --------------------------------------------------------------------------- class Signal: def __init__(self, name: str, spec: dict): self.name = name self.type = spec.get("type", "continuous") self.label = spec.get("label", name) nrm = spec.get("norm", {}) mode = nrm.get("mode", "minmax") fixed = None if "lo" in nrm and "hi" in nrm: fixed = (nrm["lo"], nrm["hi"]) mode = "fixed" elif not nrm and (name.startswith(("midi.", "hand.", "tof.", "light.")) or self.type == "event"): # Control inputs (knobs, hand, sensors) and bare events already # speak 0..1 — adaptively rescaling a constant stream would pin it # to 0.5 (a held knob) or halve every event. Pass them through. # ponytail: prefix allowlist; give any exception an explicit norm. fixed = (0.0, 1.0) mode = "fixed" self.norm = AdaptiveNormalizer(nrm.get("window", 600), mode, fixed) flt = spec.get("filter", {}) self.filter = OneEuroFilter(flt.get("min_cutoff", 1.2), flt.get("beta", 0.02)) self.env = ImpulseEnvelope(spec.get("attack", 0.01), spec.get("decay", 0.6)) self.raw = 0.0 self.value = 0.0 def update_continuous(self, raw: float, t: float) -> float: self.raw = raw n = self.norm(raw) self.value = self.filter(n, t) return self.value def trigger(self, magnitude: float, t: float): self.raw = magnitude self.env.trigger(self.norm(magnitude), t) def event_value(self, t: float) -> float: self.value = self.env.value(t) return self.value # --------------------------------------------------------------------------- # The hub # --------------------------------------------------------------------------- class Hub: def __init__(self, cfg: dict): self.cfg = cfg self.specs = cfg.get("signals", {}) self.matrix = ModMatrix.from_config(cfg) self.tweaks = TweakBank(cfg) self.rate = cfg.get("control_rate_hz", 60) # read-only mode: broadcast to viewers but ignore every incoming command # (the ws control channel has no auth — safe to expose publicly this way). # Set via config "readonly": true or env GODSTRUMENT_READONLY=1. self.readonly = bool(cfg.get("readonly")) or \ os.environ.get("GODSTRUMENT_READONLY", "").strip().lower() in ("1", "true", "yes", "on") self.signals: dict[str, Signal] = {} self._raw_lock = threading.Lock() self._raw_inbox: dict[str, tuple[float, float]] = {} # key -> (val,t) self._last_raw: dict[str, float] = {} # last value fed to the normalizer self._event_q: "queue.Queue[tuple[str,float,float]]" = queue.Queue() self.out = SimpleUDPClient(cfg.get("osc_out_host", "127.0.0.1"), cfg.get("osc_out_port", 9001)) self.warp = cfg.get("warp", {}) self._jog_clients: dict[int, SimpleUDPClient] = {} self.orbit_base = cfg.get("orbit_base_seconds", 60) self.patches_dir = os.path.join(HERE, "patches") self._patches_dirty = False self._scan_patches() self.midi = _open_midi(cfg.get("midi", {})) self.clients: set = set() self.cnames: dict = {} # ws -> identity (session username, or claimed on local hubs) self.communes: dict = {} # room -> {host, members:set, invited:set, alloc:{section:user}} self.recent_events: list[dict] = [] self._last_notes: dict[str, int] = {} self.route_meta = [ {"source": r.get("source"), "dest": r.get("dest"), "label": r.get("label", "")} for r in cfg.get("routes", []) ] # ---- OSC ingest (runs in the OSC server thread) -------------------- def _osc_handler(self, address: str, *args): if not args: return try: val = float(args[0]) except (TypeError, ValueError): return # /gs/a/b -> "a.b" ; /gs/quake/event -> event on "quake.event" parts = address.strip("/").split("/") if parts and parts[0] == "gs": parts = parts[1:] key = ".".join(parts) t = now() if key.endswith(".event"): self._event_q.put((key, val, t)) else: with self._raw_lock: self._raw_inbox[key] = (val, t) def _ensure_signal(self, key: str) -> Signal: sig = self.signals.get(key) if sig is None: spec = self.specs.get(key, {}) if key.endswith(".event") and "type" not in spec: spec = {**spec, "type": "event"} sig = Signal(key, spec) self.signals[key] = sig return sig # ---- control-rate loop (runs in asyncio) -------------------------- async def control_loop(self): period = 1.0 / self.rate while True: t = now() # drain events -> triggers while True: try: key, mag, et = self._event_q.get_nowait() except queue.Empty: break sig = self._ensure_signal(key) sig.trigger(mag, et) ev = {"src": key.rsplit(".", 1)[0], "mag": round(mag, 3), "t": et} self.recent_events.append(ev) self.recent_events = self.recent_events[-40:] # snapshot raw inbox with self._raw_lock: inbox = dict(self._raw_inbox) sources: dict[str, float] = {} wire_sources: dict[str, dict] = {} for key, (raw, rt) in inbox.items(): sig = self._ensure_signal(key) if self._last_raw.get(key) != raw: # only feed the normalizer/filter when the value actually # changed. Workers re-emit their last value on a cadence with # a fresh timestamp each time, so gating on arrival time would # still pad an adaptive window with duplicates; gating on the # value means the window holds N distinct samples (not N ticks # of polling). ponytail: windows now forget by sample, not # wall-clock — the right call for this instrument. v = sig.update_continuous(raw, t) self._last_raw[key] = raw else: v = sig.value # unchanged datum; hold the last value sources[key] = v wire_sources[key] = {"raw": round(raw, 4), "norm": round(v, 4), "label": sig.label, "muted": self.tweaks.is_muted(key)} # event envelopes decay every tick for key, sig in self.signals.items(): if sig.type == "event": v = sig.event_value(t) sources[key] = v wire_sources[key] = {"raw": round(sig.raw, 4), "norm": round(v, 4), "label": sig.label, "event": True, "muted": self.tweaks.is_muted(key)} # planetary orbits — each orbit-group's gain waxes/wanes at its # planet's real relative rate (Mercury whips, Neptune barely drifts) for g, planet in self.tweaks.group_orbit.items(): orb = PLANET_PERIODS.get(planet, 1.0) * self.orbit_base phase = (t / orb) % 1.0 self.tweaks.set_orbit_lfo( g, 0.25 + 0.75 * (0.5 + 0.5 * math.sin(2 * math.pi * phase))) # scrub any time-warped replay workers from live controls self._forward_jog(sources) # performance layer: live controls -> tweaks -> groups -> matrix self.tweaks.apply_controls(sources) tweaked = self.tweaks.transform(sources) dests, active = self.matrix.evaluate(tweaked) self._emit(dests) await self._broadcast(t, wire_sources, dests, active, self.tweaks.macros()) await asyncio.sleep(period) # ---- time-warp jog forwarding ------------------------------------- def _forward_jog(self, sources: dict[str, float]): for w in self.warp.values(): port = w.get("port") if not port: continue cl = self._jog_clients.get(port) if cl is None: cl = SimpleUDPClient("127.0.0.1", port) self._jog_clients[port] = cl rs, ns = w.get("rate_src"), w.get("nudge_src") if rs and rs in sources: cl.send_message("/jog/rate", float(sources[rs])) if ns and ns in sources: cl.send_message("/jog/nudge", float(sources[ns])) # ---- outputs ------------------------------------------------------- def _emit(self, dests: dict[str, float]): midi_map = self.cfg.get("midi", {}).get("map", {}) notes_live = set() # note dests sounding this frame for dest, val in dests.items(): self.out.send_message(f"/out/{dest}", float(val)) m = midi_map.get(dest) if not (m and self.midi): continue if m.get("note"): # note-mapped: val IS a note number note = int(round(val)) if not 24 <= note <= 127: # matches the viz Web-MIDI guard continue notes_live.add(dest) if self._last_notes.get(dest) != note: if self._last_notes.get(dest) is not None: self.midi.send_message([0x80 | m.get("channel", 0), self._last_notes[dest], 0]) self.midi.send_message([0x90 | m.get("channel", 0), note, 100]) self._last_notes[dest] = note elif "cc" in m: cc = int(_clamp01(val) * 127) self.midi.send_message([0xB0 | m.get("channel", 0), m["cc"], cc]) # release any held note whose source went silent (muted/gated) this frame if self.midi: for dest, note in list(self._last_notes.items()): if note is not None and dest not in notes_live: ch = midi_map.get(dest, {}).get("channel", 0) self.midi.send_message([0x80 | ch, note, 0]) self._last_notes[dest] = None async def _broadcast(self, t, sources, dests, active, macros=None): if not self.clients: return routes = [] for i, r in enumerate(self.matrix.routes): routes.append({"source": r.get("source"), "dest": r.get("dest"), "label": r.get("label", ""), "amount": round(r.get("amount", 0), 3), "active": round(active.get(i, 0.0), 4)}) p = { "t": round(t, 3), "sources": sources, "dests": {k: (round(v, 4) if isinstance(v, float) else v) for k, v in dests.items()}, "routes": routes, "macros": macros or {}, "events": self.recent_events[-12:], } if self._patches_dirty: p["patches"] = self.patches self._patches_dirty = False payload = json.dumps(p) websockets.broadcast(self.clients, payload) async def ws_handler(self, ws): self.clients.add(ws) # who is this? the session cookie rides the websocket handshake, so a # signed-in user's identity is known here and cannot be spoofed. try: hdrs = getattr(ws, "request_headers", None) if hdrs is None and getattr(ws, "request", None) is not None: hdrs = ws.request.headers cookie = hdrs.get("Cookie", "") if hdrs else "" if cookie: import auth as _auth uid = _auth._uid_from_cookies(cookie, _auth.DEFAULT_DB) info = _auth.user_info(uid, _auth.DEFAULT_DB) if uid else None if info and info.get("username"): self.cnames[ws] = info["username"] except Exception: # noqa pass try: await ws.send(json.dumps({ "hello": "godstrument", "readonly": self.readonly, # full route spec so a logged-in client can compute its own mix "routes": [{"source": r.get("source"), "dest": r.get("dest"), "label": r.get("label", ""), "amount": round(r.get("amount", 0), 3), "curve": r.get("curve", "lin"), "gate": r.get("gate"), "quantize": r.get("quantize")} for r in self.matrix.routes], "groups": self.cfg.get("groups", {}), "midi": self.cfg.get("midi", {}).get("map", {}), "patches": self.patches})) async for msg in ws: # browser -> hub commands try: m = json.loads(msg) c = m.get("commune") if isinstance(c, dict): self._commune(ws, c) continue self._handle_cmd(m) except Exception: # noqa pass finally: self.clients.discard(ws) self.cnames.pop(ws, None) # ---- the commune: hosted rooms, invitations, host-allocated sections ---- # Identity comes from the session cookie (public hub: signed-in users only); # membership is by invitation with accept/decline; the host allocates # sections; voice-state relays only from whoever actually holds the section. def _commune(self, ws, c): user = self.cnames.get(ws) if self.readonly and not user: websockets.broadcast([ws], json.dumps( {"commune": {"error": "the commune is for signed-in players — log in first"}})) return if not user: # local/trusted hub: take the claimed name user = str(c.get("from") or "guest")[:24] self.cnames[ws] = user room = str(c.get("room") or "")[:40].strip().lower() R = self.communes def out(names, obj): targets = [w for w in self.clients if self.cnames.get(w) in names] if targets: websockets.broadcast(targets, json.dumps({"commune": obj})) def roominfo(r): g2 = R.get(r) if g2: out(g2["members"], {"roominfo": {"room": r, "host": g2["host"], "members": sorted(g2["members"]), "alloc": g2["alloc"]}}) if c.get("hello"): # presence tag only return if c.get("create") and room: g = R.get(room) if g and g["members"] and user not in g["members"]: out([user], {"error": "“" + room + "” is already alive — ask its host for an invitation"}) return R[room] = {"host": user, "members": {user}, "invited": set(), "alloc": {}} roominfo(room) return g = R.get(room) if not g: out([user], {"error": "no such room — its host may have closed it"}) return if c.get("invite") and user == g["host"]: tgt = str(c["invite"]).strip()[:24] if not any(self.cnames.get(w) == tgt for w in self.clients): out([user], {"error": tgt + " isn't on the site right now — invitations reach the present"}) return g["invited"].add(tgt) out([tgt], {"invite": {"room": room, "host": user}}) out([user], {"ok": "invitation sent to " + tgt}) return if c.get("accept"): if user in g["invited"] or user in g["members"]: g["invited"].discard(user) g["members"].add(user) roominfo(room) else: out([user], {"error": "no standing invitation to that room"}) return if c.get("decline"): g["invited"].discard(user) out([g["host"]], {"declined": user, "room": room}) return if user not in g["members"]: out([user], {"error": "you are not in that room"}) return if c.get("leave"): g["members"].discard(user) for s in [s for s, o in list(g["alloc"].items()) if o == user]: del g["alloc"][s] if user == g["host"] or not g["members"]: out(g["members"] | {user}, {"disbanded": room}) R.pop(room, None) else: roominfo(room) return if c.get("allocate") and user == g["host"]: a = c["allocate"] if isinstance(c["allocate"], dict) else {} s, to = str(a.get("section", ""))[:40], str(a.get("to", ""))[:24] if s and to in g["members"]: g["alloc"][s] = to roominfo(room) return if c.get("claim"): s = str(c["claim"])[:40] held = g["alloc"].get(s) if held and held != user and user != g["host"]: out([user], {"error": "that section is held by " + held + " — ask the host"}) else: g["alloc"][s] = user roominfo(room) return if c.get("release"): s = str(c["release"])[:40] if g["alloc"].get(s) == user: del g["alloc"][s] roominfo(room) return if c.get("voice") and c.get("state") is not None: s = str(c["voice"])[:40] if g["alloc"].get(s) == user: # only the section's holder speaks for it out(g["members"] - {user}, {"voice": s, "state": c["state"], "from": user, "room": room}) return # ---- named templates (patches) ------------------------------------ def _scan_patches(self): try: self.patches = sorted(f[:-5] for f in os.listdir(self.patches_dir) if f.endswith(".json")) except OSError: self.patches = [] def _save_patch(self, name): if not name: return safe = _safe_name(name) or "patch" os.makedirs(self.patches_dir, exist_ok=True) data = {"routes": self.matrix.routes, "groups": self.tweaks.dump_groups(), "state": self.tweaks.dump_state()} with open(os.path.join(self.patches_dir, safe + ".json"), "w") as f: json.dump(data, f, indent=2) self._scan_patches() self._patches_dirty = True print(f" saved template '{safe}'") def _load_patch(self, name): safe = _safe_name(name) # no path traversal / absolute paths if not safe: return try: with open(os.path.join(self.patches_dir, safe + ".json")) as f: data = json.load(f) except (OSError, ValueError): return if "routes" in data: self.matrix.routes = [dict(r) for r in data["routes"]] if "groups" in data: self.tweaks.load_groups(data["groups"]) if "state" in data: self.tweaks.load_state(data["state"]) print(f" loaded spell '{safe}'") def _handle_cmd(self, m: dict): if self.readonly: # public spectator: ignore all control commands return cmd = m.get("cmd") if cmd == "mute_toggle": self.tweaks.toggle(m.get("target"), "mute") elif cmd == "toggle": self.tweaks.toggle(m.get("target"), m.get("param", "mute")) elif cmd == "set": self.tweaks.set_param(m.get("target"), m.get("param", "gain"), _ffloat(m.get("value"), 0.0)) elif cmd == "add_route": self.matrix.add_route(m.get("source"), m.get("dest"), _ffloat(m.get("amount"), 0.5)) elif cmd == "remove_route": self.matrix.remove_route(m.get("source"), m.get("dest")) elif cmd == "set_route": self.matrix.set_route_amount(m.get("source"), m.get("dest"), _ffloat(m.get("amount"), 0.5)) elif cmd == "make_group": self.tweaks.add_group(m.get("name"), m.get("members", [])) elif cmd == "save_patch": self._save_patch(m.get("name")) elif cmd == "load_patch": self._load_patch(m.get("name")) # ---- run ----------------------------------------------------------- async def run(self): # OSC server thread disp = Dispatcher() disp.set_default_handler(self._osc_handler) osc_srv = ThreadingOSCUDPServer( (self.cfg.get("osc_in_host", "127.0.0.1"), self.cfg.get("osc_in_port", 9000)), disp) threading.Thread(target=osc_srv.serve_forever, daemon=True).start() # HTTP static server for the visualizer _serve_http(os.path.join(HERE, "viz"), self.cfg.get("http_port", 8080)) ws_port = self.cfg.get("ws_port", 8765) print("╔══════════════════════════════════════════════╗") print("║ GODSTRUMENT hub online ") print(f"║ OSC in udp/{self.cfg.get('osc_in_port',9000)} " f"OSC out udp/{self.cfg.get('osc_out_port',9001)}") print(f"║ WebSocket ws/{ws_port}") print(f"║ Visualizer -> http://localhost:" f"{self.cfg.get('http_port',8080)}") print(f"║ MIDI out: {'connected' if self.midi else 'off (optional)'}") print(f"║ {len(self.route_meta)} routes patched, " f"clock {self.rate}Hz") print("╚══════════════════════════════════════════════╝") async with websockets.serve(self.ws_handler, "127.0.0.1", ws_port): await self.control_loop() def _open_midi(midi_cfg: dict): if not midi_cfg.get("enabled"): return None try: import rtmidi out = rtmidi.MidiOut() ports = out.get_ports() target = midi_cfg.get("port_name", "IAC") idx = next((i for i, p in enumerate(ports) if target.lower() in p.lower()), None) if idx is None and ports: idx = 0 if idx is None: out.open_virtual_port("Godstrument") else: out.open_port(idx) print(f" MIDI -> {ports[idx]}") return out except Exception as e: # noqa print(f" MIDI unavailable ({e}); continuing without it.") return None def _serve_http(directory: str, port: int): try: import auth # invite-only accounts + presets, mounted at /api/* except Exception as e: # noqa auth = None print(f" (accounts disabled: {e})") class Handler(SimpleHTTPRequestHandler): def __init__(self, *a, **k): super().__init__(*a, directory=directory, **k) def log_message(self, *a, **k): pass def end_headers(self): # never let a browser serve a stale console after a deploy if not self.path.startswith("/api/"): self.send_header("Cache-Control", "no-cache, must-revalidate") super().end_headers() def _api(self, method): path = self.path.split("?", 1)[0] if auth is None: return self.send_error(503, "accounts unavailable") # CSRF: a browser always sends Origin on a cross-site state-changing request. # Block those (incl. login-CSRF); a same-origin request or a non-browser # client (no Origin) passes. Allowlist the canonical host + localhost so a # legitimate request always passes even if the tunnel rewrites Host. if method in ("POST", "PUT", "DELETE"): origin = self.headers.get("Origin", "") if origin: from urllib.parse import urlparse o = urlparse(origin) ok = (o.hostname in ("godstrument.pro", "localhost", "127.0.0.1") or o.netloc == self.headers.get("Host", "")) if not ok: blob = json.dumps({"error": "cross-site request blocked"}).encode() self.send_response(403) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(blob))) self.end_headers(); self.wfile.write(blob) return length = int(self.headers.get("Content-Length", 0) or 0) if length > 512 * 1024: return self.send_error(413, "too large") body = self.rfile.read(length) if length else b"" try: status, resp, set_cookie = auth.handle_api( method, path, body, self.headers.get("Cookie", "")) except Exception: # never leak a stack trace to the client status, resp, set_cookie = 500, {"error": "server error"}, None payload = json.dumps(resp).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") if set_cookie: self.send_header("Set-Cookie", set_cookie) self.end_headers() self.wfile.write(payload) def _force_https(self): # visitors on http can't keep a Secure session cookie -> send them to # https (Cloudflare sets X-Forwarded-Proto to the visitor's scheme). if self.headers.get("X-Forwarded-Proto", "").lower() == "http": host = self.headers.get("Host", "") if host: self.send_response(301) self.send_header("Location", f"https://{host}{self.path}") self.send_header("Content-Length", "0") self.end_headers() return True return False def do_GET(self): if self._force_https(): return self._api("GET") if self.path.startswith("/api/") else super().do_GET() def do_POST(self): self._api("POST") if self.path.startswith("/api/") else self.send_error(501) def do_PUT(self): self._api("PUT") if self.path.startswith("/api/") else self.send_error(501) def do_DELETE(self): self._api("DELETE") if self.path.startswith("/api/") else self.send_error(501) httpd = ThreadingHTTPServer(("127.0.0.1", port), Handler) threading.Thread(target=httpd.serve_forever, daemon=True).start() def _clamp01(x: float) -> float: return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x def _safe_name(name) -> str: """A patch name safe to use as a filename: basename only, alnum/-/_/space. Defeats path traversal ('../x') and absolute paths from the ws command.""" base = os.path.basename(str(name or "")) return "".join(c for c in base if c.isalnum() or c in "-_ ").strip() def _ffloat(x, default: float = 0.0) -> float: """float() that rejects NaN/Infinity (json.loads accepts them) so a crafted command can't poison the matrix or blind the console.""" try: v = float(x) except (TypeError, ValueError): return default return v if math.isfinite(v) else default def main(): ap = argparse.ArgumentParser() ap.add_argument("--config", default=os.path.join(HERE, "config.json")) args = ap.parse_args() with open(args.config) as f: cfg = json.load(f) hub = Hub(cfg) try: asyncio.run(hub.run()) except KeyboardInterrupt: print("\n godstrument sleeps.") if __name__ == "__main__": main()