Newer
Older
TIASjudge / tests / core / test_svm.py
"""`core/svm.py` の検証(SPEC_02 (3)「SVM 分類」・「診断失敗の判定」・PLAN_02 ステップ 4).

仕様上の要件は以下.

- 出力 190 列(医師 38 名 × 5 色)それぞれについて 2 クラス SVM を診断のたびに学習し,
  入力 L\\*a\\*b\\* を予測する(正例が 2 未満の列は学習せずスコア 0)
- SVM パラメータは C_SVC / RBF・γ=0.011・C=12.0・終了条件 EPS 1000 回・FLT_EPSILON
- 色ごとに 38 医師分の予測値を平均 → 5 色の一致率(0〜1)を百分率で表示する
- スコア降順で上位 2 色を主要ラベルとする
- 学習不能・予測失敗は直接検出して「診断失敗」として返す(旧実装のマジックナンバー
  比較 42/0/5/11/3 は移植しない.ただし表示仕様=失敗判定の結果は同じであること)

旧実装との結果同一性が成功基準(PLAN_01)のため,実 `data/data.csv` を用いた
入出力ペアを回帰テストとして固定する(testing.md「回帰テストを最重視する」).
"""

from __future__ import annotations

from pathlib import Path

import cv2
import numpy as np
import pytest

from tiasjudge.core.color import Lab, rgb_to_lab
from tiasjudge.core.csv_io import (
    COLOR_COUNT,
    DOCTOR_COUNT,
    EVALUATION_COLUMN_COUNT,
    SAMPLE_COUNT,
    TrainingData,
    load_training_data,
)
from tiasjudge.core.roi import mean_rgb
from tiasjudge.core.svm import (
    COLOR_KEYS,
    MIN_POSITIVE_SAMPLES,
    SVM_C,
    SVM_GAMMA,
    DiagnosisResult,
    FailureReason,
    create_model,
    diagnose,
)

#: 旧実装が「Diagnosis failed」と表示するスコア組(SPEC_02「診断失敗の判定」).
#: 実測でこれは Lab=(-16, 0, 0)(XYZ が全て 0)の出力であることを確認している
FAILED_PERCENTAGES = {"purple": 42, "dark_red": 0, "crimson": 5, "pink": 11, "pale": 3}
DEGENERATE_LAB = Lab(l_star=-16.0, a_star=0.0, b_star=0.0)

#: 実 data.csv での回帰値(入力 Lab -> (5 色の %, 上位 2 色))
REGRESSION_CASES: dict[str, tuple[tuple[float, float, float], dict[str, int], tuple[str, str]]] = {
    "purple_dark_red": (
        (60.0, 15.0, 10.0),
        {"purple": 58, "dark_red": 37, "crimson": 0, "pink": 0, "pale": 0},
        ("purple", "dark_red"),
    ),
    "dark_red_crimson": (
        (50.0, 20.0, 5.0),
        {"purple": 5, "dark_red": 55, "crimson": 21, "pink": 8, "pale": 3},
        ("dark_red", "crimson"),
    ),
    "crimson_pink": (
        (45.0, 25.0, -5.0),
        {"purple": 21, "dark_red": 8, "crimson": 29, "pink": 24, "pale": 5},
        ("crimson", "pink"),
    ),
    "high_lightness": (
        (65.0, 20.0, 8.0),
        {"purple": 53, "dark_red": 45, "crimson": 5, "pink": 0, "pale": 3},
        ("purple", "dark_red"),
    ),
}


# --- フィクスチャ(SVM 学習は 1 診断あたり 190 列なので結果を使い回す) ---


@pytest.fixture(scope="module")
def training(data_csv_path: Path) -> TrainingData:
    """同梱の実学習データ(84 サンプル × 190 列)."""
    return load_training_data(data_csv_path)


@pytest.fixture(scope="module")
def results(training: TrainingData) -> dict[str, DiagnosisResult]:
    """回帰ケースの診断結果(モジュール内で共有する)."""
    return {name: diagnose(lab, training) for name, (lab, _, _) in REGRESSION_CASES.items()}


