""" ConvMLP Models ============== Conv2D-integrated MLPClassifier compatible with sklearn. """ import numpy as np from typing import List, Tuple, Optional, Union from sklearn.base import BaseEstimator, ClassifierMixin, RegressorMixin from sklearn.neural_network import MLPClassifier, MLPRegressor from sklearn.preprocessing import LabelEncoder from sklearn.utils.multiclass import unique_labels from sklearn.utils.validation import check_X_y, check_array, check_is_fitted import warnings from .layers import ( Conv2D, MaxPool2D, AvgPool2D, Flatten, Dropout2D, BatchNorm2D, GlobalAvgPool2D, ReLU, Sigmoid, Tanh, Softmax ) from .utils import im2col, col2im, calculate_receptive_field class ConvMLPClassifier(BaseEstimator, ClassifierMixin): """ Convolutional MLP Classifier with sklearn compatibility. Combines Conv2D layers with MLPClassifier backend for end-to-end training. Parameters ---------- conv_layers : list, optional List of convolutional layer configurations. Each config is a dict with keys: type, filters, kernel_size, etc. mlp_layers : list, optional Hidden layer sizes for the MLP backend. conv_method : str, default='im2col' Convolution method: 'im2col', 'fft', 'winograd', 'direct'. max_iter : int, default=100 Maximum number of iterations. batch_size : int, default=32 Mini-batch size. learning_rate : float, default=0.001 Initial learning rate. momentum : float, default=0.9 Momentum for SGD. random_state : int, optional Random seed. verbose : bool, default=False Print training progress. Example ------- >>> model = ConvMLPClassifier( ... conv_layers=[ ... {'type': 'conv2d', 'filters': 32, 'kernel_size': 3}, ... {'type': 'maxpool2d', 'pool_size': 2}, ... {'type': 'conv2d', 'filters': 64, 'kernel_size': 3}, ... ], ... mlp_layers=[128, 64], ... max_iter=50 ... ) >>> model.fit(X_train, y_train) >>> predictions = model.predict(X_test) """ def __init__( self, conv_layers: Optional[List[dict]] = None, mlp_layers: Optional[List[int]] = None, conv_method: str = 'im2col', max_iter: int = 100, batch_size: int = 32, learning_rate: float = 0.001, learning_rate_init: float = 0.001, momentum: float = 0.9, nesterov_momentum: bool = True, early_stopping: bool = False, validation_fraction: float = 0.1, random_state: Optional[int] = None, verbose: bool = False, input_shape: Optional[Tuple[int, ...]] = None, ): self.conv_layers = conv_layers or [ {'type': 'conv2d', 'filters': 32, 'kernel_size': 3, 'padding': 'same', 'activation': 'relu'}, {'type': 'maxpool2d', 'pool_size': 2}, {'type': 'conv2d', 'filters': 64, 'kernel_size': 3, 'padding': 'same', 'activation': 'relu'}, {'type': 'maxpool2d', 'pool_size': 2}, ] self.mlp_layers = mlp_layers or [128, 64] self.conv_method = conv_method self.max_iter = max_iter self.batch_size = batch_size self.learning_rate = learning_rate self.learning_rate_init = learning_rate_init self.momentum = momentum self.nesterov_momentum = nesterov_momentum self.early_stopping = early_stopping self.validation_fraction = validation_fraction self.random_state = random_state self.verbose = verbose self.input_shape = input_shape # Private attributes set during fit self._conv_net = None self._mlp = None self._label_encoder = None self._input_size = None self._built = False def _build_conv_layers(self, input_shape: Tuple[int, ...]): """Build the convolutional network from configuration.""" self._conv_net = [] current_shape = input_shape for layer_config in self.conv_layers: layer_type = layer_config.get('type', 'conv2d') if layer_type == 'conv2d': layer = Conv2D( filters=layer_config['filters'], kernel_size=layer_config.get('kernel_size', 3), strides=layer_config.get('strides', 1), padding=layer_config.get('padding', 'same'), activation=layer_config.get('activation', 'relu'), use_bias=layer_config.get('use_bias', True), method=self.conv_method ) elif layer_type == 'maxpool2d': layer = MaxPool2D( pool_size=layer_config.get('pool_size', 2), strides=layer_config.get('strides'), padding=layer_config.get('padding', 0) ) elif layer_type == 'avgpool2d': layer = AvgPool2D( pool_size=layer_config.get('pool_size', 2), strides=layer_config.get('strides'), padding=layer_config.get('padding', 0) ) elif layer_type == 'global_avgpool': layer = GlobalAvgPool2D() elif layer_type == 'flatten': layer = Flatten() elif layer_type == 'dropout': layer = Dropout2D(rate=layer_config.get('rate', 0.5)) elif layer_type == 'batchnorm': layer = BatchNorm2D( momentum=layer_config.get('momentum', 0.9), epsilon=layer_config.get('epsilon', 1e-5) ) else: warnings.warn(f"Unknown layer type: {layer_type}") continue if not layer.built: current_shape = layer.build(current_shape) else: current_shape = getattr(layer, 'output_shape', current_shape) self._conv_net.append(layer) return current_shape def _reshape_input(self, X: np.ndarray) -> np.ndarray: """Reshape input to (batch, channels, height, width).""" n_samples = X.shape[0] if self.input_shape: expected = (n_samples,) + self.input_shape if X.shape == expected: return X.astype(np.float32, copy=False) else: return X.reshape(expected).astype(np.float32, copy=False) # Try to infer shape total_size = X.shape[1] # Common aspect ratios candidates = [ (1, 1, total_size), # (C, 1, 1) (3, 1, total_size // 3), # (C, H, W) where H=1 (total_size, 1, 1), # (H, 1, 1) ] # Try to find valid shape for c, h, w in candidates: if c * h * w == total_size: return X.reshape(n_samples, c, h, w).astype(np.float32, copy=False) # Default: assume square size = int(np.sqrt(total_size)) if size * size == total_size: return X.reshape(n_samples, 1, size, size).astype(np.float32, copy=False) # Last resort: use first dimension as channels return X.reshape(n_samples, X.shape[1], 1, -1).astype(np.float32, copy=False) def _feature_batch_size(self, n_samples: int) -> int: """Return the batch size used for conv feature extraction.""" if self.batch_size in (None, 'auto'): return min(32, n_samples) return max(1, min(int(self.batch_size), n_samples)) def _transform_conv_batch(self, X_batch: np.ndarray, training: bool = False) -> np.ndarray: """Run a single batch through the conv stack and flatten the result.""" X_conv = X_batch for layer in self._conv_net: X_conv = layer.forward(X_conv, training=training) if len(X_conv.shape) > 2: return X_conv.reshape(X_conv.shape[0], -1) return X_conv def _iter_conv_features(self, X: np.ndarray, training: bool = False): """Yield flattened conv features batch by batch.""" batch_size = self._feature_batch_size(X.shape[0]) for start in range(0, X.shape[0], batch_size): stop = min(start + batch_size, X.shape[0]) yield self._transform_conv_batch(X[start:stop], training=training) def _initialize_classifier(self, X: np.ndarray, classes: np.ndarray): """Initialize encoder, conv stack, and MLP backend for incremental training.""" self._label_encoder = LabelEncoder() self._label_encoder.fit(classes) self._input_size = X.shape[1:] conv_output_shape = self._build_conv_layers(self._input_size) if len(conv_output_shape) > 2: self._mlp_input_size = np.prod(conv_output_shape[1:]) else: self._mlp_input_size = conv_output_shape[-1] if self.verbose: print(f"Input shape: {X.shape}") print(f"MLP input size: {self._mlp_input_size}") self._mlp = MLPClassifier( hidden_layer_sizes=list(self.mlp_layers), activation='relu', solver='adam', alpha=0.0001, batch_size=self.batch_size, learning_rate_init=self.learning_rate_init, max_iter=1, shuffle=True, random_state=self.random_state, early_stopping=self.early_stopping, validation_fraction=self.validation_fraction, nesterovs_momentum=self.nesterov_momentum, verbose=self.verbose ) def fit(self, X: np.ndarray, y: np.ndarray): """ Fit the model to data matrices X and target y. Parameters ---------- X : array-like of shape (n_samples, n_features) Training data. y : array-like of shape (n_samples,) Target values. Returns ------- self : object Fitted estimator. """ # Set random state if self.random_state is not None: np.random.seed(self.random_state) # Validate data X, y = check_X_y(X, y, multi_output=False) n_input_features = X.shape[1] # Reshape input X = self._reshape_input(X) classes = unique_labels(y) self._initialize_classifier(X, classes) y_encoded = self._label_encoder.transform(y) batch_size = self._feature_batch_size(X.shape[0]) if self.verbose: print(f"Feature batch size: {batch_size}") for _ in range(self.max_iter): for start in range(0, X.shape[0], batch_size): stop = min(start + batch_size, X.shape[0]) X_flat = self._transform_conv_batch(X[start:stop], training=True) self._mlp.partial_fit(X_flat, y_encoded[start:stop], classes=np.arange(len(classes))) self._built = True self._n_features_in_ = n_input_features return self def partial_fit(self, X: np.ndarray, y: np.ndarray, classes=None): """Incrementally fit the model on a batch of samples.""" if self.random_state is not None and not self._built: np.random.seed(self.random_state) X, y = check_X_y(X, y, multi_output=False) n_input_features = X.shape[1] X = self._reshape_input(X) if not self._built: if classes is None: classes = unique_labels(y) classes = np.asarray(classes) self._initialize_classifier(X, classes) self._built = True else: if classes is not None: classes = np.asarray(classes) if not np.array_equal(classes, self._label_encoder.classes_): raise ValueError("`classes` must match the classes from the first partial_fit call.") elif np.any(~np.isin(unique_labels(y), self._label_encoder.classes_)): raise ValueError("New classes found in `y`. Pass all classes in the first partial_fit call.") y_encoded = self._label_encoder.transform(y) encoded_classes = np.arange(len(self._label_encoder.classes_)) for X_flat in self._iter_conv_features(X, training=True): batch_len = X_flat.shape[0] y_batch = y_encoded[:batch_len] y_encoded = y_encoded[batch_len:] self._mlp.partial_fit(X_flat, y_batch, classes=encoded_classes) self._n_features_in_ = n_input_features return self def predict(self, X: np.ndarray) -> np.ndarray: """ Predict class labels for samples in X. Parameters ---------- X : array-like of shape (n_samples, n_features) Samples. Returns ------- y : ndarray of shape (n_samples,) Class labels for samples in X. """ check_is_fitted(self) X = check_array(X) # Reshape input X = self._reshape_input(X) # Forward pass through conv layers y_pred = np.concatenate([ self._mlp.predict(X_flat) for X_flat in self._iter_conv_features(X, training=False) ]) # Decode labels return self._label_encoder.inverse_transform(y_pred) def predict_proba(self, X: np.ndarray) -> np.ndarray: """Return probability estimates for samples.""" check_is_fitted(self) X = check_array(X) # Reshape input X = self._reshape_input(X) # Forward pass through conv layers return np.vstack([ self._mlp.predict_proba(X_flat) for X_flat in self._iter_conv_features(X, training=False) ]) def score(self, X: np.ndarray, y: np.ndarray) -> float: """Return the mean accuracy on the given test data and labels.""" from sklearn.metrics import accuracy_score return accuracy_score(y, self.predict(X)) def get_conv_layers(self) -> List: """Return the list of convolutional layers.""" return self._conv_net def get_feature_extractor(self): """Return a function that extracts features from the conv layers.""" def extractor(X): X = self._reshape_input(X) for layer in self._conv_net: X = layer.forward(X, training=False) return X.reshape(X.shape[0], -1) return extractor class ConvMLPRegressor(BaseEstimator, RegressorMixin): """ Convolutional MLP Regressor with sklearn compatibility. Combines Conv2D layers with MLPRegressor backend. Parameters ---------- (Same as ConvMLPClassifier, with target type regression instead of classification) """ def __init__( self, conv_layers: Optional[List[dict]] = None, mlp_layers: Optional[List[int]] = None, conv_method: str = 'im2col', max_iter: int = 100, batch_size: int = 32, learning_rate_init: float = 0.001, momentum: float = 0.9, random_state: Optional[int] = None, verbose: bool = False, input_shape: Optional[Tuple[int, ...]] = None, ): self.conv_layers = conv_layers or [ {'type': 'conv2d', 'filters': 32, 'kernel_size': 3, 'padding': 'same', 'activation': 'relu'}, {'type': 'maxpool2d', 'pool_size': 2}, {'type': 'conv2d', 'filters': 64, 'kernel_size': 3, 'padding': 'same', 'activation': 'relu'}, ] self.mlp_layers = mlp_layers or [128, 64] self.conv_method = conv_method self.max_iter = max_iter self.batch_size = batch_size self.learning_rate_init = learning_rate_init self.momentum = momentum self.random_state = random_state self.verbose = verbose self.input_shape = input_shape self._conv_net = None self._mlp = None self._built = False def _reshape_input(self, X: np.ndarray) -> np.ndarray: """Reshape input to (batch, channels, height, width).""" n_samples = X.shape[0] if self.input_shape: expected = (n_samples,) + self.input_shape if X.shape == expected: return X.astype(np.float32, copy=False) else: return X.reshape(expected).astype(np.float32, copy=False) total_size = X.shape[1] size = int(np.sqrt(total_size)) if size * size == total_size: return X.reshape(n_samples, 1, size, size).astype(np.float32, copy=False) return X.reshape(n_samples, X.shape[1], 1, -1).astype(np.float32, copy=False) def _feature_batch_size(self, n_samples: int) -> int: if self.batch_size in (None, 'auto'): return min(32, n_samples) return max(1, min(int(self.batch_size), n_samples)) def _transform_conv_batch(self, X_batch: np.ndarray, training: bool = False) -> np.ndarray: X_conv = X_batch for layer in self._conv_net: X_conv = layer.forward(X_conv, training=training) return X_conv.reshape(X_conv.shape[0], -1) def _iter_conv_features(self, X: np.ndarray, training: bool = False): batch_size = self._feature_batch_size(X.shape[0]) for start in range(0, X.shape[0], batch_size): stop = min(start + batch_size, X.shape[0]) yield self._transform_conv_batch(X[start:stop], training=training) def _build_conv_layers(self, input_shape: Tuple[int, ...]): """Build the convolutional network.""" self._conv_net = [] current_shape = input_shape for layer_config in self.conv_layers: layer_type = layer_config.get('type', 'conv2d') if layer_type == 'conv2d': layer = Conv2D( filters=layer_config['filters'], kernel_size=layer_config.get('kernel_size', 3), strides=layer_config.get('strides', 1), padding=layer_config.get('padding', 'same'), activation=layer_config.get('activation', 'relu'), use_bias=layer_config.get('use_bias', True), method=self.conv_method ) elif layer_type == 'maxpool2d': layer = MaxPool2D( pool_size=layer_config.get('pool_size', 2), strides=layer_config.get('strides'), padding=layer_config.get('padding', 0) ) elif layer_type == 'avgpool2d': layer = AvgPool2D( pool_size=layer_config.get('pool_size', 2), strides=layer_config.get('strides'), padding=layer_config.get('padding', 0) ) elif layer_type == 'flatten': layer = Flatten() elif layer_type == 'dropout': layer = Dropout2D(rate=layer_config.get('rate', 0.5)) else: continue if not layer.built: current_shape = layer.build(current_shape) else: current_shape = getattr(layer, 'output_shape', current_shape) self._conv_net.append(layer) return current_shape def fit(self, X: np.ndarray, y: np.ndarray): """Fit the model.""" if self.random_state is not None: np.random.seed(self.random_state) X, y = check_X_y(X, y) n_input_features = X.shape[1] X = self._reshape_input(X) self._input_size = X.shape[1:] conv_output_shape = self._build_conv_layers(self._input_size) if len(conv_output_shape) > 2: self._mlp_input_size = np.prod(conv_output_shape[1:]) else: self._mlp_input_size = conv_output_shape[-1] mlp_layers = list(self.mlp_layers) self._mlp = MLPRegressor( hidden_layer_sizes=mlp_layers, activation='relu', solver='adam', max_iter=1, batch_size=self.batch_size, learning_rate_init=self.learning_rate_init, random_state=self.random_state, verbose=self.verbose ) batch_size = self._feature_batch_size(X.shape[0]) for _ in range(self.max_iter): for start in range(0, X.shape[0], batch_size): stop = min(start + batch_size, X.shape[0]) X_flat = self._transform_conv_batch(X[start:stop], training=True) self._mlp.partial_fit(X_flat, y[start:stop]) self._built = True self._n_features_in_ = n_input_features return self def predict(self, X: np.ndarray) -> np.ndarray: """Predict target values.""" check_is_fitted(self) X = check_array(X) X = self._reshape_input(X) predictions = [ self._mlp.predict(X_flat) for X_flat in self._iter_conv_features(X, training=False) ] return np.concatenate(predictions).flatten() def score(self, X: np.ndarray, y: np.ndarray) -> float: """Return R² score.""" from sklearn.metrics import r2_score return r2_score(y, self.predict(X))