<jupyter_start><jupyter_text>Caution: important experiment, please do not adjustThe contents within this file are for running a publication-quality experiment and producing the associated figures and results. As such, it is finely tuned for specific high-quality graphics using the `primary.mplstyle` file. Please do not adjust. If you need to replicate results on your local system, please create a new notebook and copy contents from here.<jupyter_code>import sys
sys.path.append('../..')  # Expose top level program access

import time
import pickle
import logging
import torch
import numpy as np
import pandas as pd
import pytorch_lightning as pl
import matplotlib.pyplot as plt

from functools import reduce
from pathlib import Path

from modules.cae_base_module import CAEBaseModule
from modules.vae_base_module import VAEBaseModule
from modules.aae_base_module import AAEBaseModule
from modules.pca_base_module import PCABaseModule
from models import supported_models
from datasets import supported_datamodules
from utils import tools, metrics, supported_preprocessing_transforms

logging.getLogger("matplotlib").setLevel(logging.WARNING)  # Suppress verbose font warnings from matplotlib
# Import configurations and paths to logged models
root = Path.cwd().parents[1]
log_path = root / 'logs' / 'NoveltyMNISTDataModule'
paths_to_archived_models = list(Path(log_path).glob('**/archive_v*'))

print('Found archived models:\n------')
print('\n'.join([f'{p.parent.name}/{p.name}' for p in paths_to_archived_models]))
# Load the checkpoints for all the training modules and save them in a dictionary
module_catalog = {}

for pth in paths_to_archived_models:
    
    model_type = pth.parent.name
    model_name = pth.name
    config = tools.load_config(pth / 'configuration.yaml', silent=True)
    
    # Unsupervising region proposal is called implicity in this line, see utils/preprocessing.py
    preprocessing_transforms = supported_preprocessing_transforms[config['data-parameters']['preprocessing']]

    datamodule = supported_datamodules[config['experiment-parameters']['datamodule']](
        data_transforms=preprocessing_transforms,
        **config['data-parameters'])
    
    if 'PCA' in model_type:

        _, pca_test_labels = datamodule.split(train=False)
        
        with open(pth / 'fitted_model.p', 'rb') as f:
            model = pickle.load(f)
            module = PCABaseModule(model)
    else:
        ckpt_path = next(iter((pth / 'checkpoints').glob('val_*')))

        datamodule.setup('test')

        # Handle the various model instantiations
        if 'AAE' in model_type:
            model = supported_models[model_type](
                in_shape=datamodule.data_shape,
                latent_nodes=config['module-parameters']['latent_nodes'])
            module = AAEBaseModule(model, **config['module-parameters'])
        elif 'VAE' in model_type:
            model = supported_models[model_type](
                in_shape=datamodule.data_shape,
                latent_nodes=config['module-parameters']['latent_nodes'])
            module = VAEBaseModule(model, **config['module-parameters'])
        elif 'CAE' in model_type:
            model = supported_models[model_type](in_shape=datamodule.data_shape)
            module = CAEBaseModule(model, **config['module-parameters'])
        else:
            raise ValueError(f'Model substring not found, got {model_type}')

        # Load the state_dict into the module architecture
        checkpoint = torch.load(ckpt_path)
        module.load_state_dict(checkpoint['state_dict'])
    
    if model_type not in module_catalog:
        module_catalog[model_type] = {}
    module_catalog[model_type][model_name] = module
    print(f'Loaded state dict for: {model_type}/{model_name}')
# Collect scores and labels on test set for all models and save to a dictionary
result_catalog = {}

for model_type, model_variants in module_catalog.items():
    start_time = time.time()
    
    for model_name, module in model_variants.items():
        
        test_novelty_scores = []
        test_novelty_labels = []
        short_name = model_type + '-' + model_name.split('_')[1]
        if short_name not in result_catalog:
            result_catalog[short_name] = {
                'scores': [], 
                'labels': []
            }
        print(f'------\n{short_name}\n------')
        
        # Handle the PCA and autoencoder frameworks separately
        if 'PCA' in short_name:
            test_novelty_scores = module.transform_pipeline(datamodule)
            test_novelty_labels = pca_test_labels
        else:
        
            # Set the model/components in evaluation mode
            if 'AAE' in model_type:
                module.encoder.eval()
                module.decoder.eval()
            else:
                module.model.eval()
        
            with torch.no_grad():
                for batch_nb, batch in enumerate(datamodule.test_dataloader()):

                    result = module.test_step(batch, batch_nb)

                    test_novelty_scores.extend(result['scores'])
                    test_novelty_labels.extend(result['labels'].cpu().numpy())
                    print(f'[BATCH {batch_nb}] Mean score: {np.mean(result["scores"])}')
                    
        result_catalog[short_name]['scores'] = test_novelty_scores
        result_catalog[short_name]['labels'] = test_novelty_labels
        print(f'{short_name} Time: {time.time() - start_time:.2f} s\n')