@pytest.fixture(scope="module")
def degenerate_result(training: TrainingData) -> DiagnosisResult:
    """旧実装が「Diagnosis failed」を表示する入力(Lab=(-16, 0, 0))の診断結果."""
    return diagnose(DEGENERATE_LAB, training)


# --- 合成学習データのヘルパー(正例数の条件を検証するために使う) ---


def synthetic_training(positive_columns: dict[int, int] | None = None) -> TrainingData:
    """仕様どおりの形(84 サンプル × 190 列)の合成学習データを作る.

    既定では全列の正例が 1 個(= `MIN_POSITIVE_SAMPLES` 未満)になる.
    `positive_columns` に {列: 正例数} を渡すと,その列だけ先頭サンプル群を正例にする
    (先頭 42 サンプルは Lab クラスタ A,後半 42 サンプルはクラスタ B).
    """
    half = SAMPLE_COUNT // 2
    lab = np.empty((SAMPLE_COUNT, 3), dtype=np.float32)
    lab[:half] = np.array([60.0, 20.0, 10.0], dtype=np.float32)
    lab[half:] = np.array([45.0, 5.0, 0.0], dtype=np.float32)
    # サンプルが完全重複しないように微小なばらつきを与える
    lab += (np.arange(SAMPLE_COUNT, dtype=np.float32) * 0.01)[:, np.newaxis]

    evaluations = np.zeros((SAMPLE_COUNT, EVALUATION_COLUMN_COUNT), dtype=np.int32)
    evaluations[0, :] = 1  # 全列に正例 1 個(学習対象外の状態)
    for column, count in (positive_columns or {}).items():
        evaluations[:count, column] = 1
    return TrainingData(lab=lab, evaluations=evaluations)


CLUSTER_A_LAB = (60.0, 20.0, 10.0)


# --- SVM パラメータ(SPEC_02 (3)) ---


def test_svm_parameters_match_the_spec() -> None:
    """γ=0.011・C=12.0 は仕様どおりの値である."""
    assert SVM_GAMMA == 0.011
    assert SVM_C == 12.0


def test_model_uses_c_svc_with_rbf_kernel() -> None:
    """生成されるモデルは C_SVC・RBF カーネルで,γ・C が設定されている."""
    model = create_model()

    assert model.getType() == cv2.ml.SVM_C_SVC
    assert model.getKernelType() == cv2.ml.SVM_RBF
    assert model.getGamma() == pytest.approx(0.011)
    assert model.getC() == pytest.approx(12.0)


def test_termination_criteria_uses_eps() -> None:
    """終了条件は EPS(FLT_EPSILON)・反復 1000 である(SPEC_02 (3))."""
    criteria_type, max_count, epsilon = create_model().getTermCriteria()

    assert criteria_type & cv2.TERM_CRITERIA_EPS
    assert max_count == 1000
    assert epsilon == pytest.approx(float(np.finfo(np.float32).eps))


def test_minimum_positive_samples_is_two() -> None:
    """学習に必要な正例数は 2(これ未満は学習せずスコア 0)."""
    assert MIN_POSITIVE_SAMPLES == 2


def test_color_keys_follow_the_csv_column_order() -> None:
    """5 色のキーは data.csv の列順(紫・暗赤・紅・淡紅・淡白)に対応する."""
    assert COLOR_KEYS == ("purple", "dark_red", "crimson", "pink", "pale")
    assert len(COLOR_KEYS) == COLOR_COUNT


