reppi

reppi — Representation Learning Algorithms

A Python library implementing classical sparse representation and dictionary learning algorithms.

Modules

sparse Sparse coding (OMP, Batch-OMP). dictionary Dictionary learning (K-SVD, LC-KSVD1, LC-KSVD2).

 1"""
 2reppi — Representation Learning Algorithms
 3==========================================
 4
 5A Python library implementing classical sparse representation and
 6dictionary learning algorithms.
 7
 8Modules
 9-------
10sparse
11    Sparse coding (OMP, Batch-OMP).
12dictionary
13    Dictionary learning (K-SVD, LC-KSVD1, LC-KSVD2).
14"""
15
16from reppi.sparse import OMP
17from reppi.dictionary import KSVD, LCKSVD, FrozenDictionaryLearner, IncrementalFrozenDictionary
18
19__all__ = ["OMP", "KSVD", "LCKSVD", "FrozenDictionaryLearner", "IncrementalFrozenDictionary"]
20__version__ = "0.1.2"
class OMP(reppi.base.BaseSparseCoder):
144class OMP(BaseSparseCoder):
145    """
146    Orthogonal Matching Pursuit sparse coder.
147
148    Parameters
149    ----------
150    n_nonzero_coefs : int
151        Target sparsity — maximum number of non-zero coefficients per signal.
152    mode : {'batch', 'cholesky'}
153        Implementation variant.
154        'batch'     — Batch-OMP; requires the full Gram matrix G = D'D.
155                      Fastest when encoding many signals at once.
156        'cholesky'  — Single-signal OMP-Cholesky; lower memory footprint.
157    check_dict : bool
158        Whether to verify that dictionary atoms are unit-norm (default True).
159    """
160
161    def __init__(
162        self,
163        n_nonzero_coefs: int,
164        mode: str = "batch",
165        check_dict: bool = True,
166    ) -> None:
167        if n_nonzero_coefs < 1:
168            raise ValueError("n_nonzero_coefs must be >= 1.")
169        if mode not in ("batch", "cholesky"):
170            raise ValueError("mode must be 'batch' or 'cholesky'.")
171        self.n_nonzero_coefs = n_nonzero_coefs
172        self.mode = mode
173        self.check_dict = check_dict
174
175    def encode(
176        self,
177        X: np.ndarray,
178        D: np.ndarray,
179        G: np.ndarray | None = None,
180    ) -> np.ndarray:
181        """
182        Compute sparse codes for each column of X.
183
184        Parameters
185        ----------
186        X : np.ndarray, shape (n_features, n_samples)
187        D : np.ndarray, shape (n_features, n_atoms)
188        G : np.ndarray or None, shape (n_atoms, n_atoms)
189            Precomputed Gram matrix D.T @ D.  Required for 'batch' mode;
190            computed internally if not supplied.
191
192        Returns
193        -------
194        Gamma : np.ndarray, shape (n_atoms, n_samples)
195        """
196        X = np.asarray(X, dtype=float)
197        D = np.asarray(D, dtype=float)
198
199        if X.ndim == 1:
200            X = X[:, np.newaxis]
201
202        if self.check_dict:
203            _check_dict_normalized(D)
204
205        T = self.n_nonzero_coefs
206
207        if self.mode == "batch":
208            if G is None:
209                G = D.T @ D
210            DtX = D.T @ X
211            return batch_omp(DtX, G, T)
212
213        # cholesky mode — signal by signal
214        n_atoms = D.shape[1]
215        n_samples = X.shape[1]
216        Gamma = np.zeros((n_atoms, n_samples))
217        for i in range(n_samples):
218            Gamma[:, i] = omp_cholesky(D, X[:, i], T)
219        return Gamma

Orthogonal Matching Pursuit sparse coder.

Parameters

n_nonzero_coefs : int Target sparsity — maximum number of non-zero coefficients per signal. mode : {'batch', 'cholesky'} Implementation variant. 'batch' — Batch-OMP; requires the full Gram matrix G = D'D. Fastest when encoding many signals at once. 'cholesky' — Single-signal OMP-Cholesky; lower memory footprint. check_dict : bool Whether to verify that dictionary atoms are unit-norm (default True).

OMP(n_nonzero_coefs: int, mode: str = 'batch', check_dict: bool = True)
161    def __init__(
162        self,
163        n_nonzero_coefs: int,
164        mode: str = "batch",
165        check_dict: bool = True,
166    ) -> None:
167        if n_nonzero_coefs < 1:
168            raise ValueError("n_nonzero_coefs must be >= 1.")
169        if mode not in ("batch", "cholesky"):
170            raise ValueError("mode must be 'batch' or 'cholesky'.")
171        self.n_nonzero_coefs = n_nonzero_coefs
172        self.mode = mode
173        self.check_dict = check_dict
n_nonzero_coefs
mode
check_dict
def encode( self, X: numpy.ndarray, D: numpy.ndarray, G: numpy.ndarray | None = None) -> numpy.ndarray:
175    def encode(
176        self,
177        X: np.ndarray,
178        D: np.ndarray,
179        G: np.ndarray | None = None,
180    ) -> np.ndarray:
181        """
182        Compute sparse codes for each column of X.
183
184        Parameters
185        ----------
186        X : np.ndarray, shape (n_features, n_samples)
187        D : np.ndarray, shape (n_features, n_atoms)
188        G : np.ndarray or None, shape (n_atoms, n_atoms)
189            Precomputed Gram matrix D.T @ D.  Required for 'batch' mode;
190            computed internally if not supplied.
191
192        Returns
193        -------
194        Gamma : np.ndarray, shape (n_atoms, n_samples)
195        """
196        X = np.asarray(X, dtype=float)
197        D = np.asarray(D, dtype=float)
198
199        if X.ndim == 1:
200            X = X[:, np.newaxis]
201
202        if self.check_dict:
203            _check_dict_normalized(D)
204
205        T = self.n_nonzero_coefs
206
207        if self.mode == "batch":
208            if G is None:
209                G = D.T @ D
210            DtX = D.T @ X
211            return batch_omp(DtX, G, T)
212
213        # cholesky mode — signal by signal
214        n_atoms = D.shape[1]
215        n_samples = X.shape[1]
216        Gamma = np.zeros((n_atoms, n_samples))
217        for i in range(n_samples):
218            Gamma[:, i] = omp_cholesky(D, X[:, i], T)
219        return Gamma

Compute sparse codes for each column of X.

Parameters

X : np.ndarray, shape (n_features, n_samples) D : np.ndarray, shape (n_features, n_atoms) G : np.ndarray or None, shape (n_atoms, n_atoms) Precomputed Gram matrix D.T @ D. Required for 'batch' mode; computed internally if not supplied.

Returns

Gamma : np.ndarray, shape (n_atoms, n_samples)

class KSVD(reppi.base.BaseDictionaryLearner):
 26class KSVD(BaseDictionaryLearner):
 27    """
 28    K-SVD dictionary learner.
 29
 30    Alternates between:
 31      1. Sparse coding — encode each training signal over the current D.
 32      2. Dictionary update — update each atom (and its coefficients) via a
 33         rank-1 approximation of the residual matrix.
 34
 35    Parameters
 36    ----------
 37    n_components : int
 38        Number of dictionary atoms to learn.
 39    n_nonzero_coefs : int
 40        Sparsity target T: each signal is represented with at most T atoms.
 41    n_iter : int
 42        Number of K-SVD iterations (default 10).
 43    exact_svd : bool
 44        If True, use full SVD for the atom update (exact K-SVD).
 45        If False (default), use the faster approximate update.
 46    mu_thresh : float
 47        Mutual-incoherence threshold in (0, 1].  Atoms whose pairwise
 48        correlation exceeds this value are replaced.  Set to 1.0 to
 49        disable (default 0.99).
 50    mem_usage : str
 51        One of 'high', 'normal' (default), 'low'.
 52        Controls whether G = D'D (and DtX = D'X) are precomputed.
 53    random_state : int or None
 54        Seed for reproducible atom initialisation.
 55    verbose : bool
 56        Print iteration progress (default False).
 57    """
 58
 59    def __init__(
 60        self,
 61        n_components: int,
 62        n_nonzero_coefs: int,
 63        n_iter: int = 10,
 64        exact_svd: bool = False,
 65        mu_thresh: float = 0.99,
 66        mem_usage: str = "normal",
 67        random_state: int | None = None,
 68        verbose: bool = False,
 69    ) -> None:
 70        if mem_usage not in ("high", "normal", "low"):
 71            raise ValueError("mem_usage must be 'high', 'normal', or 'low'.")
 72        self.n_components = n_components
 73        self.n_nonzero_coefs = n_nonzero_coefs
 74        self.n_iter = n_iter
 75        self.exact_svd = exact_svd
 76        self.mu_thresh = mu_thresh
 77        self.mem_usage = mem_usage
 78        self.random_state = random_state
 79        self.verbose = verbose
 80
 81        # Set after fit
 82        self.D_: np.ndarray | None = None
 83        self.errors_: list[float] = []
 84
 85    # ------------------------------------------------------------------
 86    # Public API
 87    # ------------------------------------------------------------------
 88
 89    def fit(self, X: np.ndarray, D_init: np.ndarray | None = None) -> "KSVD":
 90        """
 91        Learn a dictionary from training signals.
 92
 93        Parameters
 94        ----------
 95        X : np.ndarray, shape (n_features, n_samples)
 96        D_init : np.ndarray or None, shape (n_features, n_components)
 97            Optional initial dictionary.  If None, random training signals
 98            are chosen as initial atoms.
 99
100        Returns
101        -------
102        self
103        """
104        X = np.asarray(X, dtype=float)
105        rng = np.random.default_rng(self.random_state)
106
107        D = self._init_dict(X, D_init, rng)
108        self.errors_ = []
109
110        for it in range(self.n_iter):
111            G = D.T @ D if self.mem_usage in ("high", "normal") else None
112            Gamma = self._sparse_code(X, D, G)
113
114            unused = np.arange(X.shape[1])
115            replaced = np.zeros(self.n_components, dtype=bool)
116
117            for j in range(self.n_components):
118                D[:, j], gamma_j, idx, unused, replaced = _optimize_atom(
119                    X, D, j, Gamma, unused, replaced, self.exact_svd
120                )
121                Gamma[j, idx] = gamma_j
122
123            err = float(np.sqrt(rep_error_squared(X, D, Gamma).sum() / X.size))
124            self.errors_.append(err)
125
126            D, _ = _clear_dict(D, Gamma, X, self.mu_thresh, unused, replaced)
127
128            if self.verbose:
129                print(f"Iter {it + 1}/{self.n_iter}  RMSE={err:.6f}")
130
131        self.D_ = D
132        return self
133
134    def transform(self, X: np.ndarray) -> np.ndarray:
135        """Encode X using the learned dictionary."""
136        if self.D_ is None:
137            raise DictionaryLearningError("Call fit() before transform().")
138        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
139        return coder.encode(X, self.D_)
140
141    # ------------------------------------------------------------------
142    # Internal helpers
143    # ------------------------------------------------------------------
144
145    def _init_dict(
146        self,
147        X: np.ndarray,
148        D_init: np.ndarray | None,
149        rng: np.random.Generator,
150    ) -> np.ndarray:
151        n_features, n_samples = X.shape
152        k = self.n_components
153
154        if D_init is not None:
155            D = np.asarray(D_init, dtype=float)
156            if D.shape != (n_features, k):
157                raise DictionaryLearningError(
158                    f"D_init shape {D.shape} does not match "
159                    f"(n_features={n_features}, n_components={k})."
160                )
161        else:
162            valid = np.where(col_norms_squared(X) > 1e-6)[0]
163            if len(valid) < k:
164                raise DictionaryLearningError(
165                    "Not enough non-zero training signals to initialise the dictionary."
166                )
167            chosen = rng.choice(valid, size=k, replace=False)
168            D = X[:, chosen].copy()
169
170        return normalize_columns(D)
171
172    def _sparse_code(
173        self,
174        X: np.ndarray,
175        D: np.ndarray,
176        G: np.ndarray | None,
177    ) -> np.ndarray:
178        if self.mem_usage == "high" and G is not None:
179            return batch_omp(D.T @ X, G, self.n_nonzero_coefs)
180        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
181        return coder.encode(X, D, G=G)

