Godstrument/transform.py

"""
transform.py — the performance layer. The human hands on the machine.

Raw sources (planet, room, markets) are honest but mechanical. This layer lets
you *play* them live: bend, offset, freeze, or smooth a signal on the fly — the
way you'd nudge a CDJ platter or lean on a pitch-bend wheel. And you can bind one
control to a whole GROUP of sources, so a single knob leans on four data streams
at once.

A Tweak sits between a source's normalized 0..1 value and the modulation matrix:

    value ── group tweak ── source tweak ──► matrix

Parameters (each live-settable by a MIDI knob, a hand sensor, or the UI):
    gain    multiply (0..~2)          — depth / "how much this source matters now"
    offset  add (-1..1)               — bias / bend / lean
    invert  flip (0/1)                — mirror the signal
    freeze  hold current value (0/1)  — "stop the planet here" (CDJ pause)
    smooth  0..1 extra lag            — make it liquid, or snap it tight
"""

from __future__ import annotations


def _clamp01(x: float) -> float:
    return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x


class Tweak:
    """A live-adjustable transform on one signal (or a whole group)."""

    DEFAULTS = {"gain": 1.0, "offset": 0.0, "invert": 0.0,
                "freeze": 0.0, "smooth": 0.0, "mute": 0.0}

    def __init__(self, spec: dict | None = None):
        self.p = dict(self.DEFAULTS)
        if spec:
            for k, v in spec.items():
                if k in self.p:
                    self.p[k] = float(v)
        self._held: dict[str, float] = {}    # freeze holds, keyed by signal name
        self._lag: dict[str, float] = {}     # smoothing state, keyed by signal name
        self.orbit_lfo = 1.0                 # external planetary-orbit gain (hub-set)

    def set(self, param: str, value: float):
        if param in self.p:
            self.p[param] = float(value)

    @property
    def active(self) -> bool:
        p = self.p
        return (p["gain"] != 1.0 or p["offset"] != 0.0 or p["invert"] >= 0.5
                or p["freeze"] >= 0.5 or p["smooth"] > 0.0 or p["mute"] >= 0.5)

    def apply(self, x: float, key: str = "") -> float:
        # `key` is the signal name: a group Tweak is applied to every member, so
        # freeze/smooth state must be kept per-signal, not once for the group.
        p = self.p
        # mute: silence this source/group entirely
        if p["mute"] >= 0.5:
            return 0.0
        # freeze: hold each signal's own last value (a CDJ pause on the data)
        if p["freeze"] >= 0.5:
            h = self._held.get(key)
            if h is None:
                h = self._lag.get(key, x)
                self._held[key] = h
            return h
        self._held.pop(key, None)

        v = (1.0 - x) if p["invert"] >= 0.5 else x
        v = v * p["gain"] * self.orbit_lfo + p["offset"]
        v = _clamp01(v)

        s = p["smooth"]
        if s > 0.0:
            alpha = 1.0 - min(0.985, s * 0.985)   # smooth=0 -> 1 (pass), 1 -> ~0.015
            prev = self._lag.get(key)
            self._lag[key] = v if prev is None else prev + alpha * (v - prev)
            v = self._lag[key]
        else:
            self._lag[key] = v
        return _clamp01(v)


