Godstrument/matrix.py

"""
matrix.py — the modulation matrix. The actual instrument.

Everything upstream turns the world into a flat bag of normalized 0..1 signals:

    sources = {"sun.speed": 0.62, "crypto.vel": 0.11, "tof.cx": 0.8, ...}

The matrix is a patchbay: N sources x M destinations, each connection a "cable"
with an amount, a response curve, and a polarity. Cables can optionally:

    * gate     — only pass when the source crosses a threshold (an event fires
                 a sample only when a real quake is big enough).
    * quantize — snap a continuous source to notes in a musical scale, so the
                 price of bitcoin plays a melody in D-dorian instead of a
                 formless wobble.

A source and a satellite are the same kind of thing here. That is the whole
point of the Godstrument.

Pure stdlib.
"""

from __future__ import annotations

# ---------------------------------------------------------------------------
# Musical scales (semitone offsets within an octave)
# ---------------------------------------------------------------------------
SCALES: dict[str, list[int]] = {
    "chromatic":  [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
    "major":      [0, 2, 4, 5, 7, 9, 11],
    "minor":      [0, 2, 3, 5, 7, 8, 10],
    "dorian":     [0, 2, 3, 5, 7, 9, 10],
    "phrygian":   [0, 1, 3, 5, 7, 8, 10],
    "lydian":     [0, 2, 4, 6, 7, 9, 11],
    "mixolydian": [0, 2, 4, 5, 7, 9, 10],
    "pentatonic": [0, 2, 4, 7, 9],
    "minor_pent": [0, 3, 5, 7, 10],
    "wholetone":  [0, 2, 4, 6, 8, 10],
    "hirajoshi":  [0, 2, 3, 7, 8],
}


def quantize_to_scale(x01: float, scale: str = "minor_pent",
                      root: int = 48, octaves: int = 3) -> int:
    """Map 0..1 to a MIDI note constrained to `scale`, starting at `root`."""
    steps = SCALES.get(scale, SCALES["chromatic"])
    n = len(steps) * octaves
    idx = int(_clamp01(x01) * (n - 1) + 0.5)
    octave, degree = divmod(idx, len(steps))
    return root + 12 * octave + steps[degree]


# ---------------------------------------------------------------------------
# Response curves
# ---------------------------------------------------------------------------
def apply_curve(x: float, curve: str = "lin") -> float:
    x = _clamp01(x)
    if curve == "lin":
        return x
    if curve == "exp":          # slow start, fast finish (perceptual filter sweep)
        return x * x
    if curve == "exp3":
        return x * x * x
    if curve == "log":          # fast start, slow finish
        return x ** 0.5
    if curve == "scurve":       # ease-in-out
        return x * x * (3 - 2 * x)
    if curve == "inv":
        return 1.0 - x
    return x


# ---------------------------------------------------------------------------
# The matrix
# ---------------------------------------------------------------------------
class ModMatrix:
    """
    routes: list of dicts, each a cable:
        {
          "source": "sun.speed",          # source signal name (dotted)
          "dest":   "filter.cutoff",      # destination parameter name
          "amount": 0.9,                  # -1..1 depth
          "curve":  "exp",                # response curve (see apply_curve)
          "gate":   0.7,                  # optional: only pass if source >= gate
          "quantize": {"scale": "dorian", # optional: snap to notes -> writes
                        "root": 48,       #   dest as a MIDI note number instead
                        "octaves": 2},    #   of a 0..1 value
          "label":  "the sun opens the filter"   # optional, for the visualizer
        }
    """

    def __init__(self, routes: list[dict]):
        self.routes = [dict(r) for r in routes]

    @classmethod
    def from_config(cls, cfg: dict) -> "ModMatrix":
        return cls(cfg.get("routes", []))

    # ---- live editing (browser -> hub) --------------------------------
    def add_route(self, source, dest, amount=0.5, curve="lin", label=""):
        if not source or not dest:
            return
        if any(r.get("source") == source and r.get("dest") == dest
               for r in self.routes):
            return                                  # already patched
        self.routes.append({"source": source, "dest": dest,
                            "amount": float(amount), "curve": curve,
                            "label": label or (source + " → " + dest)})

    def remove_route(self, source, dest):
        self.routes = [r for r in self.routes
                       if not (r.get("source") == source and r.get("dest") == dest)]

    def set_route_amount(self, source, dest, amount):
        for r in self.routes:
            if r.get("source") == source and r.get("dest") == dest:
                r["amount"] = float(amount)

    def evaluate(self, sources: dict[str, float]):
        """Returns (dests, active) where:
             dests  = {dest_name: value}     summed & clamped contributions
             active = {route_index: 0..1}    how 'lit' each cable is (for viz)
        """
        dests: dict[str, float] = {}
        notes: dict[str, int] = {}
        active: dict[int, float] = {}

        for i, r in enumerate(self.routes):
            src = r.get("source")
            dst = r.get("dest")
            if src is None or dst is None:
                active[i] = 0.0
                continue
            raw = sources.get(src)
            if raw is None:
                active[i] = 0.0
                continue

            # gate: a threshold the source must cross to pass at all
            gate = r.get("gate")
            if gate is not None and raw < gate:
                active[i] = 0.0
                continue

            shaped = apply_curve(raw, r.get("curve", "lin"))
            amount = float(r.get("amount", 1.0))

            q = r.get("quantize")
            if q:
                note = quantize_to_scale(
                    shaped,
                    q.get("scale", "minor_pent"),
                    q.get("root", 48),
                    q.get("octaves", 3),
                )
                notes[dst] = note
                active[i] = abs(amount) * shaped
                continue

            contrib = amount * shaped
            dests[dst] = dests.get(dst, 0.0) + contrib
            active[i] = abs(contrib)

        # clamp summed continuous destinations
        for k in list(dests.keys()):
            dests[k] = _clamp01(dests[k])
        # note destinations override as integers
        for k, v in notes.items():
            dests[k] = v

        return dests, active


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