230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""
|
||
Тесты ручной разметки: проверка вердикта, хранение, наложение и выгрузка.
|
||
|
||
Ключевое свойство, которое здесь фиксируется: файл ручных вердиктов должен
|
||
читаться тем же загрузчиком, что и построенная разметка (`load_labels_csv`),
|
||
иначе его нельзя передать в обучение как `--labels-csv` и вся затея с
|
||
подтверждением специалистом теряет смысл.
|
||
"""
|
||
import csv
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from src.dxa.excel_labels import load_labels_csv
|
||
from src.dxa.labels import QUALITY_BAD, QUALITY_GOOD, ImageRecord
|
||
from src.dxa.manual_labels import (
|
||
MANUAL_FIELDS,
|
||
ManualVerdict,
|
||
VerdictError,
|
||
apply_manual,
|
||
export_rows,
|
||
load_verdicts,
|
||
make_verdict,
|
||
remove_verdict,
|
||
review_progress,
|
||
save_verdicts,
|
||
upsert_verdict,
|
||
validate_verdict,
|
||
)
|
||
|
||
|
||
def _record(path, region="spine", label=QUALITY_GOOD, marker=None, study="S1"):
|
||
return ImageRecord(path=Path(path), study=study, region=region, label=label, marker=marker)
|
||
|
||
|
||
class TestValidateVerdict:
|
||
def test_accepts_good_and_bad(self):
|
||
assert validate_verdict("spine", QUALITY_GOOD, "") == ("spine", 0, "")
|
||
assert validate_verdict("hip_left", QUALITY_BAD, "rotation") == ("hip_left", 1, "rotation")
|
||
|
||
def test_legacy_alias_is_canonicalized(self):
|
||
"""Устаревшие значения внешних источников приводятся к канону, а не отклоняются."""
|
||
assert validate_verdict("spine", 1, "position_error")[2] == "positioning"
|
||
assert validate_verdict("spine", 1, "roi_error")[2] == "roi_incorrect"
|
||
|
||
def test_rejects_unknown_violation_instead_of_silent_unspecified(self):
|
||
with pytest.raises(VerdictError, match="Неизвестный тип нарушения"):
|
||
validate_verdict("spine", 1, "ukladka_plohaya")
|
||
|
||
def test_rejects_region_outside_selectable(self):
|
||
with pytest.raises(VerdictError, match="Область должна быть"):
|
||
validate_verdict("unknown", 0, "")
|
||
|
||
def test_rejects_quality_outside_binary(self):
|
||
with pytest.raises(VerdictError, match="0 или 1"):
|
||
validate_verdict("spine", 2, "")
|
||
|
||
def test_rejects_violation_for_good_image(self):
|
||
with pytest.raises(VerdictError, match="не может быть типа нарушения"):
|
||
validate_verdict("spine", 0, "artifact")
|
||
|
||
def test_rejects_violation_from_other_region(self):
|
||
"""Ротация — критерий бедра; на позвоночнике это признак ошибки ввода."""
|
||
with pytest.raises(VerdictError, match="относится к области"):
|
||
validate_verdict("spine", 1, "rotation")
|
||
|
||
def test_hip_scope_covers_both_hips(self):
|
||
"""В словаре нарушений бедро — один код `hip`, а область всегда конкретная."""
|
||
assert validate_verdict("hip_right", 1, "rotation")[2] == "rotation"
|
||
assert validate_verdict("hip_left", 1, "rotation")[2] == "rotation"
|
||
with pytest.raises(VerdictError, match="относится к области"):
|
||
validate_verdict("hip_right", 1, "axis_deviation")
|
||
|
||
|
||
class TestStore:
|
||
def test_missing_file_is_empty(self, tmp_path):
|
||
assert load_verdicts(tmp_path / "manual_labels.csv") == {}
|
||
|
||
def test_roundtrip_and_upsert(self, tmp_path):
|
||
store = tmp_path / "manual_labels.csv"
|
||
upsert_verdict(store, make_verdict("/s/a.dcm", "spine", 1, "artifact", comment="мусор"))
|
||
upsert_verdict(store, make_verdict("/s/a.dcm", "spine", 0, "", reviewer="Иванов"))
|
||
|
||
verdicts = load_verdicts(store)
|
||
assert len(verdicts) == 1
|
||
assert verdicts["/s/a.dcm"].quality == QUALITY_GOOD
|
||
assert verdicts["/s/a.dcm"].reviewer == "Иванов"
|
||
|
||
def test_remove_verdict_returns_image_to_built_labels(self, tmp_path):
|
||
store = tmp_path / "manual_labels.csv"
|
||
upsert_verdict(store, make_verdict("/s/a.dcm", "spine", 1, "artifact"))
|
||
upsert_verdict(store, make_verdict("/s/b.dcm", "spine", 1, "artifact"))
|
||
|
||
assert set(remove_verdict(store, "/s/a.dcm")) == {"/s/b.dcm"}
|
||
with pytest.raises(VerdictError, match="Нет ручного вердикта"):
|
||
remove_verdict(store, "/s/a.dcm")
|
||
|
||
def test_file_has_declared_columns(self, tmp_path):
|
||
store = tmp_path / "manual_labels.csv"
|
||
upsert_verdict(store, make_verdict("/s/a.dcm", "spine", 1, "artifact"))
|
||
with store.open(newline="", encoding="utf-8") as fh:
|
||
assert tuple(next(csv.reader(fh))) == MANUAL_FIELDS
|
||
|
||
def test_reads_file_saved_by_excel_with_bom(self, tmp_path):
|
||
"""Excel сохраняет CSV с BOM; разметку после ручной правки надо читать."""
|
||
store = tmp_path / "manual_labels.csv"
|
||
body = (
|
||
"path_to_image,anatomical_region,quality_class,violation_type,comment,reviewer,reviewed_at\n"
|
||
"/s/a.dcm,spine,1,artifact,,,\n"
|
||
)
|
||
store.write_bytes(b"\xef\xbb\xbf" + body.encode("utf-8"))
|
||
assert set(load_verdicts(store)) == {"/s/a.dcm"}
|
||
|
||
def test_corrupt_row_is_skipped_not_fatal(self, tmp_path):
|
||
store = tmp_path / "manual_labels.csv"
|
||
store.write_text(
|
||
"path_to_image,anatomical_region,quality_class,violation_type,comment,reviewer,reviewed_at\n"
|
||
"/s/ok.dcm,spine,1,artifact,,,\n"
|
||
"/s/bad_region.dcm,hips,1,artifact,,,\n"
|
||
"/s/bad_quality.dcm,spine,7,artifact,,,\n",
|
||
encoding="utf-8",
|
||
)
|
||
assert set(load_verdicts(store)) == {"/s/ok.dcm"}
|
||
|
||
def test_readable_by_training_loader(self, tmp_path):
|
||
"""Файл вердиктов должен приниматься обучением как `--labels-csv`."""
|
||
store = tmp_path / "manual_labels.csv"
|
||
upsert_verdict(store, make_verdict("/s/a.dcm", "spine", 1, "artifact"))
|
||
upsert_verdict(store, make_verdict("/s/b.dcm", "hip_left", 0, ""))
|
||
|
||
table = load_labels_csv(store)
|
||
assert table["/s/a.dcm"]["quality_class"] == "1"
|
||
assert table["/s/b.dcm"]["anatomical_region"] == "hip_left"
|
||
|
||
|
||
class TestOverlay:
|
||
def test_overrides_only_reviewed_images(self):
|
||
records = [
|
||
_record("/s/a.dcm", label=QUALITY_GOOD),
|
||
_record("/s/b.dcm", label=QUALITY_BAD, marker="bad"),
|
||
]
|
||
verdicts = {"/s/a.dcm": make_verdict("/s/a.dcm", "hip_right", 1, "rotation")}
|
||
|
||
updated = apply_manual(records, verdicts)
|
||
assert [(r.label, r.region) for r in updated] == [(1, "hip_right"), (1, "spine")]
|
||
assert updated[1].region == "spine"
|
||
assert updated[1].marker == "bad"
|
||
|
||
def test_partial_review_keeps_other_images_intact(self):
|
||
records = [_record(f"/s/{i}.dcm") for i in range(5)]
|
||
updated = apply_manual(records, {"/s/2.dcm": make_verdict("/s/2.dcm", "spine", 1, "artifact")})
|
||
assert [r.label for r in updated] == [0, 0, 1, 0, 0]
|
||
|
||
|
||
class TestProgress:
|
||
def test_counts_reviewed_changed_and_conflicts(self):
|
||
records = [
|
||
_record("/s/a.dcm", region="spine", label=QUALITY_GOOD),
|
||
_record("/s/b.dcm", region="spine", label=QUALITY_GOOD, marker="bad"),
|
||
_record("/s/c.dcm", region="hip_left", label=QUALITY_BAD, marker="bad"),
|
||
]
|
||
verdicts = {"/s/a.dcm": make_verdict("/s/a.dcm", "spine", 0, "")}
|
||
|
||
progress = review_progress(records, verdicts)
|
||
assert progress["total"] == 3
|
||
assert progress["reviewed"] == 1
|
||
assert progress["remaining"] == 2
|
||
assert progress["confirmed"] == 1
|
||
assert progress["changed"] == 0
|
||
# /s/b.dcm: разметка «годное», пометка в имени «bad»
|
||
assert progress["filename_conflict"] == 1
|
||
assert progress["by_region"]["spine"] == {"total": 2, "reviewed": 1}
|
||
assert progress["by_region"]["hip_left"] == {"total": 1, "reviewed": 0}
|
||
|
||
def test_changed_counts_label_corrections(self):
|
||
records = [_record("/s/a.dcm", label=QUALITY_GOOD)]
|
||
verdicts = {"/s/a.dcm": make_verdict("/s/a.dcm", "spine", 1, "artifact")}
|
||
progress = review_progress(records, verdicts)
|
||
assert progress["changed"] == 1
|
||
assert progress["confirmed"] == 0
|
||
|
||
|
||
class TestExport:
|
||
def test_scope_reviewed_contains_only_reviewed(self):
|
||
records = [_record("/s/a.dcm"), _record("/s/b.dcm", label=QUALITY_BAD)]
|
||
verdicts = {"/s/a.dcm": make_verdict("/s/a.dcm", "spine", 1, "artifact", reviewer="Иванов")}
|
||
|
||
rows = export_rows(records, verdicts, scope="reviewed")
|
||
assert len(rows) == 1
|
||
assert rows[0]["label_rule"] == "manual"
|
||
assert rows[0]["quality_class"] == "1"
|
||
assert rows[0]["reviewer"] == "Иванов"
|
||
|
||
def test_scope_all_merges_and_marks_origin(self):
|
||
records = [_record("/s/a.dcm", label=QUALITY_BAD), _record("/s/b.dcm")]
|
||
verdicts = {}
|
||
|
||
rows = export_rows(records, verdicts, scope="all", expert_paths=["/s/a.dcm"])
|
||
rules = {r["path_to_image"]: r["label_rule"] for r in rows}
|
||
assert rules == {"/s/a.dcm": "table", "/s/b.dcm": "filename"}
|
||
|
||
def test_unknown_scope_rejected(self):
|
||
with pytest.raises(VerdictError, match="scope должен быть"):
|
||
export_rows([], {}, scope="everything")
|
||
|
||
def test_missing_dicom_does_not_break_export(self):
|
||
"""Пути вне датасета (или битый DICOM) дают пустые UID, а не исключение."""
|
||
rows = export_rows([_record("/no/such/file.dcm")], {}, scope="all")
|
||
assert rows[0]["study_uid"] == ""
|
||
assert rows[0]["image_uid"] == ""
|
||
|
||
def test_empty_verdict_is_not_confused_with_good(self):
|
||
"""Словарь вердиктов пуст — строка всё равно описывает метку из разметки."""
|
||
rows = export_rows([_record("/s/a.dcm", label=QUALITY_BAD)], {}, scope="all")
|
||
assert rows[0]["quality_class"] == "1"
|
||
assert rows[0]["violation_type"] == ""
|
||
assert rows[0]["reviewed_at"] == ""
|
||
|
||
|
||
def test_verdict_carries_timestamp():
|
||
verdict: ManualVerdict = make_verdict("/s/a.dcm", "spine", 1, "artifact")
|
||
assert verdict.reviewed_at.endswith("+00:00")
|
||
|
||
|
||
def test_saving_empty_store_creates_header(tmp_path):
|
||
store = tmp_path / "manual_labels.csv"
|
||
save_verdicts(store, {})
|
||
assert store.read_text(encoding="utf-8").startswith("path_to_image,")
|
||
assert load_verdicts(store) == {}
|