<jupyter_start><jupyter_text>Tuning Tolerance(Experimentation playground for CLI tune command. See `abfs tune -h` for running this command.)<jupyter_code>from collections import namedtuple as Struct
from toolz import curry, topk, second
from time import time

import matplotlib.pyplot as plt
import numpy as np
import cv2

import abfs.cli
import abfs.keras.metrics

import tensorflow as tf

CLIParams = Struct('CLIParams', [
    'weights_path', 'size',
    'epochs', 'gpu_count', 'learning_rate', 'max_batches', 'batch_size'
])

params = CLIParams(
    'checkpoints/<INSERT WEIGHTS HERE>.hdf5', 512,
    None, 2, 0, 9999999, 99999998
)

u_net = abfs.cli.model(params)

print(f'Batch Count: {len(u_net.val_generator)}')
images, truth_masks = u_net.val_generator.__getitem__(0)

print(f'Tuning of the tolerance parameter will occur on {images.shape[0]} images')
predictions = u_net.model.predict(images)<jupyter_output>Loading weights from checkpoints/<INSERT WEIGHTS HERE>.hdf5<jupyter_text>Sample Tolerance<jupyter_code>%matplotlib inline

plt.imshow(np.squeeze(predictions[0]), cmap='gray')
plt.show()<jupyter_output><empty_output><jupyter_text>Using a sample threshold of 0.5, here's the associated mask.<jupyter_code>_, threshold = cv2.threshold(predictions[0], 1 - 0.5, 1, cv2.THRESH_BINARY)
plt.imshow(np.squeeze(threshold), cmap='gray')
plt.show()
tf.InteractiveSession()

abfs.keras.metrics.fbeta_score(
    tf.stack(np.squeeze(truth_masks[0].astype(np.float32))),
    tf.stack(np.squeeze(threshold.astype(np.float32)))
).eval()<jupyter_output><empty_output><jupyter_text>Tuning<jupyter_code>combined = np.array(list(zip(truth_masks, predictions)))

def threshold_tf(prediction, tolerance):
    return tf.cast(prediction > (1 - tolerance), tf.float32)

@curry
def f1_score_tf(tolerance, truth_mask_and_prediction):
    truth_mask = truth_mask_and_prediction[0]
    prediction_mask = threshold_tf(tolerance, truth_mask_and_prediction[1])
    
    return abfs.keras.metrics.fbeta_score(
        tf.squeeze(tf.cast(truth_mask, tf.float32)),
        tf.squeeze(prediction_mask)
    )

@curry
def f1_scores_tf(truth_masks_and_predictions, tolerance):
    return tf.map_fn(f1_score_tf(tolerance), truth_masks_and_predictions)

def f1_scores_per_tolerances_tf(tolerances, truth_masks_and_predictions):
    return tf.map_fn(f1_scores_tf(truth_masks_and_predictions), tolerances)

start = time()

TOLERANCES = np.linspace(0.02, 1, 50)

results = f1_scores_per_tolerances_tf(
    tf.constant(TOLERANCES, tf.float32),
    tf.stack(combined)
).eval()

print(f'Duration: {time() - start}')
results.shape
from matplotlib.ticker import FuncFormatter

def format_ticks(x, pos):
    return f'{x:.2f}' if pos % 3 == 0 else ''

fig, ax = plt.subplots(figsize=(10, 6))
plt.boxplot(list(results), positions=TOLERANCES, widths=0.015, sym='.')
ax.set_ylim([0, 1])
ax.set_xlim([0, 1])
ax.set_xlabel('Tolerance')
ax.set_ylabel('F1-Score')
ax.set_title('F1-Score by Adjusting Tolerance')
plt.gca().xaxis.set_major_formatter(FuncFormatter(format_ticks))
plt.show()
third = lambda value: value[2]

tolerance_populations = [(f'{tolerance:.2f}', np.median(scores), np.std(scores))
                         for tolerance, scores in zip(TOLERANCES, results)]

min(topk(5, tolerance_populations, key=second), key=third)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13060487511284982, 'fraction_numerical': 0.024074631357207343, 'mean_word_length': 6.087420042643923, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '486919', 'license': 'MIT', 'n_tokens_mistral': 1319, 'n_tokens_neox': 1241, 'n_words': 226, 'path': 'tune.ipynb', 'repo_name': 'rcdilorenzo/abfs', 'size': '3323'}