K-SVD dictionary learner.

Alternates between:

  1. Sparse coding — encode each training signal over the current D.
  2. Dictionary update — update each atom (and its coefficients) via a rank-1 approximation of the residual matrix.

Parameters

n_components : int Number of dictionary atoms to learn. n_nonzero_coefs : int Sparsity target T: each signal is represented with at most T atoms. n_iter : int Number of K-SVD iterations (default 10). exact_svd : bool If True, use full SVD for the atom update (exact K-SVD). If False (default), use the faster approximate update. mu_thresh : float Mutual-incoherence threshold in (0, 1]. Atoms whose pairwise correlation exceeds this value are replaced. Set to 1.0 to disable (default 0.99). mem_usage : str One of 'high', 'normal' (default), 'low'. Controls whether G = D'D (and DtX = D'X) are precomputed. random_state : int or None Seed for reproducible atom initialisation. verbose : bool Print iteration progress (default False).

KSVD( n_components: int, n_nonzero_coefs: int, n_iter: int = 10, exact_svd: bool = False, mu_thresh: float = 0.99, mem_usage: str = 'normal', random_state: int | None = None, verbose: bool = False)
59    def __init__(
60        self,
61        n_components: int,
62        n_nonzero_coefs: int,
63        n_iter: int = 10,
64        exact_svd: bool = False,
65        mu_thresh: float = 0.99,
66        mem_usage: str = "normal",
67        random_state: int | None = None,
68        verbose: bool = False,
69    ) -> None:
70        if mem_usage not in ("high", "normal", "low"):
71            raise ValueError("mem_usage must be 'high', 'normal', or 'low'.")
72        self.n_components = n_components
73        self.n_nonzero_coefs = n_nonzero_coefs
74        self.n_iter = n_iter
75        self.exact_svd = exact_svd
76        self.mu_thresh = mu_thresh
77        self.mem_usage = mem_usage
78        self.random_state = random_state
79        self.verbose = verbose
80
81        # Set after fit
82        self.D_: np.ndarray | None = None
83        self.errors_: list[float] = []
n_components
n_nonzero_coefs
n_iter
exact_svd
mu_thresh
mem_usage
random_state
verbose
D_: numpy.ndarray | None
errors_: list[float]
def fit( self, X: numpy.ndarray, D_init: numpy.ndarray | None = None) -> KSVD:
 89    def fit(self, X: np.ndarray, D_init: np.ndarray | None = None) -> "KSVD":
 90        """
 91        Learn a dictionary from training signals.
 92
 93        Parameters
 94        ----------
 95        X : np.ndarray, shape (n_features, n_samples)
 96        D_init : np.ndarray or None, shape (n_features, n_components)
 97            Optional initial dictionary.  If None, random training signals
 98            are chosen as initial atoms.
 99
100        Returns
101        -------
102        self
103        """
104        X = np.asarray(X, dtype=float)
105        rng = np.random.default_rng(self.random_state)
106
107        D = self._init_dict(X, D_init, rng)
108        self.errors_ = []
109
110        for it in range(self.n_iter):
111            G = D.T @ D if self.mem_usage in ("high", "normal") else None
112            Gamma = self._sparse_code(X, D, G)
113
114            unused = np.arange(X.shape[1])
115            replaced = np.zeros(self.n_components, dtype=bool)
116
117            for j in range(self.n_components):
118                D[:, j], gamma_j, idx, unused, replaced = _optimize_atom(
119                    X, D, j, Gamma, unused, replaced, self.exact_svd
120                )
121                Gamma[j, idx] = gamma_j
122
123            err = float(np.sqrt(rep_error_squared(X, D, Gamma).sum() / X.size))
124            self.errors_.append(err)
125
126            D, _ = _clear_dict(D, Gamma, X, self.mu_thresh, unused, replaced)
127
128            if self.verbose:
129                print(f"Iter {it + 1}/{self.n_iter}  RMSE={err:.6f}")
130
131        self.D_ = D
132        return self

Learn a dictionary from training signals.

Parameters

X : np.ndarray, shape (n_features, n_samples) D_init : np.ndarray or None, shape (n_features, n_components) Optional initial dictionary. If None, random training signals are chosen as initial atoms.

Returns

self

def transform(self, X: numpy.ndarray) -> numpy.ndarray:
134    def transform(self, X: np.ndarray) -> np.ndarray:
135        """Encode X using the learned dictionary."""
136        if self.D_ is None:
137            raise DictionaryLearningError("Call fit() before transform().")
138        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
139        return coder.encode(X, self.D_)

Encode X using the learned dictionary.

