""" normalize.py — turning messy real-world numbers into clean 0..1 control signals. Every source in the Godstrument (your hand, the room mic, a satellite, the price of bitcoin) speaks in its own units and its own range. Before any of it can modulate sound, it has to be tamed: * OneEuroFilter — kills jitter without adding lag (the gold standard for interactive/sensor data: still when you're still, snappy when you move). * AdaptiveNormalizer — auto-scales an unknown, drifting range (a gold price, a solar-wind speed) into 0..1 using a rolling window. * ImpulseEnvelope — turns a discrete event (an earthquake, a wiki edit) into a decaying 0..1 signal you can actually hear. * Slew — smoothly chases a target so slow data doesn't step/click. Pure stdlib. No numpy required. """ from __future__ import annotations import math from collections import deque # --------------------------------------------------------------------------- # One Euro Filter (Casiez, Roussel, Vogel 2012) # --------------------------------------------------------------------------- class _LowPass: def __init__(self): self.y = None self.s = None def __call__(self, x: float, alpha: float) -> float: if self.s is None: self.s = x else: self.s = alpha * x + (1.0 - alpha) * self.s self.y = x return self.s def _alpha(cutoff: float, dt: float) -> float: tau = 1.0 / (2.0 * math.pi * cutoff) return 1.0 / (1.0 + tau / dt) class OneEuroFilter: """Adaptive low-pass. Low speed -> heavy smoothing; high speed -> low lag. f = OneEuroFilter(min_cutoff=1.0, beta=0.007) y = f(x, t_seconds) """ def __init__(self, min_cutoff: float = 1.0, beta: float = 0.007, d_cutoff: float = 1.0): self.min_cutoff = float(min_cutoff) self.beta = float(beta) self.d_cutoff = float(d_cutoff) self._x = _LowPass() self._dx = _LowPass() self._t_prev: float | None = None self._x_prev: float | None = None def __call__(self, x: float, t: float) -> float: if self._t_prev is None: self._t_prev = t self._x_prev = x self._x(x, 1.0) return x dt = t - self._t_prev if dt <= 0: dt = 1e-3 # derivative, low-passed dx = (x - self._x_prev) / dt edx = self._dx(dx, _alpha(self.d_cutoff, dt)) # signal, low-passed with speed-dependent cutoff cutoff = self.min_cutoff + self.beta * abs(edx) y = self._x(x, _alpha(cutoff, dt)) self._t_prev = t self._x_prev = x return y # --------------------------------------------------------------------------- # Adaptive normalizer — unknown, drifting ranges -> 0..1 # --------------------------------------------------------------------------- class AdaptiveNormalizer: """Auto-scales a stream into 0..1 against a rolling window. mode: "minmax" — scale between the window's min and max "zscore" — map (x-mean)/std through a sigmoid (robust to outliers) "fixed" — use an explicit (lo, hi); no adaptation """ def __init__(self, window: int = 600, mode: str = "minmax", fixed: tuple[float, float] | None = None): self.mode = mode self.fixed = fixed self.buf: deque[float] = deque(maxlen=window) self._last = 0.0 def __call__(self, x: float) -> float: try: x = float(x) except (TypeError, ValueError): return self._last if math.isnan(x) or math.isinf(x): return self._last if self.mode == "fixed" and self.fixed: lo, hi = self.fixed self._last = _clamp01((x - lo) / (hi - lo) if hi != lo else 0.0) return self._last self.buf.append(x) if len(self.buf) < 3: self._last = 0.5 return self._last if self.mode == "zscore": mean = sum(self.buf) / len(self.buf) var = sum((v - mean) ** 2 for v in self.buf) / len(self.buf) std = math.sqrt(var) or 1e-6 self._last = _clamp01(0.5 + 0.5 * math.tanh((x - mean) / (2.0 * std))) return self._last lo, hi = min(self.buf), max(self.buf) self._last = _clamp01((x - lo) / (hi - lo) if hi != lo else 0.5) return self._last # --------------------------------------------------------------------------- # Impulse envelope — discrete events -> decaying 0..1 # --------------------------------------------------------------------------- class ImpulseEnvelope: """A quake or a wiki edit is a single instant. This gives it a body: a fast attack and an exponential decay you can route to anything.""" def __init__(self, attack: float = 0.01, decay: float = 0.6): self.attack = max(attack, 1e-4) self.decay = max(decay, 1e-3) self._peak = 0.0 self._t_trigger: float | None = None self._amount = 0.0 def trigger(self, amount: float = 1.0, t: float | None = None): amt = _clamp01(amount) # peak-hold: a smaller event arriving mid-decay must not chop a bigger # one that's still ringing (nor zero it out on a negative magnitude). if t is not None and self._t_trigger is not None and amt <= self.value(t): return self._amount = amt self._t_trigger = t self._peak = 0.0 def value(self, t: float) -> float: if self._t_trigger is None: return 0.0 age = t - self._t_trigger if age < 0: return 0.0 if age < self.attack: return self._amount * (age / self.attack) decay_age = age - self.attack v = self._amount * math.exp(-decay_age / self.decay) return v if v > 1e-4 else 0.0 # --------------------------------------------------------------------------- # Slew — chase a target smoothly (fills gaps between slow polls) # --------------------------------------------------------------------------- class Slew: """rate = units per second the value is allowed to move toward target.""" def __init__(self, rate: float = 4.0): self.rate = rate self.value: float | None = None def __call__(self, target: float, dt: float) -> float: if self.value is None: self.value = target return target step = self.rate * dt delta = target - self.value if abs(delta) <= step: self.value = target else: self.value += step * (1 if delta > 0 else -1) return self.value def _clamp01(x: float) -> float: return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x