import streamlit as st import pandas as pd import numpy as np import time from ucimlrepo import fetch_ucirepo from sklearn.preprocessing import LabelEncoder, StandardScaler from sklearn.metrics.pairwise import euclidean_distances from sklearn.neighbors import KDTree # -------------- Load Dataset -------------- @st.cache_data def load_data(): dataset = fetch_ucirepo(id=544) X = dataset.data.features y = dataset.data.targets df = pd.concat([X, y], axis=1) return df # ------------------- Init Session State ------------------- if 'case_base' not in st.session_state: df_original = load_data() st.session_state.case_base = df_original.copy() # ------------------- Preprocessing ------------------- df = st.session_state.case_base df_encoded = df.copy() label_encoders = {} for col in df_encoded.select_dtypes(include='object').columns: le = LabelEncoder() df_encoded[col] = le.fit_transform(df_encoded[col]) label_encoders[col] = le features = df_encoded.drop(columns=['NObeyesdad']) target = df_encoded['NObeyesdad'] scaler = StandardScaler() features_scaled = scaler.fit_transform(features) # ------------------- HEOM Function ------------------- def heom_distance(x1, x2, numerical_cols, categorical_cols, ranges): dist = 0 for col in numerical_cols: r = ranges[col] if r > 0: d = ((x1[col] - x2[col]) / r) ** 2 dist += d for col in categorical_cols: dist += 0 if x1[col] == x2[col] else 1 return np.sqrt(dist) # ------------------- Adaptability Score ------------------- def calculate_adaptability_score(new_case_df, case_base_df): info_score = 0 epsilon = 1e-9 for col in new_case_df.columns: freq = case_base_df[col].value_counts(normalize=True) p = freq.get(new_case_df.iloc[0][col], epsilon) info_score += -np.log2(p) return info_score # ------------------- Threshold Adaptif ------------------- @st.cache_data def get_adaptive_threshold(case_base_df, percentile=50): all_scores = [ calculate_adaptability_score(pd.DataFrame([row]), case_base_df) for _, row in case_base_df.iterrows() ] return np.percentile(all_scores, percentile) # ------------------- Retain Case ------------------- def retain_case(new_case_dict, case_base_df, distance_threshold=0.5, adaptive_threshold=30, force=False): new_case_df = pd.DataFrame([new_case_dict]) for col in new_case_df.select_dtypes(include='object').columns: if col in label_encoders: new_case_df[col] = label_encoders[col].transform(new_case_df[col]) new_case_encoded = new_case_df.copy() new_case_scaled = scaler.transform(new_case_encoded) numerical_cols = features.select_dtypes(include=np.number).columns.tolist() categorical_cols = [col for col in features.columns if col not in numerical_cols] feature_ranges = {col: df_encoded[col].max() - df_encoded[col].min() for col in numerical_cols} raw_input = new_case_encoded.iloc[0] heom_distances = [ heom_distance(raw_input, row, numerical_cols, categorical_cols, feature_ranges) for _, row in features.iterrows() ] min_dist = min(heom_distances) adaptability_score = calculate_adaptability_score(new_case_encoded, df_encoded.drop(columns=['NObeyesdad'])) retain_flag = (min_dist > distance_threshold and adaptability_score < adaptive_threshold) or force return retain_flag, min_dist, adaptability_score, adaptive_threshold # ------------------- CBR ------------------- def case_based_reasoning(new_input_dict): new_input_df = pd.DataFrame([new_input_dict]) for col in new_input_df.select_dtypes(include='object').columns: if col in label_encoders: new_input_df[col] = label_encoders[col].transform(new_input_df[col]) new_input_scaled = scaler.transform(new_input_df) # ========== Euclidean ========== # start_time = time.time() eucl_distances = euclidean_distances(new_input_scaled, features_scaled) eucl_closest_index = eucl_distances.argmin() eucl_score = float(eucl_distances[0][eucl_closest_index]) eucl_time = time.time() - start_time st.write(f"🧮 Euclidean Similarity Score: `{eucl_score:.4f}`") st.write(f"⏱️ Euclidean Search Time: `{eucl_time:.6f}` seconds") # ========== Euclidean + KD-Tree ========== # tree = KDTree(features_scaled) # Build outside timing block start_time = time.time() kd_dist, kd_idx = tree.query(new_input_scaled, k=1) kd_index = kd_idx[0][0] kd_score = float(kd_dist[0][0]) kd_time = time.time() - start_time st.write(f"🧮 Euclidean (KD-Tree) Similarity Score: `{kd_score:.4f}`") st.write(f"⏱️ Euclidean (KD-Tree) Search Time: `{kd_time:.6f}` seconds") # ========== HEOM ========== # raw_input = new_input_df.iloc[0] numerical_cols = features.select_dtypes(include=np.number).columns.tolist() categorical_cols = [col for col in features.columns if col not in numerical_cols] feature_ranges = {col: df_encoded[col].max() - df_encoded[col].min() for col in numerical_cols} start_time = time.time() heom_distances = [ heom_distance(raw_input, row, numerical_cols, categorical_cols, feature_ranges) for _, row in features.iterrows() ] heom_closest_index = int(np.argmin(heom_distances)) heom_score = float(heom_distances[heom_closest_index]) heom_time = time.time() - start_time st.write(f"🧮 HEOM Similarity Score: `{heom_score:.4f}`") st.write(f"⏱️ HEOM Search Time: `{heom_time:.6f}` seconds") # ========== HEOM + KD-Tree (Hybrid) ========== # k_candidates = 50 kd_tree_indices = tree.query(new_input_scaled, k=k_candidates)[1][0] start_time = time.time() heom_candidate_dists = [] for idx in kd_tree_indices: row = features.iloc[idx] dist = heom_distance(raw_input, row, numerical_cols, categorical_cols, feature_ranges) heom_candidate_dists.append(dist) heom_kdtree_index = int(kd_tree_indices[int(np.argmin(heom_candidate_dists))]) heom_kdtree_score = float(min(heom_candidate_dists)) heom_kdtree_time = time.time() - start_time st.write(f"🧮 HEOM (KD-Tree Hybrid) Similarity Score: `{heom_kdtree_score:.4f}`") st.write(f"⏱️ HEOM (KD-Tree Hybrid) Search Time: `{heom_kdtree_time:.6f}` seconds") return { "euclidean": { "index": int(eucl_closest_index), "distance": eucl_score, }, "kdtree": { "index": int(kd_index), "distance": kd_score, }, "heom": { "index": heom_closest_index, "case": df.iloc[heom_closest_index].to_dict() }, "heom_kdtree": { "index": heom_kdtree_index, "case": df.iloc[heom_kdtree_index].to_dict() } } # ------------------- Streamlit UI ------------------- st.title("🧠 CBR Obesitas + Retain Adaptif (Live Session)") st.markdown(f"Jumlah kasus dalam database saat ini: **{len(st.session_state.case_base)} kasus**") user_input = { 'Gender': st.selectbox("Gender", ['Male', 'Female']), 'Age': st.number_input("Age", 10, 100, 25), 'Height': st.number_input("Height (in meters)", 1.0, 2.5, 1.70), 'Weight': st.number_input("Weight (in kg)", 30.0, 200.0, 70.0), 'family_history_with_overweight': st.selectbox("Family history with overweight", ['yes', 'no']), 'FAVC': st.selectbox("Frequent consumption of high caloric food", ['yes', 'no']), 'FCVC': st.slider("Vegetable consumption (0–3)", 0.0, 3.0, 2.0), 'NCP': st.slider("Number of main meals", 1.0, 5.0, 3.0), 'CAEC': st.selectbox("Food between meals", ['no', 'Sometimes', 'Frequently', 'Always']), 'SMOKE': st.selectbox("Do you smoke?", ['yes', 'no']), 'CH2O': st.slider("Daily water intake", 0.0, 3.0, 2.0), 'SCC': st.selectbox("Calories monitoring", ['yes', 'no']), 'FAF': st.slider("Physical activity (hrs/week)", 0.0, 5.0, 1.0), 'TUE': st.slider("Tech usage (hrs/day)", 0.0, 5.0, 1.0), 'CALC': st.selectbox("Alcohol consumption", ['no', 'Sometimes', 'Frequently', 'Always']), 'MTRANS': st.selectbox("Transport", ['Automobile', 'Motorbike', 'Bike', 'Public_Transportation', 'Walking']) } force_save = st.checkbox("💾 Simpan paksa jika ditolak?", value=False) if st.button("🔍 Prediksi dan Evaluasi Retain"): result = case_based_reasoning(user_input) st.markdown("### 🔢 Kasus Paling Mirip") st.json(result['heom_kdtree']['case']) # Ambil prediksi dan tambahkan ke input predicted_label = result['heom_kdtree']['case']['NObeyesdad'] user_input_with_label = user_input.copy() user_input_with_label['NObeyesdad'] = predicted_label # Tampilkan label prediksi st.markdown("### 🧾 Prediksi Kategori Obesitas") st.write(f"🎯 **{predicted_label}** (berdasarkan kasus paling mirip)") # Evaluasi Retain adaptive_threshold = get_adaptive_threshold(df_encoded.drop(columns=['NObeyesdad']), percentile=50) retain_flag, min_dist, adaptability_score, threshold_used = retain_case( user_input, df_encoded, distance_threshold=0.5, adaptive_threshold=adaptive_threshold, force=force_save ) st.markdown("### 📥 Evaluasi Retain") st.write(f"HEOM Distance Terdekat: `{min_dist:.3f}`") st.write(f"Skor Adaptabilitas: `{adaptability_score:.2f}` (Threshold adaptif: `{threshold_used:.2f}`)") if retain_flag: st.session_state.case_base = pd.concat([st.session_state.case_base, pd.DataFrame([user_input_with_label])], ignore_index=True) st.success(f"✅ Kasus berhasil disimpan. Jumlah kasus sekarang: {len(st.session_state.case_base)}") else: st.info("❌ Kasus tidak disimpan otomatis (tidak adaptif & terlalu mirip).") if force_save: st.session_state.case_base = pd.concat([st.session_state.case_base, pd.DataFrame([user_input_with_label])], ignore_index=True) st.warning(f"⚠️ Simpan paksa dilakukan. Jumlah kasus sekarang: {len(st.session_state.case_base)}")