class LCKSVD(reppi.base.BaseDiscriminativeDictionaryLearner):
239class LCKSVD(BaseDiscriminativeDictionaryLearner):
240    """
241    Label Consistent K-SVD dictionary learner (LC-KSVD1 and LC-KSVD2).
242
243    Parameters
244    ----------
245    n_components : int
246        Number of dictionary atoms.
247    n_nonzero_coefs : int
248        Sparsity level T.
249    alpha : float
250        Weight for the label-consistency term (sqrt_alpha in the paper).
251    beta : float
252        Weight for the classifier term (sqrt_beta; LC-KSVD2 only).
253    variant : {'lcksvd1', 'lcksvd2'}
254        Which variant to train.
255    n_iter : int
256        Number of LC-KSVD iterations (default 50).
257    n_iter_init : int
258        K-SVD iterations for the initialisation phase (default 20).
259    exact_svd : bool
260        Use exact SVD in the atom-update step (slower but slightly better).
261    mu_thresh : float
262        Mutual-incoherence threshold (default 0.99).
263    random_state : int or None
264    verbose : bool
265
266    Attributes
267    ----------
268    D_ : np.ndarray, shape (n_features, n_components)
269        Learned dictionary.
270    W_ : np.ndarray, shape (n_classes, n_components)
271        Learned linear classifier weights.
272    A_ : np.ndarray, shape (n_components, n_components)
273        Learned label-consistency transform.
274    errors_ : list of float
275        Per-iteration RMSE on training data.
276    """
277
278    def __init__(
279        self,
280        n_components: int,
281        n_nonzero_coefs: int,
282        alpha: float = 4.0,
283        beta: float = 2.0,
284        variant: str = "lcksvd2",
285        n_iter: int = 50,
286        n_iter_init: int = 20,
287        exact_svd: bool = False,
288        mu_thresh: float = 0.99,
289        random_state: int | None = None,
290        verbose: bool = False,
291    ) -> None:
292        if variant not in ("lcksvd1", "lcksvd2"):
293            raise ValueError("variant must be 'lcksvd1' or 'lcksvd2'.")
294        self.n_components = n_components
295        self.n_nonzero_coefs = n_nonzero_coefs
296        self.alpha = alpha
297        self.beta = beta
298        self.variant = variant
299        self.n_iter = n_iter
300        self.n_iter_init = n_iter_init
301        self.exact_svd = exact_svd
302        self.mu_thresh = mu_thresh
303        self.random_state = random_state
304        self.verbose = verbose
305
306        self.D_: np.ndarray | None = None
307        self.W_: np.ndarray | None = None
308        self.A_: np.ndarray | None = None
309        self.errors_: list[float] = []
310        self.class_boundaries_: dict[int, tuple[int, int]] | None = None
311
312    # ------------------------------------------------------------------
313    # Public API
314    # ------------------------------------------------------------------
315
316    def fit(
317        self,
318        X: np.ndarray,
319        H: np.ndarray,
320        D_init: np.ndarray | None = None,
321        A_init: np.ndarray | None = None,
322        W_init: np.ndarray | None = None,
323        Q: np.ndarray | None = None,
324    ) -> "LCKSVD":
325        """
326        Learn a discriminative dictionary from labelled training data.
327
328        Parameters
329        ----------
330        X : np.ndarray, shape (n_features, n_samples)
331            Training signals.
332        H : np.ndarray, shape (n_classes, n_samples)
333            One-hot label matrix.
334        D_init : np.ndarray or None
335            Initial dictionary. If None, a K-SVD initialisation is run.
336        A_init : np.ndarray or None
337            Initial label-consistency transform.
338        W_init : np.ndarray or None
339            Initial classifier weights (required / used for LC-KSVD2).
340        Q : np.ndarray or None
341            Label-consistent target matrix. Computed from H if None.
342
343        Returns
344        -------
345        self
346        """
347        X = np.asarray(X, dtype=float)
348        H = np.asarray(H, dtype=float)
349        n_features, n_samples = X.shape
350        n_classes = H.shape[0]
351
352        # ---- Initialisation ----
353        if D_init is None or A_init is None or W_init is None or Q is None:
354            if self.verbose:
355                print("Running initialisation K-SVD...")
356            D_init, A_init, W_init, Q = initialization4lcksvd(
357                X, H,
358                self.n_components,
359                self.n_iter_init,
360                self.n_nonzero_coefs,
361                random_state=self.random_state,
362                verbose=self.verbose,
363            )
364
365        D = normalize_columns(D_init.copy())
366        A = A_init.copy()
367        W = W_init.copy()
368
369        sqrt_alpha = self.alpha
370        sqrt_beta = self.beta
371
372        use_classifier_term = (self.variant == "lcksvd2")
373
374        # ---- Build augmented training data ----
375        # Y_aug = [X ; sqrt_alpha*Q ; sqrt_beta*H]  (LC-KSVD2)
376        # Y_aug = [X ; sqrt_alpha*Q]                 (LC-KSVD1)
377        H_aug = H if use_classifier_term else None
378        X_aug, _, _ = _augment_data(X, Q, H_aug, sqrt_alpha, sqrt_beta)
379
380        self.errors_ = []
381
382        for it in range(self.n_iter):
383
384            # ---- Build augmented dictionary ----
385            # D_aug = [D ; sqrt_alpha*A ; sqrt_beta*W]
386            D_aug = self._build_aug_dict(D, A, W, sqrt_alpha, sqrt_beta, use_classifier_term)
387            D_aug_norm = normalize_columns(D_aug)
388
389            # ---- Sparse coding on augmented system ----
390            G_aug = D_aug_norm.T @ D_aug_norm
391            Gamma = batch_omp(D_aug_norm.T @ X_aug, G_aug, self.n_nonzero_coefs)
392
393            # ---- Dictionary update (on original data only) ----
394            # We update D, A (and W for LC-KSVD2) jointly via the
395            # augmented residual, but evaluate coherence/usage on original X.
396            unused = np.arange(n_samples)
397            replaced = np.zeros(self.n_components, dtype=bool)
398
399            for j in range(self.n_components):
400                D_aug_norm[:, j], gamma_j, idx, unused, replaced = _optimize_atom(
401                    X_aug, D_aug_norm, j, Gamma, unused, replaced, self.exact_svd
402                )
403                Gamma[j, idx] = gamma_j
404
405            # De-augment: extract D, A, W from D_aug_norm
406            D, A, W = self._split_aug_dict(
407                D_aug_norm, n_features, n_classes, sqrt_alpha, sqrt_beta, use_classifier_term
408            )
409            D = normalize_columns(D)
410
411            # ---- Update classifier W (LC-KSVD2) via least squares ----
412            if use_classifier_term:
413                W = H @ np.linalg.pinv(Gamma)
414
415            # ---- Update A via least squares ----
416            A = Q @ np.linalg.pinv(Gamma)
417
418            # ---- Clear incoherent / rarely-used atoms ----
419            # Rebuild normalised augmented dict for coherence checking
420            D_aug_rebuilt = self._build_aug_dict(D, A, W, sqrt_alpha, sqrt_beta, use_classifier_term)
421            D_aug_rebuilt = normalize_columns(D_aug_rebuilt)
422            D_aug_rebuilt, _ = _clear_dict(
423                D_aug_rebuilt, Gamma, X_aug, self.mu_thresh,
424                unused, replaced
425            )
426            D, A, W = self._split_aug_dict(
427                D_aug_rebuilt, n_features, n_classes, sqrt_alpha, sqrt_beta, use_classifier_term
428            )
429            D = normalize_columns(D)
430
431            # ---- Track RMSE on original X ----
432            err = float(np.sqrt(rep_error_squared(X, D, Gamma).sum() / X.size))
433            self.errors_.append(err)
434
435            if self.verbose:
436                print(f"[{self.variant.upper()}] Iter {it + 1}/{self.n_iter}  RMSE={err:.6f}")
437
438        self.D_ = D
439        self.A_ = A
440        self.W_ = W
441
442        # Record per-class atom ranges matching _build_label_consistent_target
443        n_classes = H.shape[0]
444        atoms_per_class = self.n_components // n_classes
445        boundaries: dict[int, tuple[int, int]] = {}
446        for c in range(n_classes):
447            start = c * atoms_per_class
448            end = start + atoms_per_class if c < n_classes - 1 else self.n_components
449            boundaries[c] = (start, end)
450        self.class_boundaries_ = boundaries
451
452        return self
453
454    def transform(self, X: np.ndarray) -> np.ndarray:
455        """
456        Encode X using the learned dictionary D.
457
458        Parameters
459        ----------
460        X : np.ndarray, shape (n_features, n_samples)
461
462        Returns
463        -------
464        Gamma : np.ndarray, shape (n_components, n_samples)
465        """
466        self._check_fitted()
467        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
468        return coder.encode(X, self.D_)
469
470    def predict(self, X: np.ndarray) -> np.ndarray:
471        """
472        Classify test signals using the learned classifier W.
473
474        The predicted class for each signal is the argmax of W @ gamma.
475
476        Parameters
477        ----------
478        X : np.ndarray, shape (n_features, n_samples)
479
480        Returns
481        -------
482        labels : np.ndarray, shape (n_samples,)  integer class indices
483        """
484        self._check_fitted()
485        if self.W_ is None:
486            raise DictionaryLearningError(
487                "Classifier W is not available. "
488                "Use variant='lcksvd2' or access sparse codes via transform()."
489            )
490        Gamma = self.transform(X)
491        scores = self.W_ @ Gamma          # (n_classes, n_samples)
492        return np.argmax(scores, axis=0)
493
494    def score(self, X: np.ndarray, H: np.ndarray) -> float:
495        """
496        Classification accuracy on (X, H).
497
498        Parameters
499        ----------
500        X : np.ndarray, shape (n_features, n_samples)
501        H : np.ndarray, shape (n_classes, n_samples) — one-hot labels
502
503        Returns
504        -------
505        accuracy : float in [0, 1]
506        """
507        true_labels = np.argmax(H, axis=0)
508        pred_labels = self.predict(X)
509        return float(np.mean(pred_labels == true_labels))
510
511    @staticmethod
512    def _build_aug_dict(
513        D: np.ndarray,
514        A: np.ndarray,
515        W: np.ndarray,
516        sqrt_alpha: float,
517        sqrt_beta: float,
518        use_classifier: bool,
519    ) -> np.ndarray:
520        """Stack [D ; sqrt_alpha*A ; (sqrt_beta*W)]."""
521        parts = [D, sqrt_alpha * A]
522        if use_classifier:
523            parts.append(sqrt_beta * W)
524        return np.vstack(parts)
525
526    @staticmethod
527    def _split_aug_dict(
528        D_aug: np.ndarray,
529        n_features: int,
530        n_classes: int,
531        sqrt_alpha: float,
532        sqrt_beta: float,
533        use_classifier: bool,
534    ) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
535        """
536        Recover (D, A, W) from the augmented dictionary D_aug.
537
538        D_aug rows are: n_features | n_components | (n_classes if lcksvd2).
539        """
540        n_components = D_aug.shape[1]
541        D = D_aug[:n_features, :]
542        A_rows = n_components
543        A = D_aug[n_features: n_features + A_rows, :] / max(sqrt_alpha, 1e-14)
544        if use_classifier:
545            W = D_aug[n_features + A_rows:, :] / max(sqrt_beta, 1e-14)
546        else:
547            W = np.zeros((n_classes, n_components))
548        return D, A, W

Label Consistent K-SVD dictionary learner (LC-KSVD1 and LC-KSVD2).

Parameters

n_components : int Number of dictionary atoms. n_nonzero_coefs : int Sparsity level T. alpha : float Weight for the label-consistency term (sqrt_alpha in the paper). beta : float Weight for the classifier term (sqrt_beta; LC-KSVD2 only). variant : {'lcksvd1', 'lcksvd2'} Which variant to train. n_iter : int Number of LC-KSVD iterations (default 50). n_iter_init : int K-SVD iterations for the initialisation phase (default 20). exact_svd : bool Use exact SVD in the atom-update step (slower but slightly better). mu_thresh : float Mutual-incoherence threshold (default 0.99). random_state : int or None verbose : bool

Attributes

D_ : np.ndarray, shape (n_features, n_components) Learned dictionary. W_ : np.ndarray, shape (n_classes, n_components) Learned linear classifier weights. A_ : np.ndarray, shape (n_components, n_components) Learned label-consistency transform. errors_ : list of float Per-iteration RMSE on training data.