class TweakBank:
    """Owns every group + per-source tweak and resolves live control bindings.

    config sections:
      groups:   { "planet": {"members": ["weather.temp", ...], "label": "..."} }
      tweaks:   { "crypto.vel": {gain,offset,...}, "@planet": {gain,...} }
      controls: [ { "src": "hand.open", "target": "@planet",
                    "param": "gain", "min": 0.2, "max": 1.3 }, ... ]
                (src is any signal name — a MIDI CC like "midi.cc.1", a sensor
                 like "hand.open", or another data source. target "@name" = group.)
    """

    def __init__(self, cfg: dict):
        self.groups: dict[str, Tweak] = {}
        self.group_members: dict[str, list[str]] = {}
        self.member_groups: dict[str, list] = {}   # a source can be in many groups
        self.source_tweaks: dict[str, Tweak] = {}
        self.controls: list[dict] = list(cfg.get("controls", []))
        self.group_orbit: dict[str, str] = {}     # group -> planet name
        self.group_label: dict[str, str] = {}     # group -> display label

        tw = cfg.get("tweaks", {})
        for gname, g in cfg.get("groups", {}).items():
            members = g.get("members", [])
            self.group_members[gname] = members
            self.groups[gname] = Tweak(tw.get("@" + gname))
            self.group_label[gname] = g.get("label", gname)
            if g.get("orbit"):
                self.group_orbit[gname] = g["orbit"]
            for m in members:
                self.member_groups.setdefault(m, []).append(gname)
        for name, spec in tw.items():
            if not name.startswith("@"):
                self.source_tweaks[name] = Tweak(spec)

        # ensure a tweak object exists for every control target
        for b in self.controls:
            self._resolve(b.get("target", ""))

    def _resolve(self, target: str) -> Tweak | None:
        if not target:
            return None
        if target.startswith("@"):
            name = target[1:]
            return self.groups.setdefault(name, Tweak())
        return self.source_tweaks.setdefault(target, Tweak())

    def apply_controls(self, sources: dict[str, float]):
        """Read live control signals and push them onto tweak params."""
        for b in self.controls:
            cv = sources.get(b.get("src"))
            if cv is None:
                continue
            lo, hi = b.get("min", 0.0), b.get("max", 1.0)
            mapped = lo + (hi - lo) * cv
            tw = self._resolve(b.get("target", ""))
            if tw:
                tw.set(b.get("param", "gain"), mapped)

    def transform(self, sources: dict[str, float]) -> dict[str, float]:
        out: dict[str, float] = {}
        for name, val in sources.items():
            v = val
            for g in self.member_groups.get(name, ()):   # stack every concept-group
                v = self.groups[g].apply(v, name)
            st = self.source_tweaks.get(name)
            if st:
                v = st.apply(v, name)
            out[name] = v
        return out

    # ---- live control from the UI (browser -> hub) --------------------
    def set_param(self, target: str, param: str, value: float):
        tw = self._resolve(target)
        if tw:
            tw.set(param, value)

    def toggle(self, target: str, param: str = "mute") -> float:
        tw = self._resolve(target)
        if not tw:
            return 0.0
        cur = tw.p.get(param, 0.0)
        newv = 0.0 if cur >= 0.5 else 1.0
        tw.set(param, newv)
        return newv

    def is_muted(self, name: str) -> bool:
        st = self.source_tweaks.get(name)
        if st and st.p.get("mute", 0.0) >= 0.5:
            return True
        return any(self.groups[g].p.get("mute", 0.0) >= 0.5
                   for g in self.member_groups.get(name, ()))

    def set_orbit_lfo(self, gname: str, v: float):
        tw = self.groups.get(gname)
        if tw:
            tw.orbit_lfo = v

    def add_group(self, name: str, members: list, label: str = None,
                  orbit: str = None):
        """Create/replace a group at runtime (browser multi-select -> group)."""
        if not name or not members:
            return
        # drop stale memberships from a previous definition so removed sources
        # don't keep getting muted/tweaked by this group forever
        for m in self.group_members.get(name, []):
            gl = self.member_groups.get(m)
            if gl and name in gl:
                gl.remove(name)
        self.group_members[name] = list(members)
        self.groups.setdefault(name, Tweak())
        self.group_label[name] = label or name
        if orbit:
            self.group_orbit[name] = orbit
        for m in members:
            gl = self.member_groups.setdefault(m, [])
            if name not in gl:
                gl.append(name)

    def dump_groups(self) -> dict:
        """Serialize current groups for saving a template."""
        return {g: {"members": self.group_members[g],
                    "label": self.group_label.get(g, g),
                    "orbit": self.group_orbit.get(g)}
                for g in self.group_members}

    def load_groups(self, groups: dict):
        """Rebuild groups from a saved template."""
        for name, gd in (groups or {}).items():
            self.add_group(name, gd.get("members", []),
                           gd.get("label"), gd.get("orbit"))

    def dump_state(self) -> dict:
        """Serialize live tweak settings (gains/mutes/freezes) — a spell's mix."""
        return {
            "groups": {g: dict(self.groups[g].p) for g in self.group_members},
            "sources": {n: dict(t.p) for n, t in self.source_tweaks.items()
                        if t.active},
        }

    def load_state(self, state: dict):
        for g, params in (state.get("groups") or {}).items():
            tw = self.groups.get(g)
            if tw:
                for k, v in params.items():
                    tw.set(k, v)
        for n, params in (state.get("sources") or {}).items():
            tw = self.source_tweaks.setdefault(n, Tweak())
            for k, v in params.items():
                tw.set(k, v)

    def macros(self) -> dict:
        """Summary of group tweak state for the UI."""
        return {g: {"gain": round(t.p["gain"], 3),
                    "offset": round(t.p["offset"], 3),
                    "freeze": t.p["freeze"] >= 0.5,
                    "mute": t.p["mute"] >= 0.5,
                    "orbit": self.group_orbit.get(g),
                    "lfo": round(t.orbit_lfo, 3)}
                for g, t in self.groups.items() if g in self.group_members}