# ====================================================================== # --- main.py (FULL VERSION 1.7.0) --- # ====================================================================== import os import json import networkx as nx import pandas as pd import skops.io as sio from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Any # --- IMPOR MODUL LOKAL --- # Pastikan file explanation_builder.py dan graduation_logic.py ada di folder yang sama from .explanation_builder import build_full_response from .graduation_logic import predict_graduation_status # ====================================================================== # 1. Inisialisasi Aplikasi FastAPI # ====================================================================== app = FastAPI( title="GCOMPRO API Service", description="API untuk Prediksi Risiko Akademik, Rekomendasi Mata Kuliah, dan Cek Kelulusan Tepat Waktu.", version="1.7.0" ) # ====================================================================== # 2. Struktur Data Input/Output (Pydantic Models) # ====================================================================== # --- [APP 1] Model untuk Prediksi Risiko --- class StudentFeatures(BaseModel): IPK_Terakhir: float IPS_Terakhir: float Total_SKS: int IPS_Tertinggi: float IPS_Terendah: float Rentang_IPS: float Jumlah_MK_Gagal: int Total_SKS_Gagal: int Tren_IPS_Slope: float Perubahan_Kinerja_Terakhir: float IPK_Ternormalisasi_SKS: float Profil_Tren: str class PredictionExplanation(BaseModel): opening_line: str factors: List[str] recommendation: str class PredictionResponse(BaseModel): prediction: str probabilities: Dict[str, float] explanation: PredictionExplanation # --- [APP 2] Model untuk Rekomendasi MK --- class RecommendationRequest(BaseModel): current_semester: int courses_passed: List[str] mk_pilihan_failed: List[str] = [] class PrerequisiteInfo(BaseModel): code: str name: str class CourseRecommendation(BaseModel): rank: int code: str name: str sks: int semester_plan: int reason: str is_tertinggal: bool priority_score: float prerequisites: List[PrerequisiteInfo] # --- [APP 3] Model untuk Prediksi Kelulusan (LTW) --- class GraduationCheckRequest(BaseModel): current_semester: int total_sks_passed: int ipk_last_semester: float courses_passed: List[str] = [] # Optional, tapi disarankan diisi untuk validasi graf class GraduationCheckResponse(BaseModel): status: str color: str description: str stats: Dict[str, Any] # ====================================================================== # 3. Variabel Global & Database Hardcode # ====================================================================== # Database Mata Kuliah Pilihan (Hardcoded untuk fallback nama/sks) ELECTIVE_COURSES_DB = { "AAK4ABB3": {"name": "New Generation Network", "sks": 3}, "AAK4BBB3": {"name": "Software Defined Network", "sks": 3}, "AAK4CBB3": {"name": "Rekayasa Jaringan", "sks": 3}, "AAK4DBB3": {"name": "Aplikasi Cyber Security", "sks": 3}, "AAK4EBB3": {"name": "Manajemen Telekomunikasi dan Transformasi Digital", "sks": 3}, "AAK4FBB3": {"name": "Adaptive Network", "sks": 3}, "AAK4GBB3": {"name": "Cloud Computing", "sks": 3}, "AAK4HBB3": {"name": "Koding dan Kompresi", "sks": 3}, "AAK4IBB3": {"name": "Steganografi dan Watermarking", "sks": 3}, "AAK4JBB3": {"name": "Mobile Application", "sks": 3}, "AAK4KBB3": {"name": "Speech Signal Processing", "sks": 3}, "AAK4LBB3": {"name": "Komunikasi Akses Wireless", "sks": 3}, "AAK4MBB3": {"name": "Wireless Optical Communication", "sks": 3}, "AAK4NBB3": {"name": "Broadband Optical Network", "sks": 3}, "AAK4OBB3": {"name": "Sistem Komunikasi Satelit", "sks": 3}, "AAK4PBB3": {"name": "Rekayasa Radio", "sks": 3}, "AAK4QBB3": {"name": "Radar, Navigasi dan Remote Sensing", "sks": 3}, "AAK4RBB3": {"name": "5G and Beyond", "sks": 3}, "AAK4SBB3": {"name": "Software Defined Radio", "sks": 3}, "AAK4TBB3": {"name": "Robotic Process Automation", "sks": 3}, "AAK4UBB3": {"name": "Rekayasa Frekuensi Radio dalam Komunikasi Selular", "sks": 3}, "AAK4VBB3": {"name": "Teknologi Radio Access Network (RAN)", "sks": 3}, "AAK4WBB3": {"name": "Internet of Things: Protokol, Platform, dan AI", "sks": 3}, "AAK4XBB3": {"name": "Jaringan Core Telekomunikasi", "sks": 3}, "AAK4YBB3": {"name": "Ethical Hacking", "sks": 3}, "AAK4ZBB3": {"name": "Keamanan Komunikasi Data", "sks": 3}, "AAK47BB3": {"name": "Rekayasa Penyiaran Digital", "sks": 3} } # Variabel Global ML ml_model = None MODEL_FEATURES = [ 'IPK_Terakhir', 'IPS_Terakhir', 'Total_SKS', 'IPS_Tertinggi', 'IPS_Terendah', 'Rentang_IPS', 'Jumlah_MK_Gagal', 'Total_SKS_Gagal', 'Tren_IPS_Slope', 'Perubahan_Kinerja_Terakhir', 'IPK_Ternormalisasi_SKS', 'Tren_Menaik', 'Tren_Menurun', 'Tren_Stabil' ] # Variabel Global Graph G = nx.DiGraph() course_details_map = {} prereq_map = {} out_degree_map = {} # --- Fungsi Pemuatan Data --- def load_ml_model(): """Memuat model ML dari file .skops""" global ml_model MODEL_PATH = os.path.join(os.path.dirname(__file__), "model_risiko_akademik.skops") print(f"Mencoba memuat model ML dari: {MODEL_PATH}") try: trusted_types = [ "numpy.ndarray", "numpy.core.multiarray.scalar", "sklearn.tree._classes.DecisionTreeClassifier", "_codecs.encode", "joblib.numpy_pickle.NumpyArrayWrapper", "numpy.core.multiarray._reconstruct", "numpy.dtype", "sklearn.tree._tree.Tree" ] ml_model = sio.load(MODEL_PATH, trusted=trusted_types) print("Model ML berhasil dimuat.") except Exception as e: print(f"ERROR: Gagal memuat model ML dari {MODEL_PATH}: {e}") def load_graph_data(): """Memuat dan memproses data graf kurikulum dari JSON""" global G, course_details_map, prereq_map, out_degree_map JSON_PATH = os.path.join(os.path.dirname(__file__), "OK_matkul_graph.json") print(f"Mencoba memuat data graf dari: {JSON_PATH}") prereq_edge_count = 0 try: with open(JSON_PATH, "r") as f: data = json.load(f) for node in data["nodes"]: course_details_map[node["code"]] = node G.add_node(node["code"]) for edge in data["edges"]: if edge["type"] == "prereq": prereq_edge_count += 1 G.add_edge(edge["from"], edge["to"]) if edge["to"] not in prereq_map: prereq_map[edge["to"]] = [] prereq_map[edge["to"]].append(edge["from"]) for node_code in G.nodes(): out_degree_map[node_code] = G.out_degree(node_code) print(f"Data graf berhasil dimuat. Total Edges: {prereq_edge_count}") except FileNotFoundError: print(f"ERROR: {JSON_PATH} tidak ditemukan!") except Exception as e: print(f"Error saat memuat graf: {e}") @app.on_event("startup") def on_startup(): load_ml_model() load_graph_data() # ====================================================================== # 4. Helper Functions (Business Logic) # ====================================================================== def get_recommendations_logic(current_semester: int, courses_passed_list: List[str], mk_pilihan_failed_list: List[str]) -> List[Dict[str, Any]]: """Logika utama untuk rekomendasi mata kuliah.""" passed_set = set(courses_passed_list) all_courses_set = set(course_details_map.keys()) not_passed_courses = all_courses_set - passed_set raw_candidates = [] for course_code in not_passed_courses: prereqs = prereq_map.get(course_code, []) if all(p_code in passed_set for p_code in prereqs): details = course_details_map.get(course_code) if not details: continue out_degree = out_degree_map.get(course_code, 0) semester = details.get("semester_plan", 1) priority_score = (out_degree / semester) if semester > 0 else 0 candidate_data = details.copy() candidate_data["priority_score"] = priority_score candidate_data["is_retake_elective"] = False raw_candidates.append(candidate_data) elective_slots = [] regular_candidates = [] for cand in raw_candidates: if cand["code"].startswith("MK_PILIHAN"): elective_slots.append(cand) else: regular_candidates.append(cand) elective_slots.sort(key=lambda x: x["semester_plan"]) processed_electives = [] failed_idx = 0 while failed_idx < len(mk_pilihan_failed_list) and len(elective_slots) > 0: slot = elective_slots.pop(0) failed_code = mk_pilihan_failed_list[failed_idx] if failed_code in ELECTIVE_COURSES_DB: real_name = ELECTIVE_COURSES_DB[failed_code]["name"] real_sks = ELECTIVE_COURSES_DB[failed_code]["sks"] else: real_name = "Mata Kuliah Pilihan (Unknown)" real_sks = 3 slot["code"] = failed_code slot["name"] = f"{real_name} (Mengulang)" slot["sks"] = real_sks slot["priority_score"] += 1.0 slot["is_retake_elective"] = True processed_electives.append(slot) failed_idx += 1 processed_electives.extend(elective_slots) final_pool = regular_candidates + processed_electives final_ranked_list = sorted( final_pool, key=lambda x: (-x["priority_score"], x["semester_plan"]) ) return final_ranked_list def apply_prediction_overrides(original_prediction: str, student_data: StudentFeatures) -> str: """Guardrails: Menerapkan aturan bisnis manual untuk override prediksi ML.""" new_prediction = original_prediction # Aturan 1: FALSE NEGATIVE (Model optimis, padahal IPK rendah) if (student_data.IPK_Terakhir < 2.40 or student_data.Jumlah_MK_Gagal >= 3) and \ (original_prediction in ["Aman", "Resiko Rendah"]): new_prediction = "Resiko Sedang" # Aturan 1B: Varian Parah if (student_data.IPK_Terakhir < 2.10 or student_data.Jumlah_MK_Gagal >= 5): new_prediction = "Resiko Tinggi" # Aturan 2: FALSE POSITIVE (Model pesimis, padahal performa naik) if (student_data.IPK_Terakhir > 2.75 and student_data.Jumlah_MK_Gagal == 0 and student_data.Tren_IPS_Slope > 0.05) and \ (original_prediction == "Resiko Tinggi" or original_prediction == "Resiko Sedang"): new_prediction = "Resiko Rendah" # Aturan 2B: Varian Sangat Baik if (student_data.IPK_Terakhir > 3.25 and student_data.Jumlah_MK_Gagal == 0) and \ (original_prediction != "Aman"): new_prediction = "Aman" return new_prediction # ====================================================================== # 5. Endpoints API # ====================================================================== @app.get("/") def read_root(): return { "message": "Selamat Datang di API Layanan Akademik Mahasiswa", "status": "ready", "endpoints": ["/predict/", "/recommend/", "/predict-graduation/"] } # --- ENDPOINT 1: PREDIKSI RISIKO AKADEMIK --- @app.post("/predict/", response_model=PredictionResponse) def predict_risk(student_data: StudentFeatures): if ml_model is None: raise HTTPException(status_code=503, detail="Model ML belum siap. Silakan coba lagi nanti.") data = student_data.dict() input_df = pd.DataFrame([data]) input_encoded = pd.get_dummies(input_df, columns=['Profil_Tren'], prefix='Tren') input_encoded = input_encoded.reindex(columns=MODEL_FEATURES, fill_value=False) try: # 1. Prediksi ML Dasar prediction_val = ml_model.predict(input_encoded)[0] prediction_proba = ml_model.predict_proba(input_encoded) classes = ml_model.classes_ probabilities = dict(zip(classes, prediction_proba[0])) structured_rules = [] # 2. Ekstraksi Decision Path if hasattr(ml_model, 'tree_'): try: tree = ml_model.tree_ feature_names = MODEL_FEATURES path = ml_model.decision_path(input_encoded) node_indices = path.indices[path.indptr[0]:path.indptr[1]] for node_id in node_indices[:-1]: feature_index = tree.feature[node_id] feature_name = feature_names[feature_index] threshold = tree.threshold[node_id] sample_value = input_encoded.iloc[0, feature_index] condition_str = "rendah" if sample_value <= threshold else "tinggi" structured_rules.append({ "feature": feature_name, "condition": condition_str, "threshold": threshold, "value": sample_value }) except Exception: pass # Lanjut tanpa path jika error # 3. Terapkan Override (Guardrails) final_prediction = apply_prediction_overrides(prediction_val, student_data) # 4. Sesuaikan Probabilitas dengan Override final_probabilities = {key: 0.0 for key in probabilities.keys()} if final_prediction in final_probabilities: final_probabilities[final_prediction] = 1.0 else: first_key = next(iter(final_probabilities)) final_probabilities[first_key] = 1.0 # 5. Bangun Penjelasan Teks explanation_obj = build_full_response(structured_rules, final_prediction) return PredictionResponse( prediction=final_prediction, probabilities=final_probabilities, explanation=explanation_obj ) except Exception as e: raise HTTPException(status_code=500, detail=f"Terjadi kesalahan saat prediksi: {e}") # --- ENDPOINT 2: REKOMENDASI MATA KULIAH --- @app.post("/recommend/", response_model=List[CourseRecommendation]) async def recommend_courses(request: RecommendationRequest): if not course_details_map: raise HTTPException(status_code=503, detail="Data kurikulum belum siap. Silakan coba lagi nanti.") ranked_candidates = get_recommendations_logic( request.current_semester, request.courses_passed, request.mk_pilihan_failed ) top_3_candidates = ranked_candidates[:3] response_output = [] for i, course in enumerate(top_3_candidates): rank = i + 1 is_tertinggal_status = False reason = "Rekomendasi semester ini" if course.get("is_retake_elective"): reason = "Wajib Mengulang (MK Pilihan Gagal)" is_tertinggal_status = True elif course["semester_plan"] < request.current_semester: reason = f"Mata kuliah tertinggal (Semester {course['semester_plan']})" is_tertinggal_status = True elif course["semester_plan"] > request.current_semester: reason = f"Akselerasi (Semester {course['semester_plan']})" prereq_codes = prereq_map.get(course["code"], []) prereq_details_list = [] for p_code in prereq_codes: if p_code in course_details_map: prereq_details_list.append( PrerequisiteInfo(code=p_code, name=course_details_map[p_code]["name"]) ) response_output.append( CourseRecommendation( rank=rank, code=course["code"], name=course["name"], sks=course["sks"], semester_plan=course["semester_plan"], reason=reason, is_tertinggal=is_tertinggal_status, priority_score=course["priority_score"], prerequisites=prereq_details_list ) ) return response_output # --- ENDPOINT 3: PREDIKSI KELULUSAN TEPAT WAKTU (LTW) --- @app.post("/predict-graduation/", response_model=GraduationCheckResponse) def check_graduation_status(request: GraduationCheckRequest): """ Endpoint untuk mengecek apakah mahasiswa masih on-track lulus di Semester 8 berdasarkan sisa SKS, kapasitas IPK, dan rantai prasyarat (Graf). """ result = predict_graduation_status( current_semester=request.current_semester, total_sks_passed=request.total_sks_passed, last_gpa=request.ipk_last_semester, graph_G=G, # Pass Graf Global (Reference) passed_courses=request.courses_passed # Pass data matkul user ) return GraduationCheckResponse(**result)