LCKSVD( n_components: int, n_nonzero_coefs: int, alpha: float = 4.0, beta: float = 2.0, variant: str = 'lcksvd2', n_iter: int = 50, n_iter_init: int = 20, exact_svd: bool = False, mu_thresh: float = 0.99, random_state: int | None = None, verbose: bool = False)
278    def __init__(
279        self,
280        n_components: int,
281        n_nonzero_coefs: int,
282        alpha: float = 4.0,
283        beta: float = 2.0,
284        variant: str = "lcksvd2",
285        n_iter: int = 50,
286        n_iter_init: int = 20,
287        exact_svd: bool = False,
288        mu_thresh: float = 0.99,
289        random_state: int | None = None,
290        verbose: bool = False,
291    ) -> None:
292        if variant not in ("lcksvd1", "lcksvd2"):
293            raise ValueError("variant must be 'lcksvd1' or 'lcksvd2'.")
294        self.n_components = n_components
295        self.n_nonzero_coefs = n_nonzero_coefs
296        self.alpha = alpha
297        self.beta = beta
298        self.variant = variant
299        self.n_iter = n_iter
300        self.n_iter_init = n_iter_init
301        self.exact_svd = exact_svd
302        self.mu_thresh = mu_thresh
303        self.random_state = random_state
304        self.verbose = verbose
305
306        self.D_: np.ndarray | None = None
307        self.W_: np.ndarray | None = None
308        self.A_: np.ndarray | None = None
309        self.errors_: list[float] = []
310        self.class_boundaries_: dict[int, tuple[int, int]] | None = None
n_components
n_nonzero_coefs
alpha
beta
variant
n_iter
n_iter_init
exact_svd
mu_thresh
random_state
verbose
D_: numpy.ndarray | None
W_: numpy.ndarray | None
A_: numpy.ndarray | None
errors_: list[float]
class_boundaries_: dict[int, tuple[int, int]] | None
def fit( self, X: numpy.ndarray, H: numpy.ndarray, D_init: numpy.ndarray | None = None, A_init: numpy.ndarray | None = None, W_init: numpy.ndarray | None = None, Q: numpy.ndarray | None = None) -> LCKSVD:
316    def fit(
317        self,
318        X: np.ndarray,
319        H: np.ndarray,
320        D_init: np.ndarray | None = None,
321        A_init: np.ndarray | None = None,
322        W_init: np.ndarray | None = None,
323        Q: np.ndarray | None = None,
324    ) -> "LCKSVD":
325        """
326        Learn a discriminative dictionary from labelled training data.
327
328        Parameters
329        ----------
330        X : np.ndarray, shape (n_features, n_samples)
331            Training signals.
332        H : np.ndarray, shape (n_classes, n_samples)
333            One-hot label matrix.
334        D_init : np.ndarray or None
335            Initial dictionary. If None, a K-SVD initialisation is run.
336        A_init : np.ndarray or None
337            Initial label-consistency transform.
338        W_init : np.ndarray or None
339            Initial classifier weights (required / used for LC-KSVD2).
340        Q : np.ndarray or None
341            Label-consistent target matrix. Computed from H if None.
342
343        Returns
344        -------
345        self
346        """
347        X = np.asarray(X, dtype=float)
348        H = np.asarray(H, dtype=float)
349        n_features, n_samples = X.shape
350        n_classes = H.shape[0]
351
352        # ---- Initialisation ----
353        if D_init is None or A_init is None or W_init is None or Q is None:
354            if self.verbose:
355                print("Running initialisation K-SVD...")
356            D_init, A_init, W_init, Q = initialization4lcksvd(
357                X, H,
358                self.n_components,
359                self.n_iter_init,
360                self.n_nonzero_coefs,
361                random_state=self.random_state,
362                verbose=self.verbose,
363            )
364
365        D = normalize_columns(D_init.copy())
366        A = A_init.copy()
367        W = W_init.copy()
368
369        sqrt_alpha = self.alpha
370        sqrt_beta = self.beta
371
372        use_classifier_term = (self.variant == "lcksvd2")
373
374        # ---- Build augmented training data ----
375        # Y_aug = [X ; sqrt_alpha*Q ; sqrt_beta*H]  (LC-KSVD2)
376        # Y_aug = [X ; sqrt_alpha*Q]                 (LC-KSVD1)
377        H_aug = H if use_classifier_term else None
378        X_aug, _, _ = _augment_data(X, Q, H_aug, sqrt_alpha, sqrt_beta)
379
380        self.errors_ = []
381
382        for it in range(self.n_iter):
383
384            # ---- Build augmented dictionary ----
385            # D_aug = [D ; sqrt_alpha*A ; sqrt_beta*W]
386            D_aug = self._build_aug_dict(D, A, W, sqrt_alpha, sqrt_beta, use_classifier_term)
387            D_aug_norm = normalize_columns(D_aug)
388
389            # ---- Sparse coding on augmented system ----
390            G_aug = D_aug_norm.T @ D_aug_norm
391            Gamma = batch_omp(D_aug_norm.T @ X_aug, G_aug, self.n_nonzero_coefs)
392
393            # ---- Dictionary update (on original data only) ----
394            # We update D, A (and W for LC-KSVD2) jointly via the
395            # augmented residual, but evaluate coherence/usage on original X.
396            unused = np.arange(n_samples)
397            replaced = np.zeros(self.n_components, dtype=bool)
398
399            for j in range(self.n_components):
400                D_aug_norm[:, j], gamma_j, idx, unused, replaced = _optimize_atom(
401                    X_aug, D_aug_norm, j, Gamma, unused, replaced, self.exact_svd
402                )
403                Gamma[j, idx] = gamma_j
404
405            # De-augment: extract D, A, W from D_aug_norm
406            D, A, W = self._split_aug_dict(
407                D_aug_norm, n_features, n_classes, sqrt_alpha, sqrt_beta, use_classifier_term
408            )
409            D = normalize_columns(D)
410
411            # ---- Update classifier W (LC-KSVD2) via least squares ----
412            if use_classifier_term:
413                W = H @ np.linalg.pinv(Gamma)
414
415            # ---- Update A via least squares ----
416            A = Q @ np.linalg.pinv(Gamma)
417
418            # ---- Clear incoherent / rarely-used atoms ----
419            # Rebuild normalised augmented dict for coherence checking
420            D_aug_rebuilt = self._build_aug_dict(D, A, W, sqrt_alpha, sqrt_beta, use_classifier_term)
421            D_aug_rebuilt = normalize_columns(D_aug_rebuilt)
422            D_aug_rebuilt, _ = _clear_dict(
423                D_aug_rebuilt, Gamma, X_aug, self.mu_thresh,
424                unused, replaced
425            )
426            D, A, W = self._split_aug_dict(
427                D_aug_rebuilt, n_features, n_classes, sqrt_alpha, sqrt_beta, use_classifier_term
428            )
429            D = normalize_columns(D)
430
431            # ---- Track RMSE on original X ----
432            err = float(np.sqrt(rep_error_squared(X, D, Gamma).sum() / X.size))
433            self.errors_.append(err)
434
435            if self.verbose:
436                print(f"[{self.variant.upper()}] Iter {it + 1}/{self.n_iter}  RMSE={err:.6f}")
437
438        self.D_ = D
439        self.A_ = A
440        self.W_ = W
441
442        # Record per-class atom ranges matching _build_label_consistent_target
443        n_classes = H.shape[0]
444        atoms_per_class = self.n_components // n_classes
445        boundaries: dict[int, tuple[int, int]] = {}
446        for c in range(n_classes):
447            start = c * atoms_per_class
448            end = start + atoms_per_class if c < n_classes - 1 else self.n_components
449            boundaries[c] = (start, end)
450        self.class_boundaries_ = boundaries
451
452        return self

Learn a discriminative dictionary from labelled training data.

Parameters

X : np.ndarray, shape (n_features, n_samples) Training signals. H : np.ndarray, shape (n_classes, n_samples) One-hot label matrix. D_init : np.ndarray or None Initial dictionary. If None, a K-SVD initialisation is run. A_init : np.ndarray or None Initial label-consistency transform. W_init : np.ndarray or None Initial classifier weights (required / used for LC-KSVD2). Q : np.ndarray or None Label-consistent target matrix. Computed from H if None.

Returns

self

def transform(self, X: numpy.ndarray) -> numpy.ndarray:
454    def transform(self, X: np.ndarray) -> np.ndarray:
455        """
456        Encode X using the learned dictionary D.
457
458        Parameters
459        ----------
460        X : np.ndarray, shape (n_features, n_samples)
461
462        Returns
463        -------
464        Gamma : np.ndarray, shape (n_components, n_samples)
465        """
466        self._check_fitted()
467        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
468        return coder.encode(X, self.D_)

Encode X using the learned dictionary D.

Parameters

X : np.ndarray, shape (n_features, n_samples)

Returns

Gamma : np.ndarray, shape (n_components, n_samples)

def predict(self, X: numpy.ndarray) -> numpy.ndarray:
470    def predict(self, X: np.ndarray) -> np.ndarray:
471        """
472        Classify test signals using the learned classifier W.
473
474        The predicted class for each signal is the argmax of W @ gamma.
475
476        Parameters
477        ----------
478        X : np.ndarray, shape (n_features, n_samples)
479
480        Returns
481        -------
482        labels : np.ndarray, shape (n_samples,)  integer class indices
483        """
484        self._check_fitted()
485        if self.W_ is None:
486            raise DictionaryLearningError(
487                "Classifier W is not available. "
488                "Use variant='lcksvd2' or access sparse codes via transform()."
489            )
490        Gamma = self.transform(X)
491        scores = self.W_ @ Gamma          # (n_classes, n_samples)
492        return np.argmax(scores, axis=0)

Classify test signals using the learned classifier W.

The predicted class for each signal is the argmax of W @ gamma.

Parameters

X : np.ndarray, shape (n_features, n_samples)

Returns

labels : np.ndarray, shape (n_samples,) integer class indices

def score(self, X: numpy.ndarray, H: numpy.ndarray) -> float:
494    def score(self, X: np.ndarray, H: np.ndarray) -> float:
495        """
496        Classification accuracy on (X, H).
497
498        Parameters
499        ----------
500        X : np.ndarray, shape (n_features, n_samples)
501        H : np.ndarray, shape (n_classes, n_samples) — one-hot labels
502
503        Returns
504        -------
505        accuracy : float in [0, 1]
506        """
507        true_labels = np.argmax(H, axis=0)
508        pred_labels = self.predict(X)
509        return float(np.mean(pred_labels == true_labels))

Classification accuracy on (X, H).

