67 KiB
67 KiB
In [ ]:
!pip install -r requirements.txt
In [ ]:
import os, shutil, zipfile, random, time, json
from collections import Counter
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from PIL import Image
import tensorflow as tf
from tensorflow.keras import layers, models, callbacks
from tensorflow.keras.applications import EfficientNetV2B0
from tensorflow.keras.applications.efficientnet_v2 import preprocess_input
from tensorflow.keras.optimizers import AdamW
from tensorflow.keras.optimizers.schedules import CosineDecay
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.utils.class_weight import compute_class_weight
from sklearn.model_selection import train_test_split
# Optional: perceptual hashing for dedup (pip install imagehash)
try:
import imagehash
HAS_IMAGEHASH = True
except ImportError:
HAS_IMAGEHASH = False
# Optional: scipy for temperature optimization
try:
from scipy.optimize import minimize_scalar
HAS_SCIPY = True
except ImportError:
HAS_SCIPY = False
# Detect environment
try:
from google.colab import drive
IS_COLAB = True
print("Running on Google Colab")
except ModuleNotFoundError:
IS_COLAB = False
print(f"Running locally (TF {tf.__version__}, GPU: {tf.config.list_physical_devices('GPU')})")
tf.keras.mixed_precision.set_global_policy('float32')
# Hyperparams
IMG_SIZE = (224, 224)
BATCH_SIZE = 32
SEED = 42
random.seed(SEED)
np.random.seed(SEED)
tf.random.set_seed(SEED)
AUTOTUNE = tf.data.AUTOTUNE
print(f"Setup OK. IMG={IMG_SIZE}, BATCH={BATCH_SIZE}")
In [ ]:
if IS_COLAB:
drive.mount('/content/drive')
archive_path = '/content/drive/MyDrive/jagung/dataset.zip'
destination_path = '/content/dataset.zip'
extract_path = '/content/dataset'
else:
base = os.getcwd()
archive_path = os.path.join(base, 'dataset.zip')
destination_path = archive_path
extract_path = os.path.join(base, 'dataset')
if os.path.exists(destination_path):
if not os.path.exists(extract_path) or len(os.listdir(extract_path)) == 0:
os.makedirs(extract_path, exist_ok=True)
print("Extracting dataset...")
try:
with zipfile.ZipFile(destination_path, 'r') as zip_ref:
zip_ref.extractall(path=extract_path)
print("Extraction completed!")
except Exception as e:
print(f"Extraction failed: {e}")
else:
print("Dataset ready.")
else:
print(f"dataset.zip not found at {destination_path}. Upload dataset.zip to Google Drive / MyDrive/jagung/")
print("Or run preprocessing.py locally and upload the resulting dataset.zip")In [ ]:
# --- Determine dataset path ---
dataset_path = extract_path
print(f"Dataset path: {dataset_path}")
# Check if dataset already validated and split exists — skip if so
split_output_dir = "/content/dataset_split" if IS_COLAB else os.path.join(os.getcwd(), "dataset_split")
if os.path.exists(os.path.join(split_output_dir, 'train')):
print("Split dataset already exists. Skipping validation and split.")
# Still need class_names and counts for downstream cells
train_dir = os.path.join(split_output_dir, 'train')
val_dir = os.path.join(split_output_dir, 'val')
test_dir = os.path.join(split_output_dir, 'test')
print(f'Train: {sum(len(files) for _, _, files in os.walk(train_dir))} | '
f'Val: {sum(len(files) for _, _, files in os.walk(val_dir))} | '
f'Test: {sum(len(files) for _, _, files in os.walk(test_dir))}')
else:
MIN_FILE_SIZE = 512
MIN_DIM = 32
MAX_ASPECT = 5.0
def remove_augmented_duplicates(directory):
removed = 0
for root, dirs, files in os.walk(directory):
for file in files:
if file.startswith("augmented_"):
try:
os.remove(os.path.join(root, file))
removed += 1
except OSError:
pass
return removed
def clean_and_validate_images(directory):
stats = {"too_small": 0, "corrupt": 0, "small_dims": 0, "extreme_aspect": 0, "low_var": 0, "ok": 0}
for root, dirs, files in os.walk(directory):
for file in files:
fp = os.path.join(root, file)
try:
if os.path.getsize(fp) < MIN_FILE_SIZE:
os.remove(fp); stats["too_small"] += 1; continue
except OSError:
continue
try:
img = Image.open(fp); img.verify()
except Exception:
try: os.remove(fp); stats["corrupt"] += 1
except OSError: pass
continue
try:
img = Image.open(fp)
w, h = img.size
if w < MIN_DIM or h < MIN_DIM:
os.remove(fp); stats["small_dims"] += 1; continue
aspect = w / max(h, 1)
if aspect > MAX_ASPECT or aspect < 1.0 / MAX_ASPECT:
os.remove(fp); stats["extreme_aspect"] += 1; continue
if img.mode not in ('RGB', 'RGBA'):
img = img.convert('RGB'); img.save(fp)
arr = np.array(img).astype(np.float32)
if np.std(arr) < 2.0:
os.remove(fp); stats["low_var"] += 1; continue
stats["ok"] += 1
except Exception:
try: os.remove(fp); stats["corrupt"] += 1
except OSError: pass
return stats
print("1. Removing augmented duplicates...")
n_aug = remove_augmented_duplicates(dataset_path)
print(f" Removed {n_aug} augmented_* files")
print("2. Validating images...")
stats = clean_and_validate_images(dataset_path)
print(f" OK: {stats['ok']} | Removed: too_small={stats['too_small']} corrupt={stats['corrupt']} "
f"small_dims={stats['small_dims']} aspect={stats['extreme_aspect']} low_var={stats['low_var']}")
if HAS_IMAGEHASH:
print("3. Perceptual hash dedup...")
seen, removed = {}, 0
for cn in sorted(os.listdir(dataset_path)):
cp = os.path.join(dataset_path, cn)
if not os.path.isdir(cp): continue
for f in sorted(os.listdir(cp)):
fp = os.path.join(cp, f)
if not os.path.isfile(fp): continue
try:
ah = imagehash.average_hash(Image.open(fp).convert('RGB'))
for sk, (sp, sc) in seen.items():
if ah - imagehash.hex_to_hash(sk) <= 5:
try: os.remove(fp); removed += 1
except OSError: pass
break
else:
seen[str(ah)] = (fp, cn)
except Exception:
pass
print(f" Removed {removed} near-duplicates")
else:
print("3. Perceptual hash dedup SKIPPED (pip install imagehash)")
total = sum(len(files) for _, _, files in os.walk(dataset_path))
print(f"\nTotal clean images: {total}")In [ ]:
def extract_source_prefix(filename):
f = os.path.splitext(filename)[0]
if f.startswith('IMG_'): return 'phone'
if f.startswith('Corn_'): return 'lab_corn'
for prefix in ['CBS', 'GLS', 'NLS', 'CLS']:
if f.startswith(prefix): return 'lab_disease'
for prefix in ['SCR', 'CR', 'NLB', 'SLB', 'SRS']:
if f.startswith(prefix): return 'lab_rust_blight'
return 'other'
def stratified_split_by_source(dataset_path, output_dir, ratios=(0.7, 0.15, 0.15), seed=42):
class_images = {}
for cn in sorted(os.listdir(dataset_path)):
cp = os.path.join(dataset_path, cn)
if not os.path.isdir(cp): continue
class_images[cn] = []
for f in os.listdir(cp):
fp = os.path.join(cp, f)
if os.path.isfile(fp):
class_images[cn].append((fp, f, extract_source_prefix(f)))
for split in ['train', 'val', 'test']:
for cn in class_images:
os.makedirs(os.path.join(output_dir, split, cn), exist_ok=True)
rng = np.random.RandomState(seed)
for cn, images in class_images.items():
by_source = {}
for fp, fn, src in images:
by_source.setdefault(src, []).append((fp, fn))
train_files, val_files, test_files = [], [], []
for src, src_images in by_source.items():
n = len(src_images)
rng.shuffle(src_images)
n_train = max(1, int(n * ratios[0]))
n_val = max(1, int(n * ratios[1]))
train_files.extend(src_images[:n_train])
val_files.extend(src_images[n_train:n_train + n_val])
test_files.extend(src_images[n_train + n_val:])
for fp, fn in train_files:
shutil.copy(fp, os.path.join(output_dir, 'train', cn, fn))
for fp, fn in val_files:
shutil.copy(fp, os.path.join(output_dir, 'val', cn, fn))
for fp, fn in test_files:
shutil.copy(fp, os.path.join(output_dir, 'test', cn, fn))
train_srcs = Counter(extract_source_prefix(fn) for _, fn in train_files)
print(f" {cn}: train={len(train_files)} val={len(val_files)} test={len(test_files)} | "
f"sources={dict(train_srcs)}")
output_dir = split_output_dir # Already defined in previous cell
if os.path.exists(os.path.join(output_dir, 'train')):
print("Split already exists. Skipping.")
else:
if os.path.exists(output_dir):
shutil.rmtree(output_dir)
print("Splitting dataset 70:15:15 (stratified by source)...")
stratified_split_by_source(dataset_path, output_dir, seed=SEED)
print("Done.")
train_dir = os.path.join(output_dir, 'train')
val_dir = os.path.join(output_dir, 'val')
test_dir = os.path.join(output_dir, 'test')
def count_images(path):
return sum(len(files) for _, _, files in os.walk(path))
print(f'Train: {count_images(train_dir)} | Val: {count_images(val_dir)} | Test: {count_images(test_dir)}')In [ ]:
# ─── RandAugment Utilities ───
def sample_beta_distribution(size, a=0.2, b=0.2):
g1 = tf.random.gamma([size], a, dtype=tf.float32)
g2 = tf.random.gamma([size], b, dtype=tf.float32)
return g2 / (g1 + g2 + 1e-8)
def _apply_contrast(img, factor):
mean = tf.reduce_mean(tf.cast(img, tf.float32), axis=(0, 1), keepdims=True)
return mean + factor * (tf.cast(img, tf.float32) - mean)
def _apply_brightness(img, delta):
return tf.clip_by_value(tf.cast(img, tf.float32) + delta, 0.0, 255.0)
def _apply_hue(img, delta):
hsv = tf.image.rgb_to_hsv(tf.cast(img, tf.float32) / 255.0)
h = hsv[..., 0] + delta
h = h - tf.floor(h)
hsv_h = tf.stack([h, hsv[..., 1], hsv[..., 2]], axis=-1)
return tf.image.hsv_to_rgb(hsv_h) * 255.0
@tf.function(reduce_retracing=True)
def _randaug_select(images, ops_per_image=3, magnitude=0.7):
# Graph-mode-safe RandAugment. No tf.image.random_* or Keras layers
# (both trace as Python bool checks on tensor bounds internally).
def _zoom(img):
h = tf.cast(tf.shape(img)[0], tf.float32)
w = tf.cast(tf.shape(img)[1], tf.float32)
z = tf.random.uniform([], 1.0 - 0.2 * magnitude, 1.0 + 0.2 * magnitude)
zh = tf.cast(h / z, tf.int32); zw = tf.cast(w / z, tf.int32)
zoomed = tf.image.resize(tf.expand_dims(img, 0), [zh, zw], method='bilinear')[0]
return tf.image.resize_with_crop_or_pad(zoomed, tf.cast(h, tf.int32), tf.cast(w, tf.int32))
def _translate(img):
h, w = tf.shape(img)[0], tf.shape(img)[1]
tx = tf.cast(tf.random.uniform([], -0.15 * magnitude, 0.15 * magnitude) * tf.cast(w, tf.float32), tf.int32)
ty = tf.cast(tf.random.uniform([], -0.15 * magnitude, 0.15 * magnitude) * tf.cast(h, tf.float32), tf.int32)
return tf.roll(img, [ty, tx], axis=[0, 1])
def op_flip(img): return tf.image.random_flip_left_right(img)
def op_flip_v(img): return tf.image.random_flip_up_down(img)
def op_rotate_90(img):
k = tf.random.uniform([], 0, 4, dtype=tf.int32)
return tf.image.rot90(img, k)
def op_zoom(img): return _zoom(img)
def op_translate(img): return _translate(img)
def op_contrast(img):
f = tf.random.uniform([], 1.0 - 0.5 * magnitude, 1.0 + 0.5 * magnitude)
return _apply_contrast(img, f)
def op_brightness(img):
d = tf.random.uniform([], -0.3 * magnitude * 255.0, 0.3 * magnitude * 255.0)
return _apply_brightness(img, d)
def op_hue(img):
d = tf.random.uniform([], -0.08 * tf.maximum(magnitude, 0.01), 0.08 * tf.maximum(magnitude, 0.01))
return _apply_hue(img, d)
def op_saturation(img):
lo = tf.maximum(0.5, 1.0 - 0.8 * magnitude)
hi = 1.0 + 0.8 * magnitude
f = tf.random.uniform([], lo, hi)
hsv = tf.image.rgb_to_hsv(tf.cast(img, tf.float32) / 255.0)
s = tf.clip_by_value(hsv[..., 1] * f, 0.0, 1.0)
return tf.image.hsv_to_rgb(tf.stack([hsv[..., 0], s, hsv[..., 2]], axis=-1)) * 255.0
def op_solarize(img):
thresh = tf.random.uniform([], 0.3, 0.8) * 255.0
f = tf.cast(img, tf.float32)
return tf.where(f < thresh, f, 255.0 - f)
def op_blur(img):
h, w = tf.shape(img)[0], tf.shape(img)[1]
sf = tf.random.uniform([], 2, 4, dtype=tf.int32)
small = tf.image.resize(tf.expand_dims(tf.cast(img, tf.float32), 0), [h // sf, w // sf], method='bilinear')
return tf.image.resize(small, [h, w], method='bilinear')[0]
def op_fog(img):
fl = tf.random.uniform([], 0.1, 0.1 + 0.4 * magnitude)
fc = tf.random.uniform([3], 0.7, 1.0) * 255.0
return tf.cast(img, tf.float32) * (1.0 - fl) + tf.reshape(fc, [1, 1, 3]) * fl
def op_shadow(img):
op = tf.random.uniform([], 0.2, 0.2 + 0.5 * magnitude)
return tf.cast(img, tf.float32) * (1.0 - op * 0.6)
def op_resolution_drop(img):
h, w = tf.shape(img)[0], tf.shape(img)[1]
sf = tf.random.uniform([], 2, 5, dtype=tf.int32)
small = tf.image.resize(tf.expand_dims(tf.cast(img, tf.float32), 0), [h // sf, w // sf], method='bilinear')
return tf.image.resize(small, [h, w], method='nearest')[0]
def op_identity(img): return tf.cast(img, tf.float32)
ops = [op_flip, op_flip_v, op_rotate_90, op_zoom, op_translate,
op_contrast, op_brightness, op_hue, op_saturation,
op_solarize, op_blur, op_fog, op_shadow, op_resolution_drop, op_identity]
def apply_randaug_single(img3d):
indices = tf.random.shuffle(tf.range(15))[:3] # ops_per_image=3, static
result = tf.cast(img3d, tf.float32)
# Unrolled static 3 iterations — avoids TF shape invariance error from tf.range loop
def _apply_one(r, idx):
return tf.switch_case(idx, {j: lambda j=j: ops[j](r) for j in range(15)})
i0, i1, i2 = indices[0], indices[1], indices[2]
result = _apply_one(result, i0)
result = _apply_one(result, i1)
result = _apply_one(result, i2)
return tf.clip_by_value(result, 0.0, 255.0)
return tf.map_fn(apply_randaug_single, images, dtype=tf.float32, parallel_iterations=8)
# Compatibility wrapper using Keras Sequential for basic geometric ops (kept for visualization)
geo_aug = tf.keras.Sequential([
layers.RandomFlip("horizontal_and_vertical"),
layers.RandomRotation(0.15),
layers.RandomZoom(0.15),
layers.RandomTranslation(0.1, 0.1),
layers.RandomContrast(0.15),
layers.RandomBrightness(0.15),
], name="geo_aug")
# ─── MixUp & CutMix (unchanged from original) ───
def mix_up(images, labels, alpha=0.2):
bs = tf.shape(images)[0]
lam = sample_beta_distribution(bs, alpha, alpha)
lam_img = tf.reshape(lam, [bs, 1, 1, 1])
ri = tf.random.shuffle(tf.range(bs))
mixed_img = lam_img * images + (1 - lam_img) * tf.gather(images, ri)
labels = tf.cast(labels, tf.float32)
lam_lbl = tf.reshape(lam, [-1, 1])
mixed_lbl = lam_lbl * labels + (1 - lam_lbl) * tf.gather(labels, ri)
return mixed_img, mixed_lbl
def cut_mix(images, labels, alpha=0.2):
bs = tf.shape(images)[0]; h = tf.shape(images)[1]; w = tf.shape(images)[2]
lam = sample_beta_distribution(bs, alpha, alpha); ri = tf.random.shuffle(tf.range(bs))
cr = tf.sqrt(1.0 - lam)
rh = tf.cast(cr * tf.cast(h, tf.float32), tf.int32); rw = tf.cast(cr * tf.cast(w, tf.float32), tf.int32)
cx = tf.random.uniform([bs], 0, w, tf.int32); cy = tf.random.uniform([bs], 0, h, tf.int32)
hh = rh // 2; hw = rw // 2
x1 = tf.clip_by_value(cx - hw, 0, w); x2 = tf.clip_by_value(cx + hw, 0, w)
y1 = tf.clip_by_value(cy - hh, 0, h); y2 = tf.clip_by_value(cy + hh, 0, h)
col = tf.range(w, dtype=tf.int32); row = tf.range(h, dtype=tf.int32)
in_x = tf.logical_and(tf.reshape(col, [1, 1, w]) >= tf.reshape(x1, [bs, 1, 1]),
tf.reshape(col, [1, 1, w]) < tf.reshape(x2, [bs, 1, 1]))
in_y = tf.logical_and(tf.reshape(row, [1, h, 1]) >= tf.reshape(y1, [bs, 1, 1]),
tf.reshape(row, [1, h, 1]) < tf.reshape(y2, [bs, 1, 1]))
cm = tf.cast(tf.logical_and(in_y, in_x), tf.float32); cm = tf.expand_dims(cm, -1)
shuf = tf.gather(images, ri)
mi = (1.0 - cm) * images + cm * shuf
labels = tf.cast(labels, tf.float32); lr = tf.reshape(lam, [-1, 1])
ml = lr * labels + (1.0 - lr) * tf.gather(labels, ri)
return mi, ml
def random_erasing(images, probability=0.25, scale=(0.02, 0.25)):
bs = tf.shape(images)[0]; h = tf.shape(images)[1]; w = tf.shape(images)[2]
ta = tf.random.uniform([], scale[0], scale[1]) * tf.cast(h * w, tf.float32)
ar = tf.random.uniform([], 0.3, 3.3)
eh = tf.cast(tf.math.sqrt(ta / ar), tf.int32); ew = tf.cast(tf.math.sqrt(ta * ar), tf.int32)
eh = tf.clip_by_value(eh, 1, h - 1); ew = tf.clip_by_value(ew, 1, w - 1)
cx = tf.random.uniform([], 0, w - ew, tf.int32); cy = tf.random.uniform([], 0, h - eh, tf.int32)
col = tf.range(w, dtype=tf.int32); row = tf.range(h, dtype=tf.int32)
ix = tf.logical_and(col >= cx, col < cx + ew)
iy = tf.logical_and(row >= cy, row < cy + eh)
em = tf.cast(tf.expand_dims(iy, 1) & tf.expand_dims(ix, 0), tf.float32)
em = tf.expand_dims(tf.expand_dims(em, 0), -1)
noise = tf.random.uniform([bs, eh, ew, 3], 0.0, 255.0, dtype=tf.float32)
pads = [[0, 0], [cy, h - (cy + eh)], [cx, w - (cx + ew)], [0, 0]]
npad = tf.pad(noise, pads, constant_values=0.0)
erased = images * (1.0 - em) + npad * em
return tf.cond(tf.random.uniform([]) < probability, lambda: erased, lambda: images)
# ─── Main augmentation pipeline ───
def augment_and_mix(images, labels):
# 1. RandAugment (geometric + color + weather + degradation)
images = tf.cast(images, tf.float32)
images = _randaug_select(images, ops_per_image=3, magnitude=0.7)
# 2. MixUp or CutMix (40% chance total: 20% MixUp, 20% CutMix)
choice = tf.random.uniform([])
labels_oh = tf.one_hot(labels, NUM_CLASSES)
images, labels_oh = tf.cond(
choice < 0.2, lambda: mix_up(images, labels_oh),
lambda: tf.cond(choice < 0.4, lambda: cut_mix(images, labels_oh),
lambda: (images, labels_oh)))
# 3. Random Erasing
images = random_erasing(images, probability=0.2)
return images, labels_oh
# ─── Preprocessing ───
def preprocess_fn(image, label):
return preprocess_input(image), label
# ─── Class weights ───
train_class_counts = Counter()
for cn in sorted(os.listdir(train_dir)):
p = os.path.join(train_dir, cn)
if os.path.isdir(p):
train_class_counts[cn] = len(os.listdir(p))
y_int = []
for i, cn in enumerate(sorted(os.listdir(train_dir))):
cp = os.path.join(train_dir, cn)
if os.path.isdir(cp):
y_int.extend([i] * len(os.listdir(cp)))
cw_array = compute_class_weight('balanced', classes=np.unique(y_int), y=y_int)
cw_capped = [min(w, 3.0) for w in cw_array]
class_weights_tensor = tf.constant(cw_capped, dtype=tf.float32)
def add_sample_weight(image, label):
ci = tf.argmax(label, axis=-1)
sw = tf.gather(class_weights_tensor, ci)
return image, label, sw
# ─── Build datasets ───
train_ds = tf.keras.utils.image_dataset_from_directory(
train_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_SIZE)
val_ds = tf.keras.utils.image_dataset_from_directory(
val_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_SIZE)
test_ds = tf.keras.utils.image_dataset_from_directory(
test_dir, shuffle=False, batch_size=BATCH_SIZE, image_size=IMG_SIZE)
class_names = train_ds.class_names
NUM_CLASSES = len(class_names)
print(f"Classes ({NUM_CLASSES}): {class_names}")
train_ds = (train_ds
.map(augment_and_mix, num_parallel_calls=AUTOTUNE)
.map(preprocess_fn, num_parallel_calls=AUTOTUNE)
.map(add_sample_weight, num_parallel_calls=AUTOTUNE)
.prefetch(AUTOTUNE))
val_ds = (val_ds
.map(lambda img, lbl: (preprocess_input(tf.cast(img, tf.float32)), lbl), num_parallel_calls=AUTOTUNE)
.map(lambda img, lbl: (img, tf.one_hot(lbl, NUM_CLASSES)), num_parallel_calls=AUTOTUNE)
.prefetch(AUTOTUNE))
test_ds = (test_ds
.map(lambda img, lbl: (preprocess_input(tf.cast(img, tf.float32)), lbl), num_parallel_calls=AUTOTUNE)
.map(lambda img, lbl: (img, tf.one_hot(lbl, NUM_CLASSES)), num_parallel_calls=AUTOTUNE)
.prefetch(AUTOTUNE))
print("Class weights (capped at 3.0):")
for i, cn in enumerate(class_names):
if i < len(cw_capped):
print(f" {cn}: {cw_capped[i]:.4f}")
print("Data pipelines ready.")
In [ ]:
vis_ds = tf.keras.utils.image_dataset_from_directory(
train_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_SIZE)
plt.figure(figsize=(16, 10))
for images, labels in vis_ds.take(1):
# Original (4)
for i in range(4):
plt.subplot(3, 4, i + 1)
plt.imshow(images[i].numpy().astype("uint8"))
plt.title(f"Asli: {class_names[labels[i].numpy()]}", fontsize=11)
plt.axis("off")
# RandAugment (4)
aug = _randaug_select(tf.cast(images, tf.float32), ops_per_image=3, magnitude=0.7)
for i in range(4):
plt.subplot(3, 4, i + 5)
plt.imshow(tf.clip_by_value(aug[i], 0, 255).numpy().astype("uint8"))
plt.title(f"RandAug: {class_names[labels[i].numpy()]}", fontsize=11)
plt.axis("off")
# MixUp result (4)
aug_f = tf.cast(images, tf.float32)
mixed, _ = mix_up(aug_f, tf.one_hot(labels, NUM_CLASSES))
for i in range(4):
plt.subplot(3, 4, i + 9)
plt.imshow(tf.clip_by_value(mixed[i], 0, 255).numpy().astype("uint8"))
plt.title("MixUp/Weather", fontsize=11)
plt.axis("off")
plt.suptitle("RandAugment + MixUp — Real-World Simulation", fontsize=16)
plt.tight_layout()
plt.show()
In [ ]:
def cbam_block(x, ratio=8, name="cbam"):
# Convolutional Block Attention Module — ringan, fokus ke foreground.
channels = x.shape[-1]
# Channel Attention
avg_pool = layers.GlobalAveragePooling2D()(x)
max_pool = layers.GlobalMaxPooling2D()(x)
ca = layers.Dense(channels // ratio, activation='swish', name=f"{name}_ca1")(avg_pool)
ca = layers.Dense(channels, activation='sigmoid', name=f"{name}_ca2")(ca)
ca2 = layers.Dense(channels // ratio, activation='swish', name=f"{name}_ca3")(max_pool)
ca2 = layers.Dense(channels, activation='sigmoid', name=f"{name}_ca4")(ca2)
ca_out = layers.Add(name=f"{name}_ca_add")([ca, ca2])
ca_out = layers.Reshape((1, 1, channels), name=f"{name}_ca_reshape")(ca_out)
x = layers.Multiply(name=f"{name}_ca_mul")([x, ca_out])
# Spatial Attention
avg_sp = layers.Lambda(lambda t: tf.reduce_mean(t, axis=-1, keepdims=True), name=f"{name}_sa_avg")(x)
max_sp = layers.Lambda(lambda t: tf.reduce_max(t, axis=-1, keepdims=True), name=f"{name}_sa_max")(x)
sp = layers.Concatenate(name=f"{name}_sa_cat")([avg_sp, max_sp])
sp = layers.Conv2D(1, 7, padding='same', activation='sigmoid', name=f"{name}_sa_conv")(sp)
x = layers.Multiply(name=f"{name}_sa_mul")([x, sp])
return x
def build_model(num_classes, target_size=(224, 224)):
# Accept any input size via variable input; Resizing handles progressive resolution.
inputs = tf.keras.Input(shape=(None, None, 3), name="input")
x = layers.Resizing(target_size[0], target_size[1], interpolation='bilinear',
name="resize_input")(inputs)
base_model = EfficientNetV2B0(
input_shape=target_size + (3,),
include_top=False,
weights='imagenet',
)
base_model.trainable = False
# Gaussian noise untuk regularisasi
x = layers.GaussianNoise(0.05, name="gauss_noise")(x)
x = base_model(x, training=False)
# CBAM attention — fokus ke region daun
x = cbam_block(x, ratio=8, name="cbam")
x = layers.GlobalAveragePooling2D(name="gap")(x)
x = layers.Dropout(0.3, name="drop_gap")(x)
x = layers.Dense(512, activation='swish', name="dense_head")(x)
x = layers.BatchNormalization(name="bn_head")(x)
x = layers.Dropout(0.4, name="drop_head")(x)
outputs = layers.Dense(num_classes, activation='linear', dtype='float32', name="logits")(x)
return models.Model(inputs, outputs), base_model
# Checkpoint
ckpt_dir = '/content/best_model' if IS_COLAB else os.path.join(os.getcwd(), 'best_model')
checkpoint_path = os.path.join(ckpt_dir, 'best_model.keras')
# Hapus checkpoint lama (arsitektur berbeda — tidak kompatibel)
if os.path.exists(checkpoint_path):
print(f"Removing old checkpoint (incompatible architecture)...")
os.remove(checkpoint_path)
if os.path.exists(checkpoint_path):
print(f"Loading checkpoint: {checkpoint_path}")
try:
model = models.load_model(checkpoint_path, compile=False)
except Exception as e:
print(f"Load failed: {e}. Building fresh.")
model, base_model = build_model(NUM_CLASSES)
else:
print("No checkpoint. Building fresh model.")
model, base_model = build_model(NUM_CLASSES)
os.makedirs(ckpt_dir, exist_ok=True)
model.summary()
In [ ]:
class WarmupCosineDecay(tf.keras.optimizers.schedules.LearningRateSchedule):
# Cosine decay with linear warmup.
def __init__(self, warmup_steps, total_steps, peak_lr, min_lr=1e-7):
super().__init__()
self.warmup_steps = warmup_steps
self.total_steps = total_steps
self.peak_lr = peak_lr
self.min_lr = min_lr
def __call__(self, step):
step = tf.cast(step, tf.float32)
warmup_steps = tf.cast(self.warmup_steps, tf.float32)
total_steps = tf.cast(self.total_steps, tf.float32)
# Warmup phase
warmup_lr = self.peak_lr * (step / warmup_steps)
# Cosine decay phase
progress = (step - warmup_steps) / tf.maximum(total_steps - warmup_steps, 1.0)
cosine_lr = self.min_lr + 0.5 * (self.peak_lr - self.min_lr) * (1.0 + tf.cos(np.pi * progress))
return tf.where(step < warmup_steps, warmup_lr, cosine_lr)
def get_config(self):
return {
"warmup_steps": self.warmup_steps, "total_steps": self.total_steps,
"peak_lr": self.peak_lr, "min_lr": self.min_lr,
}
class SWACallback(tf.keras.callbacks.Callback):
# Stochastic Weight Averaging — averages weights over final epochs.
def __init__(self, start_epoch, swa_lr=1e-5):
super().__init__()
self.start_epoch = start_epoch
self.swa_lr = swa_lr
self.swa_weights = None
self.n_models = 0
def on_epoch_begin(self, epoch, logs=None):
if epoch >= self.start_epoch and self.swa_weights is None:
self.swa_weights = [w.numpy() for w in self.model.weights]
print(f"\nSWA: starting weight averaging at epoch {epoch+1}")
def on_epoch_end(self, epoch, logs=None):
if epoch >= self.start_epoch and self.swa_weights is not None:
for i, w in enumerate(self.model.weights):
self.swa_weights[i] = (self.swa_weights[i] * self.n_models + w.numpy()) / (self.n_models + 1)
self.n_models += 1
def apply_swa_weights(self):
if self.swa_weights is None:
print("SWA: no weights to average (skipped)")
return
for w, swa_w in zip(self.model.weights, self.swa_weights):
w.assign(swa_w)
print(f"SWA weights applied ({self.n_models} models averaged).")
# Shared callbacks
checkpoint_cb = callbacks.ModelCheckpoint(
checkpoint_path, save_best_only=True, monitor="val_accuracy",
mode="max", verbose=1)
csv_logger = callbacks.CSVLogger(os.path.join(ckpt_dir, 'training_log.csv'))
def make_callbacks(swa_start=None):
cbs = [checkpoint_cb, csv_logger]
if swa_start is not None:
cbs.append(SWACallback(swa_start))
return cbs
print("Callbacks ready.")
print(f"Checkpoint path: {checkpoint_path}")
In [ ]:
IMG_128 = (128, 128)
# Rebuild datasets at 128x128
train_ds_128 = tf.keras.utils.image_dataset_from_directory(
train_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_128)
val_ds_128 = tf.keras.utils.image_dataset_from_directory(
val_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_128)
train_ds_128 = (train_ds_128.map(augment_and_mix, num_parallel_calls=AUTOTUNE)
.map(preprocess_fn, num_parallel_calls=AUTOTUNE)
.map(add_sample_weight, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
val_ds_128 = (val_ds_128.map(lambda img, lbl: (preprocess_input(tf.cast(img, tf.float32)), lbl), num_parallel_calls=AUTOTUNE)
.map(lambda img, lbl: (img, tf.one_hot(lbl, NUM_CLASSES)), num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
EPOCHS_P1 = 25
steps_per_epoch = tf.data.experimental.cardinality(train_ds_128).numpy() or 100
total_steps = steps_per_epoch * EPOCHS_P1
warmup_steps = steps_per_epoch * 3 # 3 epoch warmup
lr_schedule_p1 = WarmupCosineDecay(warmup_steps, total_steps, peak_lr=1e-3, min_lr=1e-5)
model.compile(
optimizer=AdamW(
learning_rate=lr_schedule_p1, weight_decay=1e-4),
loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True, label_smoothing=0.15),
metrics=['accuracy']
)
print("Phase 1: Head training at 128×128...")
history_1 = model.fit(train_ds_128, validation_data=val_ds_128,
epochs=EPOCHS_P1, callbacks=make_callbacks())
In [ ]:
IMG_192 = (192, 192)
train_ds_192 = tf.keras.utils.image_dataset_from_directory(
train_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_192)
val_ds_192 = tf.keras.utils.image_dataset_from_directory(
val_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=IMG_192)
train_ds_192 = (train_ds_192.map(augment_and_mix, num_parallel_calls=AUTOTUNE)
.map(preprocess_fn, num_parallel_calls=AUTOTUNE)
.map(add_sample_weight, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
val_ds_192 = (val_ds_192.map(lambda img, lbl: (preprocess_input(tf.cast(img, tf.float32)), lbl), num_parallel_calls=AUTOTUNE)
.map(lambda img, lbl: (img, tf.one_hot(lbl, NUM_CLASSES)), num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
# Unfreeze top 100 layers
base_model.trainable = True
for layer in base_model.layers[:-100]:
layer.trainable = False
EPOCHS_P2 = 30
steps_p2 = tf.data.experimental.cardinality(train_ds_192).numpy() or 100
total_p2 = steps_p2 * EPOCHS_P2
warmup_p2 = steps_p2 * 2
lr_schedule_p2 = WarmupCosineDecay(warmup_p2, total_p2, peak_lr=5e-4, min_lr=1e-6)
model.compile(
optimizer=AdamW(
learning_rate=lr_schedule_p2, weight_decay=1e-4),
loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True, label_smoothing=0.15),
metrics=['accuracy']
)
print("Phase 2: Fine-tuning top 100 layers at 192×192...")
history_2 = model.fit(train_ds_192, validation_data=val_ds_192,
epochs=EPOCHS_P1 + EPOCHS_P2, initial_epoch=history_1.epoch[-1] + 1,
callbacks=make_callbacks())
In [ ]:
img_size = IMG_SIZE # (224, 224)
train_ds_full = tf.keras.utils.image_dataset_from_directory(
train_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=img_size)
val_ds_full = tf.keras.utils.image_dataset_from_directory(
val_dir, shuffle=True, batch_size=BATCH_SIZE, image_size=img_size)
train_ds_full = (train_ds_full.map(augment_and_mix, num_parallel_calls=AUTOTUNE)
.map(preprocess_fn, num_parallel_calls=AUTOTUNE)
.map(add_sample_weight, num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
val_ds_full = (val_ds_full.map(lambda img, lbl: (preprocess_input(tf.cast(img, tf.float32)), lbl), num_parallel_calls=AUTOTUNE)
.map(lambda img, lbl: (img, tf.one_hot(lbl, NUM_CLASSES)), num_parallel_calls=AUTOTUNE).prefetch(AUTOTUNE))
# Full unfreeze
for layer in base_model.layers:
layer.trainable = True
EPOCHS_P3 = 30
steps_p3 = tf.data.experimental.cardinality(train_ds_full).numpy() or 100
total_p3 = steps_p3 * EPOCHS_P3
warmup_p3 = steps_p3 * 2
lr_schedule_p3 = WarmupCosineDecay(warmup_p3, total_p3, peak_lr=1e-4, min_lr=1e-7)
model.compile(
optimizer=AdamW(
learning_rate=lr_schedule_p3, weight_decay=1e-4),
loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True, label_smoothing=0.10),
metrics=['accuracy']
)
print("Phase 3: Full fine-tuning at 224×224...")
history_3 = model.fit(train_ds_full, validation_data=val_ds_full,
epochs=EPOCHS_P1 + EPOCHS_P2 + EPOCHS_P3, initial_epoch=(history_2.epoch[-1] + 1) if history_2.epoch else EPOCHS_P1 + EPOCHS_P2,
callbacks=make_callbacks())
In [ ]:
EPOCHS_SWA = 15
swa_start_epoch = (history_3.epoch[-1] + 1) if history_3.epoch else EPOCHS_P1 + EPOCHS_P2 + EPOCHS_P3
swa_cb = SWACallback(start_epoch=swa_start_epoch, swa_lr=1e-5)
model.compile(
optimizer=AdamW(
learning_rate=1e-5, weight_decay=1e-4),
loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True, label_smoothing=0.10),
metrics=['accuracy']
)
print(f"SWA: {EPOCHS_SWA} epochs starting at epoch {swa_start_epoch + 1}...")
history_swa = model.fit(train_ds_full, validation_data=val_ds_full,
epochs=swa_start_epoch + EPOCHS_SWA, initial_epoch=swa_start_epoch,
callbacks=make_callbacks() + [swa_cb])
# Apply SWA weights
swa_cb.apply_swa_weights()
print(f"SWA complete. Final model has SWA weights applied.")
In [ ]:
# Gabungkan semua history
acc = (history_1.history['accuracy'] + history_2.history['accuracy'] +
history_3.history['accuracy'] + history_swa.history['accuracy'])
val_acc = (history_1.history['val_accuracy'] + history_2.history['val_accuracy'] +
history_3.history['val_accuracy'] + history_swa.history['val_accuracy'])
loss = (history_1.history['loss'] + history_2.history['loss'] +
history_3.history['loss'] + history_swa.history['loss'])
val_loss = (history_1.history['val_loss'] + history_2.history['val_loss'] +
history_3.history['val_loss'] + history_swa.history['val_loss'])
b1 = len(history_1.history['accuracy']) - 1
b2 = b1 + len(history_2.history['accuracy'])
b3 = b2 + len(history_3.history['accuracy'])
plt.figure(figsize=(16, 6))
plt.subplot(1, 2, 1)
plt.plot(acc, label='Training Accuracy', linewidth=2)
plt.plot(val_acc, label='Validation Accuracy', linewidth=2)
plt.axvline(x=b1, color='gray', linestyle='--', alpha=0.7, label='P2 (192)')
plt.axvline(x=b2, color='black', linestyle='--', alpha=0.7, label='P3 (224)')
plt.axvline(x=b3, color='blue', linestyle='--', alpha=0.7, label='SWA start')
plt.legend(fontsize=10)
plt.title('Training & Validation Accuracy', fontsize=14)
plt.xlabel('Epoch'); plt.ylabel('Accuracy'); plt.grid(alpha=0.3)
plt.subplot(1, 2, 2)
plt.plot(loss, label='Training Loss', linewidth=2)
plt.plot(val_loss, label='Validation Loss', linewidth=2)
plt.axvline(x=b1, color='gray', linestyle='--', alpha=0.7, label='P2 (192)')
plt.axvline(x=b2, color='black', linestyle='--', alpha=0.7, label='P3 (224)')
plt.axvline(x=b3, color='blue', linestyle='--', alpha=0.7, label='SWA start')
plt.legend(fontsize=10)
plt.title('Training & Validation Loss', fontsize=14)
plt.xlabel('Epoch'); plt.ylabel('Loss'); plt.grid(alpha=0.3)
plt.tight_layout()
plt.show()
In [ ]:
def compute_ece(probs, true_labels, n_bins=15):
# Expected Calibration Error.
confs = np.max(probs, axis=1)
preds = np.argmax(probs, axis=1)
true = np.argmax(true_labels, axis=1)
accs = (preds == true).astype(np.float32)
bins = np.linspace(0, 1, n_bins + 1)
ece = 0.0
bin_stats = []
for i in range(n_bins):
in_bin = (confs > bins[i]) & (confs <= bins[i + 1])
n = np.sum(in_bin)
if n > 0:
bin_acc = np.mean(accs[in_bin])
bin_conf = np.mean(confs[in_bin])
ece += (n / len(confs)) * np.abs(bin_acc - bin_conf)
bin_stats.append((bins[i], n, bin_acc, bin_conf))
return ece, bin_stats
# Collect logits and labels from validation set
print("Collecting validation logits...")
logits_model = tf.keras.Model(model.input, model.output)
all_logits = []
all_labels = []
for images, labels in val_ds_full.unbatch().batch(BATCH_SIZE):
all_logits.append(logits_model.predict_on_batch(images))
all_labels.append(labels.numpy())
all_logits = np.concatenate(all_logits, axis=0)
all_labels = np.concatenate(all_labels, axis=0)
# ECE before scaling (T=1)
probs_raw = tf.nn.softmax(all_logits).numpy()
ece_raw, _ = compute_ece(probs_raw, all_labels)
print(f"ECE before scaling (T=1.0): {ece_raw:.4f}")
# Optimize T on validation set
if HAS_SCIPY:
def nll_temperature(T):
scaled = all_logits / float(T)
probs = tf.nn.softmax(scaled).numpy()
probs = np.clip(probs, 1e-7, 1.0 - 1e-7)
return -np.mean(np.log(np.sum(all_labels * probs, axis=1)))
result = minimize_scalar(nll_temperature, bounds=(0.1, 5.0), method='bounded')
T_opt = result.x
print(f"Optimal temperature: T = {T_opt:.4f}")
else:
# Grid search fallback
best_nll, T_opt = float('inf'), 1.0
for T in np.linspace(0.5, 4.0, 36):
scaled = all_logits / T
probs = tf.nn.softmax(scaled).numpy()
probs = np.clip(probs, 1e-7, 1.0 - 1e-7)
nll = -np.mean(np.log(np.sum(all_labels * probs, axis=1)))
if nll < best_nll:
best_nll = nll
T_opt = T
print(f"Optimal temperature (grid): T = {T_opt:.4f}")
# ECE after scaling
probs_cal = tf.nn.softmax(all_logits / T_opt).numpy()
ece_cal, bin_stats = compute_ece(probs_cal, all_labels)
print(f"ECE after scaling (T={T_opt:.4f}): {ece_cal:.4f}")
# Save calibration metadata
calibration_meta = {
"temperature": float(T_opt),
"conf_threshold_high": 0.70,
"conf_threshold_low": 0.45,
"ece_raw": float(ece_raw),
"ece_calibrated": float(ece_cal),
}
with open(os.path.join(ckpt_dir, "calibration.json"), "w") as f:
json.dump(calibration_meta, f, indent=2)
print(f"Calibration metadata saved to {os.path.join(ckpt_dir, 'calibration.json')}")
# Reliability diagram
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
if bin_stats:
bin_mids = [(s[0] + s[0] + 1/n_bins)/2 for s in bin_stats]
bin_accs = [s[2] for s in bin_stats]
bin_confs = [s[3] for s in bin_stats]
plt.bar(bin_mids, bin_accs, width=0.05, alpha=0.5, label='Accuracy')
plt.bar(bin_mids, bin_confs, width=0.05, alpha=0.3, label='Confidence')
plt.plot([0, 1], [0, 1], 'k--', alpha=0.3)
plt.xlabel('Confidence'); plt.ylabel('Accuracy')
plt.title(f'Reliability Diagram (T={T_opt:.2f})')
plt.legend(); plt.grid(alpha=0.3)
plt.subplot(1, 2, 2)
conf_raw = np.max(probs_raw, axis=1)
conf_cal = np.max(probs_cal, axis=1)
plt.hist(conf_raw, bins=30, alpha=0.5, label='Before scaling', density=True)
plt.hist(conf_cal, bins=30, alpha=0.5, label='After scaling', density=True)
plt.xlabel('Max Confidence'); plt.ylabel('Density')
plt.title('Confidence Distribution')
plt.legend(); plt.grid(alpha=0.3)
plt.tight_layout()
plt.show()
In [ ]:
# Load best model (dengan SWA weights)
print(f"Loading best model from {checkpoint_path}...")
best_model = tf.keras.models.load_model(checkpoint_path, compile=False)
# Collect test images
raw_test_ds = tf.keras.utils.image_dataset_from_directory(
test_dir, shuffle=False, batch_size=BATCH_SIZE, image_size=IMG_SIZE)
test_images = []
test_labels_raw = []
for images, labels in raw_test_ds.unbatch():
test_images.append(images.numpy())
test_labels_raw.append(labels.numpy())
test_images = np.array(test_images)
test_labels_true = tf.one_hot(np.array(test_labels_raw), NUM_CLASSES).numpy()
# ── TTA 5x ──
TTA_STEPS = 5
tta_logits = []
for i in range(TTA_STEPS):
aug_images = geo_aug(test_images, training=True)
aug_images = preprocess_input(aug_images)
logits = best_model.predict(aug_images, batch_size=BATCH_SIZE, verbose=0)
tta_logits.append(logits)
print(f" TTA step {i+1}/{TTA_STEPS}")
mean_logits = np.mean(tta_logits, axis=0)
# Apply temperature scaling
mean_cal_probs = tf.nn.softmax(mean_logits / T_opt).numpy()
test_preds = np.argmax(mean_cal_probs, axis=1)
test_true = np.argmax(test_labels_true, axis=1)
tta_acc = np.mean(test_preds == test_true)
print(f"\nTest Accuracy (TTA {TTA_STEPS}x, T={T_opt:.2f}): {tta_acc*100:.2f}%")
# ECE on test set
ece_test, _ = compute_ece(mean_cal_probs, test_labels_true)
print(f"ECE on test set: {ece_test:.4f}")
In [ ]:
def get_source(filename):
f = os.path.splitext(filename)[0]
if f.startswith('IMG_'): return 'Phone'
if f.startswith('Corn_'): return 'Lab_Corn'
for p in ['CBS', 'GLS', 'NLS', 'CLS']:
if f.startswith(p): return 'Lab_Disease'
for p in ['SCR', 'CR', 'NLB', 'SLB', 'SRS']:
if f.startswith(p): return 'Lab_RustBlight'
return 'Other'
# Map each test image to its source
test_files = []
for cn in class_names:
cp = os.path.join(test_dir, cn)
if os.path.isdir(cp):
test_files.extend([(f, cn, get_source(f)) for f in sorted(os.listdir(cp))])
sources = set(s for _, _, s in test_files)
print(f"Sources found: {sorted(sources)}")
print(f"{'Source':<18} {'Count':>6} {'Accuracy':>10}")
print("-" * 38)
for src in sorted(sources):
indices = [i for i, (_, _, s) in enumerate(test_files) if s == src]
if not indices: continue
n = len(indices)
acc = np.mean(test_preds[indices] == test_true[indices])
print(f"{src:<18} {n:>6} {acc*100:>9.1f}%")
# Overall with count
overall_acc = np.mean(test_preds == test_true)
print("-" * 38)
print(f"{'ALL':<18} {len(test_preds):>6} {overall_acc*100:>9.1f}%")
In [ ]:
print("\nClassification Report:\n")
print(classification_report(test_true, test_preds, target_names=class_names))
cm = confusion_matrix(test_true, test_preds)
plt.figure(figsize=(10, 8))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues',
xticklabels=class_names, yticklabels=class_names)
plt.title(f'Confusion Matrix (TTA {TTA_STEPS}x)', fontsize=14)
plt.xlabel('Predicted', fontsize=12)
plt.ylabel('True', fontsize=12)
plt.tight_layout()
plt.show()
In [ ]:
# Save final model
model.save(final_path := os.path.join(ckpt_dir, 'best_model.keras'))
print(f"Model saved to {final_path}")
# Copy calibration metadata to model directory
model_export_dir = os.path.join(os.getcwd(), 'model')
os.makedirs(model_export_dir, exist_ok=True)
# Export labels.json with calibration metadata
cal_path = os.path.join(ckpt_dir, 'calibration.json')
if os.path.exists(cal_path):
with open(cal_path) as f:
cal_meta = json.load(f)
labels_json = {
"version": "3.0",
"labels": class_names,
"temperature": cal_meta["temperature"],
"conf_threshold_high": cal_meta["conf_threshold_high"],
"conf_threshold_low": cal_meta["conf_threshold_low"],
"input_size": [224, 224],
"input_range": [0, 255],
"preprocessing": "resize_bilinear_224x224_no_normalization",
"architecture": "EfficientNetV2B0 + CBAM + Dense(512)",
"output_type": "logits",
}
else:
labels_json = {
"version": "3.0",
"labels": class_names,
"temperature": 1.0,
"input_size": [224, 224],
"input_range": [0, 255],
"output_type": "logits",
}
with open(os.path.join(model_export_dir, 'labels.json'), 'w') as f:
json.dump(labels_json, f, indent=2)
print("Labels + calibration metadata saved to model/labels.json")
# Export class names list (legacy)
with open(os.path.join(model_export_dir, 'labels.json'), 'r') as f:
pass # already written above
print(f"Classes: {class_names}")
print(f"Temperature: {labels_json['temperature']:.4f}")