🔬 การวิเคราะห์ภาพ X-ray ด้วย CNN และ LLM

# แบ่งข้อมูลเป็น 3 ส่วน
train_dir = 'train'      # 70% สำหรับ training (4,192 ภาพ)
val_dir = 'val'          # 15% สำหรับ validation (1,040 ภาพ)  
test_dir = 'test'        # 15% สำหรับ testing (624 ภาพ)

# Data Augmentation - สร้างภาพเพิ่มจากภาพเดิม
train_datagen = ImageDataGenerator(
    rescale=1./255,           # ปรับค่าพิกเซล 0-255 → 0-1
    shear_range=0.2,          # บิดภาพ
    zoom_range=0.2,           # ซูมเข้า-ออก
    horizontal_flip=True      # พลิกภาพซ้าย-ขวา
)

# คำนวณ class weights เพื่อแก้ปัญหา imbalanced data
# Normal: 1,082 ภาพ (weight = 1.94)
# Pneumonia: 3,110 ภาพ (weight = 0.67)

weights = compute_class_weight(
    class_weight='balanced',
    classes=np.unique(train_generator.classes),
    y=train_generator.classes
)

# Compile model
model.compile(
    optimizer='adam',              # Algorithm สำหรับ optimize
    loss='binary_crossentropy',    # Loss function สำหรับ binary classification
    metrics=['accuracy']            # ติดตามความแม่นยำ
)

# Train model
epochs = 30
history = model.fit(
    train_generator,
    callbacks=[learning_rate_reduction],
    steps_per_epoch=train_generator.samples // batch_size,
    epochs=epochs,                      # จำนวนรอบการเรียนรู้ 30 รอบ
    validation_data=validation_generator,
    validation_steps=validation_generator.samples // batch_size,
    class_weight=cw                      # ใช้ class weights
)

learning_rate_reduction = ReduceLROnPlateau(
    monitor='val_loss',    # ติดตาม validation loss
    patience=2,            # รอ 2 epochs ถ้าไม่ดีขึ้น
    factor=0.1,           # ลด learning rate 10 เท่า
    min_lr=0.000001       # ค่าต่ำสุดของ learning rate
)

# คำนวณ ROC และ AUC
fpr, tpr, threshold = roc_curve(test_generator.classes, y_score)
roc_auc = auc(fpr, tpr)  # AUC = 0.9564

# การใช้ MedGemma วิเคราะห์ภาพ X-ray
pipe = pipeline(
    "image-text-to-text",
    model="google/medgemma-4b-it",
    torch_dtype=torch.bfloat16,
    device="cuda"
)

messages = [{
    "role": "user",
    "content": [
        {"type": "text", "text": "Describe this X-ray"},
        {"type": "image", "image": xray_image}
    ]
}]

# โมเดลจะวิเคราะห์และให้คำอธิบายทางการแพทย์

import tensorflow as tf
from tensorflow.keras.preprocessing.image import ImageDataGenerator
import matplotlib.pyplot as plt
from PIL import Image
import numpy as np

# Download dataset จาก Kaggle
import kagglehub
path = kagglehub.dataset_download("pcbreviglieri/pneumonia-xray-images")

# กำหนด path ของแต่ละ class
normal_path = os.path.join(train_path, 'normal')
opacity_path = os.path.join(train_path, 'opacity')

# แสดงภาพตัวอย่าง
plt.figure(figsize=(10, 5))
normal_img = Image.open(normal_img_path)
plt.imshow(normal_img, cmap='gray')
plt.title('Normal')
plt.show()

# สร้าง ImageDataGenerator สำหรับ training
train_datagen = ImageDataGenerator(
    rescale=1./255,        # Normalize pixels to [0,1]
    shear_range=0.2,       # Random shearing
    zoom_range=0.2,        # Random zoom
    horizontal_flip=True   # Random horizontal flip
)

# Load images from directories
train_generator = train_datagen.flow_from_directory(
    train_dir,
    target_size=(350, 350),  # Resize all images
    batch_size=32,           # Process 32 images at a time
    class_mode='binary',     # Binary classification
    color_mode='grayscale'   # Use grayscale images
)

# สร้างโมเดลด้วย Functional API
# Input
inputs = tf.keras.Input(shape=(350, 350, 1))