Parameters

X : np.ndarray, shape (n_features, n_samples) H : np.ndarray, shape (n_classes, n_samples) — one-hot labels

Returns

accuracy : float in [0, 1]

class FrozenDictionaryLearner:
132class FrozenDictionaryLearner:
133    """
134    Learn a residual dictionary D_active given a frozen dictionary D_frozen.
135
136    The learner encodes all signals over D_frozen first, then trains
137    D_active on the reconstruction residual.  The combined dictionary
138
139        D_combined = [ D_frozen | D_active ]
140
141    is exposed as ``D_combined_`` after fitting.
142
143    This class handles a single residual learning step.  For the full
144    sequential pipeline, see ``IncrementalFrozenDictionary``.
145
146    Parameters
147    ----------
148    D_frozen : np.ndarray, shape (n_features, n_frozen_atoms)
149        Pre-trained frozen dictionary.  Never modified.
150    learner_class : type[BaseDiscriminativeDictionaryLearner]
151        Discriminative dictionary learning class to use for the residual.
152    learner_kwargs : dict
153        Keyword arguments forwarded to ``learner_class.__init__``.
154    n_nonzero_coefs : int
155        Sparsity level used when encoding over the frozen dictionary to
156        compute the residual, and when encoding over the combined dictionary
157        for downstream tasks.
158    learn_on_residual : bool
159        If True (default), train D_active on the reconstruction residual
160        R = X - D_frozen @ Gamma_frozen.  If False, train on the original
161        X — useful when the frozen dict is very small and you want the
162        active dict to model the full signal, not just what D_frozen misses.
163    refit_classifier : bool
164        If True (default), re-learn W over the full combined dictionary
165        after fitting D_active.
166
167    Attributes
168    ----------
169    D_combined_ : np.ndarray, shape (n_features, n_frozen + n_active)
170    W_          : np.ndarray, shape (n_classes, n_frozen + n_active)
171    learner_    : fitted instance of ``learner_class``
172    n_frozen_   : int  number of frozen atoms
173    n_active_   : int  number of active (residual) atoms
174    class_boundaries_ : dict[int, tuple[int, int]]
175        Atom ranges for each class in the *combined* dictionary, merging
176        frozen boundaries (if any) with the active learner's boundaries.
177    """
178
179    def __init__(
180        self,
181        D_frozen: np.ndarray,
182        learner_class: type[BaseDiscriminativeDictionaryLearner],
183        learner_kwargs: dict,
184        n_nonzero_coefs: int,
185        learn_on_residual: bool = True,
186        refit_classifier: bool = True,
187    ) -> None:
188        self.D_frozen = np.asarray(D_frozen, dtype=float)
189        self.learner_class = learner_class
190        self.learner_kwargs = learner_kwargs
191        self.n_nonzero_coefs = n_nonzero_coefs
192        self.learn_on_residual = learn_on_residual
193        self.refit_classifier = refit_classifier
194
195        self.D_combined_: np.ndarray | None = None
196        self.W_: np.ndarray | None = None
197        self.learner_: BaseDiscriminativeDictionaryLearner | None = None
198        self.n_frozen_: int = self.D_frozen.shape[1]
199        self.n_active_: int = 0
200        self.class_boundaries_: dict[int, tuple[int, int]] | None = None
201
202    # ------------------------------------------------------------------
203    # Public API
204    # ------------------------------------------------------------------
205
206    def fit(
207        self,
208        X: np.ndarray,
209        H: np.ndarray,
210        frozen_class_boundaries: dict[int, tuple[int, int]] | None = None,
211    ) -> "FrozenDictionaryLearner":
212        """
213        Fit the residual dictionary on (X, H).
214
215        Parameters
216        ----------
217        X : np.ndarray, shape (n_features, n_samples)
218        H : np.ndarray, shape (n_classes, n_samples)  one-hot labels
219        frozen_class_boundaries : dict or None
220            ``class_boundaries_`` from earlier frozen stages, used to build
221            the merged ``class_boundaries_`` on the combined dictionary.
222            Pass None if D_frozen has no class structure (e.g. base stage).
223
224        Returns
225        -------
226        self
227        """
228        X = np.asarray(X, dtype=float)
229        H = np.asarray(H, dtype=float)
230
231        # --- Train active dict on residual (or full X) ---
232        X_train = (
233            _encode_residual(X, self.D_frozen, self.n_nonzero_coefs)
234            if self.learn_on_residual
235            else X
236        )
237
238        learner = self.learner_class(**self.learner_kwargs)
239        learner.fit(X_train, H)
240        self.learner_ = learner
241
242        D_active = learner.D_
243        self.n_active_ = D_active.shape[1]
244
245        # --- Combine dictionaries ---
246        self.D_combined_ = np.hstack([self.D_frozen, D_active])
247
248        # --- Build merged class_boundaries_ ---
249        n_frozen = self.n_frozen_
250        active_boundaries = learner.class_boundaries_ or {}
251        merged: dict[int, tuple[int, int]] = {}
252
253        # Carry over frozen boundaries unchanged
254        if frozen_class_boundaries:
255            merged.update(frozen_class_boundaries)
256
257        # Shift active boundaries by n_frozen columns
258        for c, (s, e) in active_boundaries.items():
259            merged[c] = (s + n_frozen, e + n_frozen)
260
261        self.class_boundaries_ = merged
262
263        # --- Re-learn classifier over full combined dict ---
264        if self.refit_classifier:
265            coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
266            Gamma_full = coder.encode(X, self.D_combined_)
267            self.W_ = _fit_classifier(Gamma_full, H)
268        else:
269            # Pad learner's W with zeros for the frozen columns
270            W_active = learner.W_
271            if W_active is not None:
272                pad = np.zeros((W_active.shape[0], n_frozen))
273                self.W_ = np.hstack([pad, W_active])
274            else:
275                self.W_ = None
276
277        return self
278
279    def transform(self, X: np.ndarray) -> np.ndarray:
280        """Encode X over the combined dictionary."""
281        self._check_fitted()
282        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
283        return coder.encode(X, self.D_combined_)
284
285    def predict(self, X: np.ndarray) -> np.ndarray:
286        """Classify X using W_ and the combined dictionary."""
287        self._check_fitted()
288        Gamma = self.transform(X)
289        return np.argmax(self.W_ @ Gamma, axis=0)
290
291    def score(self, X: np.ndarray, H: np.ndarray) -> float:
292        """Classification accuracy on (X, H)."""
293        true = np.argmax(H, axis=0)
294        pred = self.predict(X)
295        return float(np.mean(pred == true))
296
297    def _check_fitted(self) -> None:
298        if self.D_combined_ is None:
299            raise DictionaryLearningError(
300                "Call fit() before transform() / predict()."
301            )

Learn a residual dictionary D_active given a frozen dictionary D_frozen.

The learner encodes all signals over D_frozen first, then trains D_active on the reconstruction residual. The combined dictionary

D_combined = [ D_frozen | D_active ]

is exposed as D_combined_ after fitting.

This class handles a single residual learning step. For the full sequential pipeline, see IncrementalFrozenDictionary.

Parameters

D_frozen : np.ndarray, shape (n_features, n_frozen_atoms) Pre-trained frozen dictionary. Never modified. learner_class : type[BaseDiscriminativeDictionaryLearner] Discriminative dictionary learning class to use for the residual. learner_kwargs : dict Keyword arguments forwarded to learner_class.__init__. n_nonzero_coefs : int Sparsity level used when encoding over the frozen dictionary to compute the residual, and when encoding over the combined dictionary for downstream tasks. learn_on_residual : bool If True (default), train D_active on the reconstruction residual R = X - D_frozen @ Gamma_frozen. If False, train on the original X — useful when the frozen dict is very small and you want the active dict to model the full signal, not just what D_frozen misses. refit_classifier : bool If True (default), re-learn W over the full combined dictionary after fitting D_active.

Attributes

D_combined_ : np.ndarray, shape (n_features, n_frozen + n_active) W_ : np.ndarray, shape (n_classes, n_frozen + n_active) learner_ : fitted instance of learner_class n_frozen_ : int number of frozen atoms n_active_ : int number of active (residual) atoms class_boundaries_ : dict[int, tuple[int, int]] Atom ranges for each class in the combined dictionary, merging frozen boundaries (if any) with the active learner's boundaries.

