import numpy as np from scipy.io.wavfile import read # Same MNIST .wav loading as before X_train = read('../X_train.wav')[1].reshape(-1, 784) y_train = (read('../y_train.wav')[1] * 9).astype(int) X_test = read('../X_test.wav')[1].reshape(-1, 784) y_test = (read('../y_test.wav')[1] * 9).astype(int) X20 = X_test[:1000] yt20 = y_test[:1000] class SinCosMatrixClassifier: """ Graphical classifier using the Sin/Cos Matrix theory. Static graphical maps: One prototype P_c and one frequency omega_c per class. For input x and predicted label y_t, the per-element "pixel errors" e_k = x_k - P_y[k] are pushed through a sin/cos coupling: score_c(x) = sum_k cos(omega_c * (x_k - P_c[k])). Pixel errors + label errors drive the parameter update. The recurrent predicted label y_t is the per-class bookkeeping label used as the *target* the parameters are pulled toward. When y_t == argmax(score), that class's parameters are frozen for that sample (the "static classification map" interpretation). """ def __init__(self, n_classes=10, input_dim=784, base_omega=0.02, lr=(0.05, 1e-3), label_weight=1.0, pixel_weight=0.5, seed=0): self.n_classes = n_classes self.input_dim = input_dim self.base_omega = base_omega self.lr_p, self.lr_w = lr self.lambda_label = label_weight self.lambda_pixel = pixel_weight rng = np.random.default_rng(seed) self.P = rng.standard_normal((n_classes, input_dim)) * 0.1 # (C, D) self.omega = np.full(n_classes, base_omega) # (C,) # ---------------------------------------------------------------- @staticmethod def _softmax(s): s = s - s.max(1, keepdims=True) e = np.exp(s) return e / e.sum(1, keepdims=True) def _scores(self, X): """Sin/cos matrix diagonal-row flux: cos(omega_c * (x_k - P_c[k])) summed.""" diff = X[:, None, :] - self.P[None, :, :] # (m, C, D) return np.cos(self.omega[None, :, None] * diff).sum(axis=2) # (m, C) def predict(self, X): return np.argmax(self._scores(X), axis=1) def score(self, X, y_true): return np.mean(self.predict(X) == y_true) # ---------------------------------------------------------------- def update(self, X, y_true, y_recurrent): """ One parameter-update step using pixel errors + label errors. Pixel error channel: d score_c / d P_c[k] = -omega_c * sin(omega_c * diff_k) Label error channel: softmax(score) - onehot(y_recurrent) Both multiplied, gated by the "freeze" mask so a class whose predicted label agrees with the recurrent bookkeeping label is treated as a STATIC MAP for that sample. """ m, D = X.shape diff = X[:, None, :] - self.P[None, :, :] # (m, C, D) scores = np.cos(self.omega[None, :, None] * diff).sum(axis=2) # (m, C) probs = self._softmax(scores) # (m, C) # ---- label-error vector (m, C) ---- y_onehot = np.eye(self.n_classes)[y_recurrent] # (m, C) label_err = (probs - y_onehot) / m # (m, C) # ---- pixel-error dscore/dP (m, C, D) ---- dscore_dP = -self.omega[None, :, None] * np.sin(self.omega[None, :, None] * diff) # (m, C, D) # ---- gate: freeze the map whose recurrent label agrees with argmax ---- y_pred = np.argmax(scores, axis=1) # (m,) freeze = (y_pred == y_recurrent).astype(float) # (m,) gate = np.ones((m, self.n_classes)) gate[np.arange(m), y_recurrent] -= freeze # (m, C) # ---- per-class prototype gradient ---- # pull P_c toward reducing pixel diff (shrinks the cos-pixel error) pixel_grad = label_err[:, :, None] * dscore_dP # (m, C, D) dP_pull_pixels = pixel_grad * gate[:, :, None] # (m, C, D) # pull P_c to make the predicted score match the recurrent label dP_label = (label_err[:, :, None] * np.cos(self.omega[None, :, None] * diff) * gate[:, :, None]) # (m, C, D) total_dP = (self.lambda_label * dP_label.sum(axis=0) + self.lambda_pixel * dP_pull_pixels.sum(axis=0)) # (C, D) # ---- omega gradient ---- dscore_dw = (diff * np.sin(self.omega[None, :, None] * diff)).sum(axis=2) # (m, C) d_omega_vec = (label_err * dscore_dw * gate).sum(axis=0) / m # (C,) self.P -= self.lr_p * total_dP self.omega -= self.lr_w * d_omega_vec # numerical safety self.omega = np.clip(self.omega, 1e-3, 5.0) return scores, probs # ---------------------------------------------------------------- def train(self, X, y_true, iters=300, batch=128, eval_every=25, eval_train_n=5000): """ Recurrent training loop. Each step: 1. Sample a batch. 2. Compute y_t = predict(X_batch) (the recurrent predicted label). 3. Call update with (y_true=y_batch, y_recurrent=y_t). 4. Monitor loss vs y_true and accuracy on y_t == y_true. The "fit" signal pushes P_c so its class's cos-coupling flux is large when y_t is the recurrent bookkeeping label, with the pixel-error term encouraging the prototype to resemble its class's typical image. """ history = [] for t in range(1, iters + 1): idx = np.random.randint(0, X.shape[0], batch) Xb = X[idx] yb = y_true[idx] y_t = self.predict(Xb) scores, probs = self.update(Xb, y_true=yb, y_recurrent=y_t) # cross-entropy vs the TRUE label (monitoring only) loss = -np.log(probs[np.arange(batch), yb] + 1e-12).mean() history.append({ "step": t, "loss": float(loss), "y_t_eq_y_true": float(np.mean(y_t == yb)), }) if t % eval_every == 0 or t == 1: tr = self.score(X[:eval_train_n], y_true[:eval_train_n]) te = self.score(X20, yt20) agree = history[-1]["y_t_eq_y_true"] print(f"step {t:4d}: loss={loss:.4f} " f"train={tr:.4f} test={te:.4f} " f"y_t==y_true={agree:.4f}") return history # ============================================================ if __name__ == "__main__": np.random.seed(0) print("=" * 64) print("Sin/Cos Matrix Graphical Classifier (pixel + label errors)") print("=" * 64) clf = SinCosMatrixClassifier( n_classes=10, input_dim=784, base_omega=0.02, lr=(0.05, 1e-3), label_weight=1.0, pixel_weight=0.5, seed=0, ) print(f"P shape: {clf.P.shape}, omega shape: {clf.omega.shape}") print(f"init test acc: {clf.score(X20, yt20):.4f}") history = clf.train( X_train, y_train, iters=300, batch=128, eval_every=25, eval_train_n=5000, ) print(f"\nFinal train acc (5000) : " f"{clf.score(X_train[:5000], y_train[:5000]):.4f}") print(f"Final test acc (X20) : " f"{clf.score(X20, yt20):.4f}") # show the last few omega values for sanity print(f"\nFinal omega per class: {np.round(clf.omega, 4)}")