252 lines
9.7 KiB
Python
252 lines
9.7 KiB
Python
import os
|
|
import joblib
|
|
import numpy as np
|
|
import pandas as pd
|
|
import cv2
|
|
from sklearn.model_selection import train_test_split
|
|
from sklearn.preprocessing import StandardScaler, LabelEncoder
|
|
import matplotlib.pyplot as plt
|
|
from utils.preprocessing import preprocess_image, preprocess_pipeline
|
|
from utils.feature_extraction import FeatureExtractor
|
|
|
|
|
|
def _save_preprocessed_training_images(img_path, class_name, original_filename, img_resized, gray_processed,
|
|
resize_dir, threshold_dir):
|
|
"""Save resized and thresholded training images for later inspection."""
|
|
os.makedirs(resize_dir, exist_ok=True)
|
|
os.makedirs(threshold_dir, exist_ok=True)
|
|
|
|
base_name, ext = os.path.splitext(original_filename)
|
|
safe_base = f"{class_name}_{base_name}" if class_name else base_name
|
|
safe_base = safe_base.replace(' ', '_')
|
|
ext = ext.lower() if ext else '.png'
|
|
|
|
resized_path = os.path.join(resize_dir, f"{safe_base}{ext}")
|
|
threshold_path = os.path.join(threshold_dir, f"{safe_base}{ext}")
|
|
|
|
try:
|
|
cv2.imwrite(resized_path, img_resized)
|
|
except Exception as e:
|
|
print(f" ⚠️ Gagal menyimpan resized training image {img_path}: {e}")
|
|
|
|
try:
|
|
cv2.imwrite(threshold_path, gray_processed)
|
|
except Exception as e:
|
|
print(f" ⚠️ Gagal menyimpan threshold training image {img_path}: {e}")
|
|
|
|
def prepare_dataset(data_dir='dataset'):
|
|
"""
|
|
Prepare dataset from directory structure using enhanced preprocessing pipeline
|
|
|
|
Auto-detects class directories:
|
|
- 'healthy' → label 'sehat'
|
|
- Any directory starting with 'pmk_' → label = directory name
|
|
- 'sick' → label 'sakit' (backward compat)
|
|
"""
|
|
features = []
|
|
labels = []
|
|
image_paths = []
|
|
|
|
extractor = FeatureExtractor()
|
|
|
|
# Auto-detect class directories
|
|
class_dirs = {}
|
|
if not os.path.exists(data_dir):
|
|
raise ValueError(f"Directory {data_dir} tidak ditemukan!")
|
|
for entry in sorted(os.listdir(data_dir)):
|
|
entry_path = os.path.join(data_dir, entry)
|
|
if os.path.isdir(entry_path):
|
|
if entry == 'healthy':
|
|
class_dirs[entry] = 'sehat'
|
|
elif entry.startswith('pmk_'):
|
|
class_dirs[entry] = entry
|
|
elif entry == 'sick':
|
|
class_dirs[entry] = 'sakit'
|
|
|
|
if not class_dirs:
|
|
raise ValueError(f"Tidak ada folder kelas yang ditemukan di {data_dir}!")
|
|
|
|
resize_dir = os.path.join('uploads', 'resize')
|
|
threshold_dir = os.path.join('uploads', 'threshold')
|
|
|
|
for class_name, label in class_dirs.items():
|
|
class_dir = os.path.join(data_dir, class_name)
|
|
|
|
if os.path.exists(class_dir):
|
|
print(f"Memproses gambar dari: {class_dir}")
|
|
|
|
for img_file in os.listdir(class_dir):
|
|
if img_file.lower().endswith(('.jpg', '.jpeg', '.png', '.bmp')):
|
|
img_path = os.path.join(class_dir, img_file)
|
|
try:
|
|
_, img_resized, _ = preprocess_image(img_path, target_size=(256, 256))
|
|
# Use threshold-based preprocessing pipeline
|
|
# Returns: img_rgb (for RGB features), gray_processed (for GLCM)
|
|
img_rgb, gray_eq = preprocess_pipeline(img_path, target_size=(256, 256))
|
|
|
|
_save_preprocessed_training_images(
|
|
img_path=img_path,
|
|
class_name=class_name,
|
|
original_filename=img_file,
|
|
img_resized=img_resized,
|
|
gray_processed=gray_eq,
|
|
resize_dir=resize_dir,
|
|
threshold_dir=threshold_dir,
|
|
)
|
|
|
|
# Extract features (RGB averages + GLCM dari preprocessing pipeline)
|
|
img_features = extractor.extract_all_features(img_rgb, gray_eq)
|
|
|
|
features.append(img_features)
|
|
labels.append(label)
|
|
image_paths.append(img_path)
|
|
|
|
print(f" ✓ {img_file}")
|
|
|
|
except Exception as e:
|
|
print(f" ✗ Error processing {img_file}: {e}")
|
|
|
|
if len(features) == 0:
|
|
raise ValueError("Tidak ada gambar yang ditemukan untuk training!")
|
|
|
|
return np.array(features), np.array(labels), image_paths
|
|
|
|
def save_model(model, scaler, label_encoder, prefix=''):
|
|
"""Save trained model and preprocessing objects"""
|
|
os.makedirs('models', exist_ok=True)
|
|
|
|
joblib.dump(model, f'models/{prefix}knn_model.pkl')
|
|
joblib.dump(scaler, f'models/{prefix}scaler.pkl')
|
|
joblib.dump(label_encoder, f'models/{prefix}label_encoder.pkl')
|
|
|
|
print(f"Model disimpan di folder 'models/' (prefix='{prefix}')")
|
|
|
|
def load_model(prefix=''):
|
|
"""Load trained model and preprocessing objects"""
|
|
model = joblib.load(f'models/{prefix}knn_model.pkl')
|
|
scaler = joblib.load(f'models/{prefix}scaler.pkl')
|
|
label_encoder = joblib.load(f'models/{prefix}label_encoder.pkl')
|
|
|
|
return model, scaler, label_encoder
|
|
|
|
|
|
def estimate_prediction_confidence(model, features_scaled):
|
|
"""Estimate a conservative prediction confidence for the current model."""
|
|
try:
|
|
features_array = np.asarray(features_scaled)
|
|
if features_array.ndim == 1:
|
|
features_array = features_array.reshape(1, -1)
|
|
|
|
if hasattr(model, 'kneighbors') and hasattr(model, 'classes_') and hasattr(model, '_y'):
|
|
n_neighbors = int(getattr(model, 'n_neighbors', 5))
|
|
distances, indices = model.kneighbors(features_array, n_neighbors=n_neighbors)
|
|
neighbor_labels = np.asarray(model._y)[indices[0]]
|
|
weights = 1.0 / (distances[0] + 1e-6)
|
|
|
|
class_weights = {}
|
|
for label, weight in zip(neighbor_labels, weights):
|
|
class_weights[label] = class_weights.get(label, 0.0) + float(weight)
|
|
|
|
sorted_weights = sorted(class_weights.values(), reverse=True)
|
|
top_weight = sorted_weights[0]
|
|
second_weight = sorted_weights[1] if len(sorted_weights) > 1 else 0.0
|
|
total_weight = sum(sorted_weights)
|
|
n_classes = max(len(getattr(model, 'classes_', [])), 2)
|
|
|
|
support = (top_weight + 1.0) / (total_weight + n_classes)
|
|
margin = (top_weight - second_weight) / total_weight if total_weight > 0 else 0.0
|
|
confidence = (0.85 * support + 0.15 * margin) * 100.0
|
|
return float(np.clip(confidence, 50.0, 89.5))
|
|
|
|
probabilities = model.predict_proba(features_array)[0]
|
|
probabilities = np.asarray(probabilities, dtype=float)
|
|
top = float(np.max(probabilities))
|
|
if probabilities.size >= 2:
|
|
second = float(np.partition(probabilities, -2)[-2])
|
|
else:
|
|
second = 0.0
|
|
|
|
confidence = (0.9 * top + 0.1 * max(top - second, 0.0)) * 100.0
|
|
return float(np.clip(confidence, 50.0, 89.5))
|
|
except Exception:
|
|
return None
|
|
|
|
def get_scaler_params(scaler):
|
|
"""
|
|
Extract and return scaler parameters (mean, scale, variance)
|
|
|
|
Parameters:
|
|
- scaler: StandardScaler object from sklearn
|
|
|
|
Return: dictionary dengan scaler parameters
|
|
"""
|
|
if scaler is None:
|
|
return None
|
|
|
|
try:
|
|
params = {
|
|
'type': scaler.__class__.__name__,
|
|
'mean': scaler.mean_.tolist() if hasattr(scaler, 'mean_') else None,
|
|
'scale': scaler.scale_.tolist() if hasattr(scaler, 'scale_') else None,
|
|
'variance': scaler.var_.tolist() if hasattr(scaler, 'var_') else None,
|
|
'n_features': scaler.n_features_in_ if hasattr(scaler, 'n_features_in_') else None,
|
|
}
|
|
return params
|
|
except Exception as e:
|
|
print(f"Error extracting scaler params: {e}")
|
|
return None
|
|
|
|
def analyze_features():
|
|
"""Analyze and visualize feature distributions"""
|
|
os.makedirs('results', exist_ok=True)
|
|
|
|
# Load dataset features (prefer the new consolidated export)
|
|
dataset_path = 'features/dataset.csv'
|
|
legacy_path = 'features/all_features.csv'
|
|
|
|
if os.path.exists(dataset_path):
|
|
df = pd.read_csv(dataset_path)
|
|
elif os.path.exists(legacy_path):
|
|
df = pd.read_csv(legacy_path)
|
|
else:
|
|
return
|
|
|
|
extractor = FeatureExtractor()
|
|
feature_columns = [feature for feature in extractor.feature_names if feature in df.columns]
|
|
if not feature_columns:
|
|
return
|
|
|
|
label_column = 'label_name' if 'label_name' in df.columns else 'label'
|
|
|
|
# Create feature comparison plot
|
|
n_features = min(len(feature_columns), 30)
|
|
n_cols = 5
|
|
n_rows = int(np.ceil(n_features / n_cols))
|
|
fig, axes = plt.subplots(n_rows, n_cols, figsize=(20, 4 * n_rows))
|
|
axes = axes.ravel()
|
|
|
|
for idx in range(n_features):
|
|
feature = feature_columns[idx]
|
|
unique_labels = sorted(df[label_column].unique())
|
|
cmap = plt.cm.Set1
|
|
colors = [cmap(i % 9) for i in range(len(unique_labels))]
|
|
|
|
for i, label_val in enumerate(unique_labels):
|
|
vals = df[df[label_column] == label_val][feature]
|
|
axes[idx].hist(vals, alpha=0.6, label=label_val, bins=20,
|
|
color=colors[i])
|
|
axes[idx].set_title(feature)
|
|
axes[idx].legend()
|
|
|
|
for idx in range(n_features, len(axes)):
|
|
axes[idx].axis('off')
|
|
|
|
plt.tight_layout()
|
|
plt.savefig('results/feature_distribution.png', dpi=150)
|
|
plt.close()
|
|
|
|
# Create statistical summary
|
|
stats = df.groupby(label_column)[feature_columns].agg(['mean', 'std', 'min', 'max'])
|
|
stats.to_csv('results/feature_statistics.csv')
|
|
|
|
print("Analisis fitur disimpan di folder 'results/'") |