FrozenDictionaryLearner( D_frozen: numpy.ndarray, learner_class: type[reppi.base.BaseDiscriminativeDictionaryLearner], learner_kwargs: dict, n_nonzero_coefs: int, learn_on_residual: bool = True, refit_classifier: bool = True)
179    def __init__(
180        self,
181        D_frozen: np.ndarray,
182        learner_class: type[BaseDiscriminativeDictionaryLearner],
183        learner_kwargs: dict,
184        n_nonzero_coefs: int,
185        learn_on_residual: bool = True,
186        refit_classifier: bool = True,
187    ) -> None:
188        self.D_frozen = np.asarray(D_frozen, dtype=float)
189        self.learner_class = learner_class
190        self.learner_kwargs = learner_kwargs
191        self.n_nonzero_coefs = n_nonzero_coefs
192        self.learn_on_residual = learn_on_residual
193        self.refit_classifier = refit_classifier
194
195        self.D_combined_: np.ndarray | None = None
196        self.W_: np.ndarray | None = None
197        self.learner_: BaseDiscriminativeDictionaryLearner | None = None
198        self.n_frozen_: int = self.D_frozen.shape[1]
199        self.n_active_: int = 0
200        self.class_boundaries_: dict[int, tuple[int, int]] | None = None
D_frozen
learner_class
learner_kwargs
n_nonzero_coefs
learn_on_residual
refit_classifier
D_combined_: numpy.ndarray | None
W_: numpy.ndarray | None
learner_: reppi.base.BaseDiscriminativeDictionaryLearner | None
n_frozen_: int
n_active_: int
class_boundaries_: dict[int, tuple[int, int]] | None
def fit( self, X: numpy.ndarray, H: numpy.ndarray, frozen_class_boundaries: dict[int, tuple[int, int]] | None = None) -> FrozenDictionaryLearner:
206    def fit(
207        self,
208        X: np.ndarray,
209        H: np.ndarray,
210        frozen_class_boundaries: dict[int, tuple[int, int]] | None = None,
211    ) -> "FrozenDictionaryLearner":
212        """
213        Fit the residual dictionary on (X, H).
214
215        Parameters
216        ----------
217        X : np.ndarray, shape (n_features, n_samples)
218        H : np.ndarray, shape (n_classes, n_samples)  one-hot labels
219        frozen_class_boundaries : dict or None
220            ``class_boundaries_`` from earlier frozen stages, used to build
221            the merged ``class_boundaries_`` on the combined dictionary.
222            Pass None if D_frozen has no class structure (e.g. base stage).
223
224        Returns
225        -------
226        self
227        """
228        X = np.asarray(X, dtype=float)
229        H = np.asarray(H, dtype=float)
230
231        # --- Train active dict on residual (or full X) ---
232        X_train = (
233            _encode_residual(X, self.D_frozen, self.n_nonzero_coefs)
234            if self.learn_on_residual
235            else X
236        )
237
238        learner = self.learner_class(**self.learner_kwargs)
239        learner.fit(X_train, H)
240        self.learner_ = learner
241
242        D_active = learner.D_
243        self.n_active_ = D_active.shape[1]
244
245        # --- Combine dictionaries ---
246        self.D_combined_ = np.hstack([self.D_frozen, D_active])
247
248        # --- Build merged class_boundaries_ ---
249        n_frozen = self.n_frozen_
250        active_boundaries = learner.class_boundaries_ or {}
251        merged: dict[int, tuple[int, int]] = {}
252
253        # Carry over frozen boundaries unchanged
254        if frozen_class_boundaries:
255            merged.update(frozen_class_boundaries)
256
257        # Shift active boundaries by n_frozen columns
258        for c, (s, e) in active_boundaries.items():
259            merged[c] = (s + n_frozen, e + n_frozen)
260
261        self.class_boundaries_ = merged
262
263        # --- Re-learn classifier over full combined dict ---
264        if self.refit_classifier:
265            coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
266            Gamma_full = coder.encode(X, self.D_combined_)
267            self.W_ = _fit_classifier(Gamma_full, H)
268        else:
269            # Pad learner's W with zeros for the frozen columns
270            W_active = learner.W_
271            if W_active is not None:
272                pad = np.zeros((W_active.shape[0], n_frozen))
273                self.W_ = np.hstack([pad, W_active])
274            else:
275                self.W_ = None
276
277        return self

Fit the residual dictionary on (X, H).

Parameters

X : np.ndarray, shape (n_features, n_samples) H : np.ndarray, shape (n_classes, n_samples) one-hot labels frozen_class_boundaries : dict or None class_boundaries_ from earlier frozen stages, used to build the merged class_boundaries_ on the combined dictionary. Pass None if D_frozen has no class structure (e.g. base stage).

Returns

self

def transform(self, X: numpy.ndarray) -> numpy.ndarray:
279    def transform(self, X: np.ndarray) -> np.ndarray:
280        """Encode X over the combined dictionary."""
281        self._check_fitted()
282        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
283        return coder.encode(X, self.D_combined_)

Encode X over the combined dictionary.

def predict(self, X: numpy.ndarray) -> numpy.ndarray:
285    def predict(self, X: np.ndarray) -> np.ndarray:
286        """Classify X using W_ and the combined dictionary."""
287        self._check_fitted()
288        Gamma = self.transform(X)
289        return np.argmax(self.W_ @ Gamma, axis=0)

Classify X using W_ and the combined dictionary.

def score(self, X: numpy.ndarray, H: numpy.ndarray) -> float:
291    def score(self, X: np.ndarray, H: np.ndarray) -> float:
292        """Classification accuracy on (X, H)."""
293        true = np.argmax(H, axis=0)
294        pred = self.predict(X)
295        return float(np.mean(pred == true))

Classification accuracy on (X, H).

