stouputils.data_science.models.base_keras module#

Keras-specific model implementation with TensorFlow integration. Provides concrete implementations for Keras model operations.

Features:

  • Transfer learning layer freezing/unfreezing

  • Keras-specific callbacks (early stopping, LR reduction)

  • Model checkpointing/weight management

  • GPU-optimized prediction pipelines

  • Keras metric/loss configuration

  • Model serialization/deserialization

Implements ModelInterface for Keras-based models.

class BaseKeras(num_classes: int, kfold: int = 0, transfer_learning: str = 'imagenet', **override_params: Any)[source]#

Bases: ModelInterface

Base class for Keras models with common functionality.

class_load() → None[source]#

Clear the session and collect garbage, reset random seeds and call the parent class method.

_fit(model: Model, x: Any, y: Any | None = None, validation_data: tuple[Any, Any] | None = None, shuffle: bool = True, batch_size: int | None = None, epochs: int = 1, callbacks: list[Callback] | None = None, class_weight: dict[int, float] | None = None, verbose: int = 0, *args: Any, **kwargs: Any) → History[source]#

Manually fit the model with a custom training loop instead of using model.fit().

This method implements a custom training loop for more control over the training process. It’s useful for implementing custom training behaviors that aren’t easily done with model.fit() such as unfreezing layers during training, resetting the optimizer, etc.

Parameters:

  • model (Model) – The model to train

  • x (Any) – Training data inputs

  • y (Any | None) – Training data targets

  • validation_data (tuple[Any, Any] | None) – Validation data as a tuple of (inputs, targets)

  • shuffle (bool) – Whether to shuffle the training data every epoch

  • batch_size (int | None) – Number of samples per gradient update.

  • epochs (int) – Number of epochs to train the model.

  • callbacks (list[Callback] | None) – List of callbacks to apply during training.

  • class_weight (dict[int, float] | None) – Optional dictionary mapping class indices to weights.

  • verbose (int) – Verbosity mode.

Returns:

Training history

Return type:

History

_get_architectures(optimizer: Any | None = None, loss: Any | None = None, metrics: list[Any] | None = None) → tuple[Model, Model][source]#

Get the model architecture and compile it if enough information is provided.

This method builds and returns the model architecture. If optimizer, loss, and (optionally) metrics are provided, the model will be compiled.

Parameters:

  • optimizer (Any) – The optimizer to use for training

  • loss (Any) – The loss function to use for training

  • metrics (list[Any] | None) – The metrics to use for evaluation

Returns:

The final model and the base model

Return type:

tuple[Model, Model]

_get_callbacks() → list[Callback][source]#

Get the callbacks for training.

_get_metrics() → list[Metric][source]#

Get the metrics for training.

Returns:

List of metrics to track during training including accuracy, AUC, etc.

Return type:

list

_get_optimizer(learning_rate: float = 0.0, mode: int = 1) → Optimizer[source]#

Get the optimizer for training.

Parameters:

  • learning_rate (float) – Learning rate

  • mode (int) – Mode to use

Returns:

Optimizer

Return type:

Optimizer

_get_loss(mode: int = 0) → Loss[source]#

Get the loss function for training depending on the mode.

  • 0: CategoricalCrossentropy (default)

  • 1: CategoricalFocalCrossentropy

  • 2: Next Generation Loss (with alpha = 2.4092)

Parameters:

mode (int) – Mode to use

Returns:

Loss function

Return type:

Loss

_find_best_learning_rate_subprocess(dataset: Dataset, queue: Queue | None = None, verbose: int = 0) → dict[str, Any] | None[source]#

Helper to run learning rate finder, potentially in a subprocess.

Parameters:

  • dataset (Dataset) – Dataset to use for training.

  • queue (multiprocessing.Queue | None) – Queue to put results in (if running in subprocess).

  • verbose (int) – Verbosity level.

Returns:

Return values

Return type:

dict[str, Any] | None

_find_best_unfreeze_percentage_subprocess(dataset: Dataset, queue: Queue | None = None, verbose: int = 0) → dict[str, Any] | None[source]#

Helper to run unfreeze percentage finder, potentially in a subprocess.

Parameters:

  • dataset (Dataset) – Dataset to use for training.

  • queue (multiprocessing.Queue | None) – Queue to put results in (if running in subprocess).

  • verbose (int) – Verbosity level.

Returns:

Return values

Return type:

dict[str, Any] | None

_train_subprocess(dataset: Dataset, checkpoint_path: str, temp_dir: TemporaryDirectory[str] | None = None, queue: Queue | None = None, verbose: int = 0) → dict[str, Any] | None[source]#

Train the model in a subprocess.

The reason for this is that when training too much models on the same process, your process may be killed by the OS since it used too much resources over time. So we train each model in a separate process to avoid this issue.

Parameters:

  • model (Model) – Model to train

  • dataset (Dataset) – Dataset to train on

  • checkpoint_path (str) – Path to save the best model checkpoint

  • temp_dir (TemporaryDirectory[str] | None) – Temporary directory to save the visualizations

  • queue (multiprocessing.Queue | None) – Queue to put the history in

  • verbose (int) – Verbosity level

Returns:

Return values

Return type:

dict[str, Any]

class_predict(X_test: Iterable[ndarray[Any, dtype[Any]]] | DatasetV2) → Iterable[ndarray[Any, dtype[Any]]][source]#

Predict the class for the given input data.

Parameters:

X_test (Iterable[NDArray[Any]]) – List of inputs to predict (e.g. a batch of images)

Returns:

A batch of predictions (model.predict())

Return type:

Iterable[NDArray[Any]]

_log_final_model() → None[source]#

Log the best model (and its weights).

class_evaluate(dataset: Dataset, metrics_names: tuple[str, ...] = (), save_model: bool = False, verbose: int = 0) → bool[source]#

Evaluate the model using the given predictions and labels.

Parameters:

  • dataset (Dataset) – Dataset containing the training and testing data

  • metrics_names (list[str]) – List of metrics to plot (default to all metrics)

  • save_model (bool) – Whether to save the best model

  • verbose (int) – Level of verbosity

Returns:

True if evaluation was successful

Return type:

bool