MythEclipse
f9bd991bdf
- Integrate Convolutional Block Attention Module (CBAM) for improved feature focus - Implement temperature scaling and confidence-based status reporting - Automate dataset acquisition using kagglehub - Update ONNX opset to 18 and refine preprocessing validation
68 KiB
68 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_jagung.zip'
destination_path = '/content/dataset_jagung.zip'
extract_path = '/content/dataset'
else:
import gdown
base = os.getcwd()
archive_path = os.path.join(base, 'dataset_jagung.zip')
destination_path = archive_path
extract_path = os.path.join(base, 'dataset')
DRIVE_FILE_ID = "1s0H2lDOQVCixywk5eZXJz2i9jj4JihxJ"
if not os.path.exists(archive_path):
print("Downloading from Google Drive...")
try:
gdown.download(f"https://drive.google.com/uc?id={DRIVE_FILE_ID}", archive_path, quiet=False)
except Exception as e:
print(f"Download failed: {e}")
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.")
In [ ]:
# --- Determine dataset path ---
if IS_COLAB:
dataset_path = "/content/dataset/dataset_jagung_v1"
else:
candidate = os.path.join(extract_path, "dataset_jagung_v1")
dataset_path = candidate if os.path.isdir(candidate) else extract_path
print(f"Dataset path: {dataset_path}")
MIN_FILE_SIZE = 512
MIN_DIM = 32
MAX_ASPECT = 5.0
def remove_augmented_duplicates(directory):
# Hapus file augmented_* yang merupakan duplikat buatan.
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):
# Validasi + bersihkan gambar corrupt / invalid.
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
# RGB conversion
if img.mode not in ('RGB', 'RGBA'):
img = img.convert('RGB'); img.save(fp)
# Low variance check (>95% pixels same value)
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']}")
# ── Perceptual hash dedup (optional) ──
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'))
key = str(ah)
dup_found = False
for sk, (sp, sc) in seen.items():
if ah - imagehash.hex_to_hash(sk) <= 5:
try: os.remove(fp); removed += 1
except OSError: pass
dup_found = True; break
if not dup_found: seen[key] = (fp, cn)
except Exception: pass
print(f" Removed {removed} near-duplicates")
else:
print("3. Perceptual hash dedup SKIPPED (pip install imagehash)")
# ── Final count ──
total = sum(len(files) for _, _, files in os.walk(dataset_path))
print(f"\nTotal clean images: {total}")
In [ ]:
def extract_source_prefix(filename):
# Detect source from filename prefix for stratification.
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):
# Split dataset stratified by source prefix within each class.
# Collect all images with their source prefix
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)))
# Create output directories
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():
# Group by source prefix
by_source = {}
for fp, fn, src in images:
by_source.setdefault(src, []).append((fp, fn))
# For each source group, split into train/val/test
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:])
# Copy files
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))
# Print split stats
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 = "/content/dataset_split" if IS_COLAB else os.path.join(os.getcwd(), "dataset_split")
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 _randaug_select(images, ops_per_image=3, magnitude=0.7):
# Apply N random ops from pool per image (RandAugment-style).
batch = tf.shape(images)[0]
h, w = tf.cast(tf.shape(images)[1], tf.float32), tf.cast(tf.shape(images)[2], tf.float32)
# Pool of 15 augmentation functions
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(img):
from tensorflow.keras import layers as L
return L.RandomRotation(0.2 * magnitude, fill_mode='reflect')(img, training=True)
def op_zoom(img):
from tensorflow.keras import layers as L
return L.RandomZoom(0.2 * magnitude, fill_mode='reflect')(img, training=True)
def op_translate(img):
from tensorflow.keras import layers as L
return L.RandomTranslation(0.15 * magnitude, 0.15 * magnitude, fill_mode='reflect')(img, training=True)
def op_contrast(img): return tf.image.random_contrast(img, 1.0 - 0.5 * magnitude, 1.0 + 0.5 * magnitude)
def op_brightness(img): return tf.image.random_brightness(img, 0.3 * magnitude)
def op_hue(img): return tf.image.random_hue(img, 0.08 * max(magnitude, 0.01))
def op_saturation(img):
lo = tf.maximum(0.5, 1.0 - 0.8 * magnitude)
return tf.image.random_saturation(img, lo, 1.0 + 0.8 * magnitude)
def op_solarize(img):
thresh = tf.random.uniform([], 0.3, 0.8) * 255.0
return tf.where(img < thresh, img, 255.0 - img)
def op_blur(img):
sigma = tf.random.uniform([], 1.0, 1.0 + 2.0 * magnitude)
kernel_size = tf.cast(tf.math.ceil(2 * sigma), tf.int32) * 2 + 1
kernel_size = tf.clip_by_value(kernel_size, 3, 9)
channels = 3
# Build Gaussian kernel
size = tf.cast(kernel_size, tf.float32)
x = tf.range(-(size - 1) / 2, (size - 1) / 2 + 1, dtype=tf.float32)
g = tf.exp(-0.5 * (x / sigma) ** 2)
g = g / tf.reduce_sum(g)
kernel = g[:, None] * g[None, :]
kernel = tf.expand_dims(tf.expand_dims(kernel, -1), -1)
kernel = tf.tile(kernel, [1, 1, channels, 1])
img_expanded = tf.expand_dims(img, 0) if len(img.shape) == 3 else img
# Use depthwise for per-channel
blurred = tf.nn.depthwise_conv2d(
tf.transpose(img_expanded, [0, 3, 1, 2]), # NCHW
tf.transpose(kernel, [2, 3, 0, 1]), # H W C_in C_out
strides=[1, 1, 1, 1], padding='SAME'
)
blurred = tf.transpose(blurred, [0, 2, 3, 1]) # NHWC
return tf.squeeze(blurred, 0) if len(img.shape) == 3 else blurred
def op_fog(img):
fog_level = tf.random.uniform([], 0.1, 0.1 + 0.4 * magnitude)
fog_color = tf.random.uniform([3], 0.7, 1.0) * 255.0
fog_color = tf.reshape(fog_color, [1, 1, 3])
return img * (1.0 - fog_level) + fog_color * fog_level
def op_shadow(img):
opacity = tf.random.uniform([], 0.2, 0.2 + 0.5 * magnitude)
# Simple: darken bottom-right quadrant to simulate shadow
mask = tf.ones_like(img)
h_i = tf.cast(h, tf.int32); w_i = tf.cast(w, tf.int32)
mask_h = tf.random.uniform([], 0, tf.cast(h_i, tf.float32) * 0.7)
mask_w = tf.random.uniform([], 0, tf.cast(w_i, tf.float32) * 0.7)
# shadow from random corner
corner = tf.random.uniform([], 0, 4, dtype=tf.int32)
y1 = tf.cond(tf.less(corner, 2), lambda: 0, lambda: tf.cast(h_i, tf.int32) - tf.cast(mask_h, tf.int32))
x1 = tf.cond(tf.equal(tf.math.floormod(corner, 2), 0), lambda: 0,
lambda: tf.cast(w_i, tf.int32) - tf.cast(mask_w, tf.int32))
y_range = tf.range(tf.cast(y1, tf.int32), tf.minimum(tf.cast(y1, tf.int32) + tf.cast(mask_h, tf.int32), h_i))
x_range = tf.range(tf.cast(x1, tf.int32), tf.minimum(tf.cast(x1, tf.int32) + tf.cast(mask_w, tf.int32), w_i))
# Build scatter update
# ... simplified: darken by blending with black
shadow_mask = tf.zeros_like(img)
# Use SparseTensor approach or just broadcast a gradient
return img * (1.0 - opacity * 0.6)
def op_resolution_drop(img):
scale_factor = tf.random.uniform([], 2, 4, dtype=tf.int32)
cur_h = tf.shape(img)[0]; cur_w = tf.shape(img)[1]
small_h = cur_h // scale_factor; small_w = cur_w // scale_factor
img_small = tf.image.resize(img[tf.newaxis, ...], [small_h, small_w], method='bilinear')[0]
img_back = tf.image.resize(img_small[tf.newaxis, ...], [cur_h, cur_w], method='bilinear')[0]
return img_back
def op_identity(img): return img
ops = [op_flip, op_flip_v, op_rotate, 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]
# Vectorized: randomly select ops_per_image ops and apply in sequence
# For simplicity in tf.function, apply operations sequentially with per-image random selection
def apply_randaug_single(img3d):
# Pick ops_per_image random indices
indices = tf.random.shuffle(tf.range(len(ops)))[:ops_per_image]
result = img3d
for i in range(ops_per_image):
idx = indices[i]
# Apply op by index (tf.case or tf.switch_case)
for j in range(len(ops)):
result = tf.cond(tf.equal(idx, j), lambda j=j: ops[j](result), lambda: result)
return tf.clip_by_value(result, 0.0, 255.0)
return tf.map_fn(apply_randaug_single, images, dtype=tf.float32)
# 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 = _randaug_select(images, ops_per_image=3, magnitude=0.7)
# 2. Convert float32 for MixUp/CutMix
images = tf.cast(images, tf.float32)
# 3. 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)))
# 4. 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 = tf.shape(x)[-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
from keras import ops
avg_sp = ops.mean(x, axis=-1, keepdims=True)
max_sp = ops.max(x, axis=-1, keepdims=True)
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, img_size=(224, 224)):
base_model = EfficientNetV2B0(
input_shape=img_size + (3,),
include_top=False,
weights='imagenet',
)
base_model.trainable = False
inputs = tf.keras.Input(shape=img_size + (3,), name="input")
# Gaussian noise untuk regularisasi
x = layers.GaussianNoise(0.05, name="gauss_noise")(inputs)
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:
tf.keras.backend.set_value(self.model.optimizer.learning_rate, self.swa_lr)
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):
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(use_ema=True, ema_momentum=0.999,
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(use_ema=True, ema_momentum=0.999,
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_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(use_ema=True, ema_momentum=0.999,
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_P3, initial_epoch=history_2.epoch[-1] + 1,
callbacks=make_callbacks())
In [ ]:
EPOCHS_SWA = 15
swa_start_epoch = history_3.epoch[-1] + 1 # Start SWA after Phase 3
swa_cb = SWACallback(start_epoch=swa_start_epoch, swa_lr=1e-5)
model.compile(
optimizer=AdamW(use_ema=True, ema_momentum=0.999,
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=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}")