class IncrementalFrozenDictionary:
309class IncrementalFrozenDictionary:
310    """
311    Incrementally learn class-specific residual dictionaries, freezing all
312    previously learned atoms before training the next class.
313
314    Pipeline
315    --------
316    1. ``fit_base(X, H)``
317       Learn a base dictionary D_n from normal/background data using
318       ``base_learner_class``.  This dictionary is frozen for all
319       subsequent steps.
320
321    2. ``add_class(X, H, class_label)``
322       Learn a residual dictionary D_a for the new class on top of
323       the currently frozen dictionary [ D_n | D_a_1 | … ].
324       Only the new residual atoms are updated; all prior atoms are frozen.
325       The combined dictionary is extended in-place.
326       W is re-learned over all classes after each addition.
327
328    3. ``predict(X)`` / ``score(X, H)``
329       Classify using the full combined dictionary and the latest W.
330
331    Parameters
332    ----------
333    base_learner_class : type[BaseDiscriminativeDictionaryLearner]
334        Learner used for the initial base dictionary.
335    base_learner_kwargs : dict
336        Init kwargs for ``base_learner_class``.
337    residual_learner_class : type[BaseDiscriminativeDictionaryLearner]
338        Learner used for each residual dictionary.  Can be the same as or
339        different from ``base_learner_class``.
340    residual_learner_kwargs : dict
341        Init kwargs for ``residual_learner_class``.  Applied identically
342        for every ``add_class`` call; override per-call via
343        ``add_class(..., learner_kwargs_override=...)``.
344    n_nonzero_coefs : int
345        Sparsity level for all encoding steps.
346    learn_on_residual : bool
347        Passed through to ``FrozenDictionaryLearner`` at each step.
348        Default True.
349    refit_classifier : bool
350        Re-learn W over the full combined dict after each add_class.
351        Default True (recommended — see module docstring).
352    freeze_classifier : bool
353        If True, W columns for previously seen classes are frozen when a
354        new class is added; only the new class's W column is learned.
355        Default False (re-learn all W columns jointly each time).
356
357    Attributes
358    ----------
359    D_  : np.ndarray  full combined dictionary after all steps
360    W_  : np.ndarray  current linear classifier
361    class_labels_ : list[int]  class labels in insertion order
362    class_boundaries_ : dict[int, tuple[int, int]]
363        Per-class atom ranges in the full combined D_.
364    stage_learners_ : list
365        Fitted learner or FrozenDictionaryLearner from each stage,
366        in order (index 0 = base stage).
367    errors_ : dict[int, list[float]]
368        Per-stage training RMSE curves keyed by class_label
369        (key -1 for the base stage).
370    """
371
372    def __init__(
373        self,
374        base_learner_class: type[BaseDiscriminativeDictionaryLearner],
375        base_learner_kwargs: dict,
376        residual_learner_class: type[BaseDiscriminativeDictionaryLearner],
377        residual_learner_kwargs: dict,
378        n_nonzero_coefs: int,
379        learn_on_residual: bool = True,
380        refit_classifier: bool = True,
381        freeze_classifier: bool = False,
382    ) -> None:
383        self.base_learner_class = base_learner_class
384        self.base_learner_kwargs = base_learner_kwargs
385        self.residual_learner_class = residual_learner_class
386        self.residual_learner_kwargs = residual_learner_kwargs
387        self.n_nonzero_coefs = n_nonzero_coefs
388        self.learn_on_residual = learn_on_residual
389        self.refit_classifier = refit_classifier
390        self.freeze_classifier = freeze_classifier
391
392        # State built incrementally
393        self.D_: np.ndarray | None = None
394        self.W_: np.ndarray | None = None
395        self.class_labels_: list[int] = []
396        self.class_boundaries_: dict[int, tuple[int, int]] = {}
397        self.stage_learners_: list = []
398        self.errors_: dict[int, list[float]] = {}
399
400        # Internal: accumulated (X, H) across all classes for W refit
401        self._X_all: list[np.ndarray] = []
402        self._H_rows: int | None = None   # n_classes total
403
404    # ------------------------------------------------------------------
405    # Public API
406    # ------------------------------------------------------------------
407
408    def fit_base(
409        self,
410        X: np.ndarray,
411        H: np.ndarray,
412    ) -> "IncrementalFrozenDictionary":
413        """
414        Learn the base dictionary from normal / background data.
415
416        Parameters
417        ----------
418        X : np.ndarray, shape (n_features, n_samples)
419        H : np.ndarray, shape (n_classes, n_samples)
420            One-hot labels for the base class(es).
421
422        Returns
423        -------
424        self
425        """
426        X = np.asarray(X, dtype=float)
427        H = np.asarray(H, dtype=float)
428
429        learner = self.base_learner_class(**self.base_learner_kwargs)
430        learner.fit(X, H)
431
432        self.D_ = learner.D_
433        self.class_boundaries_ = dict(learner.class_boundaries_ or {})
434        self.stage_learners_.append(learner)
435        self.errors_[-1] = list(getattr(learner, "errors_", []))
436
437        # Initialise W
438        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
439        Gamma = coder.encode(X, self.D_)
440        self.W_ = _fit_classifier(Gamma, H)
441
442        self._H_rows = H.shape[0]
443        self._X_all.append(X)
444
445        return self
446
447    def add_class(
448        self,
449        X: np.ndarray,
450        H: np.ndarray,
451        class_label: int,
452        learner_kwargs_override: dict | None = None,
453    ) -> "IncrementalFrozenDictionary":
454        """
455        Learn a residual dictionary for a new class and extend D_.
456
457        Parameters
458        ----------
459        X : np.ndarray, shape (n_features, n_samples)
460            Training signals for this class.
461        H : np.ndarray, shape (n_classes_so_far + 1, n_samples)
462            One-hot label matrix for *all* classes seen so far including
463            the new one.  Used to refit W after extending the dictionary.
464        class_label : int
465            Integer label for this class.  Must not have been added before.
466        learner_kwargs_override : dict or None
467            If supplied, overrides ``residual_learner_kwargs`` for this
468            call only.  Useful for adjusting n_components per class.
469
470        Returns
471        -------
472        self
473        """
474        if self.D_ is None:
475            raise DictionaryLearningError(
476                "Call fit_base() before add_class()."
477            )
478        if class_label in self.class_labels_:
479            raise ValueError(
480                f"class_label {class_label} has already been added."
481            )
482
483        X = np.asarray(X, dtype=float)
484        H = np.asarray(H, dtype=float)
485
486        kwargs = {**self.residual_learner_kwargs, **(learner_kwargs_override or {})}
487
488        # The residual learner only sees X (n_samples for this class).
489        # Extract the H columns that correspond to X — the caller passes the
490        # full H over all accumulated data, but the learner needs H for X only.
491        n_new = X.shape[1]
492        H_for_learner = H[:, -n_new:]   # last n_new columns = this class's signals
493
494        frozen_step = FrozenDictionaryLearner(
495            D_frozen=self.D_,
496            learner_class=self.residual_learner_class,
497            learner_kwargs=kwargs,
498            n_nonzero_coefs=self.n_nonzero_coefs,
499            learn_on_residual=self.learn_on_residual,
500            refit_classifier=False,  # we handle W ourselves below
501        )
502        frozen_step.fit(X, H_for_learner, frozen_class_boundaries=dict(self.class_boundaries_))
503
504        # --- Extend D_ and class_boundaries_ ---
505        n_prev = self.D_.shape[1]
506        D_active = frozen_step.learner_.D_
507        n_active = D_active.shape[1]
508        self.D_ = np.hstack([self.D_, D_active])
509
510        # Map new class atoms into the combined dictionary
511        self.class_boundaries_[class_label] = (n_prev, n_prev + n_active)
512        self.class_labels_.append(class_label)
513        self.stage_learners_.append(frozen_step)
514        self.errors_[class_label] = list(
515            getattr(frozen_step.learner_, "errors_", [])
516        )
517
518        # Accumulate training data for W refit
519        self._X_all.append(X)
520
521        # --- Refit W over all data and the full combined dict ---
522        if self.refit_classifier:
523            X_all = np.hstack(self._X_all)
524            # H must cover all columns of X_all — caller is responsible
525            # for passing the full H including all previous classes
526            self._refit_W(X_all, H)
527        elif self.freeze_classifier:
528            self._extend_W_frozen(D_active.shape[1], H)
529
530        return self
531
532    def predict(self, X: np.ndarray) -> np.ndarray:
533        """
534        Classify X using the full combined dictionary and the current W.
535
536        Parameters
537        ----------
538        X : np.ndarray, shape (n_features, n_samples)
539
540        Returns
541        -------
542        labels : np.ndarray, shape (n_samples,)
543        """
544        self._check_fitted()
545        Gamma = self._encode(X)
546        return np.argmax(self.W_ @ Gamma, axis=0)
547
548    def score(self, X: np.ndarray, H: np.ndarray) -> float:
549        """
550        Classification accuracy on (X, H).
551
552        Parameters
553        ----------
554        X : np.ndarray, shape (n_features, n_samples)
555        H : np.ndarray, shape (n_classes, n_samples)
556
557        Returns
558        -------
559        accuracy : float in [0, 1]
560        """
561        true = np.argmax(H, axis=0)
562        pred = self.predict(X)
563        return float(np.mean(pred == true))
564
565    def transform(self, X: np.ndarray) -> np.ndarray:
566        """Encode X over the full combined dictionary D_."""
567        self._check_fitted()
568        return self._encode(X)
569
570    def get_stage_dict(self, stage: int) -> np.ndarray:
571        """
572        Return the sub-dictionary learned at a given stage.
573
574        Stage 0 is the base dictionary; stage k (k >= 1) is the k-th
575        residual dictionary.
576
577        Parameters
578        ----------
579        stage : int
580
581        Returns
582        -------
583        D_stage : np.ndarray
584        """
585        self._check_fitted()
586        if stage < 0 or stage >= len(self.stage_learners_):
587            raise IndexError(
588                f"stage must be in [0, {len(self.stage_learners_) - 1}], got {stage}."
589            )
590        learner = self.stage_learners_[stage]
591        # Base stage: learner is a BaseDiscriminativeDictionaryLearner
592        if isinstance(learner, FrozenDictionaryLearner):
593            return learner.learner_.D_
594        return learner.D_
595
596    def get_class_dict(self, class_label: int) -> np.ndarray:
597        """
598        Return the sub-dictionary atoms for a given class label.
599
600        Parameters
601        ----------
602        class_label : int
603
604        Returns
605        -------
606        D_c : np.ndarray
607        """
608        self._check_fitted()
609        if class_label not in self.class_boundaries_:
610            raise KeyError(f"class_label {class_label} not found.")
611        s, e = self.class_boundaries_[class_label]
612        return self.D_[:, s:e]
613
614    # ------------------------------------------------------------------
615    # Internal helpers
616    # ------------------------------------------------------------------
617
618    def _check_fitted(self) -> None:
619        if self.D_ is None:
620            raise DictionaryLearningError(
621                "Call fit_base() before using this method."
622            )
623
624    def _encode(self, X: np.ndarray) -> np.ndarray:
625        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
626        return coder.encode(X, self.D_)
627
628    def _refit_W(self, X_all: np.ndarray, H: np.ndarray) -> None:
629        """Re-learn W jointly over all classes on the full combined dict."""
630        Gamma = self._encode(X_all)
631        self.W_ = _fit_classifier(Gamma, H)
632
633    def _extend_W_frozen(self, n_new_atoms: int, H: np.ndarray) -> None:
634        """
635        Freeze existing W columns; learn only the columns for new atoms.
636
637        This implements the ``freeze_classifier=True`` behaviour: old class
638        boundaries in W stay fixed; only weights for the n_new_atoms are
639        updated for the new class.
640        """
641        if self.W_ is None:
642            return
643        n_classes_new = H.shape[0]
644        n_classes_old = self.W_.shape[0]
645        n_atoms_old = self.W_.shape[1]
646
647        # Extend W with zero rows for any new classes and zero cols for new atoms
648        W_extended = np.zeros((n_classes_new, n_atoms_old + n_new_atoms))
649        W_extended[:n_classes_old, :n_atoms_old] = self.W_
650
651        # Learn only the new columns via least squares restricted to new atoms
652        # Encode all accumulated data over full dict, extract new-atom codes
653        X_all = np.hstack(self._X_all)
654        Gamma_full = self._encode(X_all)
655        Gamma_new = Gamma_full[n_atoms_old:, :]   # (n_new_atoms, n_samples)
656
657        # Solve W_new @ Gamma_new ≈ H - W_old @ Gamma_old
658        Gamma_old = Gamma_full[:n_atoms_old, :]
659        residual_H = H - W_extended[:, :n_atoms_old] @ Gamma_old
660        W_new_cols = residual_H @ np.linalg.pinv(Gamma_new)
661        W_extended[:, n_atoms_old:] = W_new_cols
662
663        self.W_ = W_extended

Incrementally learn class-specific residual dictionaries, freezing all previously learned atoms before training the next class.

Pipeline

  1. fit_base(X, H) Learn a base dictionary D_n from normal/background data using base_learner_class. This dictionary is frozen for all subsequent steps.

  2. add_class(X, H, class_label) Learn a residual dictionary D_a for the new class on top of the currently frozen dictionary [ D_n | D_a_1 | … ]. Only the new residual atoms are updated; all prior atoms are frozen. The combined dictionary is extended in-place. W is re-learned over all classes after each addition.

  3. predict(X) / score(X, H) Classify using the full combined dictionary and the latest W.

Parameters

base_learner_class : type[BaseDiscriminativeDictionaryLearner] Learner used for the initial base dictionary. base_learner_kwargs : dict Init kwargs for base_learner_class. residual_learner_class : type[BaseDiscriminativeDictionaryLearner] Learner used for each residual dictionary. Can be the same as or different from base_learner_class. residual_learner_kwargs : dict Init kwargs for residual_learner_class. Applied identically for every add_class call; override per-call via add_class(..., learner_kwargs_override=...). n_nonzero_coefs : int Sparsity level for all encoding steps. learn_on_residual : bool Passed through to FrozenDictionaryLearner at each step. Default True. refit_classifier : bool Re-learn W over the full combined dict after each add_class. Default True (recommended — see module docstring). freeze_classifier : bool If True, W columns for previously seen classes are frozen when a new class is added; only the new class's W column is learned. Default False (re-learn all W columns jointly each time).

Attributes

D_ : np.ndarray full combined dictionary after all steps W_ : np.ndarray current linear classifier class_labels_ : list[int] class labels in insertion order class_boundaries_ : dict[int, tuple[int, int]] Per-class atom ranges in the full combined D_. stage_learners_ : list Fitted learner or FrozenDictionaryLearner from each stage, in order (index 0 = base stage). errors_ : dict[int, list[float]] Per-stage training RMSE curves keyed by class_label (key -1 for the base stage).

