vvs/sim-numerics
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

baseline: the published sim-numerics artifact as it stands (490 tests, verify_article.py checks all 72 paper numbers against sweep_full.json)

Browse Source 
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
vvs
2026-09-20 22:38:17 +01:00
co-authored by Claude Opus 5
commit 4a23aacf96
75 changed files with 12175 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
anc/__init__.py
View File
anc/bench.py
+190
View File
@@ -0,0 +1,190 @@
"""Active noise control: the step size of an adaptive filter.
The third domain, and the one where the object we have is most exactly the
object the field argues about.
In NLMS the coefficient update is
w <- w + mu * e * x / (||x||^2 + eps)
and mu = 1 is not an arbitrary scale: it is the step that drives the a
posteriori error at the current sample exactly to zero. Every practical
system uses mu < 1, because zeroing the current sample fits the measurement
noise as well as the signal. So mu is precisely a SAFETY FACTOR on an exact
step - the same object as the safety factor in ODE step control, and the same
object as the projection budget in the root-finding safeguard.
The whole variable-step-size LMS literature exists because mu is hard to
choose, and it answers with more constants: Kwong and Johnston's rule has
four. What this file asks is whether the quantity mu is guessing can be
measured instead.
predicted a posteriori error (1 - mu) * e if the model were exact
observed a posteriori error d - w_new . x after the update
Their ratio says how much of e was signal the filter could actually remove
and how much was noise it could not. That is the safety factor, measured.
The acoustic paths and the noise regimes are taken verbatim from our earlier
ANC benchmark (branch feature/sem-anc-benchmark of the SEM repository), so
the environment is the established one and not built to suit the result.
"""
from __future__ import annotations
import math
import numpy as np
FS = 8000
PRIMARY_PATH_LEN = 64
SECONDARY_PATH_LEN = 32
FILTER_LEN = 64
# ---------------------------------------------------------------- paths ---
def make_primary_path(rng):
h = np.zeros(PRIMARY_PATH_LEN)
delay = rng.integers(5, 10)
h[delay] = 0.8 + 0.2 * rng.standard_normal()
for k in range(3):
idx = delay + rng.integers(3, 15)
if idx < PRIMARY_PATH_LEN:
h[idx] = (0.3 - 0.1 * k) * rng.standard_normal()
for i in range(delay + 5, PRIMARY_PATH_LEN):
h[i] += 0.05 * math.exp(-0.1 * (i - delay)) * rng.standard_normal()
return h
def make_secondary_path(rng):
h = np.zeros(SECONDARY_PATH_LEN)
delay = rng.integers(2, 5)
h[delay] = 0.9 + 0.1 * rng.standard_normal()
for k in range(2):
idx = delay + rng.integers(2, 8)
if idx < SECONDARY_PATH_LEN:
h[idx] = (0.2 - 0.05 * k) * rng.standard_normal()
for i in range(delay + 3, SECONDARY_PATH_LEN):
h[i] += 0.03 * math.exp(-0.15 * (i - delay)) * rng.standard_normal()
return h
# ------------------------------------------------------------- regimes ---
def gen_tonal(n, rng, f0=200.0):
t = np.arange(n) / FS
return np.sin(2 * math.pi * f0 * t + rng.uniform(0, 2 * math.pi))
def gen_broadband(n, rng):
pink = np.zeros(n)
rows = np.zeros(5)
for i in range(n):
for j in range(5):
if i % (2 ** j) == 0:
rows[j] = rng.standard_normal()
pink[i] = np.sum(rows)
return pink / (np.std(pink) + 1e-12)
def gen_multitonal(n, rng, f0=100.0):
t = np.arange(n) / FS
ph = rng.uniform(0, 2 * math.pi, 3)
sig = (np.sin(2 * math.pi * f0 * t + ph[0])
+ 0.5 * np.sin(2 * math.pi * 2 * f0 * t + ph[1])
+ 0.25 * np.sin(2 * math.pi * 3 * f0 * t + ph[2]))
return sig / (np.max(np.abs(sig)) + 1e-12)
def gen_silence(n, rng):
return rng.standard_normal(n) * 0.01
def gen_impulse_train(n, rng):
sig = np.zeros(n)
for _ in range(rng.integers(5, 12)):
pos = rng.integers(0, n)
width = rng.integers(10, 40)
amp = rng.uniform(1.0, 3.0) * rng.choice([-1, 1])
end = min(pos + width, n)
sig[pos:end] = amp * np.exp(-5.0 * np.arange(end - pos) / width)
return sig
def scen_stable(n, rng):
seg = n // 4
return np.concatenate([gen_tonal(seg, rng), gen_broadband(seg, rng),
gen_multitonal(seg, rng),
gen_silence(n - 3 * seg, rng)])
def scen_gradual(n, rng):
a = np.linspace(0, 1, n)
return (1 - a) * gen_tonal(n, rng) + a * gen_broadband(n, rng)
def scen_abrupt(n, rng):
out = np.zeros(n)
gens = [gen_tonal, gen_broadband, gen_multitonal]
pos = 0
while pos < n:
seg = min(int(rng.integers(n // 20, n // 8)), n - pos)
out[pos:pos + seg] = gens[rng.integers(0, 3)](seg, rng)
pos += seg
return out
def scen_cabin(n, rng):
engine = gen_multitonal(n, rng, f0=80.0)
road = 0.6 * gen_broadband(n, rng)
imp = 0.4 * gen_impulse_train(n, rng)
s = engine + road + imp
return s / (np.max(np.abs(s)) + 1e-12)
def scen_impulsive(n, rng):
return gen_impulse_train(n, rng) + 0.2 * gen_broadband(n, rng)
SCENARIOS = [("stable regimes", scen_stable), ("gradual transition", scen_gradual),
("abrupt switching", scen_abrupt), ("car cabin", scen_cabin),
("impulsive", scen_impulsive)]
# --------------------------------------------------------- the filter ---
def run_nlms(noise, prim, sec, rule, snr_db=30.0, seed=0):
"""One NLMS/FxLMS run. Everything is fixed except `rule`, which returns
the step size for the current sample."""
rng = np.random.default_rng(seed)
n = len(noise)
d = np.convolve(noise, prim)[:n] # what reaches the ear
xf = np.convolve(noise, sec)[:n] # filtered reference
p = float(np.mean(d * d))
d = d + rng.standard_normal(n) * math.sqrt(p / (10 ** (snr_db / 10.0)))
w = np.zeros(FILTER_LEN)
buf = np.zeros(FILTER_LEN)
err = np.zeros(n)
rule.reset()
for i in range(n):
buf[1:] = buf[:-1]
buf[0] = xf[i]
y = float(np.dot(w, buf))
e = d[i] - y
err[i] = e
nrm = float(np.dot(buf, buf)) + 1e-6
mu = rule.step(e, buf, nrm)
if mu != 0.0:
w += mu * e * buf / nrm
# the a posteriori error, which is what the rule needs to learn
# how much of e it was able to remove
rule.observe(e, d[i] - float(np.dot(w, buf)), mu)
else:
rule.observe(e, e, 0.0)
# noise reduction: power of the residual against the power of what would
# have reached the ear untreated, measured on the second half so the
# transient is not counted
half = n // 2
p_in = float(np.mean(d[half:] ** 2))
p_out = float(np.mean(err[half:] ** 2))
if p_out <= 0.0 or not math.isfinite(p_out):
return float("nan")
return 10.0 * math.log10(p_in / p_out)
anc/rules.py
+204
View File
@@ -0,0 +1,204 @@
"""Step-size rules for NLMS. Everything else in the filter is shared.
mu = 1 drives the a posteriori error of the current sample to zero
mu < 1 hedges, because that sample carries measurement noise too
So mu is a safety factor on an exact step, exactly as in ODE step control.
The variable-step-size literature answers the difficulty of choosing it with
more constants; the rule at the bottom measures it instead.
"""
from __future__ import annotations
import math
class Rule:
name = "?"
constants = 0
def reset(self):
pass
def step(self, e, x, nrm):
raise NotImplementedError
def observe(self, e_pre, e_post, mu):
pass
class Fixed(Rule):
def __init__(self, mu):
self.mu = mu
self.name = f"fixed mu={mu:g}"
self.constants = 1
def step(self, e, x, nrm):
return self.mu
class KwongJohnston(Rule):
"""The classical variable step size: mu(n+1) = a*mu(n) + g*e(n)^2, clamped.
Four constants, and the paper that introduced it spends its length on how
to pick them."""
name = "Kwong-Johnston VSS"
constants = 4
A, G, MU_MIN, MU_MAX = 0.97, 0.02, 0.001, 1.0
def reset(self):
self.mu = 0.5
def step(self, e, x, nrm):
return self.mu
def observe(self, e_pre, e_post, mu):
self.mu = self.A * self.mu + self.G * e_pre * e_pre
self.mu = min(self.MU_MAX, max(self.MU_MIN, self.mu))
class InSampleMeasured(Rule):
"""REFUTED - kept so the mistake is reproducible.
The first attempt measured the A POSTERIORI error of the sample the update
had just been fitted to. That observable is uninformative by
construction: NLMS with mu = 1 drives it to exactly zero, so the rule
reads "the model was perfect", proposes mu = 1 again, and is locked there.
Measured, it scores 18.50 dB against fixed mu = 1's 18.49 - the same rule
wearing a different name.
It is the same error as the rho detector in the root-finding work:
measuring a quantity in the one place where it is identically zero."""
name = "in-sample (refuted)"
constants = 0
def reset(self):
self.rho = 1.0
def step(self, e, x, nrm):
return min(1.0, 1.0 / self.rho) if self.rho > 0.0 else 1.0
def observe(self, e_pre, e_post, mu):
if mu <= 0.0 or e_pre == 0.0:
return
predicted = abs((1.0 - mu) * e_pre)
observed = abs(e_post)
if predicted <= 0.0:
self.rho = 1.0 + observed / (abs(e_pre) + 1e-300)
else:
r = observed / predicted
if math.isfinite(r) and r > 0.0:
self.rho = math.sqrt(self.rho * max(r, 1e-6))
class Measured(Rule):
"""The safety factor measured out of sample.
What mu is hedging against is the part of the error the filter cannot
cancel - the measurement noise. NLMS theory gives the step that minimises
the mean square deviation as the SIGNAL FRACTION of the error,
mu = 1 - p_noise / p_error,
and both quantities are observable. p_error is the running power of e.
p_noise is what e cannot be driven below, so it is estimated as the
smallest error power the run has sustained - the floor the filter keeps
hitting.
Nothing here is chosen. The two smoothing weights are 1/2, an average of
the current observation with the accumulated one, not a tuned forgetting
factor; and the first samples, having no floor to compare against, take
the exact step - the licence the root-finding safeguard grants.
"""
name = "measured (no constants)"
constants = 0
def reset(self):
self.p_err = None # running error power
self.floor = None # the smallest sustained error power seen
def step(self, e, x, nrm):
if self.p_err is None or self.floor is None or self.p_err <= 0.0:
return 1.0
frac = 1.0 - self.floor / self.p_err
return min(1.0, max(0.0, frac))
def observe(self, e_pre, e_post, mu):
p = e_pre * e_pre
self.p_err = p if self.p_err is None else 0.5 * (self.p_err + p)
if self.floor is None:
self.floor = self.p_err
elif self.p_err < self.floor:
self.floor = self.p_err
else:
# the floor may rise when the environment changes; let it drift up
# by the same averaging, so a regime change is not remembered for
# ever as a floor that no longer exists
self.floor = 0.5 * (self.floor + min(self.p_err, self.floor * 2.0))
class CosineSimilarity(Rule):
"""mu as a SIMILARITY, which is what this project was supposed to be about.
The two rules above estimate mu from variances - a noise floor, a model
discrepancy. Neither is a similarity; both are statistics of the error.
The similarity formulation asks a different question:
how much does the error RESEMBLE something the filter can remove?
The filter can only remove what is correlated with its input. So the
quantity is the cosine similarity between the error and the filtered
reference,
mu = |<e, x>| / (||e|| ||x||) in [0, 1],
which is the cosine of the original paper, applied to the two signals the
filter actually has. It equals 1 when the error is entirely explainable
by the reference - the exact step is then right - and 0 when the error is
orthogonal to it, which is measurement noise the update can only fit.
The three running means use the weight 1/2, an average of the current
observation with the accumulated one, and the first sample takes the exact
step for want of anything to compare. No constant is chosen.
"""
name = "cosine similarity (no constants)"
constants = 0
S = None # None means the identity: mu = |cos| directly
def reset(self):
self.ex = None # running <e, x>
self.ee = None # running <e, e>
self.xx = None # running <x, x>
def step(self, e, x, nrm):
# decide from what has been accumulated SO FAR, then fold in the
# current sample - so the step never uses the observation it is about
# to be judged on
if self.ex is None:
mu = 1.0
else:
den = math.sqrt(self.ee * self.xx)
if den <= 0.0 or not math.isfinite(den):
mu = 1.0
else:
cos = min(1.0, abs(self.ex) / den)
if self.S is None:
mu = cos
else:
# the general family: mu = S(d) with d = 1 - |cos| the
# dissimilarity between the error and what the filter can
# remove. d = 0 means fully removable, so S(0) = 1 and the
# exact step is taken; the identity above is S(d) = 1 - d.
mu = max(0.0, min(1.0, self.S(1.0 - cos)))
xv = float(x[0])
self.ex = e * xv if self.ex is None else 0.5 * (self.ex + e * xv)
self.ee = e * e if self.ee is None else 0.5 * (self.ee + e * e)
self.xx = xv * xv if self.xx is None else 0.5 * (self.xx + xv * xv)
return mu
ALL = [Fixed(0.1), Fixed(0.5), Fixed(1.0), KwongJohnston(),
InSampleMeasured(), Measured(), CosineSimilarity()]