source.utils.custom_classes.EvalAnalyzer

  1#!/usr/bin/env python
  2# coding: utf-8
  3
  4import os
  5import numpy as np
  6import pandas as pd
  7import matplotlib.pyplot as plt
  8import torch
  9from pathlib import Path
 10from PIL import Image
 11from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
 12from sklearn.calibration import calibration_curve
 13
 14from source.utils.custom_classes.GarbageClassifier import GarbageClassifier
 15from source.utils.custom_classes.GarbageDataModule import GarbageDataModule
 16from source.utils import config as cfg
 17
 18
 19class GarbageModelAnalyzer:
 20    """
 21    Analyzer class for evaluating and visualizing garbage
 22    classification model performance.
 23
 24    This class provides comprehensive tools for model evaluation
 25    including confusion matrices, calibration curves, and misclassified
 26    sample visualization. It handles both model loading and data
 27    module setup, with support for GPU acceleration.
 28
 29    Attributes
 30    ----------
 31    dataset_path : str
 32        Path to the dataset folder containing images organized by class.
 33    performance_path : str
 34        Path to the directory where performance figures are saved.
 35    device : torch.device
 36        Computational device (CUDA if available, otherwise CPU).
 37    df : pd.DataFrame or None
 38        Metadata DataFrame containing dataset information.
 39    model : GarbageClassifier or None
 40        Loaded model instance for inference.
 41    data_module : GarbageDataModule or None
 42        Data module for dataset loading and preprocessing.
 43    """
 44
 45    def __init__(self, dataset_path=None, performance_path=None):
 46        """
 47        Initialize the GarbageModelAnalyzer instance.
 48
 49        Loads metadata from the dataset path and sets up paths
 50        for storing performance figures. Detects CUDA availability
 51        for GPU acceleration.
 52
 53        Parameters
 54        ----------
 55        dataset_path : str, optional
 56            Path to the dataset folder. Default is
 57            "../data/raw/sample_dataset".
 58        performance_path : str, optional
 59            Path to the directory for saving performance figures. Default is
 60            "../reports/figures/performance/".
 61
 62        Returns
 63        -------
 64        None
 65        """
 66        self.dataset_path = dataset_path or os.path.join(
 67            "..", "data", "raw", "sample_dataset"
 68        )
 69        self.performance_path = performance_path or \
 70            "../reports/figures/performance/"
 71        self.device = torch.device(
 72            "cuda" if torch.cuda.is_available() else "cpu")
 73        metadata_path = Path(cfg.DATASET_PATH).parent / "metadata.csv"
 74
 75        if metadata_path.exists():
 76            self.df = pd.read_csv(metadata_path)
 77        else:
 78            print(f"Warning: metadata.csv not found at {metadata_path}")
 79            self.df = None
 80
 81        self.model = None
 82        self.data_module = None
 83
 84    def load_model(self, checkpoint_path=None, num_classes=None):
 85        """
 86        Load a trained GarbageClassifier model from a checkpoint.
 87
 88        Loads a model from a PyTorch Lightning checkpoint and moves it to the
 89        specified device in evaluation mode.
 90
 91        Parameters
 92        ----------
 93        checkpoint_path : str, optional
 94            Path to the model checkpoint file. If None, uses cfg.MODEL_PATH.
 95            Default is None.
 96        num_classes : int, optional
 97            Number of output classes. If None, uses cfg.NUM_CLASSES.
 98            Default is None.
 99
100        Returns
101        -------
102        None
103
104        Notes
105        -----
106        Model is automatically set to evaluation mode and moved
107        to the configured device.
108        """
109        checkpoint_path = checkpoint_path or cfg.MODEL_PATH
110        num_classes = num_classes or cfg.NUM_CLASSES
111        print("Loading model...")
112        self.model = GarbageClassifier.load_from_checkpoint(
113            checkpoint_path, num_classes=num_classes
114        )
115        self.model.to(self.device).eval()
116        print("Model loaded.")
117
118    def setup_data(self, batch_size=32):
119        """
120        Set up the data module and filter metadata for available samples.
121
122        Initializes the GarbageDataModule and creates a filtered subset
123        of metadata containing only files present in the dataset directory.
124
125        Parameters
126        ----------
127        batch_size : int, optional
128            Batch size for data loading. Default is 32.
129
130        Returns
131        -------
132        None
133
134        Notes
135        -----
136        Creates self.df_subset containing only samples actually
137        present in the dataset.
138        """
139
140        self.data_module = GarbageDataModule(batch_size=batch_size)
141        self.data_module.setup()
142        file_names = []
143        for root, dirs, files in os.walk(cfg.DATASET_PATH):
144            for file in files:
145                file_names.append(file)
146        self.df_subset = (
147            self.df[
148                self.df["filename"].isin(file_names)
149            ].reset_index(drop=True).copy()
150        )
151
152    def evaluate_loader(self, loader):
153        """
154        Evaluate model on a data loader and collect predictions
155        and probabilities.
156
157        Iterates through a PyTorch DataLoader, performs inference,
158        and collects predictions, true labels, and confidence scores.
159
160        Parameters
161        ----------
162        loader : torch.utils.data.DataLoader
163            DataLoader containing batches of (images, labels).
164
165        Returns
166        -------
167        tuple
168            A tuple containing:
169            - all_preds : torch.Tensor
170                Predicted class indices.
171            - all_labels : torch.Tensor
172                True class labels.
173            - all_probs : np.ndarray
174                Confidence scores for each class (shape: [N, num_classes]).
175
176        Notes
177        -----
178        Model inference is performed without gradient computation
179        for efficiency.
180        Probabilities are computed using softmax activation.
181        """
182        all_preds, all_labels, all_probs = [], [], []
183        with torch.no_grad():
184            for xb, yb in loader:
185                xb = xb.to(self.device)
186                yb = yb.to(self.device)
187                out = self.model(xb)
188                preds = out.argmax(dim=1)
189                probs = torch.softmax(out, dim=1)
190                all_preds.append(preds)
191                all_probs.append(probs.cpu())
192                all_labels.append(yb)
193        all_preds = torch.cat(all_preds)
194        all_labels = torch.cat(all_labels)
195        all_probs = torch.cat(all_probs).numpy()
196        return all_preds, all_labels, all_probs
197
198    def plot_confusion_matrix(self, labels, preds, set_name="Train"):
199        """
200        Plot and save confusion matrices (raw and normalized)
201        with class metrics.
202
203        Generates both raw and normalized confusion matrices,
204        computes TP, FP, FN, TN per class, and saves visualizations
205        as PDF files.
206
207        Parameters
208        ----------
209        labels : array-like
210            True class labels.
211        preds : array-like
212            Predicted class labels.
213        set_name : str, optional
214            Name of the dataset split (e.g., "Train", "Val", "Test")
215            for plot titles and filenames. Default is "Train".
216
217        Returns
218        -------
219        None
220
221        Side Effects
222        -----------
223        - Saves confusion_mat_{set_name.lower()}.pdf to performance_path.
224        - Saves confusion_mat_{set_name.lower()}_norm.pdf to performance_path.
225        - Prints TP, FP, FN, TN statistics for each class.
226        - Displays matplotlib figures.
227
228        Notes
229        -----
230        Normalized confusion matrix divides by row sums to show percentages
231        per class.
232        """
233        num_classes = self.data_module.num_classes
234        cm = confusion_matrix(labels, preds, labels=range(num_classes))
235        disp = ConfusionMatrixDisplay(
236            confusion_matrix=cm, display_labels=cfg.CLASS_NAMES
237        )
238        disp.plot(cmap=plt.cm.Blues)
239        plt.title(f"Confusion Matrix - {set_name} set")
240        plt.savefig(
241            os.path.join(
242                self.performance_path, f"confusion_mat_{set_name.lower()}.pdf"
243            ),
244            dpi=80,
245        )
246        plt.show()
247
248        # Normalized
249        cm_norm = cm.astype("float") / cm.sum(axis=1)[:, np.newaxis]
250        disp_norm = ConfusionMatrixDisplay(
251            confusion_matrix=cm_norm, display_labels=cfg.CLASS_NAMES
252        )
253        disp_norm.plot(cmap=plt.cm.Blues)
254        plt.title(f"Normalized Confusion Matrix - {set_name} set")
255        plt.savefig(
256            os.path.join(
257                self.performance_path,
258                f"confusion_mat_{set_name.lower()}_norm.pdf"
259            ),
260            dpi=80,
261        )
262        plt.show()
263
264        # TP, FP, FN, TN
265        TP = np.diag(cm)
266        FP = cm.sum(axis=0) - TP
267        FN = cm.sum(axis=1) - TP
268        TN = cm.sum() - (TP + FP + FN)
269        for i in range(num_classes):
270            print(f"Clase {i}: TP={TP[i]}, FP={FP[i]}, FN={FN[i]}, TN={TN[i]}")
271
272    def plot_top_misclassified(
273        self, df_set, y_true, y_pred, y_proba, N=10, filename=None
274    ):
275        """
276        Visualize the top N misclassified samples with lowest confidence.
277
278        Identifies misclassified samples, sorts them by confidence
279        on the true class, and displays the least confident (worst)
280        predictions with their images.
281
282        Parameters
283        ----------
284        df_set : pd.DataFrame
285            DataFrame containing sample metadata with 'label' and 'filename'
286            columns.
287        y_true : array-like
288            True class labels (can be integers or strings).
289        y_pred : array-like
290            Predicted class labels (integers).
291        y_proba : np.ndarray
292            Confidence scores matrix (shape: [N, num_classes]).
293        N : int, optional
294            Number of top misclassified samples to display. Default is 10.
295        filename : str, optional
296            Filename (without extension) to save the figure. If provided,
297            saves as PDF to performance_path. Default is None (no save).
298
299        Returns
300        -------
301        None
302
303        Side Effects
304        -----------
305        - Displays matplotlib figure with misclassified samples.
306        - Saves figure to performance_path/{filename}.pdf
307        if filename is provided.
308
309        Notes
310        -----
311        Samples are sorted by confidence on the true class in ascending order,
312        showing the most uncertain misclassifications first.
313        Images are loaded from dataset_path/{label}/{filename} structure.
314        """
315        y_true = np.array(y_true)
316        y_pred = np.array(y_pred)
317
318        # Handle case: y_true are integers vs strings
319        if np.issubdtype(y_true.dtype, np.integer):
320            true_indices = y_true
321            classes = sorted(df_set["label"].unique())
322        else:
323            classes = sorted(np.unique(y_true))
324            class_to_idx = {cls: i for i, cls in enumerate(classes)}
325            true_indices = np.array([class_to_idx[label] for label in y_true])
326
327        true_confidences = y_proba[np.arange(len(y_true)), true_indices]
328        misclassified_idx = np.where(y_true != y_pred)[0]
329
330        if len(misclassified_idx) == 0:
331            print("No misclassified samples found!")
332            return
333
334        sorted_idx = misclassified_idx[
335            np.argsort(true_confidences[misclassified_idx])
336        ]
337        selected_idx = sorted_idx[:N]
338
339        plt.figure(figsize=(15, 3 * (N // 5 + 1)))
340        for i, idx in enumerate(selected_idx, 1):
341            row = df_set.iloc[idx]
342            img_path = os.path.join(
343                self.dataset_path,
344                row["label"],
345                row["filename"]
346            )
347            if not os.path.exists(img_path):
348                continue
349            try:
350                img = Image.open(img_path).convert("RGB")
351            except Exception:
352                continue
353            plt.subplot(int(np.ceil(N / 5)), 5, i)
354            plt.imshow(img)
355            plt.axis("off")
356            plt.title(
357                f"True: {row['label']}\nPred: \
358                    {classes[y_pred[idx]]}\nConf True Class: \
359                        {true_confidences[idx]:.2f}",
360                fontsize=9,
361                color="red",
362            )
363        plt.tight_layout()
364        if filename:
365            plt.savefig(
366                os.path.join(
367                    self.performance_path,
368                    f"{filename}.pdf"
369                ),
370                dpi=80
371            )
372        plt.show()
373
374    def plot_calibration_curves(self, y_true, y_probs):
375        """
376        Plot calibration curves for all classes using one-vs-rest approach.
377
378        Generates calibration curves showing the relationship between predicted
379        probabilities and actual positive fractions for each class.
380
381        Parameters
382        ----------
383        y_true : array-like or torch.Tensor
384            True class labels (integers). Shape: [N,].
385        y_probs : np.ndarray or torch.Tensor
386            Predicted probability matrix. Shape: [N, num_classes].
387
388        Returns
389        -------
390        None
391
392        Side Effects
393        -----------
394        - Displays matplotlib figure with 2x3 grid of calibration curves.
395        - One subplot per class showing calibration curve and reference
396        diagonal.
397
398        Notes
399        -----
400        Uses sklearn's calibration_curve function with 10 bins for each class.
401        Each class is converted to a binary classification problem (OVR).
402        Calibration curves below the diagonal indicate:
403        -> overconfident predictions.
404        Calibration curves above the diagonal indicate:
405        -> underconfident predictions.
406        """
407        num_classes = self.data_module.num_classes
408        fig, axes = plt.subplots(2, 3, figsize=(18, 10))
409        axes = axes.flatten()
410
411        if isinstance(y_true, torch.Tensor):
412            y_true_np = y_true.cpu().numpy()
413        else:
414            y_true_np = y_true
415
416        if isinstance(y_probs, torch.Tensor):
417            y_probs = y_probs.cpu().numpy()
418
419        for c in range(num_classes):
420            ax = axes[c]
421            y_true_c = (y_true_np == c).astype(int)
422            y_prob_c = y_probs[:, c]
423
424            frac_pos, mean_pred = calibration_curve(
425                y_true_c, y_prob_c, n_bins=10
426            )
427            ax.plot(mean_pred, frac_pos, marker="o", label=f"Class {c}")
428            ax.plot(
429                [0, 1],
430                [0, 1],
431                linestyle="--",
432                color="gray",
433                label="Reference"
434            )
435            ax.set_xlabel("Mean predicted probability")
436            ax.set_ylabel("Fraction of positives")
437            ax.set_title(f"Calibration Curve: {cfg.CLASS_NAMES[c]}")
438            ax.set_xticks(np.arange(0, 1.1, 0.1))
439            ax.set_yticks(np.arange(0, 1.1, 0.1))
440            ax.set_xlim(-0.05, 1.05)
441            ax.set_ylim(-0.05, 1.05)
442            ax.grid(True)
443            ax.legend(fontsize=8)
444
445        plt.tight_layout()
446        plt.show()