Blame for app.py

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)