From 783d8d0c9795d9aeac3427b93203f344b52659f1 Mon Sep 17 00:00:00 2001 From: type-two Date: Mon, 6 Jul 2026 15:51:42 +1000 Subject: [PATCH] =?UTF-8?q?local=5Fsource.py=20=E2=80=94=20device=20firmwa?= =?UTF-8?q?re=20for=20the=20"local=20source"=20sensor=20rig?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The room as an instrument. Streams environmental sensors (light/temp/humidity/ pressure/ToF-hand/IMU-shake+tilt/mic-energy + crowd.hit events) to a hub over OSC. Confirms + documents the hardware-agnostic ingest contract: OSC /gs// -> a live source (already handled by hub._osc_handler; send RAW, the hub auto-scales). Runs with zero sensors (smooth simulated fallbacks) so the pipeline works before any hardware exists; real driver lines (BH1750/BME280/VL53L1X/MPU6050/mic) are stubbed to uncomment. Verified: --selfcheck passes; a live burst to the running hub created sources room.*/hand.height/booth.*/crowd.* end-to-end over the websocket. Any SBC/lang that speaks the OSC contract works identically — Pi4B ready today. Co-Authored-By: Claude Fable 5 --- local_source.py | 197 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 local_source.py diff --git a/local_source.py b/local_source.py new file mode 100644 index 0000000..cfad71d --- /dev/null +++ b/local_source.py @@ -0,0 +1,197 @@ +#!/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// 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.")