e69b7e2ad6
Captures backend work already running on godstrument.pro but never committed: - auth.py: invite-only accounts + per-user presets (SQLite, scrypt, signed stateless sessions), mounted by hub.py at /api/* - hub.py: /api static+API handler, readonly public mode, full route spec in hello - socio-economic / market / fire / debt / crypto workers + normalize/transform - .gitignore: never commit godstrument_users.db* or auth_secret - test_fixes.py: framework-free self-check for the review fixes Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
195 lines
6.8 KiB
Python
195 lines
6.8 KiB
Python
"""
|
|
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
|