Godstrument/workers/midi_worker.py
monsterrobotparty 8deb65b4fe Godstrument: a multi-modal sensor-fusion instrument
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>
2026-07-05 02:13:05 +10:00

83 lines
3.1 KiB
Python

#!/usr/bin/env python3
"""MIDI worker: your Ableton pad controller / keyboard — the tactile core of the rig.
Human touch (knobs, pads, keys) is the most expressive input on stage; CCs become
smooth continuous modulators and note velocity fires as percussive event magnitude.
"""
import argparse
import sys
def main():
parser = argparse.ArgumentParser(description="Forward MIDI input to Godstrument OSC hub.")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=9000)
parser.add_argument("--portname", default=None,
help="Substring to match a specific MIDI input port.")
args = parser.parse_args()
# Lazy import — this worker is optional. Never traceback if mido is missing.
try:
import mido
except Exception:
print("midi worker: mido/no MIDI input — skipping (optional)")
sys.exit(0)
# Find an input port.
try:
inputs = mido.get_input_names()
except Exception:
print("midi worker: mido/no MIDI input — skipping (optional)")
sys.exit(0)
if not inputs:
print("midi worker: mido/no MIDI input — skipping (optional)")
sys.exit(0)
portname = inputs[0]
if args.portname:
matches = [p for p in inputs if args.portname.lower() in p.lower()]
if matches:
portname = matches[0]
else:
print("midi worker: mido/no MIDI input — skipping (optional)")
sys.exit(0)
from pythonosc.udp_client import SimpleUDPClient
client = SimpleUDPClient(args.host, args.port)
print("[midi] emitting /gs/midi/... every 0s (event-driven)")
# Loop forever over the input port. Reopen on transient failure.
while True:
try:
with mido.open_input(portname) as inport:
for msg in inport:
try:
if msg.type == "control_change":
addr = "/gs/midi/cc/{}".format(msg.control)
client.send_message(addr, float(msg.value) / 127.0)
elif msg.type == "note_on" and msg.velocity > 0:
client.send_message("/gs/midi/note/event",
float(msg.velocity))
except Exception as e:
print("midi worker: send warning: {}".format(e))
except Exception as e:
# Device unplugged / transient error — warn, back off, reconnect forever.
print("midi worker: input warning: {} — reconnecting".format(e))
try:
import time
time.sleep(2.0)
fresh = mido.get_input_names()
if fresh:
if args.portname:
m = [p for p in fresh if args.portname.lower() in p.lower()]
portname = m[0] if m else fresh[0]
else:
portname = fresh[0]
except Exception:
pass
if __name__ == "__main__":
main()