# Feature Extraction
x = tf.keras.layers.Conv2D(32, (3, 3), activation='relu')(inputs)
x = tf.keras.layers.MaxPooling2D((2, 2))(x)

x = tf.keras.layers.Conv2D(32, (3, 3), activation='relu')(x)
x = tf.keras.layers.MaxPooling2D((2, 2))(x)

x = tf.keras.layers.Conv2D(64, (3, 3), activation='relu')(x)
x = tf.keras.layers.MaxPooling2D((2, 2))(x)

x = tf.keras.layers.Conv2D(64, (3, 3), activation='relu')(x)
x = tf.keras.layers.MaxPooling2D((2, 2))(x)

x = tf.keras.layers.Conv2D(128, (3, 3), activation='relu', padding='same', name='last_conv_layer')(x)

# Image Classification
x = tf.keras.layers.Flatten()(x)
x = tf.keras.layers.Dense(128, activation='relu')(x)
x = tf.keras.layers.Dense(64, activation='relu')(x)
outputs = tf.keras.layers.Dense(1, activation='sigmoid')(x)

# Model
model = tf.keras.Model(inputs=inputs, outputs=outputs)

# นับจำนวนภาพในแต่ละ class
from collections import Counter
Counter(train_generator.classes)
# Output: {0: 1082, 1: 3110}  # Normal:Opacity ratio

# คำนวณ class weights
from sklearn.utils.class_weight import compute_class_weight
weights = compute_class_weight(
    'balanced', 
    np.unique(train_generator.classes), 
    train_generator.classes
)
# weights = [1.937, 0.674]  # Give more weight to minority class

# Learning rate scheduler
learning_rate_reduction = ReduceLROnPlateau(
    monitor='val_loss',
    patience=2,         # Wait 2 epochs
    factor=0.1,        # Reduce LR by factor of 10
    verbose=1,
    min_lr=0.000001
)

# Train the model
history = model.fit(
    train_generator,
    epochs=10,
    validation_data=validation_generator,
    callbacks=[learning_rate_reduction],
    class_weight=dict(zip(np.unique(train_generator.classes), weights))
)

# Evaluate on test set
loss, accuracy = model.evaluate(test_generator, steps=test_generator.samples // batch_size)
print(f"Test Loss: {loss:.4f}")
print(f"Test Accuracy: {accuracy:.4f}")

# Generate predictions
predicted_classes = (model.predict(test_generator, verbose=1) > 0.5).astype("int32")[:,0]

# Create confusion matrix
from sklearn.metrics import confusion_matrix, classification_report
cm = confusion_matrix(test_generator.classes, predicted_classes)

# Generate classification report
report = classification_report(
    test_generator.classes, 
    predicted_classes,
    target_names=['normal', 'opacity'],
    digits=4
)

# Calculate ROC curve
from sklearn.metrics import roc_curve, auc

y_score = model.predict(test_generator)
y_score = y_score[:,0]
fpr, tpr, threshold = roc_curve(test_generator.classes, y_score)
roc_auc = auc(fpr, tpr)

# Plot ROC curve
import plotly.express as px
roc_df = pd.DataFrame({
    'False Positive Rate': fpr,
    'True Positive Rate': tpr,
    'Threshold': threshold
})

fig = px.area(
    roc_df,
    x='False Positive Rate',
    y='True Positive Rate',
    title=f'ROC Curve (AUC={roc_auc:.4f})'
)

# Save the trained model
model.save('pneumonia_detection_model.keras')

print("Model saved successfully as 'pneumonia_detection_model.keras'")

# ติดตั้ง tf-keras-vis
!pip install tf-keras-vis

# หา last convolutional layer
for layer in reversed(model.layers):
    if isinstance(layer, tf.keras.layers.Conv2D):
        last_conv_layer_name = layer.name
        break

# สร้าง GradCAM
from tf_keras_vis.gradcam import Gradcam
gradcam = Gradcam(model, model_modifier=None, clone=True)

def custom_score(output):
    return output

heatmap = gradcam(custom_score,
                  image_array,
                  penultimate_layer=last_conv_layer_name)

# แสดงผล heatmap overlay บนภาพต้นฉบับ
plt.imshow(image.convert('RGB'))
plt.imshow(heatmap[0], cmap='jet', alpha=0.5)  # Overlay heatmap
plt.title('GradCAM Heatmap')
plt.show()