# Use standard custom styling
plt.style.use('../../utils/_extras/primary.mplstyle')

# Plot the ROC curves with AUCs for each model
fig, ax = plt.subplots(figsize=(8, 8))
for sn, results in result_catalog.items():

    scores = results['scores']
    labels = results['labels']
    print(f'{sn}: {sum(scores)}')
    
    fpr, tpr, thresholds, auc = metrics.roc(scores, labels)
    
    ax.plot(fpr, tpr, label=f'{sn}: {auc:.2f}')
    
ax.set_xlabel('FPR')
ax.set_ylabel('TPR')
ax.plot([0., 1.], [0., 1.], 'k:', lw=2, label='Random: 0.5')
ax.legend(title='AUC', bbox_to_anchor=(1.05, 1.), loc='upper left')

ax.axhline(0, 0, c='k', lw=1)
ax.axvline(0, 0, c='k', lw=1)

plt.savefig('ROC-all_NoveltyMNIST.pdf', format='pdf', bbox_inches='tight')
plt.show()
print('Random ROC AUC: 0.5')<jupyter_output>SimpleVAE-v1: 3474.07666015625
SimpleAAE-v3: 5770.53466796875
BaselineCAE-v1: 2516.391357421875
CompressionCAEHighCapacity-v1: 2862.93408203125
StandardPCA-v2: 1.6779717100269127e-22
CompressionCAEMidCapacity-v1: 2968.94384765625
CompressionCAEMidCapacity-v2: 3093.654296875<jupyter_text>By evaluating the Precision(k) across all test samples, we make the assumption that all the samples are of the positive class. Clearly this iss an erroneous assumption; conceptually, we should evaluate the Precision(k) only for the novel samples. But this sort of evaluation places a bias on the use of Precision(k) to measure the performance of our detector. By the definition of a novelty, is *cannot* be known beforehand their exact number or frequency. Therefore by analyzing Precision(k) only for the samples we know are novel, because we have insider information about the test data (e.g. labels), we are only evaluating misclassifications of the positive class. Knowing the proportion of correct novelty predictions to total novelty predictions (precision) is important, but it omits a crucial aspect of the usefullness of a detector. Namely, how many *more* false positives would 'flood the results' if typical images were included in the analysis---which they *would* be in an operational scenario.<jupyter_code># Plot the Precision(k) across all test images.
fig, ax = plt.subplots(figsize=(8, 8))
for sn, results in result_catalog.items():
    
    scores = results['scores']
    labels = results['labels']
    pak = metrics.precision_at_k(scores, labels)

    uniques, counts = np.unique(labels, return_counts=True)
    random = counts.min() / counts.sum()

    ax.plot(pak, label=f'{sn}: {pak[int(sum(labels))]:.2f}')
    
ax.plot([0, len(pak)], [random, random], 'k:', lw=2, label='$k_{\mathrm{novel}}/k_{\mathrm{total}}$')

ax.set_xlabel('$k$')
ax.set_ylabel('Precision($k$)')
ax.legend(title='Precision($k_{\mathrm{novel}}$)', bbox_to_anchor=(1.05, 1.), loc='upper left')

ax.set_xticks([counts[1], int(counts.sum()/2), counts.sum()])
ax.set_xticklabels([str(counts[1])+' ($k_{\mathrm{novel}}$)', int(counts.sum()/2), counts.sum()])

# Draw axes for styling
ax.axhline(0, 0, c='k', lw=0.5)
ax.axvline(0, 0, c='k', lw=0.5)

plt.savefig('Pak-all_NoveltyMNIST.pdf', format='pdf', bbox_inches='tight')
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11689063948100092, 'fraction_numerical': 0.018651529193697867, 'mean_word_length': 3.6067235859124867, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '638638', 'license': 'Apache-2.0', 'n_tokens_mistral': 2827, 'n_tokens_neox': 2650, 'n_words': 701, 'path': 'experiments/novelty-mnist_reconstruction-detections/novelty-mnist_reconstruction-detections.ipynb', 'repo_name': 'brahste/novelty-detection', 'size': '8632'}