Godstrument/local_source.py

#!/usr/bin/env python3
"""
local_source.py — the "local source" device firmware for Godstrument.

The room becomes an instrument. Run this on a Raspberry Pi (or any laptop) wired
to environmental sensors; it streams their readings to a Godstrument hub, where
each one appears as a live SOURCE you can wire to any voice, in any skin.

────────────────────────────────────────────────────────────────────────────
THE CONTRACT (this is the whole thing — any hardware that speaks it works):

    OSC over UDP  →  hub  osc_in_port (default 9000)

    address:  /gs/<group>/<name>      value:  one float
      e.g.    /gs/room/temp  21.4     →  source  "room.temp"
              /gs/hand/height 0.62    →  source  "hand.height"

    A key ending in ".event" is an IMPULSE (a hit) carrying a magnitude:
              /gs/crowd/hit  0.8      →  a triggered burst, magnitude 0.8

    Send RAW values, not pre-normalized ones. The hub auto-scales each source
    with an adaptive min/max window (add `"norm": {"mode":"minmax"}` in the
    hub's config `signals` for a feed if you want to tune it), so the SAME
    firmware calibrates itself to a dark club or a bright festival tent.

That's it. Pick any SBC, any sensors, any language — if it emits that OSC, the
planet-instrument receives it. This file is just a convenient reference sender.
────────────────────────────────────────────────────────────────────────────

RUN IT (works with NO sensors wired — everything falls back to gentle
simulated values, so you can prove the pipeline before you own the hardware):

    python3 local_source.py --host 127.0.0.1        # send to a local hub
    python3 local_source.py --host 192.168.1.42     # send to the laptop at the booth
    python3 local_source.py --rate 30 --once        # one burst, then exit (self-check)

To receive from a SEPARATE device on the LAN, run the hub with
`"osc_in_host": "0.0.0.0"` in config.json (so it listens on the network, not
just localhost), and point --host at the hub machine's IP.

WIRING (all I2C unless noted; the readers below are stubbed — uncomment the
driver lines for the sensors you actually solder on):
    BH1750    lux / light         →  room.light
    BME280    temp+humidity+press →  room.temp, room.humidity, room.pressure
    VL53L1X   time-of-flight       →  hand.height   (the theremin over the desk)
    MPU6050   6-DOF IMU            →  booth.shake (vibration), booth.tilt
    I2S/USB mic  amplitude         →  crowd.energy  + crowd.hit (transients)
"""
import argparse
import math
import time

try:
    from pythonosc.udp_client import SimpleUDPClient
except ImportError:
    raise SystemExit("pip install python-osc  (the hub already depends on it)")


# ── sensor readers ──────────────────────────────────────────────────────────
# Each reader returns a raw float, or None if unavailable. Keep them fast and
# non-blocking. When the hardware isn't present they raise/return None and we
# fall back to a smooth simulated value so the pipeline still runs and sings.

class Sensors:
    def __init__(self):
        self.i2c = None
        # ── real drivers: uncomment what you wire (needs the matching pip pkgs)
        # import board, busio
        # self.i2c = busio.I2C(board.SCL, board.SDA)
        # import adafruit_bh1750;  self.bh1750 = adafruit_bh1750.BH1750(self.i2c)
        # import adafruit_bme280.basic as bme;  self.bme = bme.Adafruit_BME280_I2C(self.i2c)
        # import adafruit_vl53l1x;  self.tof = adafruit_vl53l1x.VL53L1X(self.i2c); self.tof.start_ranging()
        # import adafruit_mpu6050;  self.imu = adafruit_mpu6050.MPU6050(self.i2c)
        # import sounddevice as sd;  self.stream = sd.InputStream(channels=1); self.stream.start()

    def light(self):        # lux → keep raw; hub auto-scales the venue's range
        try: return self.bh1750.lux
        except Exception: return None

    def temp(self):
        try: return self.bme.temperature
        except Exception: return None

    def humidity(self):
        try: return self.bme.relative_humidity
        except Exception: return None

    def pressure(self):
        try: return self.bme.pressure
        except Exception: return None

    def hand_height(self):  # ToF distance in mm; near = hand over the desk
        try: return self.tof.distance     # cm on adafruit driver
        except Exception: return None

    def shake(self):        # IMU: magnitude of acceleration minus 1g = vibration
        try:
            x, y, z = self.imu.acceleration
            return abs(math.sqrt(x * x + y * y + z * z) - 9.81)
        except Exception: return None

    def tilt(self):
        try:
            x, y, z = self.imu.acceleration
            return math.degrees(math.atan2(x, z))
        except Exception: return None

    def crowd_energy(self):  # RMS of a short mic block
        try:
            block, _ = self.stream.read(512)
            return float((block.astype("float64") ** 2).mean() ** 0.5)
        except Exception: return None


