281 lines
12 KiB
Python
281 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from PIL import Image
|
|
|
|
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
|
|
from copienator.annotation_data import AnnotationLoadResult, _coordinate_index
|
|
from copienator.commands import reading_grouped_annotations as reader
|
|
|
|
|
|
class RefaireTests(unittest.TestCase):
|
|
def workspace(self, root, mode):
|
|
for name in ("Copies", "Par label", mode, "BRnot/Copie01"):
|
|
(root / name).mkdir(parents=True)
|
|
(root / "labels").write_text("Ex 1\nEx 2\n")
|
|
atomic_write_json(root / "correction.json", {})
|
|
atomic_write_json(root / "refaire.json", [["Copie01", ["Ex 2"]]])
|
|
for name in (
|
|
"bnote.json",
|
|
"checkboxes.json",
|
|
"Reference.jpg",
|
|
"Concat_annotated.pdf",
|
|
):
|
|
(root / "BRnot/Copie01" / name).touch()
|
|
atomic_write_json(
|
|
root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 2"}]}
|
|
)
|
|
return EvaluationWorkspace(root)
|
|
|
|
def test_reader_merges_full_copy_for_every_original_mode(self):
|
|
for mode in ("BGnot", "Bnot", "Anot"):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
|
|
workspace = self.workspace(Path(tmp), mode)
|
|
data = {"01": {"Ex 1": {}, "Ex 2": {}}, "02": {"Ex 1": {}}}
|
|
|
|
def load(_workspace, data=data, **kwargs):
|
|
return AnnotationLoadResult(
|
|
{"01": {"Ex 2": {}}} if kwargs else data, []
|
|
)
|
|
|
|
with (
|
|
patch.object(reader, "load_annotation_data", side_effect=load),
|
|
patch.object(
|
|
reader,
|
|
"_scan_annotation_directory",
|
|
return_value=(
|
|
{"01": [{"label": "Ex 2", "type": "score", "value": 3}]},
|
|
{},
|
|
),
|
|
),
|
|
patch.object(
|
|
reader,
|
|
"apply_actions_and_regenerate_grouped",
|
|
return_value=(ExitCode.SUCCESS, ""),
|
|
) as render,
|
|
):
|
|
self.assertEqual(
|
|
reader.run(workspace, refaire=True, annotation_dir=mode),
|
|
ExitCode.SUCCESS,
|
|
)
|
|
render.assert_called_once()
|
|
self.assertEqual(set(render.call_args.args[1]["01"]), {"Ex 1", "Ex 2"})
|
|
self.assertEqual(render.call_args.args[2], "01")
|
|
self.assertEqual(render.call_args.kwargs["selected_labels"], {"Ex 2"})
|
|
self.assertEqual(render.call_args.kwargs["annotation_dir"], mode)
|
|
|
|
def test_whole_copy_selection_replaces_all_old_actions(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
workspace = self.workspace(root, "BGnot")
|
|
(root / "BGnot/Ex 1").mkdir()
|
|
atomic_write_json(root / "refaire.json", [["Copie01", []]])
|
|
atomic_write_json(
|
|
root / "BRnot/Copie01/bnote.json",
|
|
{"images": [{"label": label} for label in ("Ex 1", "Ex 2")]},
|
|
)
|
|
loaded = AnnotationLoadResult({"01": {"Ex 1": {}, "Ex 2": {}}}, [])
|
|
|
|
def scan(directory, *args, **kwargs):
|
|
return (
|
|
({"01": [{"label": "Ex 1", "type": "score", "value": 1}]}, {})
|
|
if directory.parent.name == "BGnot"
|
|
else ({"01": [{"label": "Ex 2", "type": "score", "value": 4}]}, {})
|
|
)
|
|
|
|
with (
|
|
patch.object(reader, "load_annotation_data", return_value=loaded),
|
|
patch.object(reader, "_scan_annotation_directory", side_effect=scan),
|
|
patch.object(
|
|
reader,
|
|
"apply_actions_and_regenerate_grouped",
|
|
return_value=(ExitCode.SUCCESS, ""),
|
|
) as render,
|
|
):
|
|
self.assertEqual(reader.run(workspace, refaire=True), ExitCode.SUCCESS)
|
|
self.assertEqual(
|
|
render.call_args.kwargs["selected_labels"], {"Ex 1", "Ex 2"}
|
|
)
|
|
self.assertEqual(
|
|
render.call_args.args[3],
|
|
[{"label": "Ex 2", "type": "score", "value": 4}],
|
|
)
|
|
|
|
def test_stale_redo_selection_is_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
workspace = self.workspace(root, "Anot")
|
|
atomic_write_json(
|
|
root / "BRnot/Copie01/bnote.json", {"images": [{"label": "Ex 1"}]}
|
|
)
|
|
with (
|
|
patch.object(
|
|
reader,
|
|
"load_annotation_data",
|
|
return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []),
|
|
),
|
|
patch.object(reader, "apply_actions_and_regenerate_grouped") as render,
|
|
):
|
|
self.assertEqual(
|
|
reader.run(workspace, refaire=True, annotation_dir="Anot"),
|
|
ExitCode.PARTIAL,
|
|
)
|
|
render.assert_not_called()
|
|
|
|
def test_missing_redo_return_leaves_copy_untouched(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
workspace = self.workspace(root, "BGnot")
|
|
(root / "BRnot/Copie01/Concat_annotated.pdf").unlink()
|
|
with (
|
|
patch.object(
|
|
reader,
|
|
"load_annotation_data",
|
|
return_value=AnnotationLoadResult({"01": {"Ex 2": {}}}, []),
|
|
),
|
|
patch.object(reader, "apply_actions_and_regenerate_grouped") as render,
|
|
):
|
|
self.assertEqual(reader.run(workspace, refaire=True), ExitCode.PARTIAL)
|
|
render.assert_not_called()
|
|
|
|
def test_regeneration_preserves_untouched_image_and_score_and_saves_redo(self):
|
|
for mode in ("BGnot", "Bnot", "Anot"):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
workspace = self.workspace(root, mode)
|
|
output = root / mode / "Copie01"
|
|
output.mkdir()
|
|
Image.new("RGB", (40, 30), "red").save(output / "Ex 1.jpg")
|
|
before = (output / "Ex 1.jpg").read_bytes()
|
|
atomic_write_json(output / "score.json", {"Ex 1": "3.5", "Ex 2": "0"})
|
|
answer = root / "answer.pdf"
|
|
answer.touch()
|
|
data = {
|
|
"01": {
|
|
label: {
|
|
"result": {"score": 2, "feedback": []},
|
|
"pdf_path": answer,
|
|
"coordinates": (0, 0),
|
|
}
|
|
for label in ("Ex 1", "Ex 2")
|
|
}
|
|
}
|
|
with (
|
|
patch.object(
|
|
reader.annotating,
|
|
"make_base_image",
|
|
return_value=(Image.new("RGB", (40, 20)), 0, 0),
|
|
),
|
|
patch.object(
|
|
reader.annotating,
|
|
"compose_label_image",
|
|
return_value=(Image.new("RGB", (40, 20), "blue"), 0),
|
|
),
|
|
patch.object(reader, "get_extra_pdfs_as_images", return_value=[]),
|
|
):
|
|
status, _ = reader.apply_actions_and_regenerate_grouped(
|
|
workspace,
|
|
data,
|
|
"01",
|
|
[],
|
|
{},
|
|
["Ex 1", "Ex 2"],
|
|
annotation_dir=mode,
|
|
selected_labels={"Ex 2"},
|
|
)
|
|
self.assertEqual(status, ExitCode.SUCCESS)
|
|
self.assertEqual(
|
|
read_json(output / "score.json"), {"Ex 1": "3.5", "Ex 2": "2"}
|
|
)
|
|
self.assertEqual((output / "Ex 1.jpg").read_bytes(), before)
|
|
self.assertTrue((output / "Ex 2.jpg").is_file())
|
|
with Image.open(output / "Concat.jpg") as concat:
|
|
self.assertEqual(concat.size, (40, 50))
|
|
|
|
def test_simple_import_survives_successive_redos(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
workspace = self.workspace(root, "Anot")
|
|
output = root / "Anot/Copie01"
|
|
output.mkdir()
|
|
for label in ("Ex 1", "Ex 2"):
|
|
Image.new("RGB", (40, 30), "white").save(output / f"{label}.jpg")
|
|
Image.new("RGB", (40, 60), "red").save(output / "Concat_annotated.jpg")
|
|
atomic_write_json(output / "score.json", {"Ex 1": "1", "Ex 2": "1"})
|
|
answer = root / "answer.pdf"
|
|
answer.touch()
|
|
data = {
|
|
"01": {
|
|
label: {
|
|
"result": {"score": 2, "feedback": []},
|
|
"pdf_path": answer,
|
|
"coordinates": (0, 0),
|
|
}
|
|
for label in ("Ex 1", "Ex 2")
|
|
}
|
|
}
|
|
with (
|
|
patch.object(
|
|
reader.annotating,
|
|
"make_base_image",
|
|
return_value=(Image.new("RGB", (40, 20)), 0, 0),
|
|
),
|
|
patch.object(
|
|
reader.annotating,
|
|
"compose_label_image",
|
|
return_value=(Image.new("RGB", (40, 20), "blue"), 0),
|
|
),
|
|
):
|
|
for selected in ({"Ex 2"}, {"Ex 1"}):
|
|
status, _ = reader.apply_actions_and_regenerate_grouped(
|
|
workspace,
|
|
data,
|
|
"01",
|
|
[],
|
|
{},
|
|
["Ex 1", "Ex 2"],
|
|
annotation_dir="Anot",
|
|
selected_labels=selected,
|
|
)
|
|
self.assertEqual(status, ExitCode.SUCCESS)
|
|
if selected == {"Ex 2"}:
|
|
with Image.open(output / "Ex 1.jpg") as untouched:
|
|
self.assertGreater(untouched.getpixel((10, 10))[0], 240)
|
|
with Image.open(output / "Concat.jpg") as concat:
|
|
self.assertEqual(concat.size, (40, 40))
|
|
self.assertGreater(concat.getpixel((10, 30))[2], 240)
|
|
|
|
def test_latest_group_coordinates_win_numerically(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
group = root / "Par label" / "Ex 1"
|
|
group.mkdir(parents=True)
|
|
for number, height in ((2, 100), (10, 40)):
|
|
atomic_write_json(
|
|
group / f"Group_{number}.json", [["01", 0, height, "", "Ex 1"]]
|
|
)
|
|
Image.new("RGB", (20, height)).save(group / f"Group_{number}.jpg")
|
|
index, warnings = _coordinate_index(EvaluationWorkspace(root))
|
|
self.assertFalse(warnings)
|
|
self.assertEqual(index[("Ex 1", "01")].height, 40)
|
|
|
|
def test_unreadable_redo_is_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
atomic_write_json(root / "bnote.json", {"images": []})
|
|
with (
|
|
patch.object(
|
|
reader, "detect_checks_and_notes", return_value=([], None)
|
|
),
|
|
self.assertRaises(ValueError),
|
|
):
|
|
reader._scan_annotation_directory(root, required=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|