Newer
Older
TIASJudgeModel / tests / scripts / test_summarize_cv.py
"""scripts/summarize_cv.py のテスト(色特徴分類器の run の集計).

仕様(実装サマリーの設計を仕様として扱う):

- color run(environment.model_name が color_logistic / color_hist_gb)を
  「色特徴+ロジスティック回帰」「色特徴+勾配ブースティング」として集計する
- 色特徴の run には学習率を表示しない(_cv_summary.json のキーにも lr を含めない)
- 既存(CNN)の run の出力(表示名・キー)は変わらない

集計値だけを持つ最小の metrics.json・config.yaml を tmp_path に作って読む.
"""

from __future__ import annotations

import contextlib
import importlib.util
import io
import json
import sys
from pathlib import Path
from types import ModuleType

import yaml

SCRIPT_PATH = Path(__file__).resolve().parents[2] / "scripts" / "summarize_cv.py"
CLASSES = ["pale", "light_red", "red", "dark_red", "purple"]


def _load_script() -> ModuleType:
    spec = importlib.util.spec_from_file_location("summarize_cv_under_test", SCRIPT_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def _subset(f1: float) -> dict:
    return {
        "macro_f1": f1,
        "accuracy": f1,
        "per_class": {c: {"recall": f1} for c in CLASSES},
    }


def _write_run(
    root: Path,
    run_id: str,
    model: str,
    *,
    severities: list[str],
    seed: int = 42,
    lr: float = 1e-4,
    weighting: str = "none",
    f1: float = 0.5,
) -> None:
    d = root / run_id
    d.mkdir(parents=True)
    metrics = {
        "run_id": run_id,
        "severities_trained": severities,
        "imbalance": {"class_weighting": weighting, "sampler": "random"},
        "folds": [{"subsets": {"common": _subset(f1), "all": _subset(f1)}} for _ in range(5)],
        "pooled": {"mild": {"prediction_breakdown": {"overall": {"light_red": 0.3}}}},
        "timing": {"total_seconds": 10.0},
        "environment": {"model_name": model},
    }
    (d / "metrics.json").write_text(json.dumps(metrics), encoding="utf-8")
    config = {"config": {"split": {"seed": seed}, "train": {"lr": lr}}}
    (d / "config.yaml").write_text(yaml.safe_dump(config), encoding="utf-8")


def _summarize(root: Path) -> tuple[str, dict]:
    module = _load_script()
    out = io.StringIO()
    with contextlib.redirect_stdout(out):
        module.main(root)
    summary = json.loads((root / "_cv_summary.json").read_text(encoding="utf-8"))
    return out.getvalue(), summary


WITH = ["strong", "mild", "normal"]
WITHOUT = ["strong", "normal"]


def test_color_runs_are_labeled_without_lr(tmp_path: Path) -> None:
    _write_run(
        tmp_path, "20260101-000000_cv_color_logistic_with_mild_x", "color_logistic", severities=WITH
    )
    _write_run(
        tmp_path,
        "20260101-000001_cv_color_logistic_without_mild_x",
        "color_logistic",
        severities=WITHOUT,
        f1=0.6,
    )
    _write_run(
        tmp_path, "20260101-000002_cv_color_hist_gb_with_mild_x", "color_hist_gb", severities=WITH
    )
    text, summary = _summarize(tmp_path)
    assert "色特徴+ロジスティック回帰" in text
    assert "色特徴+勾配ブースティング" in text
    color_lines = [ln for ln in text.splitlines() if "色特徴" in ln]
    assert color_lines
    assert all("lr" not in ln for ln in color_lines)
    assert set(summary["summary"]) == {
        "color_logistic/none/with_mild",
        "color_logistic/none/without_mild",
        "color_hist_gb/none/with_mild",
    }
    # 込み/除外の対応比較にも lr の無いキーで入る
    assert set(summary["paired"]) == {"color_logistic/none"}
    assert summary["paired"]["color_logistic/none"]["n"] == 5


def test_color_run_with_class_weighting_label(tmp_path: Path) -> None:
    _write_run(
        tmp_path, "20260101-000000_cv_color_logistic_with_mild_x", "color_logistic", severities=WITH
    )
    _write_run(
        tmp_path,
        "20260101-000001_cv_color_logistic_with_mild_cwinv_x",
        "color_logistic",
        severities=WITH,
        weighting="inverse",
        f1=0.7,
    )
    text, summary = _summarize(tmp_path)
    assert "色特徴+ロジスティック回帰(重み inverse)" in text
    assert "color_logistic/inverse/with_mild" in summary["summary"]
    assert all("lr" not in k for k in summary["summary"])


def test_cnn_run_output_is_unchanged(tmp_path: Path) -> None:
    _write_run(
        tmp_path, "20260101-000000_cv_resnet50_with_mild_x", "resnet50", severities=WITH, lr=1e-3
    )
    _write_run(
        tmp_path,
        "20260101-000001_cv_resnet50_with_mild_cwinv_x",
        "resnet50",
        severities=WITH,
        lr=1e-3,
        weighting="inverse",
    )
    text, summary = _summarize(tmp_path)
    assert "ResNet-50(lr 0.001)" in text
    assert "ResNet-50(lr 0.001・重み inverse)" in text
    assert set(summary["summary"]) == {
        "resnet50/lr0.001/none/with_mild",
        "resnet50/lr0.001/inverse/with_mild",
    }


def test_cnn_and_color_runs_side_by_side(tmp_path: Path) -> None:
    _write_run(tmp_path, "20260101-000000_cv_with_mild_x", "convnext_tiny", severities=WITH)
    _write_run(
        tmp_path, "20260101-000001_cv_color_logistic_with_mild_x", "color_logistic", severities=WITH
    )
    _write_run(tmp_path, "20260101-000002_cv_unknown_x", "resnet18", severities=WITH)
    text, summary = _summarize(tmp_path)
    assert set(summary["summary"]) == {
        "convnext_tiny/lr0.0001/none/with_mild",
        "color_logistic/none/with_mild",
    }
    assert len(summary["runs"]) == 2  # 対象外のモデル(resnet18)は飛ばす
    # 並びは CNN → 色特徴
    assert text.index("ConvNeXt-Tiny") < text.index("色特徴+ロジスティック回帰")