# ── the source map: (osc address, reader, simulated-fallback shape) ──────────
# The fallback lets the whole rig run with zero sensors — a living demo. `shape`
# picks a distinct simulated motion so you can tell the feeds apart on screen.
FEEDS = [
    ("/gs/room/light",     "light",        "drift"),
    ("/gs/room/temp",      "temp",         "slowrise"),
    ("/gs/room/humidity",  "humidity",     "slowrise"),
    ("/gs/room/pressure",  "pressure",     "drift"),
    ("/gs/hand/height",    "hand_height",  "flux"),
    ("/gs/booth/shake",    "shake",        "flux"),
    ("/gs/booth/tilt",     "tilt",         "drift"),
    ("/gs/crowd/energy",   "crowd_energy", "pulse"),
]


def simulate(shape, t, seed):
    """Gentle stand-in values so the device 'works' before any sensor exists."""
    p = t * 0.5 + seed
    if shape == "drift":    return 0.5 + 0.4 * math.sin(p * 0.3)
    if shape == "slowrise": return 0.3 + 0.5 * (0.5 + 0.5 * math.sin(p * 0.08))
    if shape == "flux":     return abs(math.sin(p * 1.7) * math.sin(p * 0.9))
    if shape == "pulse":    return max(0.0, math.sin(p * 2.3)) ** 3
    return 0.5


def run(host, port, rate, once):
    client = SimpleUDPClient(host, port)
    sensors = Sensors()
    period = 1.0 / max(1, rate)
    t0 = time.monotonic()
    last_energy = 0.0
    live_keys = set()
    print(f"local source → OSC {host}:{port} @ {rate} Hz  (Ctrl-C to stop)")
    while True:
        t = time.monotonic() - t0
        energy = None
        for i, (addr, reader, shape) in enumerate(FEEDS):
            val = getattr(sensors, reader)()
            simulated = val is None
            if simulated:
                val = simulate(shape, t, i * 1.37)
            client.send_message(addr, float(val))
            if not simulated:
                live_keys.add(addr)
            if reader == "crowd_energy":
                energy = val
        # transient detector → a hit event when the room's energy jumps
        if energy is not None:
            if energy - last_energy > 0.18:
                client.send_message("/gs/crowd/hit", float(min(1.0, (energy - last_energy) * 3)))
            last_energy = last_energy * 0.85 + energy * 0.15
        if once:
            print(f"sent {len(FEEDS)} feeds once.  live sensors: {sorted(live_keys) or 'none (all simulated)'}")
            return
        time.sleep(period)


def _selfcheck():
    """No hub needed: prove the OSC packets serialize & the sim values are sane."""
    for shape in ("drift", "slowrise", "flux", "pulse"):
        for tt in (0.0, 1.0, 5.0, 20.0):
            v = simulate(shape, tt, 0.0)
            assert 0.0 <= v <= 1.0, f"{shape}@{tt} out of range: {v}"
    # a real client to a dead port must still build a valid packet without error
    SimpleUDPClient("127.0.0.1", 59999).send_message("/gs/room/light", 0.5)
    print("local_source self-check: OK (sim values bounded, OSC packet builds)")


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description="stream local room sensors to a Godstrument hub over OSC")
    ap.add_argument("--host", default="127.0.0.1", help="hub IP (the machine running Godstrument)")
    ap.add_argument("--port", type=int, default=9000, help="hub osc_in_port")
    ap.add_argument("--rate", type=int, default=30, help="updates per second")
    ap.add_argument("--once", action="store_true", help="send one burst and exit")
    ap.add_argument("--selfcheck", action="store_true", help="validate without a hub")
    a = ap.parse_args()
    if a.selfcheck:
        _selfcheck()
    else:
        try:
            run(a.host, a.port, a.rate, a.once)
        except KeyboardInterrupt:
            print("\nstopped.")