IncrementalFrozenDictionary( base_learner_class: type[reppi.base.BaseDiscriminativeDictionaryLearner], base_learner_kwargs: dict, residual_learner_class: type[reppi.base.BaseDiscriminativeDictionaryLearner], residual_learner_kwargs: dict, n_nonzero_coefs: int, learn_on_residual: bool = True, refit_classifier: bool = True, freeze_classifier: bool = False)
372    def __init__(
373        self,
374        base_learner_class: type[BaseDiscriminativeDictionaryLearner],
375        base_learner_kwargs: dict,
376        residual_learner_class: type[BaseDiscriminativeDictionaryLearner],
377        residual_learner_kwargs: dict,
378        n_nonzero_coefs: int,
379        learn_on_residual: bool = True,
380        refit_classifier: bool = True,
381        freeze_classifier: bool = False,
382    ) -> None:
383        self.base_learner_class = base_learner_class
384        self.base_learner_kwargs = base_learner_kwargs
385        self.residual_learner_class = residual_learner_class
386        self.residual_learner_kwargs = residual_learner_kwargs
387        self.n_nonzero_coefs = n_nonzero_coefs
388        self.learn_on_residual = learn_on_residual
389        self.refit_classifier = refit_classifier
390        self.freeze_classifier = freeze_classifier
391
392        # State built incrementally
393        self.D_: np.ndarray | None = None
394        self.W_: np.ndarray | None = None
395        self.class_labels_: list[int] = []
396        self.class_boundaries_: dict[int, tuple[int, int]] = {}
397        self.stage_learners_: list = []
398        self.errors_: dict[int, list[float]] = {}
399
400        # Internal: accumulated (X, H) across all classes for W refit
401        self._X_all: list[np.ndarray] = []
402        self._H_rows: int | None = None   # n_classes total
base_learner_class
base_learner_kwargs
residual_learner_class
residual_learner_kwargs
n_nonzero_coefs
learn_on_residual
refit_classifier
freeze_classifier
D_: numpy.ndarray | None
W_: numpy.ndarray | None
class_labels_: list[int]
class_boundaries_: dict[int, tuple[int, int]]
stage_learners_: list
errors_: dict[int, list[float]]
def fit_base( self, X: numpy.ndarray, H: numpy.ndarray) -> IncrementalFrozenDictionary:
408    def fit_base(
409        self,
410        X: np.ndarray,
411        H: np.ndarray,
412    ) -> "IncrementalFrozenDictionary":
413        """
414        Learn the base dictionary from normal / background data.
415
416        Parameters
417        ----------
418        X : np.ndarray, shape (n_features, n_samples)
419        H : np.ndarray, shape (n_classes, n_samples)
420            One-hot labels for the base class(es).
421
422        Returns
423        -------
424        self
425        """
426        X = np.asarray(X, dtype=float)
427        H = np.asarray(H, dtype=float)
428
429        learner = self.base_learner_class(**self.base_learner_kwargs)
430        learner.fit(X, H)
431
432        self.D_ = learner.D_
433        self.class_boundaries_ = dict(learner.class_boundaries_ or {})
434        self.stage_learners_.append(learner)
435        self.errors_[-1] = list(getattr(learner, "errors_", []))
436
437        # Initialise W
438        coder = OMP(self.n_nonzero_coefs, mode="batch", check_dict=False)
439        Gamma = coder.encode(X, self.D_)
440        self.W_ = _fit_classifier(Gamma, H)
441
442        self._H_rows = H.shape[0]
443        self._X_all.append(X)
444
445        return self

Learn the base dictionary from normal / background data.

Parameters

X : np.ndarray, shape (n_features, n_samples) H : np.ndarray, shape (n_classes, n_samples) One-hot labels for the base class(es).

Returns

self

def add_class( self, X: numpy.ndarray, H: numpy.ndarray, class_label: int, learner_kwargs_override: dict | None = None) -> IncrementalFrozenDictionary:
447    def add_class(
448        self,
449        X: np.ndarray,
450        H: np.ndarray,
451        class_label: int,
452        learner_kwargs_override: dict | None = None,
453    ) -> "IncrementalFrozenDictionary":
454        """
455        Learn a residual dictionary for a new class and extend D_.
456
457        Parameters
458        ----------
459        X : np.ndarray, shape (n_features, n_samples)
460            Training signals for this class.
461        H : np.ndarray, shape (n_classes_so_far + 1, n_samples)
462            One-hot label matrix for *all* classes seen so far including
463            the new one.  Used to refit W after extending the dictionary.
464        class_label : int
465            Integer label for this class.  Must not have been added before.
466        learner_kwargs_override : dict or None
467            If supplied, overrides ``residual_learner_kwargs`` for this
468            call only.  Useful for adjusting n_components per class.
469
470        Returns
471        -------
472        self
473        """
474        if self.D_ is None:
475            raise DictionaryLearningError(
476                "Call fit_base() before add_class()."
477            )
478        if class_label in self.class_labels_:
479            raise ValueError(
480                f"class_label {class_label} has already been added."
481            )
482
483        X = np.asarray(X, dtype=float)
484        H = np.asarray(H, dtype=float)
485
486        kwargs = {**self.residual_learner_kwargs, **(learner_kwargs_override or {})}
487
488        # The residual learner only sees X (n_samples for this class).
489        # Extract the H columns that correspond to X — the caller passes the
490        # full H over all accumulated data, but the learner needs H for X only.
491        n_new = X.shape[1]
492        H_for_learner = H[:, -n_new:]   # last n_new columns = this class's signals
493
494        frozen_step = FrozenDictionaryLearner(
495            D_frozen=self.D_,
496            learner_class=self.residual_learner_class,
497            learner_kwargs=kwargs,
498            n_nonzero_coefs=self.n_nonzero_coefs,
499            learn_on_residual=self.learn_on_residual,
500            refit_classifier=False,  # we handle W ourselves below
501        )
502        frozen_step.fit(X, H_for_learner, frozen_class_boundaries=dict(self.class_boundaries_))
503
504        # --- Extend D_ and class_boundaries_ ---
505        n_prev = self.D_.shape[1]
506        D_active = frozen_step.learner_.D_
507        n_active = D_active.shape[1]
508        self.D_ = np.hstack([self.D_, D_active])
509
510        # Map new class atoms into the combined dictionary
511        self.class_boundaries_[class_label] = (n_prev, n_prev + n_active)
512        self.class_labels_.append(class_label)
513        self.stage_learners_.append(frozen_step)
514        self.errors_[class_label] = list(
515            getattr(frozen_step.learner_, "errors_", [])
516        )
517
518        # Accumulate training data for W refit
519        self._X_all.append(X)
520
521        # --- Refit W over all data and the full combined dict ---
522        if self.refit_classifier:
523            X_all = np.hstack(self._X_all)
524            # H must cover all columns of X_all — caller is responsible
525            # for passing the full H including all previous classes
526            self._refit_W(X_all, H)
527        elif self.freeze_classifier:
528            self._extend_W_frozen(D_active.shape[1], H)
529
530        return self

Learn a residual dictionary for a new class and extend D_.

Parameters

X : np.ndarray, shape (n_features, n_samples) Training signals for this class. H : np.ndarray, shape (n_classes_so_far + 1, n_samples) One-hot label matrix for all classes seen so far including the new one. Used to refit W after extending the dictionary. class_label : int Integer label for this class. Must not have been added before. learner_kwargs_override : dict or None If supplied, overrides residual_learner_kwargs for this call only. Useful for adjusting n_components per class.

Returns

self

def predict(self, X: numpy.ndarray) -> numpy.ndarray:
532    def predict(self, X: np.ndarray) -> np.ndarray:
533        """
534        Classify X using the full combined dictionary and the current W.
535
536        Parameters
537        ----------
538        X : np.ndarray, shape (n_features, n_samples)
539
540        Returns
541        -------
542        labels : np.ndarray, shape (n_samples,)
543        """
544        self._check_fitted()
545        Gamma = self._encode(X)
546        return np.argmax(self.W_ @ Gamma, axis=0)

Classify X using the full combined dictionary and the current W.

Parameters

X : np.ndarray, shape (n_features, n_samples)

Returns

labels : np.ndarray, shape (n_samples,)

def score(self, X: numpy.ndarray, H: numpy.ndarray) -> float:
548    def score(self, X: np.ndarray, H: np.ndarray) -> float:
549        """
550        Classification accuracy on (X, H).
551
552        Parameters
553        ----------
554        X : np.ndarray, shape (n_features, n_samples)
555        H : np.ndarray, shape (n_classes, n_samples)
556
557        Returns
558        -------
559        accuracy : float in [0, 1]
560        """
561        true = np.argmax(H, axis=0)
562        pred = self.predict(X)
563        return float(np.mean(pred == true))

Classification accuracy on (X, H).

Parameters

X : np.ndarray, shape (n_features, n_samples) H : np.ndarray, shape (n_classes, n_samples)

Returns

accuracy : float in [0, 1]

def transform(self, X: numpy.ndarray) -> numpy.ndarray:
565    def transform(self, X: np.ndarray) -> np.ndarray:
566        """Encode X over the full combined dictionary D_."""
567        self._check_fitted()
568        return self._encode(X)

Encode X over the full combined dictionary D_.

def get_stage_dict(self, stage: int) -> numpy.ndarray:
570    def get_stage_dict(self, stage: int) -> np.ndarray:
571        """
572        Return the sub-dictionary learned at a given stage.
573
574        Stage 0 is the base dictionary; stage k (k >= 1) is the k-th
575        residual dictionary.
576
577        Parameters
578        ----------
579        stage : int
580
581        Returns
582        -------
583        D_stage : np.ndarray
584        """
585        self._check_fitted()
586        if stage < 0 or stage >= len(self.stage_learners_):
587            raise IndexError(
588                f"stage must be in [0, {len(self.stage_learners_) - 1}], got {stage}."
589            )
590        learner = self.stage_learners_[stage]
591        # Base stage: learner is a BaseDiscriminativeDictionaryLearner
592        if isinstance(learner, FrozenDictionaryLearner):
593            return learner.learner_.D_
594        return learner.D_

Return the sub-dictionary learned at a given stage.

Stage 0 is the base dictionary; stage k (k >= 1) is the k-th residual dictionary.

Parameters

stage : int

Returns

D_stage : np.ndarray

def get_class_dict(self, class_label: int) -> numpy.ndarray:
596    def get_class_dict(self, class_label: int) -> np.ndarray:
597        """
598        Return the sub-dictionary atoms for a given class label.
599
600        Parameters
601        ----------
602        class_label : int
603
604        Returns
605        -------
606        D_c : np.ndarray
607        """
608        self._check_fitted()
609        if class_label not in self.class_boundaries_:
610            raise KeyError(f"class_label {class_label} not found.")
611        s, e = self.class_boundaries_[class_label]
612        return self.D_[:, s:e]

Return the sub-dictionary atoms for a given class label.

Parameters

class_label : int

Returns

D_c : np.ndarray