Hugging Face's logo

Spaces:
KLP2-PCD
/
KLP2
Sleeping

App Files Community
KLP2 / app.py
firgiawann's picture
firgiawann
Update 41 load Best.pt from artifact repo
04cb891 verified
Raw History Blame Contribute Delete
94.2 kB
import io
import os
import shutil
import asyncio
import fnmatch
import time
import csv
import json
import sys
import subprocess
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from typing import Any, Dict, Tuple
import cv2
import gradio as gr
import matplotlib.pyplot as plt
import numpy as np
import seaborn as sns
from huggingface_hub import CommitOperationAdd, CommitOperationDelete, HfApi, hf_hub_download
from ultralytics import YOLO
CM_INIT = {"tp": 0, "fp": 0, "fn": 0, "tn": 0, "person": 0, "not_person": 0}
EVAL_CLASSES = ["Person", "Not Person/Background"]
APP_VERSION = os.getenv("APP_VERSION", "v0.9.0")
APP_MINOR_UPDATE = os.getenv("APP_MINOR_UPDATE", "41")
# Gradio/Spaces can trigger noisy asyncio cleanup traces ("Invalid file descriptor: -1")
# during loop finalization. This keeps logs clean without affecting runtime behavior.
_orig_loop_del = asyncio.BaseEventLoop.__del__
def _safe_loop_del(self):
try:
_orig_loop_del(self)
except ValueError as err:
if "Invalid file descriptor" not in str(err):
raise
asyncio.BaseEventLoop.__del__ = _safe_loop_del
def resolve_model_path(model_choice: str) -> str:
model_env = os.getenv("MODEL_PATH", "").strip()
base_dir = os.path.dirname(__file__)
if model_env:
if os.path.exists(model_env):
return model_env
raise FileNotFoundError(f"MODEL_PATH tidak valid: {model_env}")
hf_repo = os.getenv("HF_MODEL_REPO", "").strip()
hf_repo_type = os.getenv("HF_MODEL_REPO_TYPE", "space").strip() or "space"
if model_choice == "Best.pt":
best_local = os.path.join(base_dir, "best.pt")
if os.path.exists(best_local):
return best_local
hf_file = os.getenv("HF_MODEL_FILE", "best.pt").strip() or "best.pt"
artifact_repo = (
os.getenv("HF_ARTIFACT_REPO", "").strip()
or os.getenv("HF_TRAINING_ARTIFACT_REPO", "").strip()
)
artifact_repo_type = os.getenv("HF_ARTIFACT_REPO_TYPE", "dataset").strip() or "dataset"
if artifact_repo:
try:
return hf_hub_download(
repo_id=artifact_repo,
filename=hf_file,
repo_type=artifact_repo_type,
)
except Exception:
pass
if hf_repo:
try:
return hf_hub_download(repo_id=hf_repo, filename=hf_file, repo_type=hf_repo_type)
except Exception as err:
raise FileNotFoundError(
"Model Best.pt belum tersedia. Jalankan training terlebih dahulu agar "
"best.pt dibuat dan tersimpan ke Space."
) from err
raise FileNotFoundError(
"Model Best.pt belum tersedia. Jalankan training terlebih dahulu agar "
"best.pt dibuat dan tersimpan ke Space."
)
yolo_local = os.path.join(base_dir, "yolov8n.pt")
if os.path.exists(yolo_local):
return yolo_local
hf_file = os.getenv("HF_YOLO_MODEL_FILE", "yolov8n.pt").strip() or "yolov8n.pt"
if hf_repo:
try:
return hf_hub_download(repo_id=hf_repo, filename=hf_file, repo_type=hf_repo_type)
except Exception as err:
raise FileNotFoundError(
"Model YOLOv8n tidak ditemukan. Simpan `yolov8n.pt` di root project "
"atau set HF_YOLO_MODEL_FILE ke file model default di Hub."
) from err
raise FileNotFoundError(
"Model YOLOv8n tidak ditemukan. Simpan `yolov8n.pt` di root project "
"atau set MODEL_PATH / HF_MODEL_REPO."
)
@lru_cache(maxsize=2)
def load_model(model_choice: str) -> YOLO:
return YOLO(resolve_model_path(model_choice))
def _resolve_person_class_id(result: Any) -> int:
names = getattr(result, "names", {})
if isinstance(names, dict):
for idx, name in names.items():
if str(name).lower() == "person":
return int(idx)
if isinstance(names, list):
for idx, name in enumerate(names):
if str(name).lower() == "person":
return idx
return 0
def _resolve_person_class_id_from_names(names: Any) -> int:
if isinstance(names, dict):
for idx, name in names.items():
if str(name).lower() == "person":
return int(idx)
if isinstance(names, list):
for idx, name in enumerate(names):
if str(name).lower() == "person":
return idx
return 0
def _annotate_person_instances(
image_rgb: np.ndarray,
result: Any,
person_id: int,
annotation_style: str,
) -> Tuple[np.ndarray, int]:
annotated = image_rgb.copy()
if result.boxes is None:
return annotated, 0
classes = result.boxes.cls.cpu().numpy()
boxes = result.boxes.xyxy.cpu().numpy()
person_idx = np.where(classes == person_id)[0]
if len(person_idx) == 0:
return annotated, 0
person_boxes = []
for i in person_idx:
x1, y1, x2, y2 = boxes[i].tolist()
person_boxes.append((int(x1), int(y1), int(x2), int(y2)))
# Stabilkan urutan nomor: atas-ke-bawah lalu kiri-ke-kanan.
person_boxes.sort(key=lambda b: (b[1], b[0]))
for idx, (x1, y1, x2, y2) in enumerate(person_boxes, start=1):
if annotation_style == "Border + Label":
cv2.rectangle(annotated, (x1, y1), (x2, y2), (34, 197, 94), 2)
label = f"person {idx}"
(tw, th), baseline = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
if annotation_style == "Label saja":
label_x = max(0, min(x1, annotated.shape[1] - tw - 8))
label_y = max(th + 4, min(y1 + th + 4, annotated.shape[0] - 6))
text_y1 = max(0, label_y - th - baseline - 6)
text_y2 = min(annotated.shape[0], label_y + 2)
text_x1 = label_x
text_x2 = min(annotated.shape[1], label_x + tw + 10)
text_org = (label_x + 4, label_y - 4)
else:
text_y1 = max(0, y1 - th - baseline - 6)
text_y2 = y1
text_x1 = x1
text_x2 = min(annotated.shape[1], x1 + tw + 10)
text_org = (x1 + 4, max(14, y1 - 6))
cv2.rectangle(annotated, (text_x1, text_y1), (text_x2, text_y2), (34, 197, 94), -1)
cv2.putText(
annotated,
label,
text_org,
cv2.FONT_HERSHEY_SIMPLEX,
0.6,
(0, 0, 0),
2,
cv2.LINE_AA,
)
return annotated, len(person_boxes)
def _annotate_all_instances(
image_rgb: np.ndarray,
result: Any,
annotation_style: str,
) -> Tuple[np.ndarray, int]:
annotated = image_rgb.copy()
if result.boxes is None:
return annotated, 0
classes = result.boxes.cls.cpu().numpy()
boxes = result.boxes.xyxy.cpu().numpy()
names = getattr(result, "names", {})
items = []
person_id = _resolve_person_class_id(result)
person_counter = 0
other_counter = 0
for box, cls_raw in zip(boxes.tolist(), classes.tolist()):
x1, y1, x2, y2 = box
cls_id = int(cls_raw)
if isinstance(names, dict):
cls_name = str(names.get(cls_id, f"class_{cls_id}"))
elif isinstance(names, list) and 0 <= cls_id < len(names):
cls_name = str(names[cls_id])
else:
cls_name = f"class_{cls_id}"
items.append((int(x1), int(y1), int(x2), int(y2), cls_id, cls_name))
items.sort(key=lambda item: (item[1], item[0]))
for x1, y1, x2, y2, cls_id, cls_name in items:
is_person = cls_id == person_id or cls_name.lower() == "person"
if is_person:
person_counter += 1
label = f"person {person_counter}"
color = (34, 197, 94)
text_color = (0, 0, 0)
else:
other_counter += 1
label = f"{cls_name} {other_counter}"
color = (59, 130, 246)
text_color = (255, 255, 255)
if annotation_style == "Border + Label":
cv2.rectangle(annotated, (x1, y1), (x2, y2), color, 2)
(tw, th), baseline = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
if annotation_style == "Label saja":
label_x = max(0, min(x1, annotated.shape[1] - tw - 8))
label_y = max(th + 4, min(y1 + th + 4, annotated.shape[0] - 6))
text_y1 = max(0, label_y - th - baseline - 6)
text_y2 = min(annotated.shape[0], label_y + 2)
text_x1 = label_x
text_x2 = min(annotated.shape[1], label_x + tw + 10)
text_org = (label_x + 4, label_y - 4)
else:
text_y1 = max(0, y1 - th - baseline - 6)
text_y2 = y1
text_x1 = x1
text_x2 = min(annotated.shape[1], x1 + tw + 10)
text_org = (x1 + 4, max(14, y1 - 6))
cv2.rectangle(annotated, (text_x1, text_y1), (text_x2, text_y2), color, -1)
cv2.putText(
annotated,
label,
text_org,
cv2.FONT_HERSHEY_SIMPLEX,
0.6,
text_color,
2,
cv2.LINE_AA,
)
return annotated, len(items)
def get_detection_scope(model_choice: str) -> str:
return "Semua Kelas" if model_choice == "YOLOv8n" else "Person saja"
def _count_detected_objects(result: Any, person_id: int) -> Tuple[int, int]:
if result.boxes is None:
return 0, 0
classes = result.boxes.cls.cpu().numpy()
person_count = int(np.sum(classes == person_id))
total_count = int(len(classes))
non_person_count = max(0, total_count - person_count)
return person_count, non_person_count
def infer_image(
image: np.ndarray,
conf_threshold: float,
model_choice: str,
annotation_style: str,
) -> Tuple[np.ndarray, int, int]:
if image is None:
raise ValueError("Silakan unggah gambar atau gunakan kamera.")
model = load_model(model_choice)
person_id = _resolve_person_class_id_from_names(getattr(model, "names", {}))
detection_scope = get_detection_scope(model_choice)
bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
predict_kwargs = {
"source": bgr,
"conf": conf_threshold,
"verbose": False,
}
if detection_scope == "Person saja":
predict_kwargs["classes"] = [person_id]
result = model.predict(**predict_kwargs)[0]
person_count, non_person_count = _count_detected_objects(result, person_id)
if detection_scope == "Semua Kelas":
annotated, _ = _annotate_all_instances(image, result, annotation_style)
else:
person_id = _resolve_person_class_id(result)
annotated, _ = _annotate_person_instances(image, result, person_id, annotation_style)
return annotated, person_count, non_person_count
def classify_presence(detected_count: int) -> str:
return "Ada Mahasiswa" if detected_count > 0 else "Tidak Ada Mahasiswa"
def classify_person_vs_background(detected_count: int) -> str:
return "Person" if detected_count > 0 else "Not Person/Background"
def classify_scene_label(person_count: int, non_person_count: int) -> str:
if person_count > 0:
return "Person"
if non_person_count > 0:
return "Not Person/Background"
return "Not Person/Background"
def _evaluation_paths(base_dir: Path) -> Dict[str, Path]:
eval_dir = base_dir / "evaluation_outputs"
eval_dir.mkdir(parents=True, exist_ok=True)
return {
"dir": eval_dir,
"csv": eval_dir / "detection_log.csv",
}
def sync_detection_artifacts_to_hf(image_path: Path, csv_path: Path) -> str:
hf_repo = os.getenv("HF_MODEL_REPO", "").strip()
hf_repo_type = os.getenv("HF_MODEL_REPO_TYPE", "space").strip() or "space"
hf_token = os.getenv("HF_TOKEN", "").strip() or None
repo_dir = os.getenv("HF_DETECTION_DIR", "space_detection_archive").strip().strip("/")
if not hf_repo:
return "HF save dilewati: set HF_MODEL_REPO untuk auto-save hasil deteksi ke Space."
image_repo_path = f"{repo_dir}/images/{image_path.name}"
csv_repo_path = f"{repo_dir}/{csv_path.name}"
try:
api = HfApi(token=hf_token)
api.upload_file(
path_or_fileobj=str(image_path),
path_in_repo=image_repo_path,
repo_id=hf_repo,
repo_type=hf_repo_type,
commit_message=f"Auto-save detection image ({image_path.name})",
)
api.upload_file(
path_or_fileobj=str(csv_path),
path_in_repo=csv_repo_path,
repo_id=hf_repo,
repo_type=hf_repo_type,
commit_message=f"Auto-update detection log ({csv_path.name})",
)
return (
f"HF save sukses: {hf_repo} ({hf_repo_type}) -> "
f"{repo_dir}/images dan {repo_dir}/{csv_path.name}"
)
except Exception as err:
return f"HF save gagal: {err}"
def save_detection_artifacts(
annotated_image: np.ndarray,
model_choice: str,
inference_mode: str,
conf_threshold: float,
predicted_label: str,
detected_count: int,
non_person_count: int,
) -> Tuple[str | None, str | None, str]:
base_dir = Path(__file__).resolve().parent
paths = _evaluation_paths(base_dir)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
image_path = paths["dir"] / f"detection_{ts}.png"
bgr = cv2.cvtColor(annotated_image, cv2.COLOR_RGB2BGR)
cv2.imwrite(str(image_path), bgr)
csv_path = paths["csv"]
write_header = not csv_path.exists()
with csv_path.open("a", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
if write_header:
writer.writerow(
[
"timestamp",
"model_choice",
"inference_mode",
"confidence_threshold",
"predicted_label",
"person_count",
"non_person_count",
"saved_image",
]
)
writer.writerow(
[
datetime.now().isoformat(),
model_choice,
inference_mode,
f"{conf_threshold:.2f}",
predicted_label,
detected_count,
non_person_count,
str(image_path),
]
)
hf_sync_msg = sync_detection_artifacts_to_hf(image_path, csv_path)
return str(image_path), str(csv_path), hf_sync_msg
def draw_detection_chart(person_count: int, non_person_count: int) -> np.ndarray:
fig, ax = plt.subplots(figsize=(6.2, 3.6))
labels = ["Person", "Not Person"]
values = [person_count, non_person_count]
bars = ax.bar(labels, values, color=["#2563eb", "#64748b"])
ax.set_title("Grafik Hasil Pengecekan")
ax.set_ylabel("Jumlah Objek")
for b in bars:
ax.text(
b.get_x() + b.get_width() / 2,
b.get_height() + 0.05,
str(int(b.get_height())),
ha="center",
va="bottom",
)
plt.tight_layout()
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=140)
plt.close(fig)
buf.seek(0)
data = np.frombuffer(buf.read(), dtype=np.uint8)
img_bgr = cv2.imdecode(data, cv2.IMREAD_COLOR)
return cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
def draw_compare_chart(
yolo_person: int,
yolo_non_person: int,
best_person: int,
best_non_person: int,
) -> np.ndarray:
fig, ax = plt.subplots(figsize=(7.4, 3.8))
x = np.arange(2)
width = 0.36
person_values = [yolo_person, best_person]
non_person_values = [yolo_non_person, best_non_person]
bars_person = ax.bar(x - width / 2, person_values, width, label="Person", color="#2563eb")
bars_non_person = ax.bar(x + width / 2, non_person_values, width, label="Not Person", color="#64748b")
ax.set_title("Perbandingan Count Objek per Model")
ax.set_ylabel("Jumlah Objek")
ax.set_xticks(x)
ax.set_xticklabels(["YOLOv8n", "Best.pt"])
ax.legend(loc="best")
for bars in (bars_person, bars_non_person):
for b in bars:
ax.text(
b.get_x() + b.get_width() / 2,
b.get_height() + 0.05,
str(int(b.get_height())),
ha="center",
va="bottom",
)
plt.tight_layout()
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=140)
plt.close(fig)
buf.seek(0)
data = np.frombuffer(buf.read(), dtype=np.uint8)
img_bgr = cv2.imdecode(data, cv2.IMREAD_COLOR)
return cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
def _merge_annotated_images(left: np.ndarray, right: np.ndarray) -> np.ndarray:
h = max(left.shape[0], right.shape[0])
w1 = int(left.shape[1] * (h / left.shape[0]))
w2 = int(right.shape[1] * (h / right.shape[0]))
left_r = cv2.resize(left, (w1, h), interpolation=cv2.INTER_AREA)
right_r = cv2.resize(right, (w2, h), interpolation=cv2.INTER_AREA)
combined = np.concatenate([left_r, right_r], axis=1)
cv2.putText(combined, "YOLOv8n", (12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)
cv2.putText(combined, "Best.pt", (w1 + 12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 0), 2)
return combined
def build_eval_payload(
predicted_label: str,
actual_label: str,
predicted_person_count: int,
predicted_non_person_count: int,
actual_person_count: int,
actual_non_person_count: int,
) -> Dict[str, Any]:
return {
"predicted_label": predicted_label,
"actual_label": actual_label,
"predicted_person_count": int(predicted_person_count),
"predicted_non_person_count": int(predicted_non_person_count),
"actual_person_count": int(actual_person_count),
"actual_non_person_count": int(actual_non_person_count),
}
def object_confusion_delta(eval_payload: Dict[str, Any]) -> Dict[str, int]:
actual_person = int(eval_payload.get("actual_person_count", 0) or 0)
actual_non_person = int(eval_payload.get("actual_non_person_count", 0) or 0)
predicted_person = int(eval_payload.get("predicted_person_count", 0) or 0)
tp = min(actual_person, predicted_person)
fn = max(actual_person - predicted_person, 0)
fp = max(predicted_person - actual_person, 0)
tn = max(actual_non_person - fp, 0)
return {
"tp": tp,
"fp": fp,
"fn": fn,
"tn": tn,
"person": actual_person,
"not_person": actual_non_person,
}
def run_check_and_classify(
image: np.ndarray,
conf_threshold: float,
model_choice: str,
inference_mode: str,
annotation_style: str,
) -> Tuple[
str | np.ndarray | None,
str | np.ndarray | None,
str,
np.ndarray,
str,
str,
Dict[str, Any],
str,
str | None,
str | None,
]:
try:
if inference_mode == "Bandingkan (YOLOv8n vs Best.pt)":
scope_before = get_detection_scope("YOLOv8n")
scope_after = get_detection_scope("Best.pt")
ann_before, det_before, non_person_before = infer_image(
image, conf_threshold, "YOLOv8n", annotation_style
)
ann_after, det_after, non_person_after = infer_image(
image, conf_threshold, "Best.pt", annotation_style
)
pred_before = classify_person_vs_background(det_before)
pred_after = classify_person_vs_background(det_after)
chart = draw_compare_chart(det_before, non_person_before, det_after, non_person_after)
yolo_image_path, saved_csv_path, hf_sync_msg_yolo = save_detection_artifacts(
ann_before,
model_choice="YOLOv8n",
inference_mode=inference_mode,
conf_threshold=conf_threshold,
predicted_label=pred_before,
detected_count=det_before,
non_person_count=non_person_before,
)
best_image_path, saved_csv_path, hf_sync_msg_best = save_detection_artifacts(
ann_after,
model_choice="Best.pt",
inference_mode=inference_mode,
conf_threshold=conf_threshold,
predicted_label=pred_after,
detected_count=det_after,
non_person_count=non_person_after,
)
actual_label = classify_scene_label(det_before, non_person_before)
eval_payload = build_eval_payload(
predicted_label=pred_after,
actual_label=actual_label,
predicted_person_count=det_after,
predicted_non_person_count=non_person_after,
actual_person_count=det_before,
actual_non_person_count=non_person_before,
)
summary = (
"### Ringkasan Perbandingan Model\n"
f"- Confidence threshold: **{conf_threshold:.2f}**\n"
f"- **YOLOv8n**: scope={scope_before}, person={det_before}, non-person={non_person_before}, prediksi kelas={pred_before}\n"
f"- **Best.pt**: scope={scope_after}, person={det_after}, non-person={non_person_after}, prediksi kelas={pred_after}\n"
f"- Label aktual/proxy evaluasi: **{actual_label}** (berdasarkan scene reference dari YOLOv8n)\n"
f"- Arsip evaluasi lokal: **{saved_csv_path or '-'}**\n"
f"- Auto-save YOLOv8n ke Space: **{hf_sync_msg_yolo}**\n"
f"- Auto-save Best.pt ke Space: **{hf_sync_msg_best}**"
)
pred_md = f"Label prediksi evaluasi (Best.pt): **{pred_after}**"
return (
yolo_image_path or ann_before,
best_image_path or ann_after,
summary,
chart,
pred_after,
actual_label,
eval_payload,
pred_md,
best_image_path,
saved_csv_path,
)
detection_scope = get_detection_scope(model_choice)
annotated, detected, non_person_count = infer_image(
image, conf_threshold, model_choice, annotation_style
)
status_presence = classify_presence(detected)
predicted_label = classify_person_vs_background(detected)
actual_label = classify_scene_label(detected, non_person_count)
actual_person_count = detected
actual_non_person_count = non_person_count
yolo_panel_output: str | np.ndarray | None = None
best_panel_output: str | np.ndarray | None = None
if model_choice == "Best.pt":
_, ref_person_count, ref_non_person_count = infer_image(
image, conf_threshold, "YOLOv8n", annotation_style
)
actual_label = classify_scene_label(ref_person_count, ref_non_person_count)
actual_person_count = ref_person_count
actual_non_person_count = ref_non_person_count
saved_image_path, saved_csv_path, hf_sync_msg = save_detection_artifacts(
annotated,
model_choice=model_choice,
inference_mode=inference_mode,
conf_threshold=conf_threshold,
predicted_label=predicted_label,
detected_count=detected,
non_person_count=non_person_count,
)
if model_choice == "YOLOv8n":
yolo_panel_output = saved_image_path or annotated
else:
best_panel_output = saved_image_path or annotated
summary = (
"### Ringkasan Hasil Deteksi dan Klasifikasi\n"
f"- Jumlah objek `person` terdeteksi: **{detected}**\n"
f"- Jumlah objek non-person terdeteksi: **{non_person_count}**\n"
f"- Klasifikasi keberadaan: **{status_presence}**\n"
f"- Kelas prediksi evaluasi: **{predicted_label}**\n"
f"- Label aktual/proxy evaluasi: **{actual_label}**\n"
f"- Confidence threshold: **{conf_threshold:.2f}**\n"
f"- Scope deteksi: **{detection_scope}**\n"
f"- Model digunakan: **{model_choice}**\n"
f"- Gaya anotasi: **{annotation_style}**\n"
"- Label objek mengikuti kelas aktif pada model\n"
f"- Arsip evaluasi lokal: **{saved_csv_path or '-'}**\n"
f"- Auto-save ke Space: **{hf_sync_msg}**"
)
chart = draw_detection_chart(detected, non_person_count)
eval_payload = build_eval_payload(
predicted_label=predicted_label,
actual_label=actual_label,
predicted_person_count=detected,
predicted_non_person_count=non_person_count,
actual_person_count=actual_person_count,
actual_non_person_count=actual_non_person_count,
)
pred_md = f"Label prediksi evaluasi: **{predicted_label}**"
return (
yolo_panel_output,
best_panel_output,
summary,
chart,
predicted_label,
actual_label,
eval_payload,
pred_md,
saved_image_path,
saved_csv_path,
)
except Exception as err:
summary = (
"### Ringkasan Hasil Pengolahan Citra\n"
f"Proses gagal: {err}\n\n"
"Tips: pastikan model tersedia (`yolov8n.pt` atau MODEL_PATH)."
)
blank = np.full((320, 500, 3), 245, dtype=np.uint8)
cv2.putText(
blank,
"Grafik belum tersedia",
(120, 165),
cv2.FONT_HERSHEY_SIMPLEX,
0.8,
(80, 80, 80),
2,
)
return None, None, summary, blank, "", "", {}, "Label prediksi evaluasi: **-**", None, None
def toggle_model_selector(mode: str):
is_single = mode == "Single Model"
return gr.update(visible=is_single)
def _safe_div(num: float, den: float) -> float:
return 0.0 if den == 0 else num / den
def _metric_cell(value: float, den: float) -> str:
if den == 0:
return '<span class="metric-muted">Tidak terdefinisi</span>'
return f"<strong>{value:.4f}</strong>"
def draw_confusion_matrix_from_state(cm_state: Dict[str, int]) -> np.ndarray:
tp = int(cm_state.get("tp", 0))
fp = int(cm_state.get("fp", 0))
fn = int(cm_state.get("fn", 0))
tn = int(cm_state.get("tn", 0))
cm = np.array([[tp, fn], [fp, tn]], dtype=int)
person_total = int(cm_state.get("person", tp + fn))
not_person_total = int(cm_state.get("not_person", fp + tn))
sns.set_style("white")
fig, ax = plt.subplots(figsize=(7.4, 5.8))
sns.heatmap(
cm,
annot=True,
fmt="d",
cmap="Blues",
cbar=True,
xticklabels=["Prediksi Person", "Prediksi Not Person"],
yticklabels=["Aktual Person", "Aktual Not Person"],
linewidths=0.5,
linecolor="white",
annot_kws={"size": 14, "weight": "bold"},
ax=ax,
)
ax.set_title("Confusion Matrix Count Objek")
ax.set_xlabel("Prediksi")
ax.set_ylabel("Aktual")
ax.text(
0.5,
-0.22,
f"Total aktual: Person={person_total} | Not Person/Background={not_person_total}",
transform=ax.transAxes,
ha="center",
va="top",
fontsize=10,
)
plt.xticks(rotation=12, ha="right")
plt.yticks(rotation=0)
plt.tight_layout(rect=[0, 0.07, 1, 1])
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150)
plt.close(fig)
buf.seek(0)
data = np.frombuffer(buf.read(), dtype=np.uint8)
img_bgr = cv2.imdecode(data, cv2.IMREAD_COLOR)
return cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
def render_metric_markdown(cm_state: Dict[str, int]) -> str:
tp = int(cm_state.get("tp", 0))
fp = int(cm_state.get("fp", 0))
fn = int(cm_state.get("fn", 0))
tn = int(cm_state.get("tn", 0))
acc = _safe_div(tp + tn, tp + fp + fn + tn)
prec = _safe_div(tp, tp + fp)
rec = _safe_div(tp, tp + fn)
f1 = _safe_div(2 * prec * rec, prec + rec)
person_total = int(cm_state.get("person", tp + fn))
not_person_total = int(cm_state.get("not_person", fp + tn))
total = tp + fp + fn + tn
precision_den = tp + fp
recall_den = tp + fn
f1_den = prec + rec
if total == 0:
return (
"""
<section class="metric-panel">
<div class="metric-header">
<div>
<h3>Rumus Evaluasi Berbasis Count Objek</h3>
<p>Belum ada data inferensi yang masuk. Metrik akan dihitung otomatis setelah deteksi dijalankan.</p>
</div>
<span class="metric-badge">Status: menunggu data</span>
</div>
<div class="metric-grid">
<div class="metric-card"><span>TP</span><strong>0</strong><small>Person benar terdeteksi</small></div>
<div class="metric-card"><span>FP</span><strong>0</strong><small>Background terdeteksi sebagai person</small></div>
<div class="metric-card"><span>FN</span><strong>0</strong><small>Person tidak terdeteksi</small></div>
<div class="metric-card"><span>TN</span><strong>0</strong><small>Background benar bukan person</small></div>
</div>
<table class="formula-table">
<thead><tr><th>Metrik</th><th>Rumus</th><th>Keterangan</th></tr></thead>
<tbody>
<tr><td>Accuracy</td><td>(TP + TN) / (TP + FP + FN + TN)</td><td>Proporsi prediksi benar terhadap seluruh objek evaluasi.</td></tr>
<tr><td>Precision</td><td>TP / (TP + FP)</td><td>Ketepatan deteksi person dari seluruh prediksi person.</td></tr>
<tr><td>Recall</td><td>TP / (TP + FN)</td><td>Kemampuan model menemukan seluruh objek person aktual.</td></tr>
<tr><td>F1-Score</td><td>2 x (Precision x Recall) / (Precision + Recall)</td><td>Rata-rata harmonik precision dan recall.</td></tr>
</tbody>
</table>
</section>
"""
)
return (
f"""
<section class="metric-panel">
<div class="metric-header">
<div>
<h3>Rumus Evaluasi Berbasis Count Objek</h3>
<p>Kelas positif adalah <strong>Person</strong>; kelas negatif adalah <strong>Not Person/Background</strong>.</p>
</div>
<span class="metric-badge">Total objek evaluasi: {total}</span>
</div>
<div class="metric-grid">
<div class="metric-card"><span>TP</span><strong>{tp}</strong><small>Person benar terdeteksi</small></div>
<div class="metric-card"><span>FP</span><strong>{fp}</strong><small>Background terdeteksi sebagai person</small></div>
<div class="metric-card"><span>FN</span><strong>{fn}</strong><small>Person tidak terdeteksi</small></div>
<div class="metric-card"><span>TN</span><strong>{tn}</strong><small>Background benar bukan person</small></div>
</div>
<table class="formula-table">
<thead>
<tr><th>Komponen</th><th>Nilai</th><th>Keterangan</th></tr>
</thead>
<tbody>
<tr><td>Person aktual</td><td>{person_total}</td><td>Jumlah objek person sebagai acuan evaluasi.</td></tr>
<tr><td>Not Person/Background aktual</td><td>{not_person_total}</td><td>Jumlah objek selain person sebagai acuan negatif.</td></tr>
</tbody>
</table>
<table class="formula-table">
<thead>
<tr><th>Metrik</th><th>Rumus</th><th>Substitusi</th><th>Hasil</th></tr>
</thead>
<tbody>
<tr><td>Accuracy</td><td>(TP + TN) / Total</td><td>({tp} + {tn}) / {total}</td><td>{_metric_cell(acc, total)}</td></tr>
<tr><td>Precision</td><td>TP / (TP + FP)</td><td>{tp} / ({tp} + {fp})</td><td>{_metric_cell(prec, precision_den)}</td></tr>
<tr><td>Recall</td><td>TP / (TP + FN)</td><td>{tp} / ({tp} + {fn})</td><td>{_metric_cell(rec, recall_den)}</td></tr>
<tr><td>F1-Score</td><td>2 x (Precision x Recall) / (Precision + Recall)</td><td>2 x ({prec:.4f} x {rec:.4f}) / ({prec:.4f} + {rec:.4f})</td><td>{_metric_cell(f1, f1_den)}</td></tr>
</tbody>
</table>
</section>
"""
)
def update_evaluation(cm_state: Dict[str, int], eval_payload: Dict[str, Any]) -> Tuple[Dict[str, int], np.ndarray, str, str]:
state = dict(cm_state or CM_INIT)
if not eval_payload:
return (
state,
draw_confusion_matrix_from_state(state),
render_metric_markdown(state),
"Jalankan deteksi terlebih dahulu agar label prediksi tersedia.",
)
delta = object_confusion_delta(eval_payload)
for key in ("tp", "fp", "fn", "tn", "person", "not_person"):
state[key] = int(state.get(key, 0)) + int(delta.get(key, 0))
return (
state,
draw_confusion_matrix_from_state(state),
render_metric_markdown(state),
(
"Confusion matrix count objek diperbarui: "
f"Person={delta['person']}, Not Person/Background={delta['not_person']}."
),
)
def update_evaluation_auto(
cm_state: Dict[str, int], eval_payload: Dict[str, Any]
) -> Tuple[Dict[str, int], np.ndarray, str, str]:
if not eval_payload:
state = dict(cm_state or CM_INIT)
return (
state,
draw_confusion_matrix_from_state(state),
render_metric_markdown(state),
"Jalankan deteksi terlebih dahulu agar evaluasi otomatis dapat diperbarui.",
)
return update_evaluation(cm_state, eval_payload)
def reset_evaluation() -> Tuple[Dict[str, int], np.ndarray, str, str]:
state = dict(CM_INIT)
return (
state,
draw_confusion_matrix_from_state(state),
render_metric_markdown(state),
"Confusion matrix direset.",
)
def _iter_images(folder: Path) -> list[Path]:
exts = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
if not folder.exists():
return []
return sorted([p for p in folder.rglob("*") if p.is_file() and p.suffix.lower() in exts])
def _label_candidates(img_path: Path) -> list[Path]:
stem = img_path.stem
return [
img_path.with_suffix(".txt"),
img_path.parent.parent / "labels" / img_path.parent.name / f"{stem}.txt",
img_path.parent / "labels" / f"{stem}.txt",
]
def _write_pseudo_label(img_rgb: np.ndarray, label_path: Path, conf_threshold: float) -> int:
model = load_model("YOLOv8n")
h, w = img_rgb.shape[:2]
bgr = cv2.cvtColor(img_rgb, cv2.COLOR_RGB2BGR)
result = model.predict(source=bgr, conf=conf_threshold, verbose=False)[0]
person_id = _resolve_person_class_id(result)
boxes_xyxy = result.boxes.xyxy.cpu().numpy() if result.boxes is not None else np.empty((0, 4))
classes = result.boxes.cls.cpu().numpy() if result.boxes is not None else np.array([])
lines = []
for box, cls_id in zip(boxes_xyxy, classes):
if int(cls_id) != person_id:
continue
x1, y1, x2, y2 = box.tolist()
xc = ((x1 + x2) / 2) / w
yc = ((y1 + y2) / 2) / h
bw = (x2 - x1) / w
bh = (y2 - y1) / h
lines.append(f"0 {xc:.6f} {yc:.6f} {bw:.6f} {bh:.6f}")
label_path.parent.mkdir(parents=True, exist_ok=True)
label_path.write_text("\n".join(lines), encoding="utf-8")
return len(lines)
def _prepare_training_dataset(
dataset_root: Path,
use_pseudo_if_missing: bool,
pseudo_conf: float,
log_file: Path | None = None,
) -> Tuple[Path, str]:
train_src = dataset_root / "train"
test_src = dataset_root / "test"
if not train_src.exists() or not test_src.exists():
raise FileNotFoundError("Folder dataset wajib berisi `dataset/train` dan `dataset/test`.")
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
prepared_root = dataset_root.parent / "prepared_dataset" / ts
img_train = prepared_root / "images" / "train"
img_val = prepared_root / "images" / "val"
lbl_train = prepared_root / "labels" / "train"
lbl_val = prepared_root / "labels" / "val"
for p in (img_train, img_val, lbl_train, lbl_val):
p.mkdir(parents=True, exist_ok=True)
def _log(msg: str):
if log_file is None:
return
with log_file.open("a", encoding="utf-8") as lf:
lf.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
def process_split(src: Path, out_img: Path, out_lbl: Path, split_name: str) -> Tuple[int, int]:
images = _iter_images(src)
if not images:
raise ValueError(f"Tidak ada gambar di {src}")
_log(f"Prepare split `{split_name}`: total gambar={len(images)}")
copied = 0
labeled = 0
for img in images:
target_img = out_img / img.name
if target_img.exists():
target_img = out_img / f"{img.stem}_{copied}{img.suffix}"
shutil.copy2(img, target_img)
label_target = out_lbl / f"{target_img.stem}.txt"
found_label = None
for cand in _label_candidates(img):
if cand.exists():
found_label = cand
break
if found_label is not None:
shutil.copy2(found_label, label_target)
labeled += 1
elif use_pseudo_if_missing:
rgb = cv2.cvtColor(cv2.imread(str(target_img)), cv2.COLOR_BGR2RGB)
labeled += 1 if _write_pseudo_label(rgb, label_target, pseudo_conf) >= 0 else 0
copied += 1
if copied % 10 == 0 or copied == len(images):
_log(f"Prepare `{split_name}` progress: {copied}/{len(images)} (labeled={labeled})")
return copied, labeled
c_train, l_train = process_split(train_src, img_train, lbl_train, "train")
c_val, l_val = process_split(test_src, img_val, lbl_val, "val")
if l_train == 0 or l_val == 0:
raise ValueError(
"Label tidak ditemukan. Aktifkan pseudo-label atau sediakan file .txt YOLO untuk train/test."
)
yaml_path = prepared_root / "data.yaml"
yaml_path.write_text(
f"path: {prepared_root.as_posix()}\n"
"train: images/train\n"
"val: images/val\n"
"names:\n"
" 0: person\n",
encoding="utf-8",
)
info = (
f"Dataset siap: train={c_train} (labeled={l_train}), "
f"val={c_val} (labeled={l_val}), yaml={yaml_path}"
)
return yaml_path, info
def sync_best_to_hf(best_model_path: Path) -> str:
hf_repo = os.getenv("HF_MODEL_REPO", "").strip()
hf_file = os.getenv("HF_MODEL_FILE", "best.pt").strip() or "best.pt"
hf_repo_type = os.getenv("HF_MODEL_REPO_TYPE", "space").strip() or "space"
hf_token = os.getenv("HF_TOKEN", "").strip() or None
if not hf_repo:
return "HF sync dilewati: set HF_MODEL_REPO untuk upload otomatis."
if _skip_runtime_commit_to_current_space(hf_repo, hf_repo_type):
return "HF sync dilewati: commit runtime ke Space yang sama dinonaktifkan agar Space tidak rebuild sendiri."
try:
api = HfApi(token=hf_token)
api.upload_file(
path_or_fileobj=str(best_model_path),
path_in_repo=hf_file,
repo_id=hf_repo,
repo_type=hf_repo_type,
commit_message=f"Auto-update {hf_file} ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})",
)
return f"HF sync sukses: {hf_repo} ({hf_repo_type}) -> {hf_file}"
except Exception as err:
return f"HF sync gagal: {err}"
def _make_info_image(text: str, width: int = 900, height: int = 420) -> np.ndarray:
canvas = np.full((height, width, 3), 245, dtype=np.uint8)
y = 52
for line in text.split("\n")[:10]:
cv2.putText(
canvas,
line[:95],
(28, y),
cv2.FONT_HERSHEY_SIMPLEX,
0.8,
(80, 80, 80),
2,
cv2.LINE_AA,
)
y += 36
return canvas
def _read_results_rows(results_csv: Path | None) -> list[dict[str, str]]:
if results_csv is None or (not results_csv.exists()) or (not results_csv.is_file()):
return []
with results_csv.open("r", encoding="utf-8", newline="") as f:
reader = csv.DictReader(f)
return [row for row in reader]
def _to_float(value: str) -> float | None:
try:
return float(value)
except Exception:
return None
def _extract_series(rows: list[dict[str, str]], key: str) -> list[float]:
out: list[float] = []
for row in rows:
value = _to_float(row.get(key, ""))
out.append(float(value) if value is not None else float("nan"))
return out
def plot_training_curves(results_csv: Path | None) -> np.ndarray:
rows = _read_results_rows(results_csv)
if not rows:
return _make_info_image("Grafik training belum tersedia.\nMenunggu proses epoch berjalan...")
epochs: list[int] = []
if "epoch" in rows[0]:
for i, row in enumerate(rows):
v = _to_float(row.get("epoch", ""))
epochs.append((int(v) + 1) if v is not None else (i + 1))
else:
epochs = list(range(1, len(rows) + 1))
box_loss = _extract_series(rows, "train/box_loss")
cls_loss = _extract_series(rows, "train/cls_loss")
dfl_loss = _extract_series(rows, "train/dfl_loss")
precision = _extract_series(rows, "metrics/precision(B)")
recall = _extract_series(rows, "metrics/recall(B)")
map50 = _extract_series(rows, "metrics/mAP50(B)")
map5095 = _extract_series(rows, "metrics/mAP50-95(B)")
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12.0, 4.2))
if np.isfinite(np.array(box_loss)).any():
ax1.plot(epochs, box_loss, label="box_loss", linewidth=2)
if np.isfinite(np.array(cls_loss)).any():
ax1.plot(epochs, cls_loss, label="cls_loss", linewidth=2)
if np.isfinite(np.array(dfl_loss)).any():
ax1.plot(epochs, dfl_loss, label="dfl_loss", linewidth=2)
ax1.set_title("Kurva Loss Training")
ax1.set_xlabel("Epoch")
ax1.set_ylabel("Loss")
ax1.grid(alpha=0.25)
ax1.legend(loc="best")
if np.isfinite(np.array(precision)).any():
ax2.plot(epochs, precision, label="Precision", linewidth=2)
if np.isfinite(np.array(recall)).any():
ax2.plot(epochs, recall, label="Recall", linewidth=2)
if np.isfinite(np.array(map50)).any():
ax2.plot(epochs, map50, label="mAP50", linewidth=2)
if np.isfinite(np.array(map5095)).any():
ax2.plot(epochs, map5095, label="mAP50-95", linewidth=2)
ax2.set_title("Kurva Metrik Evaluasi")
ax2.set_xlabel("Epoch")
ax2.set_ylabel("Score")
ax2.set_ylim(0, 1.0)
ax2.grid(alpha=0.25)
ax2.legend(loc="best")
plt.tight_layout()
buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150)
plt.close(fig)
buf.seek(0)
data = np.frombuffer(buf.read(), dtype=np.uint8)
bgr = cv2.imdecode(data, cv2.IMREAD_COLOR)
return cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
def save_training_curve_image(chart_rgb: np.ndarray, output_path: Path) -> str | None:
if chart_rgb is None or not isinstance(chart_rgb, np.ndarray) or chart_rgb.size == 0:
return None
output_path.parent.mkdir(parents=True, exist_ok=True)
chart_bgr = cv2.cvtColor(chart_rgb, cv2.COLOR_RGB2BGR)
cv2.imwrite(str(output_path), chart_bgr)
return str(output_path)
def load_rgb_image(image_path: Path) -> np.ndarray | None:
if not image_path.exists() or not image_path.is_file():
return None
bgr = cv2.imread(str(image_path))
if bgr is None:
return None
return cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
def _latest_training_value(rows: list[dict[str, str]], key: str) -> float | None:
if not rows:
return None
for row in reversed(rows):
value = _to_float(row.get(key, ""))
if value is not None and np.isfinite(value):
return value
return None
def render_training_eval_html(rows: list[dict[str, str]], state: Dict[str, Any]) -> str:
precision = _latest_training_value(rows, "metrics/precision(B)")
recall = _latest_training_value(rows, "metrics/recall(B)")
map50 = _latest_training_value(rows, "metrics/mAP50(B)")
map5095 = _latest_training_value(rows, "metrics/mAP50-95(B)")
epoch_count = len(rows)
run_name = state.get("run_name", "-") if state else "-"
def fmt(value: float | None) -> str:
return "-" if value is None else f"{value:.4f}"
return (
f"""
<section class="metric-panel">
<div class="metric-header">
<div>
<h3>Evaluasi Training Model</h3>
<p>Ringkasan metrik validasi YOLO setelah training. Nilai mAP50 digunakan sebagai indikator akurasi deteksi untuk artikel.</p>
</div>
<span class="metric-badge">Run: {run_name}</span>
</div>
<div class="metric-grid">
<div class="metric-card"><span>Epoch</span><strong>{epoch_count}</strong><small>Jumlah epoch tercatat</small></div>
<div class="metric-card"><span>Precision</span><strong>{fmt(precision)}</strong><small>Ketepatan prediksi person</small></div>
<div class="metric-card"><span>Recall</span><strong>{fmt(recall)}</strong><small>Kemampuan menemukan person</small></div>
<div class="metric-card"><span>mAP50</span><strong>{fmt(map50)}</strong><small>Indikator akurasi deteksi</small></div>
</div>
<table class="formula-table">
<thead><tr><th>Metrik</th><th>Nilai</th><th>Keterangan</th></tr></thead>
<tbody>
<tr><td>Precision</td><td>{fmt(precision)}</td><td>Proporsi prediksi person yang benar.</td></tr>
<tr><td>Recall</td><td>{fmt(recall)}</td><td>Proporsi person aktual yang berhasil ditemukan.</td></tr>
<tr><td>mAP50</td><td>{fmt(map50)}</td><td>Rata-rata precision pada IoU 0.50; dipakai sebagai akurasi deteksi.</td></tr>
<tr><td>mAP50-95</td><td>{fmt(map5095)}</td><td>Evaluasi lebih ketat pada rentang IoU 0.50 sampai 0.95.</td></tr>
</tbody>
</table>
</section>
"""
)
def save_training_confusion_matrix(state: Dict[str, Any], output_path: Path) -> str | None:
save_dir_raw = (state.get("save_dir") or "").strip() if isinstance(state, dict) else ""
if not save_dir_raw:
return str(output_path) if output_path.exists() else None
save_dir = Path(save_dir_raw)
candidates = [
save_dir / "confusion_matrix.png",
save_dir / "confusion_matrix_normalized.png",
]
for candidate in candidates:
if candidate.exists() and candidate.is_file():
output_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(candidate, output_path)
return str(output_path)
return str(output_path) if output_path.exists() else None
def save_training_summary(paths: Dict[str, Path], state: Dict[str, Any], rows_count: int) -> str:
payload = {
"saved_at": datetime.now().isoformat(),
"status": state.get("status", "unknown"),
"stage": state.get("stage", "-"),
"run_name": state.get("run_name", "-"),
"rows_count": rows_count,
"results_csv": state.get("results_csv"),
"best_path": state.get("best_path") or str(paths["best"]) if paths["best"].exists() else None,
"hf_sync_msg": state.get("hf_sync_msg"),
"error": state.get("error"),
}
paths["summary"].write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
return str(paths["summary"])
def sync_training_artifacts_to_hf(
run_name: str,
results_csv: Path | None,
log_file: Path | None,
chart_file: Path | None,
confusion_matrix_file: Path | None,
summary_file: Path | None,
) -> str:
artifact_repo, artifact_repo_type, artifact_token = _artifact_repo_config()
hf_repo = artifact_repo or os.getenv("HF_MODEL_REPO", "").strip()
hf_repo_type = artifact_repo_type if artifact_repo else (os.getenv("HF_MODEL_REPO_TYPE", "space").strip() or "space")
hf_token = artifact_token
repo_dir = os.getenv("HF_TRAINING_DIR", "space_training_archive").strip().strip("/")
run_name = run_name or datetime.now().strftime("training_%Y%m%d_%H%M%S")
if not hf_repo:
return "HF training artifact sync dilewati: set HF_MODEL_REPO."
if _skip_runtime_commit_to_current_space(hf_repo, hf_repo_type):
return "HF training artifact sync dilewati: commit runtime ke Space yang sama dinonaktifkan agar Space tidak rebuild sendiri."
upload_items: list[tuple[Path | None, str]] = [
(results_csv, f"{repo_dir}/{run_name}/results.csv"),
(log_file, f"{repo_dir}/{run_name}/training.log"),
(chart_file, f"{repo_dir}/{run_name}/training_curve.png"),
(confusion_matrix_file, f"{repo_dir}/{run_name}/confusion_matrix.png"),
(summary_file, f"{repo_dir}/{run_name}/training_summary.json"),
(chart_file, "training_outputs/training_curve_latest.png"),
(confusion_matrix_file, "training_outputs/confusion_matrix_latest.png"),
(summary_file, "training_outputs/training_summary_latest.json"),
]
try:
api = HfApi(token=hf_token)
uploaded = 0
for local_path, repo_path in upload_items:
if local_path is None or (not local_path.exists()) or (not local_path.is_file()):
continue
api.upload_file(
path_or_fileobj=str(local_path),
path_in_repo=repo_path,
repo_id=hf_repo,
repo_type=hf_repo_type,
commit_message=f"Save training artifact {run_name}",
)
uploaded += 1
return f"HF training artifact sync sukses: {uploaded} file -> {repo_dir}/{run_name}"
except Exception as err:
return f"HF training artifact sync gagal: {err}"
def _env_true(var_name: str, default: bool = False) -> bool:
val = os.getenv(var_name)
if val is None:
return default
return val.strip().lower() in {"1", "true", "yes", "on"}
def _skip_runtime_commit_to_current_space(repo_id: str, repo_type: str) -> bool:
if repo_type != "space":
return False
if _env_true("HF_ALLOW_SPACE_RUNTIME_COMMITS", default=False):
return False
current_space = os.getenv("SPACE_ID", "").strip() or os.getenv("HF_SPACE_ID", "").strip()
return bool(current_space and repo_id.strip() == current_space)
def _parse_globs(raw_value: str) -> list[str]:
return [g.strip() for g in raw_value.split(",") if g.strip()]
def _snapshot_repo_files(
root: Path,
include_globs: list[str],
ignore_globs: list[str],
) -> Dict[str, Tuple[int, int]]:
snapshot: Dict[str, Tuple[int, int]] = {}
for p in root.rglob("*"):
if not p.is_file():
continue
rel = p.relative_to(root).as_posix()
if include_globs and not any(fnmatch.fnmatch(rel, pat) for pat in include_globs):
continue
if any(fnmatch.fnmatch(rel, pat) for pat in ignore_globs):
continue
st = p.stat()
snapshot[rel] = (int(st.st_mtime_ns), int(st.st_size))
return snapshot
def _commit_repo_deltas(
api: HfApi,
root: Path,
repo_id: str,
repo_type: str,
changed: set[str],
deleted: set[str],
max_file_mb: float,
) -> str:
ops = []
skipped: list[str] = []
for rel in sorted(changed):
src = root / rel
if not src.exists():
continue
if src.stat().st_size > max_file_mb * 1024 * 1024:
skipped.append(rel)
continue
ops.append(CommitOperationAdd(path_in_repo=rel, path_or_fileobj=str(src)))
for rel in sorted(deleted):
ops.append(CommitOperationDelete(path_in_repo=rel))
if not ops:
if skipped:
return f"HF autosync: tidak ada file di-commit (skip terlalu besar: {len(skipped)})."
return "HF autosync: tidak ada perubahan untuk di-commit."
api.create_commit(
repo_id=repo_id,
repo_type=repo_type,
operations=ops,
commit_message=(
"Auto-sync repo updates "
f"({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})"
),
)
return (
"HF autosync sukses: "
f"updated={len(changed) - len(skipped)}, deleted={len(deleted)}, skipped={len(skipped)}"
)
def _repo_autosync_loop(
root: Path,
api: HfApi,
repo_id: str,
repo_type: str,
interval_sec: int,
debounce_sec: int,
include_globs: list[str],
ignore_globs: list[str],
max_file_mb: float,
) -> None:
prev = _snapshot_repo_files(root, include_globs, ignore_globs)
pending_changed: set[str] = set()
pending_deleted: set[str] = set()
pending_since = 0.0
print(
"HF autosync aktif: "
f"repo={repo_id} type={repo_type} interval={interval_sec}s debounce={debounce_sec}s"
)
while True:
try:
time.sleep(interval_sec)
curr = _snapshot_repo_files(root, include_globs, ignore_globs)
changed = {rel for rel, sig in curr.items() if prev.get(rel) != sig}
deleted = set(prev.keys()) - set(curr.keys())
if changed or deleted:
if pending_since <= 0:
pending_since = time.time()
pending_changed |= changed
pending_deleted |= deleted
pending_changed -= pending_deleted
if pending_changed or pending_deleted:
elapsed = time.time() - pending_since if pending_since > 0 else 0.0
if elapsed >= debounce_sec:
msg = _commit_repo_deltas(
api=api,
root=root,
repo_id=repo_id,
repo_type=repo_type,
changed=pending_changed,
deleted=pending_deleted,
max_file_mb=max_file_mb,
)
print(msg)
prev = curr
pending_changed = set()
pending_deleted = set()
pending_since = 0.0
except Exception as err:
print(f"HF autosync error: {err}")
_repo_autosync_started = False
def _running_in_hf_space() -> bool:
return bool(os.getenv("SPACE_ID", "").strip())
def _sync_repo_once(
root: Path,
api: HfApi,
repo_id: str,
repo_type: str,
include_globs: list[str],
ignore_globs: list[str],
max_file_mb: float,
) -> str:
current = _snapshot_repo_files(root, include_globs, ignore_globs)
changed = set(current.keys())
if not changed:
return "HF build-sync dilewati: tidak ada file yang cocok pola include."
msg = _commit_repo_deltas(
api=api,
root=root,
repo_id=repo_id,
repo_type=repo_type,
changed=changed,
deleted=set(),
max_file_mb=max_file_mb,
)
return f"HF build-sync (one-shot): {msg}"
def start_repo_autosync(root: Path) -> str:
global _repo_autosync_started
if _repo_autosync_started:
return "HF autosync sudah aktif."
repo_id = os.getenv("HF_AUTOSYNC_REPO_ID", "").strip() or os.getenv("HF_MODEL_REPO", "").strip()
enabled = _env_true("HF_AUTOSYNC_ENABLED", default=False)
if not enabled:
return "HF autosync nonaktif (HF_AUTOSYNC_ENABLED=false)."
local_only = _env_true("HF_AUTOSYNC_LOCAL_ONLY", default=True)
if local_only and _running_in_hf_space():
return "HF autosync dilewati di HF Space (mode local-only aktif)."
if not repo_id:
return "HF autosync dilewati: set HF_AUTOSYNC_REPO_ID atau HF_MODEL_REPO."
repo_type = (
os.getenv("HF_AUTOSYNC_REPO_TYPE", "").strip()
or os.getenv("HF_MODEL_REPO_TYPE", "").strip()
or "space"
)
token = os.getenv("HF_AUTOSYNC_TOKEN", "").strip() or os.getenv("HF_TOKEN", "").strip() or None
interval_sec = max(2, int(float(os.getenv("HF_AUTOSYNC_INTERVAL_SEC", "3").strip() or "3")))
debounce_sec = max(2, int(float(os.getenv("HF_AUTOSYNC_DEBOUNCE_SEC", "8").strip() or "8")))
realtime = _env_true("HF_AUTOSYNC_REALTIME", default=False)
max_file_mb = max(1.0, float(os.getenv("HF_AUTOSYNC_MAX_FILE_MB", "200").strip() or "200"))
include_globs = _parse_globs(
os.getenv(
"HF_AUTOSYNC_INCLUDE_GLOBS",
"*.py,*.md,requirements.txt,.gitignore",
)
)
ignore_globs = _parse_globs(
os.getenv(
"HF_AUTOSYNC_IGNORE_GLOBS",
".git/**,__pycache__/**,.venv/**,venv/**,*.pyc,*.pyo,*.tmp,runs_train/**,prepared_dataset/**",
)
)
api = HfApi(token=token)
if not realtime:
try:
_repo_autosync_started = True
return _sync_repo_once(
root=root,
api=api,
repo_id=repo_id,
repo_type=repo_type,
include_globs=include_globs,
ignore_globs=ignore_globs,
max_file_mb=max_file_mb,
)
except Exception as err:
_repo_autosync_started = False
return f"HF build-sync gagal (one-shot): {err}"
worker = Thread(
target=_repo_autosync_loop,
args=(
root,
api,
repo_id,
repo_type,
interval_sec,
debounce_sec,
include_globs,
ignore_globs,
max_file_mb,
),
daemon=True,
name="hf-repo-autosync",
)
worker.start()
_repo_autosync_started = True
return (
"HF autosync realtime dinyalakan: "
f"repo={repo_id}, type={repo_type}, interval={interval_sec}s, debounce={debounce_sec}s"
)
def _train_paths(base_dir: Path) -> Dict[str, Path]:
artifact_dir = base_dir / "training_outputs"
artifact_dir.mkdir(parents=True, exist_ok=True)
return {
"status": base_dir / "training_status.json",
"log": base_dir / "training.log",
"best": base_dir / "best.pt",
"artifact_dir": artifact_dir,
"chart": artifact_dir / "training_curve_latest.png",
"train_cm": artifact_dir / "confusion_matrix_latest.png",
"summary": artifact_dir / "training_summary_latest.json",
}
def _write_status(path: Path, payload: Dict[str, Any]) -> None:
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
tmp.replace(path)
def _read_status(path: Path) -> Dict[str, Any]:
if not path.exists():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return {}
def _is_pid_running(pid: int) -> bool:
if pid <= 0:
return False
try:
os.kill(pid, 0)
return True
except Exception:
return False
def _tail_log(path: Path, max_chars: int = 12000) -> str:
if not path.exists():
return "Belum ada log training."
text = path.read_text(encoding="utf-8", errors="replace")
if len(text) <= max_chars:
return text
return text[-max_chars:]
def _safe_results_csv_path(state: Dict[str, Any]) -> Path | None:
raw = (state.get("results_csv") or "").strip() if isinstance(state, dict) else ""
if not raw:
return None
p = Path(raw)
if p.exists() and p.is_file():
return p
return None
def _artifact_repo_config() -> Tuple[str, str, str | None]:
repo = (
os.getenv("HF_ARTIFACT_REPO", "").strip()
or os.getenv("HF_TRAINING_ARTIFACT_REPO", "").strip()
)
repo_type = os.getenv("HF_ARTIFACT_REPO_TYPE", "dataset").strip() or "dataset"
token = os.getenv("HF_TOKEN", "").strip() or None
return repo, repo_type, token
def _download_hf_file(
repo_id: str,
repo_type: str,
token: str | None,
filename: str,
destination: Path,
) -> bool:
try:
downloaded = hf_hub_download(
repo_id=repo_id,
repo_type=repo_type,
filename=filename,
token=token,
)
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(downloaded, destination)
return True
except Exception:
return False
def hydrate_latest_training_artifacts(paths: Dict[str, Path]) -> Dict[str, Any]:
repo_id, repo_type, token = _artifact_repo_config()
if not repo_id:
return {}
artifact_dir = paths["artifact_dir"]
latest_results = artifact_dir / "results_latest.csv"
latest_log = artifact_dir / "training_latest.log"
_download_hf_file(
repo_id,
repo_type,
token,
"training_outputs/training_curve_latest.png",
paths["chart"],
)
_download_hf_file(
repo_id,
repo_type,
token,
"training_outputs/confusion_matrix_latest.png",
paths["train_cm"],
)
if paths["summary"].exists() is False:
_download_hf_file(
repo_id,
repo_type,
token,
"training_outputs/training_summary_latest.json",
paths["summary"],
)
state = _read_status(paths["summary"])
run_name = str(state.get("run_name", "")).strip()
if run_name and run_name != "-":
base = f"space_training_archive/{run_name}"
_download_hf_file(repo_id, repo_type, token, f"{base}/results.csv", latest_results)
_download_hf_file(repo_id, repo_type, token, f"{base}/training.log", latest_log)
if latest_results.exists() and latest_results.is_file():
state["results_csv"] = str(latest_results)
if state.get("status") in {"running", "", None}:
state["status"] = "done"
state["stage"] = "latest_artifact"
if latest_log.exists() and latest_log.is_file() and not paths["log"].exists():
shutil.copy2(latest_log, paths["log"])
if state:
state["artifacts_sync_msg"] = f"Ditampilkan dari repo artefak: {repo_id}"
_write_status(paths["status"], state)
return state
def _training_status_md(state: Dict[str, Any], rows_count: int) -> str:
if not state:
return "Status training: **Idle**"
status = state.get("status", "unknown")
run_name = state.get("run_name", "-")
if status == "running":
stage = state.get("stage", "running")
return (
f"Status training: **Running** \nRun: `{run_name}` \n"
f"Tahap: **{stage}** \nEpoch tercatat: **{rows_count}**"
)
if status == "done":
sync_msg = state.get("artifacts_sync_msg")
suffix = f" \nArtefak: `{sync_msg}`" if sync_msg else ""
return f"Status training: **Selesai** \nRun: `{run_name}`{suffix}"
if status == "failed":
err = state.get("error", "-")
sync_msg = state.get("artifacts_sync_msg")
suffix = f" \nArtefak: `{sync_msg}`" if sync_msg else ""
return f"Status training: **Gagal** \nRun: `{run_name}` \nError: `{err}`{suffix}"
return f"Status training: **{status}** \nRun: `{run_name}`"
def poll_training_status() -> Tuple[str, Any, Any, Any, Any, np.ndarray, Any, np.ndarray, str, str]:
try:
base_dir = Path(__file__).resolve().parent
paths = _train_paths(base_dir)
state = _read_status(paths["status"])
if not state and (
not paths["chart"].exists()
or not paths["train_cm"].exists()
or not paths["summary"].exists()
):
state = hydrate_latest_training_artifacts(paths)
log_text = _tail_log(paths["log"])
results_csv = _safe_results_csv_path(state)
rows = _read_results_rows(results_csv) if results_csv is not None else []
saved_chart = load_rgb_image(paths["chart"])
chart = (
plot_training_curves(results_csv)
if results_csv is not None
else saved_chart
if saved_chart is not None
else _make_info_image("Grafik training belum tersedia.")
)
chart_file = save_training_curve_image(chart, paths["chart"]) if results_csv is not None else (
str(paths["chart"]) if paths["chart"].exists() else None
)
train_cm_file = save_training_confusion_matrix(state, paths["train_cm"])
train_cm_image = load_rgb_image(Path(train_cm_file)) if train_cm_file else None
if train_cm_image is None:
train_cm_image = _make_info_image("Confusion matrix training belum tersedia.\nJalankan training sampai validasi selesai.")
if state.get("status") == "running":
pid = int(state.get("pid", 0) or 0)
if not _is_pid_running(pid):
# Fallback jika proses selesai tapi status belum ter-update.
state["status"] = "done" if paths["best"].exists() else "failed"
if state["status"] == "failed" and not state.get("error"):
state["error"] = "Proses berhenti tanpa status akhir."
_write_status(paths["status"], state)
best_file = str(paths["best"]) if paths["best"].exists() else None
metrics_file = str(results_csv) if results_csv is not None else None
log_file = str(paths["log"]) if paths["log"].exists() else None
summary_file = save_training_summary(paths, state, len(rows))
if state.get("status") in {"done", "failed"} and not state.get("artifacts_sync_attempted"):
sync_msg = sync_training_artifacts_to_hf(
run_name=str(state.get("run_name", "training_run")),
results_csv=results_csv,
log_file=Path(log_file) if log_file else None,
chart_file=Path(chart_file) if chart_file else None,
confusion_matrix_file=Path(train_cm_file) if train_cm_file else None,
summary_file=Path(summary_file) if summary_file else None,
)
state["artifacts_sync_attempted"] = True
state["artifacts_sync_msg"] = sync_msg
_write_status(paths["status"], state)
return (
log_text,
best_file,
metrics_file,
log_file,
chart_file,
chart,
train_cm_file,
train_cm_image,
render_training_eval_html(rows, state),
_training_status_md(state, len(rows)),
)
except Exception as err:
return (
f"Monitor training error: {err}",
None,
None,
None,
None,
_make_info_image("Monitor training error.\nSilakan klik Refresh Status Training."),
None,
_make_info_image("Monitor training error.\nConfusion matrix training tidak tersedia."),
render_training_eval_html([], {}),
"Status training: **Error monitor**",
)
def _append_training_log(log_path: Path, message: str) -> None:
with log_path.open("a", encoding="utf-8") as lf:
lf.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {message}\n")
def training_worker_main() -> None:
base_dir = Path(os.environ["TRAIN_BASE_DIR"]).resolve()
paths = _train_paths(base_dir)
run_name = os.environ["TRAIN_RUN_NAME"]
project_dir = Path(os.environ["TRAIN_PROJECT_DIR"])
save_dir = Path(os.environ["TRAIN_SAVE_DIR"])
results_csv = save_dir / "results.csv"
train_model_choice = os.environ.get("TRAIN_MODEL_CHOICE", "YOLOv8n")
state: Dict[str, Any] = {
"status": "running",
"stage": "preparing_dataset",
"run_name": run_name,
"pid": os.getpid(),
"save_dir": str(save_dir),
"results_csv": str(results_csv),
"started_at": datetime.now().isoformat(),
}
_write_status(paths["status"], state)
try:
_append_training_log(paths["log"], f"Worker training aktif. PID={os.getpid()}")
_append_training_log(paths["log"], "Preparing dataset dan pseudo-label di background...")
data_yaml, prep_info = _prepare_training_dataset(
dataset_root=base_dir / "dataset",
use_pseudo_if_missing=_env_true("TRAIN_USE_PSEUDO", default=True),
pseudo_conf=float(os.environ.get("TRAIN_PSEUDO_CONF", "0.30")),
log_file=paths["log"],
)
model_path = resolve_model_path("Best.pt" if train_model_choice == "Best.pt" else "YOLOv8n")
state.update(
{
"stage": "training",
"data_yaml": str(data_yaml),
"model_path": str(model_path),
}
)
_write_status(paths["status"], state)
_append_training_log(paths["log"], f"Dataset siap: {prep_info}")
_append_training_log(paths["log"], f"Start training {run_name}")
_append_training_log(paths["log"], f"Model awal training: {train_model_choice} ({model_path})")
model = YOLO(str(model_path))
res = model.train(
data=str(data_yaml),
epochs=int(os.environ["TRAIN_EPOCHS"]),
imgsz=int(os.environ["TRAIN_IMGSZ"]),
batch=int(os.environ["TRAIN_BATCH"]),
device="cpu",
project=str(project_dir),
name=run_name,
exist_ok=False,
verbose=True,
)
final_save = Path(str(res.save_dir))
final_results_csv = final_save / "results.csv"
best_pt = final_save / "weights" / "best.pt"
best_out = base_dir / "best.pt"
hf_sync_msg = "HF sync dilewati: best.pt tidak ditemukan."
if best_pt.exists() and best_pt.is_file():
shutil.copy2(best_pt, best_out)
hf_sync_msg = sync_best_to_hf(best_out)
chart_file: str | None = None
chart_src = final_save / "results.png"
if chart_src.exists() and chart_src.is_file():
paths["chart"].parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(chart_src, paths["chart"])
chart_file = str(paths["chart"])
elif final_results_csv.exists() and final_results_csv.is_file():
chart_file = save_training_curve_image(plot_training_curves(final_results_csv), paths["chart"])
cm_file = save_training_confusion_matrix({"save_dir": str(final_save)}, paths["train_cm"])
rows = _read_results_rows(final_results_csv if final_results_csv.exists() else None)
final_state = {
**state,
"status": "done",
"stage": "done",
"ended_at": datetime.now().isoformat(),
"save_dir": str(final_save),
"results_csv": str(final_results_csv),
"best_path": str(best_out if best_out.exists() else best_pt),
"hf_sync_msg": hf_sync_msg,
}
summary_file = save_training_summary(paths, final_state, len(rows))
artifact_sync_msg = sync_training_artifacts_to_hf(
run_name=run_name,
results_csv=final_results_csv if final_results_csv.exists() else None,
log_file=paths["log"] if paths["log"].exists() else None,
chart_file=Path(chart_file) if chart_file else None,
confusion_matrix_file=Path(cm_file) if cm_file else None,
summary_file=Path(summary_file),
)
final_state.update(
{
"artifacts_sync_attempted": True,
"artifacts_sync_msg": artifact_sync_msg,
}
)
save_training_summary(paths, final_state, len(rows))
_write_status(paths["status"], final_state)
_append_training_log(paths["log"], "TRAIN_DONE")
except Exception as err:
_append_training_log(paths["log"], f"TRAIN_FAILED: {err}")
failed_state = {
**state,
"status": "failed",
"stage": "failed",
"ended_at": datetime.now().isoformat(),
"error": str(err),
}
summary_file = save_training_summary(paths, failed_state, 0)
artifact_sync_msg = sync_training_artifacts_to_hf(
run_name=run_name,
results_csv=None,
log_file=paths["log"] if paths["log"].exists() else None,
chart_file=paths["chart"] if paths["chart"].exists() else None,
confusion_matrix_file=paths["train_cm"] if paths["train_cm"].exists() else None,
summary_file=Path(summary_file),
)
failed_state.update(
{
"artifacts_sync_attempted": True,
"artifacts_sync_msg": artifact_sync_msg,
}
)
save_training_summary(paths, failed_state, 0)
_write_status(paths["status"], failed_state)
raise
def run_training(
train_model_choice: str,
epochs: float,
imgsz: float,
batch: float,
use_pseudo_if_missing: bool,
pseudo_conf: float,
) -> Tuple[str, Any, Any, Any, Any, np.ndarray, Any, np.ndarray, str, str]:
base_dir = Path(__file__).resolve().parent
paths = _train_paths(base_dir)
current = _read_status(paths["status"])
with paths["log"].open("a", encoding="utf-8") as lf:
lf.write(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Trigger training request\n")
if current.get("status") == "running":
pid = int(current.get("pid", 0) or 0)
if _is_pid_running(pid):
return poll_training_status()
try:
run_name = f"classroom_train_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
project_dir = base_dir / "runs_train"
save_dir = project_dir / run_name
results_csv = save_dir / "results.csv"
env = os.environ.copy()
env["TRAIN_BASE_DIR"] = str(base_dir)
env["TRAIN_MODEL_CHOICE"] = str(train_model_choice)
env["TRAIN_EPOCHS"] = str(int(epochs))
env["TRAIN_IMGSZ"] = str(int(imgsz))
env["TRAIN_BATCH"] = str(int(batch))
env["TRAIN_USE_PSEUDO"] = "true" if use_pseudo_if_missing else "false"
env["TRAIN_PSEUDO_CONF"] = str(float(pseudo_conf))
env["TRAIN_PROJECT_DIR"] = str(project_dir)
env["TRAIN_RUN_NAME"] = run_name
env["TRAIN_SAVE_DIR"] = str(save_dir)
env["TRAIN_STATUS_PATH"] = str(paths["status"])
with paths["log"].open("a", encoding="utf-8") as lf:
lf.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Worker background dimulai: {run_name}\n")
proc = subprocess.Popen(
[sys.executable, "-u", __file__, "--train-worker"],
stdout=lf,
stderr=subprocess.STDOUT,
cwd=str(base_dir),
env=env,
start_new_session=True,
)
_write_status(
paths["status"],
{
"status": "running",
"stage": "preparing_dataset",
"run_name": run_name,
"pid": proc.pid,
"save_dir": str(save_dir),
"results_csv": str(results_csv),
"started_at": datetime.now().isoformat(),
},
)
return poll_training_status()
except Exception as err:
with paths["log"].open("a", encoding="utf-8") as lf:
lf.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Training gagal sebelum worker aktif: {err}\n")
_write_status(
paths["status"],
{
"status": "failed",
"stage": "failed_before_spawn",
"run_name": "-",
"error": str(err),
"ended_at": datetime.now().isoformat(),
},
)
return poll_training_status()
CUSTOM_CSS = """
.container {max-width: 1180px !important; margin: auto;}
.academic-hero {
padding: 22px 0 14px 0; border-bottom: 1px solid #dbe4ef; margin-bottom: 14px;
}
.academic-hero h1 {
margin: 0; text-align: center; font-size: 28px; line-height: 1.22; color: #0f172a; font-weight: 800;
}
.hero-subtitle {text-align:center;color:#475569;margin:8px auto 0 auto; max-width: 860px; line-height: 1.55;}
.section-box {
padding: 18px; border: 1px solid #d7dee8; border-radius: 8px; background: #ffffff;
box-shadow: 0 1px 2px rgba(15, 23, 42, 0.04); margin-bottom: 14px;
}
.method-box {
padding: 18px; border: 1px solid #bfd0e6; border-radius: 8px;
background: linear-gradient(180deg, #f8fbff 0%, #ffffff 100%);
margin-bottom: 14px;
}
.method-grid {display: grid; grid-template-columns: repeat(3, minmax(0, 1fr)); gap: 10px; margin-top: 10px;}
.method-item {border: 1px solid #dbe4ef; border-radius: 8px; padding: 12px; background: #ffffff;}
.method-item span {display:block; font-size:12px; font-weight:800; color:#2563eb; margin-bottom:4px;}
.method-item strong {display:block; color:#0f172a; margin-bottom:4px;}
.method-item p {margin:0; color:#475569; font-size:13px; line-height:1.45;}
.section-title {margin-bottom: 8px !important;}
.version-chip {
display:inline-block; padding:4px 10px; border:1px solid #c7d2e3; border-radius:999px;
font-size:12px; color:#334155; background:#f8fbff; margin-top:4px;
}
.version-foot {font-size:12px; color:#64748b; text-align:center; margin-top:8px;}
.advanced-box {margin-top: 8px; padding: 8px; border: 1px dashed #cbd5e1; border-radius: 8px;}
.section-box table {width: 100%; border-collapse: collapse; font-size: 14px;}
.section-box th, .section-box td {padding: 9px 11px; border-bottom: 1px solid #e2e8f0; vertical-align: top; text-align: left !important;}
.section-box th {background: #f8fafc; font-weight: 700; color:#0f172a;}
.section-box code {white-space: normal; font-size: 13px;}
.metric-panel {border:1px solid #dbe4ef; border-radius:8px; padding:14px; background:#fbfdff;}
.metric-header {display:flex; align-items:flex-start; justify-content:space-between; gap:12px; margin-bottom:12px;}
.metric-header h3 {margin:0 0 4px 0; font-size:18px; color:#0f172a;}
.metric-header p {margin:0; color:#475569; line-height:1.45;}
.metric-badge {white-space:nowrap; border:1px solid #bfdbfe; background:#eff6ff; color:#1d4ed8; border-radius:999px; padding:5px 10px; font-size:12px; font-weight:700;}
.metric-grid {display:grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap:10px; margin:10px 0 14px 0;}
.metric-card {border:1px solid #dbe4ef; border-radius:8px; background:#ffffff; padding:12px;}
.metric-card span {display:block; font-size:12px; color:#64748b; font-weight:800; letter-spacing:.02em;}
.metric-card strong {display:block; color:#0f172a; font-size:26px; line-height:1.1; margin:3px 0;}
.metric-card small {color:#64748b; line-height:1.35;}
.formula-table {margin-top:10px; overflow-wrap:anywhere;}
.formula-table td:last-child, .formula-table th:last-child {text-align:right !important;}
.metric-muted {color:#64748b; font-style:italic;}
@media (max-width: 760px) {
.method-grid, .metric-grid {grid-template-columns: 1fr;}
.metric-header {display:block;}
.metric-badge {display:inline-block; margin-top:8px;}
}
@media (prefers-color-scheme: dark) {
body, .gradio-container {background: #0f172a !important; color: #e2e8f0 !important;}
.section-box {background: #111827 !important; border-color: #334155 !important;}
.method-box {background: #0b1220 !important; border-color: #334155 !important;}
.academic-hero {border-color:#334155 !important;}
.academic-hero h1 {color:#f8fafc !important;}
.hero-subtitle {color: #94a3b8 !important;}
.version-chip {background: #111827 !important; border-color: #475569 !important; color: #cbd5e1 !important;}
.version-foot {color: #94a3b8 !important;}
.advanced-box {border-color: #475569 !important;}
.section-box th {background: #1f2937 !important;}
.section-box th, .section-box td {border-color: #334155 !important;}
.method-item, .metric-card, .metric-panel {background:#111827 !important; border-color:#334155 !important;}
.method-item strong, .metric-header h3, .metric-card strong, .section-box th {color:#e2e8f0 !important;}
.method-item p, .metric-header p, .metric-card small {color:#94a3b8 !important;}
.metric-badge {background:#172554 !important; border-color:#1d4ed8 !important; color:#bfdbfe !important;}
}
"""
with gr.Blocks(title="Pengecekan dan Klasifikasi Mahasiswa - YOLOv8") as demo:
gr.HTML(
f"""
<header class="academic-hero">
<h1>Implementasi Klasifikasi Kehadiran Mahasiswa Berbasis Citra Menggunakan YOLOv8</h1>
<p class="hero-subtitle">
Aplikasi demo penelitian untuk mendeteksi objek person pada suasana kelas, membandingkan model dasar
dan model hasil training, serta menyajikan evaluasi count objek dalam bentuk confusion matrix dan metrik.
</p>
<div style="text-align:center;">
<span class="version-chip">Versi {APP_VERSION} | Minor Update ke-{APP_MINOR_UPDATE}</span>
</div>
</header>
"""
)
with gr.Column(elem_classes=["method-box"]):
gr.HTML(
"""
<h2 style="margin:0 0 6px 0;">Metodologi Sistem</h2>
<p style="margin:0;color:#475569;line-height:1.55;">
Alur sistem disusun mengikuti tahapan penelitian: akuisisi data, inferensi model, visualisasi hasil,
evaluasi berbasis confusion matrix, dan pengembangan model melalui training.
</p>
<div class="method-grid">
<div class="method-item"><span>01</span><strong>Akuisisi Data</strong><p>Input citra kelas melalui upload file atau kamera.</p></div>
<div class="method-item"><span>02</span><strong>Inferensi YOLOv8</strong><p>Model default mendeteksi semua kelas; Best.pt difokuskan pada person.</p></div>
<div class="method-item"><span>03</span><strong>Klasifikasi Objek</strong><p>Hasil dipetakan menjadi Person dan Not Person/Background.</p></div>
<div class="method-item"><span>04</span><strong>Evaluasi</strong><p>Count objek dihitung sebagai TP, FP, FN, dan TN.</p></div>
<div class="method-item"><span>05</span><strong>Training</strong><p>Bobot model diperbarui dari dataset train dan test.</p></div>
<div class="method-item"><span>06</span><strong>Arsip Penelitian</strong><p>Gambar, CSV, grafik, dan log disimpan untuk kebutuhan artikel.</p></div>
</div>
"""
)
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 1 - Pilih Model", elem_classes=["section-title"])
inference_mode = gr.Dropdown(
choices=["Single Model", "Bandingkan (YOLOv8n vs Best.pt)"],
value="Single Model",
label="Mode Inferensi",
)
model_choice = gr.Dropdown(
choices=["YOLOv8n", "Best.pt"],
value="YOLOv8n",
label="Pilih Model",
)
conf_input = gr.Slider(0.1, 1.0, value=0.25, step=0.05, label="Parameter Confidence (Deteksi)")
with gr.Accordion("Advanced Options", open=False, elem_classes=["advanced-box"]):
annotation_style = gr.Radio(
choices=["Border + Label", "Label saja"],
value="Border + Label",
label="Gaya anotasi objek",
info="Pilih tampilan bbox atau label-only untuk label objek yang terdeteksi.",
)
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 2 - Training dan Evaluasi Model", elem_classes=["section-title"])
gr.Markdown(
"Tahap ini dijalankan sebelum uji foto. Training menghasilkan metrik validasi, grafik, "
"confusion matrix, dan bobot `best.pt`."
)
train_model_choice = gr.Dropdown(
choices=["YOLOv8n", "Best.pt"],
value="YOLOv8n",
label="Model awal training",
info="Pilih YOLOv8n untuk training awal, atau Best.pt untuk melanjutkan fine-tuning dari bobot terakhir.",
)
with gr.Row():
train_epochs = gr.Number(label="Epochs", value=10, precision=0, minimum=1)
train_imgsz = gr.Number(label="Image Size", value=640, precision=0, minimum=64)
train_batch = gr.Number(label="Batch", value=8, precision=0, minimum=1)
use_pseudo = gr.Checkbox(
label="Gunakan pseudo-label jika label YOLO tidak tersedia",
value=True,
)
pseudo_conf = gr.Slider(
0.1,
0.9,
value=0.30,
step=0.05,
label="Confidence pseudo-label",
)
train_btn = gr.Button("Mulai Training", variant="primary")
refresh_train_btn = gr.Button("Refresh Status Training")
train_status_md = gr.Markdown("Status training: **Idle**")
training_eval_html = gr.HTML(render_training_eval_html([], {}))
train_curve = gr.Image(
label="Grafik Training (Loss, Precision/Recall, mAP per Epoch)",
type="numpy",
value=_make_info_image("Grafik training akan tampil di sini."),
)
training_cm_image = gr.Image(
label="Confusion Matrix Training",
type="numpy",
value=_make_info_image("Confusion matrix training akan tampil setelah validasi selesai."),
interactive=False,
)
train_log = gr.Textbox(label="Terminal Training (Live)", lines=14, interactive=False)
best_model_file = gr.File(label="best.pt hasil training")
training_metrics_file = gr.File(label="results.csv / Metrik Training Lokal")
training_log_file = gr.File(label="training.log / Log Training Lokal")
training_chart_file = gr.File(label="training_curve_latest.png / Grafik Training Lokal")
training_cm_file = gr.File(label="confusion_matrix_latest.png / Confusion Matrix Training Lokal")
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 3 - Input Gambar", elem_classes=["section-title"])
image_input = gr.Image(
label="Input Gambar Kelas (Upload/Kamera)",
type="numpy",
sources=["upload", "webcam"],
)
run_btn = gr.Button("Jalankan Deteksi dan Klasifikasi", variant="primary")
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 4 - Grafik Hasil Deteksi", elem_classes=["section-title"])
comparison_chart = gr.Image(
label="Grafik Count Person dan Not Person",
type="numpy",
value=_make_info_image("Grafik hasil deteksi akan tampil setelah proses inferensi."),
)
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 5 - Evaluasi Foto Uji (Confusion Matrix + Rumus)", elem_classes=["section-title"])
cm_state = gr.State(dict(CM_INIT))
predicted_state = gr.State("")
actual_state = gr.State("")
eval_payload_state = gr.State({})
gr.Markdown(
"Evaluasi otomatis memakai count objek. `Person` dihitung sebagai objek positif, "
"sedangkan objek selain person dihitung sebagai `Not Person/Background`."
)
reset_eval_btn = gr.Button("Reset Evaluasi")
eval_status = gr.Markdown("Belum ada update evaluasi otomatis.")
with gr.Row():
cm_image = gr.Image(
label="Confusion Matrix Count Objek",
value=draw_confusion_matrix_from_state(CM_INIT),
type="numpy",
interactive=False,
)
metric_md = gr.HTML(render_metric_markdown(CM_INIT))
with gr.Column(elem_classes=["section-box"]):
gr.Markdown("## Tahap 6 - Hasil Gambar", elem_classes=["section-title"])
output_image_yolo = gr.Image(
label="Output Deteksi YOLOv8n",
type="filepath",
)
output_image_best = gr.Image(
label="Output Deteksi Best.pt",
type="filepath",
)
class_result_md = gr.Markdown("Ringkasan hasil deteksi akan tampil di sini.")
predicted_label_md = gr.Markdown("Label prediksi evaluasi: **-**")
saved_detection_file = gr.File(label="File Gambar Hasil Evaluasi")
evaluation_csv_file = gr.File(label="Log CSV Evaluasi Lokal")
run_btn_evt = run_btn.click(
fn=run_check_and_classify,
inputs=[image_input, conf_input, model_choice, inference_mode, annotation_style],
outputs=[
output_image_yolo,
output_image_best,
class_result_md,
comparison_chart,
predicted_state,
actual_state,
eval_payload_state,
predicted_label_md,
saved_detection_file,
evaluation_csv_file,
],
)
run_btn_evt.then(
fn=update_evaluation_auto,
inputs=[cm_state, eval_payload_state],
outputs=[cm_state, cm_image, metric_md, eval_status],
)
inference_mode.change(
fn=toggle_model_selector,
inputs=[inference_mode],
outputs=[model_choice],
)
reset_eval_btn.click(
fn=reset_evaluation,
inputs=[],
outputs=[cm_state, cm_image, metric_md, eval_status],
)
train_btn.click(
fn=run_training,
inputs=[train_model_choice, train_epochs, train_imgsz, train_batch, use_pseudo, pseudo_conf],
outputs=[
train_log,
best_model_file,
training_metrics_file,
training_log_file,
training_chart_file,
train_curve,
training_cm_file,
training_cm_image,
training_eval_html,
train_status_md,
],
)
refresh_train_btn.click(
fn=poll_training_status,
inputs=[],
outputs=[
train_log,
best_model_file,
training_metrics_file,
training_log_file,
training_chart_file,
train_curve,
training_cm_file,
training_cm_image,
training_eval_html,
train_status_md,
],
queue=False,
)
demo.load(
fn=poll_training_status,
inputs=[],
outputs=[
train_log,
best_model_file,
training_metrics_file,
training_log_file,
training_chart_file,
train_curve,
training_cm_file,
training_cm_image,
training_eval_html,
train_status_md,
],
queue=False,
)
train_timer = gr.Timer(value=2.0)
train_timer.tick(
fn=poll_training_status,
inputs=[],
outputs=[
train_log,
best_model_file,
training_metrics_file,
training_log_file,
training_chart_file,
train_curve,
training_cm_file,
training_cm_image,
training_eval_html,
train_status_md,
],
queue=False,
)
gr.HTML(
f"""
<div class="version-foot">
Build info: {APP_VERSION} (minor update ke-{APP_MINOR_UPDATE})
</div>
"""
)
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--train-worker":
training_worker_main()
raise SystemExit(0)
if _env_true("HF_AUTOSYNC_ENABLED", default=False):
print(start_repo_autosync(Path(__file__).resolve().parent))
else:
print("HF autosync nonaktif (HF_AUTOSYNC_ENABLED=false).")
demo.queue()
demo.launch(css=CUSTOM_CSS, ssr_mode=False)