diff --git a/tests/cli/test_train.py b/tests/cli/test_train.py new file mode 100644 index 0000000..513f046 --- /dev/null +++ b/tests/cli/test_train.py @@ -0,0 +1,289 @@ +"""tias-train cv の結合テスト. + +仕様: SPEC_01「CLI」「実験の出力」「回収ホワイトリスト」・SPEC_02「出力」． + +- runs/<実験ID>/ に SPEC_02「出力」の表のファイルが揃う +- metrics.json に部分集合ごとの指標・mild の内訳・所要時間・実行環境がある +- history.csv にエポックの所要秒数の列がある +- 実験 ID は `YYYYmmdd-HHMMSS_<設定ファイル名>_`．標準出力の最終行が実験 ID +- 回収対象の出力（ログを含む）に患者 ID・ファイル名が出ない +- split.json・predictions.csv・checkpoints は回収ホワイトリストの対象外（checkpoints は明示時のみ） +- 「ややあり除外」でも held_out は全程度で評価される + +合成データ（scripts/make_synthetic_dataset.py）のファイル名を目立つ文字列に付け替えて使う． +""" + +from __future__ import annotations + +import contextlib +import csv +import importlib.util +import io +import json +import logging +import re +import sys +from pathlib import Path +from types import ModuleType + +import numpy as np +import pytest +import yaml + +import tias_judge_model.cli.train as cli +from tias_judge_model.data.manifest import build_manifest_records, write_manifest +from tias_judge_model.remote.whitelist import is_pull_allowed + +pytestmark = pytest.mark.slow + +SCRIPT_PATH = Path(__file__).resolve().parents[2] / "scripts" / "make_synthetic_dataset.py" +SECRET = "SECRETPT" +GIT = "abc1234" +N_FOLDS = 5 +PER_DIR = 6 +CONFIG_NAME = "cv_tiny_nomild" +RECOVERABLE = ["config.yaml", "metrics.json", "history.csv", "confusion_matrix.csv", "log.txt"] + + +def _load_script() -> ModuleType: + spec = importlib.util.spec_from_file_location("make_synthetic_dataset_train", 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 + + +@pytest.fixture(scope="module") +def run(tmp_path_factory: pytest.TempPathFactory) -> dict: + tmp = tmp_path_factory.mktemp("train_cli") + data_root = tmp / "dataset" + _load_script().generate(data_root, per_dir=PER_DIR, seed=0) + for n, png in enumerate(sorted((data_root / "images").rglob("*.png"))): + png.rename(png.with_name(f"{SECRET}{n:04d}.png")) + write_manifest(build_manifest_records(data_root / "images"), data_root / "manifest.csv") + + config = tmp / f"{CONFIG_NAME}.yaml" + config.write_text( + yaml.safe_dump( + { + "data": {"root": data_root.as_posix(), "severities": ["strong", "normal"]}, + "split": {"seed": 3, "n_folds": N_FOLDS}, + "model": {"name": "resnet18", "pretrained": False}, + "train": { + "image_size": 32, + "batch_size": 8, + "max_epochs": 2, + "warmup_epochs": 1, + "inner_val_ratio": 0.2, + "num_workers": 0, + "amp": False, + }, + } + ), + encoding="utf-8", + ) + out_root = tmp / "runs" + + mp = pytest.MonkeyPatch() + mp.delenv("TIAS_DATA_ROOT", raising=False) + mp.setenv("TIAS_GIT_COMMIT", GIT) + root_logger = logging.getLogger() + old_level = root_logger.level + root_logger.setLevel(logging.INFO) + out, err = io.StringIO(), io.StringIO() + stderr_handler = logging.StreamHandler(err) + root_logger.addHandler(stderr_handler) + try: + with contextlib.redirect_stdout(out): + rc = cli.main(["cv", "--config", str(config), "--out-root", str(out_root)]) + finally: + root_logger.removeHandler(stderr_handler) + root_logger.setLevel(old_level) + mp.undo() + run_dirs = [p for p in out_root.iterdir() if p.is_dir()] + return { + "rc": rc, + "stdout": out.getvalue(), + "stderr": err.getvalue(), + "run_dirs": run_dirs, + "run_dir": run_dirs[0] if run_dirs else None, + "n_images": 9 * PER_DIR, + } + + +def test_exit_code_and_single_run_dir(run: dict) -> None: + assert run["rc"] == 0 + assert len(run["run_dirs"]) == 1 + + +def test_run_id_format_and_last_stdout_line(run: dict) -> None: + run_id = run["run_dir"].name + assert re.fullmatch(rf"\d{{8}}-\d{{6}}_{CONFIG_NAME}_{GIT}", run_id) + lines = [ln for ln in run["stdout"].splitlines() if ln.strip()] + assert lines[-1] == run_id + + +def test_output_files_exist(run: dict) -> None: + d = run["run_dir"] + for name in [*RECOVERABLE, "split.json", "predictions.csv"]: + assert (d / name).is_file(), name + pngs = list((d / "figures").glob("*.png")) + assert pngs + names = " ".join(p.name for p in pngs) + assert "confusion" in names # 混同行列 + assert len(pngs) >= 2 # 学習曲線 ＋ 混同行列 + ckpts = sorted(p.name for p in (d / "checkpoints").glob("*.pt")) + assert ckpts == [f"fold{k}.pt" for k in range(N_FOLDS)] + assert (d / "log.txt").stat().st_size > 0 + + +def test_config_yaml_records_git_and_seed(run: dict) -> None: + doc = yaml.safe_load((run["run_dir"] / "config.yaml").read_text(encoding="utf-8")) + text = json.dumps(doc, ensure_ascii=False) + assert GIT in text + # 解決済み設定（model・train を含む）と seed が記録される + assert doc["config"]["split"]["seed"] == 3 + assert doc["config"]["model"]["name"] == "resnet18" + assert doc["config"]["train"]["image_size"] == 32 + assert doc["seeds"]["split"] == 3 + + +def _metrics(run: dict) -> dict: + return json.loads((run["run_dir"] / "metrics.json").read_text(encoding="utf-8")) + + +def test_metrics_has_subsets_per_fold_and_mean_std(run: dict) -> None: + m = _metrics(run) + assert len(m["folds"]) == N_FOLDS + for fold in m["folds"]: + assert set(fold["subsets"]) >= {"all", "common", "mild"} + assert fold["best_epoch"] >= 1 + for subset in ("all", "common", "mild"): + s = fold["subsets"][subset] + assert {"macro_f1", "accuracy", "per_class", "confusion_matrix"} <= set(s) + for subset in ("all", "common", "mild"): + ms = m["mean_std"][subset] + assert {"mean", "std"} <= set(ms["macro_f1"]) + assert {"mean", "std"} <= set(ms["accuracy"]) + + +def test_held_out_evaluated_on_all_severities(run: dict) -> None: + # severities=[strong, normal] でも held_out は全程度（合計 = manifest 全件） + m = _metrics(run) + n_all = sum(f["subsets"]["all"]["n"] for f in m["folds"]) + n_mild = sum(f["subsets"]["mild"]["n"] for f in m["folds"]) + assert n_all == run["n_images"] + assert n_mild == 4 * PER_DIR # mild は淡紅以外の 4 クラス + + +def test_confusion_matrix_sum(run: dict) -> None: + m = _metrics(run) + all_cm = np.array(m["confusion_matrix_sum"]["all"]) + common_cm = np.array(m["confusion_matrix_sum"]["common"]) + assert all_cm.shape == (5, 5) + assert all_cm.sum() == run["n_images"] + assert common_cm.sum() == run["n_images"] - 4 * PER_DIR + # 行（正解）の合計はクラスごとの枚数（淡紅は normal のみ，他は strong + mild） + assert all_cm.sum(axis=1).tolist() == [ + 2 * PER_DIR, + PER_DIR, + 2 * PER_DIR, + 2 * PER_DIR, + 2 * PER_DIR, + ] + + with (run["run_dir"] / "confusion_matrix.csv").open(encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + subsets = {r["subset"] for r in rows} + assert subsets == {"common", "all"} + all_rows = [r for r in rows if r["subset"] == "all"] + assert [r["true"] for r in all_rows] == ["pale", "light_red", "red", "dark_red", "purple"] + csv_matrix = [[int(v) for k, v in r.items() if k.startswith("pred_")] for r in all_rows] + assert csv_matrix == all_cm.tolist() + + +def test_mild_prediction_breakdown_present(run: dict) -> None: + m = _metrics(run) + breakdown = m["pooled"]["mild"]["prediction_breakdown"] + overall = breakdown["overall"] + assert overall["n"] == 4 * PER_DIR + total = overall["light_red"] + overall["original"] + overall["other"] + assert total == pytest.approx(1.0) + assert set(breakdown["per_class"]) == {"pale", "red", "dark_red", "purple"} + for fold in m["folds"]: + assert "prediction_breakdown" in fold["subsets"]["mild"] + + +def test_metrics_has_timing_and_environment(run: dict) -> None: + m = _metrics(run) + t = m["timing"] + assert t["total_seconds"] > 0 + assert t["data_loading_seconds"] >= 0 + assert t["total_seconds"] >= t["data_loading_seconds"] + for fold in m["folds"]: + ft = fold["timing"] + assert ft["train_seconds"] >= 0 + assert ft["eval_seconds"] >= 0 + assert 1 <= ft["epochs_run"] <= 2 + assert ft["mean_epoch_seconds"] >= 0 + env = m["environment"] + assert env["device"] + assert env["model_name"] == "resnet18" + assert env["num_parameters"] > 0 + + +def test_history_has_epoch_seconds(run: dict) -> None: + with (run["run_dir"] / "history.csv").open(encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + assert rows + cols = set(rows[0]) + assert {"fold", "epoch", "train_loss", "val_loss", "val_macro_f1", "lr"} <= cols + seconds_cols = [c for c in cols if "second" in c] + assert seconds_cols + assert all(float(r[seconds_cols[0]]) >= 0 for r in rows) + assert {int(r["fold"]) for r in rows} == set(range(N_FOLDS)) + + +def test_recoverable_outputs_have_no_patient_ids(run: dict) -> None: + d = run["run_dir"] + for name in RECOVERABLE: + assert SECRET not in (d / name).read_text(encoding="utf-8"), name + for png in (d / "figures").glob("*.png"): + assert SECRET.encode() not in png.read_bytes(), png.name + assert SECRET not in run["stdout"] + assert SECRET not in run["stderr"] + + +def test_split_and_predictions_contain_ids_and_are_not_pulled(run: dict) -> None: + d = run["run_dir"] + # 個票（患者 ID を含む）はサーバ上にだけ置き，回収しない + assert SECRET in (d / "split.json").read_text(encoding="utf-8") + assert SECRET in (d / "predictions.csv").read_text(encoding="utf-8") + for name in ("split.json", "predictions.csv"): + assert not is_pull_allowed(name, with_checkpoints=False) + assert not is_pull_allowed(name, with_checkpoints=True) + + +def test_predictions_cover_all_severities(run: dict) -> None: + with (run["run_dir"] / "predictions.csv").open(encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + assert len(rows) == run["n_images"] + assert {r["severity"] for r in rows} == {"strong", "mild", "normal"} + + +def test_every_output_matches_pull_whitelist_as_spec(run: dict) -> None: + d = run["run_dir"] + recoverable = set(RECOVERABLE) + for path in d.rglob("*"): + if not path.is_file(): + continue + rel = path.relative_to(d).as_posix() + if rel in recoverable or (rel.startswith("figures/") and rel.endswith(".png")): + assert is_pull_allowed(rel, with_checkpoints=False), rel + elif rel.startswith("checkpoints/") and rel.endswith(".pt"): + assert not is_pull_allowed(rel, with_checkpoints=False), rel + assert is_pull_allowed(rel, with_checkpoints=True), rel + else: + assert not is_pull_allowed(rel, with_checkpoints=True), rel + assert rel in {"split.json", "predictions.csv"}, f"想定外の出力: {rel}" diff --git a/tests/data/test_augmentation.py b/tests/data/test_augmentation.py new file mode 100644 index 0000000..7784e13 --- /dev/null +++ b/tests/data/test_augmentation.py @@ -0,0 +1,128 @@ +"""データ拡張と Dataset のテスト（SPEC_02「データ拡張」「学習の設定」）. + +- **色を変える拡張は行わない**: 一様な色の画像に拡張をかけても，舌領域の画素は元の色のまま + （はみ出しの黒との境界の補間を除き，黒と元の色の混合＝元の色の定数倍にしかならない） +- 拡張は幾何変換（左右反転等）であり，実際に画像が変わりうる +- 評価時（augment=False）は拡張しない +- 正規化は ImageNet の mean / std +""" + +from __future__ import annotations + +import numpy as np +import pytest +import torch +from torchvision.transforms import v2 + +from tias_judge_model.data.dataset import ( + TongueImageDataset, + make_train_augmentation, + normalize_batch, +) + +COLOR = np.array([190, 110, 95], dtype=np.uint8) +N_TRIALS = 40 + + +def _uniform_tongue(size: int = 64) -> np.ndarray: + """黒背景の中央に一様な色の領域がある画像 `(H, W, 3)`.""" + img = np.zeros((size, size, 3), dtype=np.uint8) + img[8 : size - 8, 12 : size - 12] = COLOR + return img + + +def _flatten_transforms(t: object) -> list[object]: + inner = getattr(t, "transforms", None) + if inner is None: + return [t] + out: list[object] = [] + for x in inner: + out.extend(_flatten_transforms(x)) + return out + + +def test_augmentation_contains_no_color_transforms() -> None: + forbidden = ( + v2.ColorJitter, + v2.Grayscale, + v2.RandomGrayscale, + v2.RandomAutocontrast, + v2.RandomEqualize, + v2.RandomPosterize, + v2.RandomSolarize, + v2.RandomInvert, + v2.RandomAdjustSharpness, + v2.RandomPhotometricDistort, + v2.RandomChannelPermutation, + v2.AutoAugment, + v2.RandAugment, + v2.TrivialAugmentWide, + v2.AugMix, + ) + for t in _flatten_transforms(make_train_augmentation()): + assert not isinstance(t, forbidden), f"色を変える変換が含まれる: {type(t).__name__}" + + +def test_augmentation_does_not_change_pixel_colors() -> None: + torch.manual_seed(0) + ds = TongueImageDataset(_uniform_tongue()[None], np.array([2]), augment=True) + color = COLOR.astype(np.float64) + for _ in range(N_TRIALS): + image, label = ds[0] + assert label == 2 + arr = image.permute(1, 2, 0).numpy().astype(np.float64).reshape(-1, 3) + nonblack = arr[arr.any(axis=1)] + assert len(nonblack) > 0 + # 画素は黒と元の色の混合（= 元の色の t 倍）でしかない（色相・彩度が変わらない） + t = nonblack.sum(axis=1, keepdims=True) / color.sum() + assert np.abs(nonblack - t * color).max() <= 2.0 + assert t.max() <= 1.0 + 1e-2 # 明るくならない + # 舌領域の大半は元の色そのまま + exact = np.all(np.abs(nonblack - color) <= 1, axis=1) + assert exact.mean() > 0.8 + + +def test_augmentation_keeps_shape_and_dtype() -> None: + torch.manual_seed(1) + ds = TongueImageDataset(_uniform_tongue(48)[None], np.array([0]), augment=True) + image, _ = ds[0] + assert image.shape == (3, 48, 48) + assert image.dtype == torch.uint8 + + +def test_augmentation_actually_transforms_geometry() -> None: + # 左右非対称な画像で，拡張をかけると少なくとも 1 回は元と異なる + img = np.zeros((32, 32, 3), dtype=np.uint8) + img[4:28, 2:12] = COLOR + ds = TongueImageDataset(img[None], np.array([0]), augment=True) + original = torch.from_numpy(img).permute(2, 0, 1) + torch.manual_seed(2) + assert any(not torch.equal(ds[0][0], original) for _ in range(N_TRIALS)) + + +def test_eval_dataset_is_not_augmented() -> None: + img = _uniform_tongue(32) + img[0, 0] = (1, 2, 3) # 非対称にする + ds = TongueImageDataset(img[None], np.array([4]), augment=False) + original = torch.from_numpy(img).permute(2, 0, 1) + for _ in range(10): + image, label = ds[0] + assert torch.equal(image, original) + assert label == 4 + + +def test_dataset_length_mismatch_raises() -> None: + with pytest.raises(Exception): # noqa: B017 件数不一致で止まれば型は問わない + TongueImageDataset(np.zeros((2, 8, 8, 3), dtype=np.uint8), np.array([0]), augment=False) + + +def test_normalize_batch_uses_imagenet_stats() -> None: + mean = torch.tensor([0.485, 0.456, 0.406]).view(1, 3, 1, 1) + std = torch.tensor([0.229, 0.224, 0.225]).view(1, 3, 1, 1) + batch = torch.tensor(np.stack([_uniform_tongue(8)] * 2)).permute(0, 3, 1, 2) + out = normalize_batch(batch) + assert out.dtype == torch.float32 + expected = (batch.float() / 255.0 - mean) / std + assert torch.allclose(out, expected, atol=1e-5) + # 黒画素は -mean/std + assert torch.allclose(out[0, :, 0, 0], (-mean / std).view(3), atol=1e-5) diff --git a/tests/data/test_images.py b/tests/data/test_images.py new file mode 100644 index 0000000..b7f0ca1 --- /dev/null +++ b/tests/data/test_images.py @@ -0,0 +1,96 @@ +"""前処理（レターボックス）のテスト（SPEC_02「前処理」「学習の設定」の train.image_size）. + +- アスペクト比を保ち，長辺を image_size に合わせる（縮小・拡大とも） +- 余白は黒 (0, 0, 0) +- 読めない画像は止める．例外メッセージに画像のファイル名を出さない +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest +from PIL import Image + +from tias_judge_model.data.images import letterbox, load_letterboxed_images +from tias_judge_model.data.manifest import ManifestRecord + +COLOR = (200, 120, 90) + + +def _bbox_of_content(arr: np.ndarray) -> tuple[int, int]: + """非黒画素の (高さ, 幅).""" + mask = arr.any(axis=2) + rows = np.where(mask.any(axis=1))[0] + cols = np.where(mask.any(axis=0))[0] + return int(rows[-1] - rows[0] + 1), int(cols[-1] - cols[0] + 1) + + +@pytest.mark.parametrize( + ("width", "height", "size", "expected_hw"), + [ + (200, 100, 64, (32, 64)), # 横長の縮小 + (100, 200, 64, (64, 32)), # 縦長の縮小 + (20, 10, 64, (32, 64)), # 横長の拡大 + (90, 90, 32, (32, 32)), # 正方形 + ], +) +def test_letterbox_keeps_aspect_and_fits_long_side( + width: int, height: int, size: int, expected_hw: tuple[int, int] +) -> None: + out = letterbox(Image.new("RGB", (width, height), COLOR), size) + assert out.shape == (size, size, 3) + assert out.dtype == np.uint8 + assert _bbox_of_content(out) == expected_hw + + +def test_letterbox_pads_with_black_and_keeps_color() -> None: + out = letterbox(Image.new("RGB", (200, 100), COLOR), 64) + # 上下の余白 16 行ずつが黒 + assert (out[:16] == 0).all() + assert (out[48:] == 0).all() + # 内側は元の色のまま + inner = out[20:44, 4:60].reshape(-1, 3) + assert (np.abs(inner.astype(int) - np.array(COLOR)) <= 1).all() + + +def test_letterbox_content_is_centered() -> None: + out = letterbox(Image.new("RGB", (100, 200), COLOR), 64) + mask = out.any(axis=2) + cols = np.where(mask.any(axis=0))[0] + left, right = cols[0], 64 - 1 - cols[-1] + assert abs(left - right) <= 1 + + +def test_load_letterboxed_images_order_and_shape(tmp_path: Path) -> None: + images = tmp_path / "images" + colors = [(255, 0, 0), (0, 255, 0), (0, 0, 255)] + records = [] + for i, c in enumerate(colors): + d = images / "red_strong" + d.mkdir(parents=True, exist_ok=True) + Image.new("RGB", (40 + 10 * i, 30), c).save(d / f"img{i}.png") + records.append(ManifestRecord(f"red_strong/img{i}.png", f"p{i}", "red", "strong")) + out = load_letterboxed_images(records, images, 16) + assert out.shape == (3, 16, 16, 3) + for i, c in enumerate(colors): + assert tuple(out[i, 8, 8]) == c + + +def test_load_letterboxed_images_broken_image_hides_filename(tmp_path: Path) -> None: + images = tmp_path / "images" / "red_strong" + images.mkdir(parents=True) + (images / "SECRETNAME.png").write_bytes(b"not a png") + records = [ManifestRecord("red_strong/SECRETNAME.png", "SECRETNAME", "red", "strong")] + with pytest.raises(ValueError) as exc_info: + load_letterboxed_images(records, tmp_path / "images", 16) + assert "SECRETNAME" not in str(exc_info.value) + + +def test_load_letterboxed_images_missing_image_hides_filename(tmp_path: Path) -> None: + (tmp_path / "images").mkdir() + records = [ManifestRecord("red_strong/SECRETNAME.png", "SECRETNAME", "red", "strong")] + with pytest.raises(ValueError) as exc_info: + load_letterboxed_images(records, tmp_path / "images", 16) + assert "SECRETNAME" not in str(exc_info.value) diff --git a/tests/data/test_split_inner_val.py b/tests/data/test_split_inner_val.py new file mode 100644 index 0000000..da5d331 --- /dev/null +++ b/tests/data/test_split_inner_val.py @@ -0,0 +1,91 @@ +"""交差検証の fold 内の検証分割のテスト（SPEC_02「交差検証の実行」手順 2-1・「データ分割」）. + +- 学習側から `inner_val_ratio` の患者を検証用に取る（患者単位・9 グループ層化） +- 学習用と検証用の患者が重ならず，合わせると学習側の全患者になる +- 同じ seed で同じ結果．manifest の行順に依存しない +- held_out と組み合わせても，held_out・学習・検証の患者が互いに重ならない +""" + +from __future__ import annotations + +import random +from collections import Counter + +import pytest + +from tias_judge_model.data.classes import ALL_DIR_NAMES, parse_dir_name +from tias_judge_model.data.manifest import ManifestRecord +from tias_judge_model.data.split import make_kfold_splits, select_records, split_inner_val + +SEED = 42 + + +def make_records(per_group: int = 20, images_per_patient: int = 1) -> list[ManifestRecord]: + records = [] + for group in ALL_DIR_NAMES: + label, severity = parse_dir_name(group) + for p in range(per_group): + pid = f"pt_{group}_{p:03d}" + for k in range(images_per_patient): + records.append(ManifestRecord(f"{group}/{pid}_{k}.png", pid, label, severity)) + return records + + +def _group_of(records: list[ManifestRecord]) -> dict[str, str]: + return {r.patient_id: f"{r.label}_{r.severity}" for r in records} + + +def test_train_and_val_are_disjoint_and_cover_all_patients() -> None: + records = make_records() + train, val = split_inner_val(records, 0.15, SEED) + assert set(train).isdisjoint(val) + assert set(train) | set(val) == {r.patient_id for r in records} + + +def test_val_ratio_is_stratified_by_group() -> None: + records = make_records(per_group=20) + _, val = split_inner_val(records, 0.15, SEED) + group = _group_of(records) + counts = Counter(group[p] for p in val) + # 20 × 0.15 = 3 人ずつ，9 グループすべてから + assert counts == dict.fromkeys(ALL_DIR_NAMES, 3) + + +def test_multiple_images_per_patient_stay_together() -> None: + records = make_records(per_group=10, images_per_patient=3) + train, val = split_inner_val(records, 0.2, SEED) + assert set(train).isdisjoint(val) + val_images = select_records(records, val) + assert len(val_images) == 3 * len(val) + + +def test_same_seed_same_result_and_order_independent() -> None: + records = make_records() + shuffled = records[:] + random.Random(0).shuffle(shuffled) + assert split_inner_val(records, 0.15, SEED) == split_inner_val(shuffled, 0.15, SEED) + + +def test_different_seed_changes_result() -> None: + records = make_records() + assert split_inner_val(records, 0.15, 1) != split_inner_val(records, 0.15, 2) + + +@pytest.mark.parametrize("ratio", [0.0, 1.0, -0.1, 1.5]) +def test_invalid_ratio_raises(ratio: float) -> None: + with pytest.raises(ValueError): + split_inner_val(make_records(), ratio, SEED) + + +def test_held_out_train_val_are_mutually_disjoint_in_every_fold() -> None: + records = make_records(per_group=15) + all_patients = {r.patient_id for r in records} + folds = make_kfold_splits(records, 5, SEED) + for fold in folds: + train_side = select_records(records, fold.train) + train, val = split_inner_val(train_side, 0.15, SEED + fold.index) + held = set(fold.held_out) + assert held.isdisjoint(train) + assert held.isdisjoint(val) + assert set(train).isdisjoint(val) + assert held | set(train) | set(val) == all_patients diff --git a/tests/evaluation/test_metrics.py b/tests/evaluation/test_metrics.py new file mode 100644 index 0000000..066e951 --- /dev/null +++ b/tests/evaluation/test_metrics.py @@ -0,0 +1,200 @@ +"""評価指標のテスト（SPEC_02「評価指標」「交差検証の実行」の「評価の出し方」）. + +- macro F1・accuracy・クラス別 precision / recall / F1・混同行列 + （クラス ID 順固定．SPEC_01「クラス定義」） +- 部分集合 `all`（全程度）・`common`（strong と normal）・`mild` +- mild の予測の内訳（淡紅・元のクラス・その他と予測された割合．全体とクラス別） +- fold 間の平均 ± 標準偏差・混同行列の全 fold 合計 + +期待値は手計算した既知の予測・正解から求める． +""" + +from __future__ import annotations + +import math + +import numpy as np +import pytest + +from tias_judge_model.data.classes import CLASS_SLUGS, label_to_id +from tias_judge_model.evaluation.metrics import ( + aggregate_folds, + classification_metrics, + macro_f1, + mean_std, + mild_prediction_breakdown, + subset_metrics, + sum_confusion_matrices, +) + +PALE, LIGHT_RED, RED, DARK_RED, PURPLE = range(5) + +# 正解 [淡白, 淡白, 淡紅, 淡紅, 紅, 紅]・予測 [淡白, 淡紅, 淡紅, 淡紅, 紅, 淡白] +Y_TRUE = np.array([PALE, PALE, LIGHT_RED, LIGHT_RED, RED, RED]) +Y_PRED = np.array([PALE, LIGHT_RED, LIGHT_RED, LIGHT_RED, RED, PALE]) +# 手計算: 淡白 P=1/2 R=1/2 F1=1/2・淡紅 P=2/3 R=1 F1=4/5・紅 P=1 R=1/2 F1=2/3 +EXPECTED = { + "pale": (0.5, 0.5, 0.5, 2), + "light_red": (2 / 3, 1.0, 0.8, 2), + "red": (1.0, 0.5, 2 / 3, 2), +} +EXPECTED_CM = [ + [1, 1, 0, 0, 0], + [0, 2, 0, 0, 0], + [1, 0, 1, 0, 0], + [0, 0, 0, 0, 0], + [0, 0, 0, 0, 0], +] + + +class TestClassOrder: + def test_class_ids_are_fixed_as_in_spec(self) -> None: + # SPEC_01「クラス定義」: 0 淡白・1 淡紅・2 紅・3 暗赤紅・4 紫 + assert CLASS_SLUGS == ("pale", "light_red", "red", "dark_red", "purple") + assert [label_to_id(s) for s in CLASS_SLUGS] == [0, 1, 2, 3, 4] + + +class TestClassificationMetrics: + def test_accuracy(self) -> None: + m = classification_metrics(Y_TRUE, Y_PRED) + assert m["accuracy"] == pytest.approx(4 / 6) + assert m["n"] == 6 + + def test_per_class_precision_recall_f1(self) -> None: + m = classification_metrics(Y_TRUE, Y_PRED) + for slug, (p, r, f1, support) in EXPECTED.items(): + got = m["per_class"][slug] + assert got["precision"] == pytest.approx(p) + assert got["recall"] == pytest.approx(r) + assert got["f1"] == pytest.approx(f1) + assert got["support"] == support + + def test_macro_f1_averages_classes_present_in_truth(self) -> None: + expected = (0.5 + 0.8 + 2 / 3) / 3 + m = classification_metrics(Y_TRUE, Y_PRED) + assert m["macro_f1"] == pytest.approx(expected) + assert macro_f1(Y_TRUE, Y_PRED) == pytest.approx(expected) + + def test_macro_f1_perfect_prediction_is_one(self) -> None: + y = np.array([0, 1, 2, 3, 4, 4]) + assert macro_f1(y, y) == pytest.approx(1.0) + + def test_confusion_matrix_rows_true_cols_pred_in_class_id_order(self) -> None: + m = classification_metrics(Y_TRUE, Y_PRED) + assert m["confusion_matrix"] == EXPECTED_CM + + def test_confusion_matrix_is_always_5x5(self) -> None: + # 正解・予測に現れないクラスがあっても軸がぶれない + m = classification_metrics(np.array([PURPLE]), np.array([PURPLE])) + cm = np.array(m["confusion_matrix"]) + assert cm.shape == (5, 5) + assert cm[PURPLE, PURPLE] == 1 + assert cm.sum() == 1 + + def test_input_order_does_not_change_metrics(self) -> None: + rng = np.random.default_rng(0) + perm = rng.permutation(len(Y_TRUE)) + a = classification_metrics(Y_TRUE, Y_PRED) + b = classification_metrics(Y_TRUE[perm], Y_PRED[perm]) + assert a == b + + +class TestMildPredictionBreakdown: + # mild の正解 [紅, 紅, 紅, 暗赤紅, 紫, 淡白]・予測 [淡紅, 紅, 暗赤紅, 暗赤紅, 淡紅, 淡白] + TRUE = np.array([RED, RED, RED, DARK_RED, PURPLE, PALE]) + PRED = np.array([LIGHT_RED, RED, DARK_RED, DARK_RED, LIGHT_RED, PALE]) + + def test_overall_fractions(self) -> None: + b = mild_prediction_breakdown(self.TRUE, self.PRED)["overall"] + assert b["n"] == 6 + assert b["light_red"] == pytest.approx(2 / 6) + assert b["original"] == pytest.approx(3 / 6) + assert b["other"] == pytest.approx(1 / 6) + + def test_fractions_sum_to_one(self) -> None: + b = mild_prediction_breakdown(self.TRUE, self.PRED) + for entry in [b["overall"], *b["per_class"].values()]: + assert entry["light_red"] + entry["original"] + entry["other"] == pytest.approx(1.0) + + def test_per_class_fractions(self) -> None: + per_class = mild_prediction_breakdown(self.TRUE, self.PRED)["per_class"] + red = per_class["red"] + assert red["n"] == 3 + assert red["light_red"] == pytest.approx(1 / 3) + assert red["original"] == pytest.approx(1 / 3) + assert red["other"] == pytest.approx(1 / 3) + assert per_class["dark_red"]["original"] == pytest.approx(1.0) + assert per_class["purple"]["light_red"] == pytest.approx(1.0) + assert per_class["pale"]["original"] == pytest.approx(1.0) + + +class TestSubsetMetrics: + # 程度つきの held_out: strong 2・normal 2・mild 2 + TRUE = np.array([RED, RED, LIGHT_RED, LIGHT_RED, RED, PURPLE]) + PRED = np.array([RED, PALE, LIGHT_RED, LIGHT_RED, LIGHT_RED, PURPLE]) + SEV = ["strong", "strong", "normal", "normal", "mild", "mild"] + + def test_subset_sizes(self) -> None: + s = subset_metrics(self.TRUE, self.PRED, self.SEV) + assert s["all"]["n"] == 6 + assert s["common"]["n"] == 4 + assert s["mild"]["n"] == 2 + + def test_common_is_strong_and_normal_only(self) -> None: + s = subset_metrics(self.TRUE, self.PRED, self.SEV) + expected = classification_metrics(self.TRUE[:4], self.PRED[:4]) + assert s["common"]["accuracy"] == pytest.approx(expected["accuracy"]) + assert s["common"]["confusion_matrix"] == expected["confusion_matrix"] + + def test_all_uses_every_image(self) -> None: + s = subset_metrics(self.TRUE, self.PRED, self.SEV) + assert s["all"]["accuracy"] == pytest.approx(4 / 6) + assert np.array(s["all"]["confusion_matrix"]).sum() == 6 + + def test_mild_has_prediction_breakdown(self) -> None: + s = subset_metrics(self.TRUE, self.PRED, self.SEV) + overall = s["mild"]["prediction_breakdown"]["overall"] + assert overall["n"] == 2 + assert overall["light_red"] == pytest.approx(0.5) + assert overall["original"] == pytest.approx(0.5) + assert overall["other"] == pytest.approx(0.0) + assert s["mild"]["accuracy"] == pytest.approx(0.5) + + def test_empty_mild_subset_does_not_crash(self) -> None: + s = subset_metrics(self.TRUE[:4], self.PRED[:4], self.SEV[:4]) + assert s["common"]["n"] == 4 + assert s["all"]["n"] == 4 + assert not s["mild"] + + +class TestFoldAggregation: + def test_mean_std_uses_sample_std(self) -> None: + r = mean_std([0.5, 0.7, 0.9]) + assert r["mean"] == pytest.approx(0.7) + assert r["std"] == pytest.approx(0.2) # ddof=1 + + def test_mean_std_single_value(self) -> None: + r = mean_std([0.4]) + assert r["mean"] == pytest.approx(0.4) + assert r["std"] == pytest.approx(0.0) + + def test_aggregate_folds_averages_macro_f1_and_accuracy(self) -> None: + f1 = classification_metrics(np.array([0, 1]), np.array([0, 1])) # acc 1.0 + f2 = classification_metrics(np.array([0, 1]), np.array([1, 1])) # acc 0.5 + agg = aggregate_folds([f1, f2]) + assert agg["accuracy"]["mean"] == pytest.approx(0.75) + assert agg["accuracy"]["std"] == pytest.approx(math.sqrt(0.125)) + assert agg["macro_f1"]["mean"] == pytest.approx((1.0 + f2["macro_f1"]) / 2) + assert agg["per_class"]["light_red"]["recall"]["mean"] == pytest.approx(1.0) + + def test_sum_confusion_matrices(self) -> None: + a = classification_metrics(Y_TRUE, Y_PRED)["confusion_matrix"] + b = classification_metrics(np.array([PURPLE, DARK_RED]), np.array([PURPLE, RED]))[ + "confusion_matrix" + ] + total = np.array(sum_confusion_matrices([a, b])) + assert total.shape == (5, 5) + assert total.sum() == 8 + assert total[PURPLE, PURPLE] == 1 + assert total[DARK_RED, RED] == 1 + assert total[LIGHT_RED, LIGHT_RED] == 2 diff --git a/tests/models/test_factory.py b/tests/models/test_factory.py new file mode 100644 index 0000000..1e38289 --- /dev/null +++ b/tests/models/test_factory.py @@ -0,0 +1,24 @@ +"""モデル生成のテスト（SPEC_02「モデル」: timm・出力 5 クラス）. + +事前学習の重みはダウンロードしない（pretrained=False のみ）． +""" + +from __future__ import annotations + +import torch + +from tias_judge_model.models.factory import count_parameters, create_model + + +def test_create_model_outputs_five_classes() -> None: + torch.manual_seed(0) + model = create_model("resnet18", pretrained=False).eval() + with torch.no_grad(): + out = model(torch.zeros(2, 3, 32, 32)) + assert out.shape == (2, 5) + + +def test_count_parameters_matches_numel() -> None: + model = create_model("resnet18", pretrained=False) + assert count_parameters(model) == sum(p.numel() for p in model.parameters()) + assert count_parameters(model) > 0 diff --git a/tests/test_training_config.py b/tests/test_training_config.py new file mode 100644 index 0000000..7ffc451 --- /dev/null +++ b/tests/test_training_config.py @@ -0,0 +1,126 @@ +"""学習の設定（`model`・`train` 節）のテスト（SPEC_02「学習の設定」・SPEC_01「設定ファイル」）. + +- 既定値が SPEC_02 の表どおり +- `model`・`train` を省略した既存の設定（`tias-dataset` 用）も読める +- 未知のキー・不正な値は拒否する +- リポジトリの `configs/cv_*.yaml` がすべて読め，モデル比較用の設定はモデル名以外が元の設定と同一 +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from tias_judge_model.schema import ExperimentConfig, load_config + +CONFIGS_DIR = Path(__file__).resolve().parents[1] / "configs" +CV_CONFIGS = sorted(CONFIGS_DIR.glob("cv_*.yaml")) +NO_ENV: dict[str, str] = {} + + +def _minimal() -> dict: + return {"data": {"root": "somewhere"}} + + +def test_model_defaults_follow_spec() -> None: + c = ExperimentConfig.model_validate(_minimal()) + assert c.model.name == "convnext_tiny" + assert c.model.pretrained is True + + +def test_train_defaults_follow_spec() -> None: + t = ExperimentConfig.model_validate(_minimal()).train + assert t.image_size == 224 + assert t.batch_size == 32 + assert t.max_epochs == 40 + assert t.patience == 8 + assert t.lr == pytest.approx(1e-4) + assert t.weight_decay == pytest.approx(0.05) + assert t.warmup_epochs == 2 + assert t.inner_val_ratio == pytest.approx(0.15) + assert t.num_workers == 4 + assert t.amp is True + + +@pytest.mark.parametrize("name", ["synthetic.yaml", "newton.yaml"]) +def test_existing_configs_without_model_train_still_load(name: str) -> None: + c = load_config(CONFIGS_DIR / name, environ=NO_ENV) + assert c.model.name == "convnext_tiny" + assert c.train.image_size == 224 + + +@pytest.mark.parametrize( + "raw", + [ + {"model": {"nme": "resnet18"}}, + {"train": {"epochs": 3}}, + {"train": {"batch_size": 0}}, + {"train": {"inner_val_ratio": 0.0}}, + {"train": {"inner_val_ratio": 1.0}}, + {"train": {"lr": 0}}, + {"model": {"name": ""}}, + ], +) +def test_invalid_training_settings_are_rejected(raw: dict) -> None: + with pytest.raises(ValidationError): + ExperimentConfig.model_validate({**_minimal(), **raw}) + + +def test_cv_configs_exist() -> None: + names = {p.name for p in CV_CONFIGS} + assert {"cv_with_mild.yaml", "cv_without_mild.yaml", "cv_synthetic.yaml"} <= names + + +@pytest.mark.parametrize("path", CV_CONFIGS, ids=lambda p: p.name) +def test_every_cv_config_loads(path: Path) -> None: + load_config(path, environ=NO_ENV) + + +def test_with_and_without_mild_differ_only_in_severities() -> None: + w = load_config(CONFIGS_DIR / "cv_with_mild.yaml", environ=NO_ENV) + wo = load_config(CONFIGS_DIR / "cv_without_mild.yaml", environ=NO_ENV) + assert w.data.severities == ("strong", "mild", "normal") + assert wo.data.severities == ("strong", "normal") + dw = w.model_dump() + dwo = wo.model_dump() + dw["data"].pop("severities") + dwo["data"].pop("severities") + assert dw == dwo + + +def test_production_cv_configs_use_spec_defaults() -> None: + c = load_config(CONFIGS_DIR / "cv_with_mild.yaml", environ=NO_ENV) + assert c.model == ExperimentConfig.model_validate(_minimal()).model + assert c.train == ExperimentConfig.model_validate(_minimal()).train + assert c.split.n_folds == 5 + + +def test_synthetic_cv_config_does_not_download_weights() -> None: + c = load_config(CONFIGS_DIR / "cv_synthetic.yaml", environ=NO_ENV) + assert c.model.pretrained is False + + +def _model_comparison_pairs() -> list[tuple[Path, Path]]: + pairs = [] + for p in CV_CONFIGS: + if p.name in {"cv_with_mild.yaml", "cv_without_mild.yaml", "cv_synthetic.yaml"}: + continue + suffix = "with_mild" if p.stem.endswith("_with_mild") else "without_mild" + pairs.append((p, CONFIGS_DIR / f"cv_{suffix}.yaml")) + return pairs + + +@pytest.mark.parametrize(("variant", "base"), _model_comparison_pairs(), ids=lambda p: Path(p).name) +def test_model_comparison_config_differs_only_in_model_name(variant: Path, base: Path) -> None: + v = load_config(variant, environ=NO_ENV).model_dump() + b = load_config(base, environ=NO_ENV).model_dump() + assert v["model"]["name"] != b["model"]["name"] + v["model"].pop("name") + b["model"].pop("name") + assert v == b + + +def test_model_comparison_configs_are_present() -> None: + assert len(_model_comparison_pairs()) >= 2 diff --git a/tests/training/test_cross_validation.py b/tests/training/test_cross_validation.py new file mode 100644 index 0000000..6e4a97b --- /dev/null +++ b/tests/training/test_cross_validation.py @@ -0,0 +1,152 @@ +"""交差検証の手順のテスト（SPEC_02「交差検証の実行」「程度（ややあり）の扱い」）. + +最重要: +- 分割は manifest 全体で作り，`data.severities` に関わらず同一（同じ seed なら同じ分割） +- 各 fold で held_out・学習・検証の患者が互いに重ならず，held_out は fold 間で manifest を分割する +- `data.severities` は学習用と検証用にだけ適用し，held_out は全程度のまま評価する + +合成データ（scripts/make_synthetic_dataset.py）・極小設定（resnet18・pretrained なし・1 エポック）で +実際に学習を回す（`@pytest.mark.slow`）． +""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import ModuleType + +import numpy as np +import pytest + +from tias_judge_model.data.manifest import read_manifest +from tias_judge_model.schema import ExperimentConfig +from tias_judge_model.training.cross_validation import CrossValidationResult, run_cross_validation + +SCRIPT_PATH = Path(__file__).resolve().parents[2] / "scripts" / "make_synthetic_dataset.py" +N_FOLDS = 5 + + +def _load_script() -> ModuleType: + spec = importlib.util.spec_from_file_location("make_synthetic_dataset_cv", 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 _config(root: Path, severities: list[str]) -> ExperimentConfig: + return ExperimentConfig.model_validate( + { + "data": {"root": str(root), "severities": severities}, + "split": {"seed": 7, "n_folds": N_FOLDS}, + "model": {"name": "resnet18", "pretrained": False}, + "train": { + "image_size": 32, + "batch_size": 8, + "max_epochs": 1, + "warmup_epochs": 0, + "inner_val_ratio": 0.2, + "num_workers": 0, + "amp": False, + }, + } + ) + + +@pytest.fixture(scope="module") +def cv_runs( + tmp_path_factory: pytest.TempPathFactory, +) -> tuple[list, CrossValidationResult, CrossValidationResult]: + root = tmp_path_factory.mktemp("cv") / "dataset" + _load_script().generate(root, per_dir=6, seed=0) + records = read_manifest(root / "manifest.csv") + with_mild = run_cross_validation( + _config(root, ["strong", "mild", "normal"]), records, root.parent / "ck_with" + ) + without_mild = run_cross_validation( + _config(root, ["strong", "normal"]), records, root.parent / "ck_without" + ) + return records, with_mild, without_mild + + +pytestmark = pytest.mark.slow + + +def test_runs_all_folds(cv_runs) -> None: + _, with_mild, without_mild = cv_runs + assert [f.index for f in with_mild.folds] == list(range(N_FOLDS)) + assert len(without_mild.folds) == N_FOLDS + + +def test_patients_disjoint_within_each_fold(cv_runs) -> None: + records, with_mild, without_mild = cv_runs + all_patients = {r.patient_id for r in records} + for result in (with_mild, without_mild): + for f in result.folds: + train, val, held = set(f.train_patients), set(f.val_patients), set(f.held_out_patients) + assert train and val and held + assert train.isdisjoint(val) + assert train.isdisjoint(held) + assert val.isdisjoint(held) + assert train | val | held == all_patients + + +def test_held_out_partitions_manifest_across_folds(cv_runs) -> None: + records, with_mild, _ = cv_runs + held_all = [p for f in with_mild.folds for p in f.held_out_patients] + assert len(held_all) == len(set(held_all)) + assert set(held_all) == {r.patient_id for r in records} + indices = np.concatenate([f.held_out_indices for f in with_mild.folds]) + assert sorted(indices.tolist()) == list(range(len(records))) + + +def test_split_is_same_regardless_of_severities(cv_runs) -> None: + _, with_mild, without_mild = cv_runs + for a, b in zip(with_mild.folds, without_mild.folds, strict=True): + assert a.held_out_patients == b.held_out_patients + assert a.train_patients == b.train_patients + assert a.val_patients == b.val_patients + np.testing.assert_array_equal(a.held_out_indices, b.held_out_indices) + + +def test_held_out_keeps_mild_even_when_excluded_from_training(cv_runs) -> None: + records, _, without_mild = cv_runs + severities = np.array([r.severity for r in records]) + held_sev = np.concatenate([severities[f.held_out_indices] for f in without_mild.folds]) + assert (held_sev == "mild").sum() == (severities == "mild").sum() + for f in without_mild.folds: + held_ids = set(f.held_out_patients) + expected = {i for i, r in enumerate(records) if r.patient_id in held_ids} + assert set(f.held_out_indices.tolist()) == expected + + +def test_severities_applied_only_to_train_and_val(cv_runs) -> None: + _, with_mild, without_mild = cv_runs + for f in without_mild.folds: + for split in ("train", "val"): + mild_counts = {g: n for g, n in f.group_counts[split].items() if g.endswith("_mild")} + assert sum(mild_counts.values()) == 0, split + assert sum(n for g, n in f.group_counts["held_out"].items() if g.endswith("_mild")) > 0 + # ややあり込みでは学習用に mild が入る + assert any( + sum(n for g, n in f.group_counts["train"].items() if g.endswith("_mild")) > 0 + for f in with_mild.folds + ) + + +def test_held_out_probabilities_shape(cv_runs) -> None: + _, with_mild, _ = cv_runs + for f in with_mild.folds: + assert f.held_out_probs.shape == (len(f.held_out_indices), 5) + np.testing.assert_allclose(f.held_out_probs.sum(axis=1), 1.0, atol=1e-4) + assert 1 <= f.best_epoch <= len(f.history) + + +def test_checkpoints_per_fold(cv_runs, tmp_path_factory) -> None: + records, _, _ = cv_runs + # fixture と同じ tmp の下に fold.pt が書かれている + base = tmp_path_factory.getbasetemp() + found = sorted(p.name for p in base.rglob("ck_with/fold*.pt")) + assert found == [f"fold{k}.pt" for k in range(N_FOLDS)]