# --- 実データでの回帰(PLAN_01 成功基準) ---


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_percentages_match_the_reference_values(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """既知入力に対する 5 色の百分率が固定値と一致する(旧実装との同一性の回帰)."""
    _, expected, _ = REGRESSION_CASES[name]

    assert results[name].percentages == expected


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_top_labels_match_the_reference_values(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """上位 2 色が固定値と一致する."""
    _, _, expected = REGRESSION_CASES[name]

    assert results[name].top_labels == expected


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_valid_input_is_not_reported_as_failure(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """学習・予測が成立する入力は診断失敗にならない."""
    result = results[name]

    assert result.failed is False
    assert result.failure_reason is None


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_top_labels_are_the_two_highest_scores(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """上位 2 色はスコア降順の先頭 2 件である(SPEC_02 (3))."""
    scores = results[name].scores
    first, second = results[name].top_labels
    others = [key for key in COLOR_KEYS if key not in (first, second)]

    assert scores[first] >= scores[second]
    assert all(scores[second] >= scores[key] for key in others)


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_scores_are_ratios_of_doctor_agreement(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """スコアは 38 医師の予測の平均(0〜1・分母 38 の分数)である."""
    for key, score in results[name].scores.items():
        assert 0.0 <= score <= 1.0, key
        assert score * DOCTOR_COUNT == pytest.approx(round(score * DOCTOR_COUNT), abs=1e-4), key


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_percentages_are_rounded_scores(name: str, results: dict[str, DiagnosisResult]) -> None:
    """百分率は一致率を 100 倍して整数に丸めた値である."""
    result = results[name]

    for key, score in result.scores.items():
        assert isinstance(result.percentages[key], int)
        assert abs(result.percentages[key] - score * 100.0) <= 0.5 + 1e-6, key


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_predictions_are_binary_class_labels(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """予測値は 190 列分の 0/1(2 クラス SVM の出力)である."""
    predictions = results[name].predictions

    assert predictions.shape == (EVALUATION_COLUMN_COUNT,)
    assert set(np.unique(predictions).tolist()) <= {0.0, 1.0}


@pytest.mark.parametrize("name", list(REGRESSION_CASES))
def test_scores_are_the_mean_of_each_color_columns(
    name: str, results: dict[str, DiagnosisResult]
) -> None:
    """各色のスコアは,その色に対応する 38 列の予測の平均である."""
    result = results[name]

    for color_index, key in enumerate(COLOR_KEYS):
        expected = float(result.predictions[color_index::COLOR_COUNT].sum()) / DOCTOR_COUNT
        assert result.scores[key] == pytest.approx(expected, abs=1e-6), key


def test_diagnosis_is_deterministic(training: TrainingData) -> None:
    """同じ入力・同じ学習データなら結果は毎回同じ(testing.md: 乱数に依存させない)."""
    first = diagnose((50.0, 20.0, 5.0), training)
    second = diagnose((50.0, 20.0, 5.0), training)

    assert first.scores == second.scores
    assert first.top_labels == second.top_labels
    assert np.array_equal(first.predictions, second.predictions)


def test_columns_with_few_positives_are_skipped(
    training: TrainingData, results: dict[str, DiagnosisResult]
) -> None:
    """正例が 2 未満の列は学習せず,予測値 0 のままになる(SPEC_02 (3))."""
    expected_skipped = tuple(
        column
        for column in range(EVALUATION_COLUMN_COUNT)
        if int((training.evaluations[:, column] > 0).sum()) < MIN_POSITIVE_SAMPLES
    )
    result = results["purple_dark_red"]

    assert result.skipped_columns == expected_skipped
    assert all(result.predictions[column] == 0.0 for column in expected_skipped)


def test_no_column_fails_to_train_with_the_bundled_data(
    results: dict[str, DiagnosisResult],
) -> None:
    """同梱データでは学習可能な全列が学習・予測に成功する."""
    assert results["purple_dark_red"].failed_columns == ()


def test_skipped_column_count_is_fixed_for_the_bundled_data(
    results: dict[str, DiagnosisResult],
) -> None:
    """同梱データで学習されない列はちょうど 20 列である(回帰値).

    190 列のうち 20 列(正例 2 未満)が学習対象外なので,学習される列は 170 列になる.
    data.csv を差し替えた際に気付けるよう固定値で押さえる.
    """
    result = results["purple_dark_red"]

    assert len(result.skipped_columns) == 20
    assert EVALUATION_COLUMN_COUNT - len(result.skipped_columns) == 170


def test_predictions_are_float32(results: dict[str, DiagnosisResult]) -> None:
    """予測値は旧実装の `float result[190]` と同じ単精度で保持する."""
    assert results["purple_dark_red"].predictions.dtype == np.float32


def test_training_data_is_not_modified_by_diagnosis(training: TrainingData) -> None:
    """診断は学習データを書き換えない(同じ `TrainingData` を繰り返し使えること)."""
    lab_before = training.lab.copy()
    evaluations_before = training.evaluations.copy()

    diagnose((50.0, 20.0, 5.0), training)

    assert np.array_equal(training.lab, lab_before)
    assert np.array_equal(training.evaluations, evaluations_before)


def test_model_is_retrained_for_every_diagnosis(training: TrainingData) -> None:
    """SVM は診断のたびに学習する(SPEC_02 (3)).

    学習データを差し替えると結果も変わることで,学習結果を使い回していないことを示す.
    ここでは医師 1 名目の紫(列 0)だけを残し,他の列は評価なし(学習対象外)にする.
    """
    modified_evaluations = np.zeros_like(training.evaluations)
    modified_evaluations[:, 0] = training.evaluations[:, 0]
    modified = TrainingData(lab=training.lab.copy(), evaluations=modified_evaluations)

    baseline = diagnose((50.0, 20.0, 5.0), training)
    retrained = diagnose((50.0, 20.0, 5.0), modified)

    assert retrained.scores != baseline.scores
    # 列ごとに独立に学習するので,残した列 0 の予測は元の学習データのときと変わらない
    assert retrained.predictions[0] == baseline.predictions[0]
    for key in ("dark_red", "crimson", "pink", "pale"):
        assert retrained.scores[key] == 0.0, key


# --- 診断失敗の判定(SPEC_02「診断失敗の判定」) ---


def test_degenerate_input_reproduces_the_legacy_failure_percentages(
    degenerate_result: DiagnosisResult,
) -> None:
    """旧実装が失敗と判定するスコア組(紫 42・暗赤 0・紅 5・淡紅 11・淡白 3)を再現する."""
    assert degenerate_result.percentages == FAILED_PERCENTAGES


def test_degenerate_input_is_reported_as_failure(degenerate_result: DiagnosisResult) -> None:
    """L* が 0 以下の入力は診断失敗として直接検出する(マジックナンバー比較を使わない)."""
    assert degenerate_result.failed is True
    assert degenerate_result.failure_reason is FailureReason.DEGENERATE_INPUT


def test_failure_result_still_carries_scores(degenerate_result: DiagnosisResult) -> None:
    """失敗時もスコアは返す(旧実装は失敗表示に重ねてスコアを描画する)."""
    assert sum(degenerate_result.scores.values()) > 0.0
    assert degenerate_result.lab == DEGENERATE_LAB


def test_failure_reason_has_japanese_description() -> None:
    """失敗理由は GUI 表示のための説明文を持つ(core は表示自体は行わない)."""
    for reason in FailureReason:
        assert isinstance(reason.describe(), str)
        assert reason.describe() != ""


@pytest.mark.parametrize(
    "lab",
    [(float("nan"), 0.0, 0.0), (float("inf"), 10.0, 5.0), (50.0, float("nan"), 5.0)],
)
def test_non_finite_input_is_reported_as_failure(
    lab: tuple[float, float, float], training: TrainingData
) -> None:
    """NaN / inf を含む入力は予測できないため診断失敗にする."""
    result = diagnose(lab, training)

    assert result.failed is True
    assert result.failure_reason is FailureReason.INVALID_INPUT
    assert result.percentages == dict.fromkeys(COLOR_KEYS, 0)


def test_no_trainable_column_is_reported_as_failure() -> None:
    """全列の正例が 2 未満なら学習できず,診断失敗になる(スコアは全て 0)."""
    result = diagnose(CLUSTER_A_LAB, synthetic_training())

    assert result.failed is True
    assert result.failure_reason is FailureReason.NO_TRAINABLE_COLUMN
    assert result.scores == dict.fromkeys(COLOR_KEYS, 0.0)
    assert result.percentages == dict.fromkeys(COLOR_KEYS, 0)
    assert len(result.skipped_columns) == EVALUATION_COLUMN_COUNT


def test_all_zero_scores_use_the_fixed_tie_break_order() -> None:
    """全色が同点のときの上位 2 色は旧実装のキー順(紅・暗赤)になる."""
    result = diagnose(CLUSTER_A_LAB, synthetic_training())

    assert result.top_labels == ("crimson", "dark_red")


def test_tied_non_zero_scores_use_the_fixed_tie_break_order() -> None:
    """同点の色が複数あるときも旧実装のキー順(C < D < Pi < Pu < W)で順位が決まる.

    旧実装は `std::map<CString, float>`(キー昇順)を降順ソートするため,
    同点の紫(Pu)と紅(C)では紅が先に並ぶ.紫・紅の 2 色だけを 1/38 で同点にする.
    """
    training = synthetic_training({0: SAMPLE_COUNT // 2, 2: SAMPLE_COUNT // 2})

    result = diagnose(CLUSTER_A_LAB, training)

    assert result.scores["purple"] == result.scores["crimson"] == pytest.approx(1.0 / DOCTOR_COUNT)
    assert result.top_labels == ("crimson", "purple")


def test_zero_lightness_is_reported_as_failure(training: TrainingData) -> None:
    """L* = 0(Y が 0 にクリップされた状態)も色変換が成立しておらず失敗である(境界値)."""
    result = diagnose((0.0, 0.0, 0.0), training)

    assert result.failed is True
    assert result.failure_reason is FailureReason.DEGENERATE_INPUT


def test_legacy_failure_pattern_is_always_reported_as_failure(training: TrainingData) -> None:
    """旧実装が失敗表示するスコア組になる入力は,L* > 0 でも診断失敗とする(SPEC_02).

    SPEC_02「診断失敗の判定」は失敗の検出方法だけを変え,表示仕様は現行と同じとしている.
    旧実装(`ResultDlg.cpp`)はスコア組のみで判定するため,学習データから外れた入力
    (例: L* = 5)でも「Diagnosis failed」になる.
    """
    result = diagnose((5.0, 0.0, 0.0), training)

    assert result.percentages == FAILED_PERCENTAGES
    assert result.failed is True


# --- 正例数の条件と色の割り当て(SPEC_02 (3)) ---


@pytest.mark.parametrize(
    ("column", "expected_key"),
    [(0, "purple"), (1, "dark_red"), (2, "crimson"), (3, "pink"), (4, "pale"), (7, "crimson")],
)
def test_trained_column_contributes_to_its_color(column: int, expected_key: str) -> None:
    """学習された列の予測は,列インデックス % 5 の色に加算される(医師 5 列の one-hot).

    他の 189 列は正例 1 個で学習されないため,スコアは対象色のみ 1/38 になる.
    """
    training = synthetic_training({column: SAMPLE_COUNT // 2})

    result = diagnose(CLUSTER_A_LAB, training)

    assert result.failed is False
    assert result.predictions[column] == 1.0
    assert result.scores[expected_key] == pytest.approx(1.0 / DOCTOR_COUNT)
    for key in COLOR_KEYS:
        if key != expected_key:
            assert result.scores[key] == 0.0, key
    assert result.top_labels[0] == expected_key


def test_column_with_exactly_two_positives_is_trained() -> None:
    """正例がちょうど 2 個の列は学習対象になる(境界値)."""
    training = synthetic_training({0: MIN_POSITIVE_SAMPLES})

    result = diagnose(CLUSTER_A_LAB, training)

    assert 0 not in result.skipped_columns
    assert len(result.skipped_columns) == EVALUATION_COLUMN_COUNT - 1
    assert result.failure_reason is not FailureReason.NO_TRAINABLE_COLUMN


def test_column_with_one_positive_is_skipped() -> None:
    """正例が 1 個の列は学習されずスコア 0 のままである(境界値)."""
    training = synthetic_training({0: SAMPLE_COUNT // 2})

    result = diagnose(CLUSTER_A_LAB, training)

    assert tuple(range(1, EVALUATION_COLUMN_COUNT)) == result.skipped_columns
    assert np.array_equal(result.predictions[1:], np.zeros(EVALUATION_COLUMN_COUNT - 1))


# --- 入力の受け取り方 ---


def test_accepts_lab_dataclass_and_tuple(training: TrainingData) -> None:
    """`Lab` でも (L*, a*, b*) の 3 要素でも同じ結果になる."""
    lab = Lab(l_star=50.0, a_star=20.0, b_star=5.0)

    from_dataclass = diagnose(lab, training)
    from_tuple = diagnose((50.0, 20.0, 5.0), training)

    assert from_dataclass.lab == from_tuple.lab == lab
    assert from_dataclass.scores == from_tuple.scores


@pytest.mark.parametrize("values", [(50.0, 20.0), (50.0, 20.0, 5.0, 1.0), ()])
def test_wrong_number_of_components_raises(
    values: tuple[float, ...], training: TrainingData
) -> None:
    """L*a*b* は 3 要素でなければならない."""
    with pytest.raises(ValueError):
        diagnose(values, training)


# --- color.py との結合(SPEC_02 のパイプライン) ---


def linear_coefficients() -> np.ndarray:
    """線形項のみの変換係数(sRGB → XYZ 相当を 0〜255 スケールに合わせたもの)."""
    coefficients = np.zeros((17, 3), dtype=np.float64)
    coefficients[0] = (0.4124, 0.2126, 0.0193)
    coefficients[1] = (0.3576, 0.7152, 0.1192)
    coefficients[2] = (0.1805, 0.0722, 0.9505)
    return coefficients * (100.0 / 255.0)


def test_roi_to_lab_to_diagnosis_pipeline(training: TrainingData) -> None:
    """ROI 平均 RGB → XYZ → Lab → SVM 分類のパイプラインが連結する(SPEC_02 冒頭).

    ROI(51×51)が単色なのでその色がそのまま診断入力になり,同じ色の画素値を
    直接 `rgb_to_lab` に渡した場合と結果が一致する.
    """
    bgr = (110, 110, 180)  # BGR 並び(= RGB (180, 110, 110))
    image = np.empty((200, 200, 3), dtype=np.uint8)
    image[:, :] = bgr

    rgb = mean_rgb(image, [(60, 70), (140, 130)])
    lab = rgb_to_lab(rgb, linear_coefficients())
    result = diagnose(lab, training)

    assert rgb == (180.0, 110.0, 110.0)
    assert result.lab == rgb_to_lab((180.0, 110.0, 110.0), linear_coefficients())
    assert set(result.scores) == set(COLOR_KEYS)
    assert result.top_labels[0] in COLOR_KEYS


def test_lab_from_rgb_conversion_can_be_diagnosed(training: TrainingData) -> None:
    """`rgb_to_lab` の出力をそのまま `diagnose` に渡せる(ROI 平均 RGB → 分類の連結)."""
    # 線形項のみの変換係数(sRGB → XYZ 相当を 0〜255 スケールに合わせたもの)
    coefficients = np.zeros((17, 3), dtype=np.float64)
    coefficients[0] = (0.4124, 0.2126, 0.0193)
    coefficients[1] = (0.3576, 0.7152, 0.1192)
    coefficients[2] = (0.1805, 0.0722, 0.9505)
    coefficients *= 100.0 / 255.0

    lab = rgb_to_lab((180.0, 110.0, 110.0), coefficients)
    result = diagnose(lab, training)

    assert result.lab == lab
    assert lab.l_star > 0.0
    assert result.failed is False
    assert set(result.scores) == set(COLOR_KEYS)
    assert all(0.0 <= score <= 1.0 for score in result.scores.values())