253 lines
9.9 KiB
Python
253 lines
9.9 KiB
Python
"""
|
|
model_manager.py
|
|
=================
|
|
Validasi file all_results.pkl baru yang diupload admin (panel "Perbarui
|
|
Model" di views/perbandingan_konfigurasi.py), backup otomatis file lama,
|
|
dan proses merge ke file master di data/all_results.pkl.
|
|
|
|
Alur pemakaian dari sisi views/perbandingan_konfigurasi.py:
|
|
|
|
1. admin upload file lewat st.file_uploader -> dapat raw bytes
|
|
2. new_data, errors = validate_new_data(raw_bytes)
|
|
-> kalau `errors` TIDAK kosong, JANGAN lanjut ke merge sama sekali.
|
|
3. diff = preview_merge(current_all_results, new_data)
|
|
-> tampilkan ke admin: produk mana yang baru / ditimpa / tidak berubah,
|
|
sebelum admin menekan tombol konfirmasi.
|
|
4. merged, backup_path = apply_merge(current_all_results, new_data)
|
|
-> ini yang benar-benar menulis ke disk (dan sebelumnya membackup
|
|
file lama ke folder backups/).
|
|
|
|
Validasi di sini SENGAJA memakai aturan struktur yang sama seperti asumsi
|
|
di utils/data_loader.py (lihat EXPECTED_SCENARIOS, REGRESSOR_SCENARIOS) --
|
|
supaya file yang lolos validasi di sini dijamin bisa dibaca data_loader
|
|
tanpa error di kemudian hari.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import pickle
|
|
import shutil
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import streamlit as st
|
|
|
|
from utils.data_loader import (
|
|
DEFAULT_DATA_PATH,
|
|
EXPECTED_SCENARIOS,
|
|
REGRESSOR_SCENARIOS,
|
|
reload_all_results,
|
|
)
|
|
|
|
BACKUP_DIR = "backups"
|
|
|
|
# Key yang wajib ada di level produk, di level tiap skenario, dan di config skenario.
|
|
REQUIRED_PRODUCT_KEYS = {"data", "scenarios", "optimal_scenario"}
|
|
REQUIRED_SCENARIO_KEYS = {"config", "best_params", "best_metrics", "best_model", "best_forecast"}
|
|
REQUIRED_CONFIG_KEYS = {"use_regressors", "use_log"}
|
|
REQUIRED_METRIC_KEYS = {"MAE", "RMSE", "MAPE"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validasi
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def validate_new_data(raw_bytes: bytes) -> tuple[dict[str, Any] | None, list[str]]:
|
|
"""
|
|
Unpickle bytes hasil upload lalu validasi strukturnya secara menyeluruh.
|
|
|
|
Returns
|
|
-------
|
|
(data, errors)
|
|
`data` bernilai None kalau file gagal di-unpickle sama sekali
|
|
(format rusak / bukan pickle / bukan dict).
|
|
`errors` berisi daftar pesan masalah dalam Bahasa Indonesia,
|
|
siap ditampilkan langsung ke admin lewat st.error/st.write.
|
|
Kalau `errors` kosong, file AMAN untuk di-merge.
|
|
|
|
Catatan keamanan: pickle.loads() bisa menjalankan kode arbitrer kalau
|
|
file berasal dari sumber tidak terpercaya. Di aplikasi ini upload
|
|
dibatasi HANYA untuk role admin (lihat auth.py + app.py), jadi risiko
|
|
ini diterima sebagai bagian dari desain (admin dianggap tepercaya).
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
try:
|
|
data = pickle.loads(raw_bytes)
|
|
except Exception as e: # noqa: BLE001 -- sengaja luas, sumber file tidak terpercaya
|
|
return None, [f"File tidak bisa dibaca sebagai pickle: {e}"]
|
|
|
|
if not isinstance(data, dict) or len(data) == 0:
|
|
return None, ["File berhasil dibaca, tapi isi dictionary produk tidak valid (atau kosong)."]
|
|
|
|
for product, entry in data.items():
|
|
if not isinstance(entry, dict):
|
|
errors.append(f"[{product}] entry harus berupa dictionary, ditemukan {type(entry).__name__}.")
|
|
continue
|
|
|
|
missing = REQUIRED_PRODUCT_KEYS - entry.keys()
|
|
if missing:
|
|
errors.append(f"[{product}] key wajib hilang di level produk: {sorted(missing)}.")
|
|
continue
|
|
|
|
errors.extend(_validate_data_block(product, entry.get("data")))
|
|
scenario_errors, scenarios = _validate_scenarios_block(product, entry.get("scenarios"))
|
|
errors.extend(scenario_errors)
|
|
|
|
optimal = entry.get("optimal_scenario")
|
|
if scenarios and optimal not in scenarios:
|
|
errors.append(
|
|
f"[{product}] optimal_scenario ('{optimal}') tidak cocok dengan salah satu "
|
|
f"skenario yang ada: {sorted(scenarios.keys())}."
|
|
)
|
|
|
|
if optimal in REGRESSOR_SCENARIOS and "scaler" not in entry:
|
|
errors.append(
|
|
f"[{product}] skenario optimal ('{optimal}') butuh regresor eksternal, "
|
|
"tapi key 'scaler' tidak ditemukan di level produk."
|
|
)
|
|
|
|
return data, errors
|
|
|
|
|
|
def _validate_data_block(product: str, data_block: Any) -> list[str]:
|
|
errors: list[str] = []
|
|
if not isinstance(data_block, dict):
|
|
return [f"[{product}] 'data' harus berupa dictionary berisi 'train' dan 'test'."]
|
|
|
|
for split in ("train", "test"):
|
|
if split not in data_block:
|
|
errors.append(f"[{product}] 'data.{split}' tidak ditemukan.")
|
|
return errors
|
|
|
|
|
|
def _validate_scenarios_block(product: str, scenarios: Any) -> tuple[list[str], dict]:
|
|
errors: list[str] = []
|
|
if not isinstance(scenarios, dict) or not scenarios:
|
|
return [f"[{product}] 'scenarios' kosong atau bukan dictionary."], {}
|
|
|
|
missing_scenarios = EXPECTED_SCENARIOS - scenarios.keys()
|
|
if missing_scenarios:
|
|
errors.append(f"[{product}] skenario belum lengkap, hilang: {sorted(missing_scenarios)}.")
|
|
|
|
for scenario_name, scenario in scenarios.items():
|
|
if not isinstance(scenario, dict):
|
|
errors.append(f"[{product}][{scenario_name}] harus berupa dictionary.")
|
|
continue
|
|
|
|
missing_keys = REQUIRED_SCENARIO_KEYS - scenario.keys()
|
|
if missing_keys:
|
|
errors.append(f"[{product}][{scenario_name}] key hilang: {sorted(missing_keys)}.")
|
|
continue
|
|
|
|
config = scenario.get("config")
|
|
if not isinstance(config, dict) or (REQUIRED_CONFIG_KEYS - config.keys()):
|
|
missing_config = REQUIRED_CONFIG_KEYS - (config.keys() if isinstance(config, dict) else set())
|
|
errors.append(f"[{product}][{scenario_name}] 'config' tidak lengkap, hilang: {sorted(missing_config)}.")
|
|
|
|
metrics = scenario.get("best_metrics")
|
|
if not isinstance(metrics, dict) or (REQUIRED_METRIC_KEYS - metrics.keys()):
|
|
errors.append(
|
|
f"[{product}][{scenario_name}] 'best_metrics' harus memuat: {sorted(REQUIRED_METRIC_KEYS)}."
|
|
)
|
|
|
|
return errors, scenarios
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preview dampak merge (tanpa mengubah apa pun)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def preview_merge(current: dict[str, Any], new_data: dict[str, Any]) -> dict[str, list[str]]:
|
|
"""
|
|
Ringkasan dampak kalau `new_data` di-merge ke `current`, TANPA benar-benar
|
|
menulis apa pun. Dipakai untuk ditampilkan ke admin sebelum tombol
|
|
konfirmasi "Terapkan pembaruan" ditekan.
|
|
"""
|
|
existing = set(current.keys())
|
|
incoming = set(new_data.keys())
|
|
return {
|
|
"produk_baru": sorted(incoming - existing),
|
|
"produk_diperbarui": sorted(incoming & existing),
|
|
"produk_tidak_berubah": sorted(existing - incoming),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backup otomatis
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def backup_current(path: str = DEFAULT_DATA_PATH, backup_dir: str = BACKUP_DIR) -> str | None:
|
|
"""
|
|
Salin file all_results.pkl yang sedang aktif ke folder backups/, dengan
|
|
nama disisipi timestamp (format sesuai penamaan di rencana folder:
|
|
all_results_backup_YYYYMMDD_HHMMSS.pkl).
|
|
|
|
Kalau file sumber belum ada sama sekali (kasus pertama kali dipakai di
|
|
server baru), tidak ada yang perlu di-backup -- return None, BUKAN error.
|
|
"""
|
|
if not os.path.exists(path):
|
|
return None
|
|
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|
timestamp = datetime.now().strftime("%d%m%Y_%H%M%S")
|
|
backup_path = os.path.join(backup_dir, f"all_results_backup_{timestamp}.pkl")
|
|
|
|
shutil.copy2(path, backup_path)
|
|
return backup_path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Merge + tulis ke disk + refresh cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def apply_merge(
|
|
current: dict[str, Any],
|
|
new_data: dict[str, Any],
|
|
path: str = DEFAULT_DATA_PATH,
|
|
) -> tuple[dict[str, Any], str | None]:
|
|
"""
|
|
Proses lengkap penerapan pembaruan model:
|
|
1. backup file lama (kalau ada)
|
|
2. gabungkan current + new_data -- new_data MENANG untuk produk yang
|
|
namanya sama, produk lain di current tetap dipertahankan apa adanya
|
|
(bukan replace-all -- upload 1 produk tidak akan menghapus 4 produk lain)
|
|
3. tulis hasil gabungan ke disk
|
|
4. panggil reload_all_results() supaya st.cache_resource ikut di-refresh,
|
|
jadi perubahan langsung kelihatan tanpa restart aplikasi
|
|
|
|
Setiap produk yang ikut ter-update diberi stempel 'updated_at', dipakai
|
|
oleh utils.data_loader.get_last_updated_info() di tab "Status Model".
|
|
|
|
Returns
|
|
-------
|
|
(merged_data, backup_path)
|
|
`backup_path` bernilai None kalau memang belum ada file lama untuk
|
|
di-backup (lihat backup_current()).
|
|
"""
|
|
backup_path = backup_current(path)
|
|
|
|
merged = dict(current)
|
|
timestamp = datetime.now().strftime("%d-%m-%Y %H:%M:%S")
|
|
for product, entry in new_data.items():
|
|
entry = dict(entry)
|
|
entry["updated_at"] = timestamp
|
|
merged[product] = entry
|
|
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
pickle.dump(merged, f)
|
|
|
|
"""
|
|
reload_all_results() = ganti resep di buku resep (lemari dapur)
|
|
st.cache_data.clear() = buang semua masakan yang sudah jadi & tersimpan di kulkas
|
|
(supaya kalau ada yang minta masakan itu lagi,
|
|
koki TERPAKSA masak ULANG pakai resep BARU,
|
|
bukan ngasih masakan lama dari kulkas)
|
|
"""
|
|
reload_all_results(path) # model/data-nya sendiri yang diupdate (objek Prophet baru masuk ke all_results, menggantikan yang lama)
|
|
st.cache_data.clear() # menghapus/membuang hasil forecast LAMA
|
|
|
|
return merged, backup_path
|