Create app.py

app.py

@@ -0,0 +1,218 @@
+from fastapi import FastAPI, HTTPException, Query
+from functools import lru_cache
+import pandas as pd
+import os
+from typing import Dict, Any
+
+app = FastAPI(title="Dashboard Akademik API")
+
+# Path ke CSV: bisa override lewat env var CSV_PATH
+CSV_PATH = os.getenv("CSV_PATH", "generated_dummy_data.csv")
+
+MAP_NILAI = {
+    "A": 4.0,
+    "AB": 3.5,
+    "B": 3.0,
+    "BC": 2.5,
+    "C": 2.0,
+    "D": 1.0,
+    "E": 0.0
+}
+
+def _ensure_columns(df: pd.DataFrame):
+    required = {
+        "kode_mhs", "nama_prodi", "id_smt", "kode_mk", "nama_mk",
+        "RMK", "sks", "nilai_akhir", "nilai_huruf", "Tahun angkatan",
+        "Semester_sekarang", "Deskripsi Matkul"
+    }
+    missing = required - set(df.columns)
+    if missing:
+        raise ValueError(f"CSV missing required columns: {missing}")
+
+@lru_cache(maxsize=1)
+def load_data_cached() -> pd.DataFrame:
+    """
+    Load CSV into DataFrame and cache it for reuse.
+    Jika ingin reload, panggil /reload_data endpoint (opsional).
+    """
+    if not os.path.exists(CSV_PATH):
+        raise FileNotFoundError(f"CSV not found at path: {CSV_PATH}")
+    df = pd.read_csv(CSV_PATH)
+    # Normalisasi kolom whitespace
+    df.columns = [c.strip() for c in df.columns]
+    _ensure_columns(df)
+    # Pastikan tipe
+    df["sks"] = pd.to_numeric(df["sks"], errors="coerce").fillna(0).astype(int)
+    df["id_smt"] = pd.to_numeric(df["id_smt"], errors="coerce").fillna(0).astype(int)
+    # map nilai huruf -> numerik, jika kosong set NaN -> 0.0
+    df["nilai_huruf"] = df["nilai_huruf"].astype(str).str.strip()
+    df["nilai_numerik"] = df["nilai_huruf"].map(MAP_NILAI)
+    # bila nilai_huruf tidak ada / invalid -> gunakan nilai_akhir jika ada (0-100 scale)
+    def fallback_numeric(row):
+        if pd.notna(row["nilai_numerik"]):
+            return row["nilai_numerik"]
+        try:
+            v = float(row.get("nilai_akhir", None))
+            # convert angka 0-100 ke skala 0-4 berdasarkan mapping ranges
+            if v >= 86:
+                return 4.0
+            if v >= 76:
+                return 3.5
+            if v >= 66:
+                return 3.0
+            if v >= 61:
+                return 2.5
+            if v >= 56:
+                return 2.0
+            if v >= 41:
+                return 1.0
+            return 0.0
+        except Exception:
+            return 0.0
+    df["nilai_numerik"] = df.apply(fallback_numeric, axis=1)
+    return df
+
+def get_final_records(df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Ambil record terakhir tiap (kode_mhs, kode_mk) berdasarkan id_smt.
+    Ini memenuhi aturan: nilai terakhir berlaku, SKS dihitung sekali.
+    """
+    # sort by id_smt lalu ambil tail(1) per group
+    df_sorted = df.sort_values(["kode_mhs", "kode_mk", "id_smt"])
+    final = df_sorted.groupby(["kode_mhs", "kode_mk"], as_index=False).last()
+    return final
+
+@app.get("/jumlah_mahasiswa")
+def jumlah_mahasiswa(reload: bool = Query(False, description="set true to reload CSV from disk")):
+    """
+    Mengembalikan total mahasiswa unik (kode_mhs).
+    Tambah ?reload=true untuk load ulang CSV.
+    """
+    if reload:
+        load_data_cached.cache_clear()
+    try:
+        df = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    total = int(df["kode_mhs"].nunique())
+    return {"total_mahasiswa": total}
+
+@app.get("/jumlah_per_angkatan")
+def jumlah_per_angkatan(reload: bool = Query(False, description="set true to reload CSV from disk")):
+    if reload:
+        load_data_cached.cache_clear()
+    try:
+        df = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    per_ang = df.groupby("Tahun angkatan")["kode_mhs"].nunique().sort_index().to_dict()
+    # convert keys to str (JSON-friendly)
+    per_ang = {str(k): int(v) for k, v in per_ang.items()}
+    return {"mahasiswa_per_angkatan": per_ang}
+
+@app.get("/eligible_ta")
+def eligible_ta(reload: bool = Query(False, description="set true to reload CSV from disk"),
+                min_sks: int = Query(110, description="threshold SKS untuk eligible (default:110)")):
+    """
+    Mengembalikan daftar mahasiswa yang total SKS (menggunakan nilai terakhir tiap matkul) > min_sks.
+    """
+    if reload:
+        load_data_cached.cache_clear()
+    try:
+        df = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    final = get_final_records(df)
+    sks_per_mhs = final.groupby("kode_mhs")["sks"].sum()
+    eligible_series = sks_per_mhs[sks_per_mhs > min_sks]
+    eligible_list = eligible_series.sort_values(ascending=False).index.to_list()
+    # optional: juga sertakan total SKS per mahasiswa pada hasil
+    eligible_info = [{"kode_mhs": m, "total_sks": int(sks_per_mhs.loc[m])} for m in eligible_list]
+    return {"jumlah_eligible": len(eligible_list), "daftar": eligible_info}
+
+@app.get("/ipk_rata_rata")
+def ipk_rata_rata(reload: bool = Query(False, description="set true to reload CSV from disk")):
+    """
+    Menghitung IPK rata-rata seluruh mahasiswa:
+    - IPS per semester = (Σ sks * nilai_numerik) / Σ sks  (menggunakan nilai terakhir utk matkul yang diulang)
+    - IPK mahasiswa = rata-rata IPS mahasiswa
+    - IPK rata-rata = rata-rata IPK semua mahasiswa
+    """
+    if reload:
+        load_data_cached.cache_clear()
+    try:
+        df = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    final = get_final_records(df)
+    # bobot per record
+    final["total_bobot"] = final["sks"] * final["nilai_numerik"]
+    # IPS per mahasiswa per semester
+    grp = final.groupby(["kode_mhs", "id_smt"]).agg(
+        total_bobot=pd.NamedAgg(column="total_bobot", aggfunc="sum"),
+        total_sks=pd.NamedAgg(column="sks", aggfunc="sum")
+    ).reset_index()
+    # Hindari pembagian 0
+    grp["ips"] = grp.apply(lambda r: (r["total_bobot"] / r["total_sks"]) if r["total_sks"] > 0 else 0.0, axis=1)
+    # IPK = rata-rata ips tiap mahasiswa
+    ipk_series = grp.groupby("kode_mhs")["ips"].mean()
+    mean_ipk = float(round(ipk_series.mean(), 3)) if not ipk_series.empty else 0.0
+    # optional: distribusi ipk (quartiles)
+    q = ipk_series.quantile([0.25, 0.5, 0.75]).to_dict() if not ipk_series.empty else {}
+    q = {str(k): float(v) for k, v in q.items()}
+    return {"ipk_rata_rata": mean_ipk, "ipk_quartiles": q}
+
+@app.get("/dashboard_summary")
+def dashboard_summary(reload: bool = Query(False, description="set true to reload CSV from disk")):
+    if reload:
+        load_data_cached.cache_clear()
+    try:
+        df = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+    final = get_final_records(df)
+    total_mhs = int(df["kode_mhs"].nunique())
+    per_ang = df.groupby("Tahun angkatan")["kode_mhs"].nunique().sort_index().to_dict()
+    per_ang = {str(k): int(v) for k, v in per_ang.items()}
+
+    sks_per_mhs = final.groupby("kode_mhs")["sks"].sum()
+    eligible_series = sks_per_mhs[sks_per_mhs > 110]
+    eligible_list = [{"kode_mhs": m, "total_sks": int(sks_per_mhs.loc[m])} for m in eligible_series.sort_values(ascending=False).index]
+
+    # IPK
+    final["total_bobot"] = final["sks"] * final["nilai_numerik"]
+    grp = final.groupby(["kode_mhs", "id_smt"]).agg(
+        total_bobot=pd.NamedAgg(column="total_bobot", aggfunc="sum"),
+        total_sks=pd.NamedAgg(column="sks", aggfunc="sum")
+    ).reset_index()
+    grp["ips"] = grp.apply(lambda r: (r["total_bobot"] / r["total_sks"]) if r["total_sks"] > 0 else 0.0, axis=1)
+    ipk_series = grp.groupby("kode_mhs")["ips"].mean()
+    mean_ipk = float(round(ipk_series.mean(), 3)) if not ipk_series.empty else 0.0
+
+    summary: Dict[str, Any] = {
+        "total_mahasiswa": total_mhs,
+        "mahasiswa_per_angkatan": per_ang,
+        "eligible_ta": {
+            "jumlah": int(eligible_series.shape[0]),
+            "daftar": eligible_list
+        },
+        "ipk": {
+            "rata_rata_ipk": mean_ipk
+        }
+    }
+    return summary
+
+@app.post("/reload_data")
+def reload_data():
+    """
+    Endpoint untuk menghapus cache dan reload CSV.
+    (tidak perlu query param, panggil endpoint ini setelah CSV diganti)
+    """
+    load_data_cached.cache_clear()
+    try:
+        # force load to check file validity
+        _ = load_data_cached()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Reload failed: {e}")
+    return {"status": "reloaded"}