diff --git a/src/deckle/assign.py b/src/deckle/assign.py new file mode 100644 index 0000000..ebcadb9 --- /dev/null +++ b/src/deckle/assign.py @@ -0,0 +1,98 @@ +"""Assign a staged master to a canonical ID""" + +from __future__ import annotations + +import os +import shutil +from dataclasses import dataclass +from pathlib import Path + +import cv2 + +from .ids import CardId, check_custom_name, resolve_ref +from .project import CARD_BACKS_DIR, Project +from .staging import provenance_for, read_staging_index, write_staging_index + + +class AssignError(RuntimeError): + """A busted assignment.""" + + +@dataclass(frozen=True) +class Assignment: + ref: str + source: Path + dest: Path + replaced: Path | None + rotated: bool + + +def _move(src: Path, dest: Path, *, rotate_180: bool) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + if not rotate_180: + try: + os.replace(src, dest) + except OSError: # different filesystem + shutil.move(str(src), str(dest)) + return + img = cv2.imread(str(src), cv2.IMREAD_UNCHANGED) + if img is None: + raise AssignError(f"cannot read {src}") + if not cv2.imwrite( + str(dest), cv2.rotate(img, cv2.ROTATE_180), [cv2.IMWRITE_PNG_COMPRESSION, 6] + ): + raise AssignError(f"failed to write {dest}") + src.unlink() + + +def assign( + project: Project, + source: Path, + ref: str | None = None, + *, + card_back: str | None = None, + rotate_180: bool = False, +) -> Assignment: + """Assign a staged master to a card ID""" + if project.degraded: + raise AssignError( + f"cannot assign: {project.degraded_reason}. Fix or remove the state file first." + ) + if (ref is None) == (card_back is None): + raise AssignError("give exactly one of a canonical ID or --card-back ") + if not source.is_file(): + raise AssignError(f"no such file: {source}") + + card: CardId | None = None + if card_back is not None: + key = check_custom_name(card_back, "card back design key") + ref_out = f"{CARD_BACKS_DIR}.{key}" + dest = project.back_path(key) + else: + card, variant = resolve_ref(ref) # type: ignore[arg-type] + ref_out = ref # type: ignore[assignment] + dest = project.master_path(card, variant) + + if source.resolve() == dest.resolve(): + raise AssignError(f"{source} is already the master for {ref_out}") + + replaced: Path | None = None + previous = project.cards.get(ref_out, {}).get("master") + if previous: + old = project.root / previous + if old.exists() and old.resolve() != dest.resolve(): + old.unlink() + replaced = old + + _move(source, dest, rotate_180=rotate_180) + + index = read_staging_index(project) + prov = provenance_for(source.name, index, rotate_180=rotate_180) + prov["master"] = dest.relative_to(project.root).as_posix() + project.record_card(ref_out, prov) + project.save() + + if index.pop(source.name, None) is not None: + write_staging_index(project, index) + + return Assignment(ref=ref_out, source=source, dest=dest, replaced=replaced, rotated=rotate_180) diff --git a/src/deckle/staging.py b/src/deckle/staging.py new file mode 100644 index 0000000..15e0387 --- /dev/null +++ b/src/deckle/staging.py @@ -0,0 +1,104 @@ +"""Staging index""" + +from __future__ import annotations + +import os +import tomllib +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from . import build_id, tomlout +from .project import Project + +if TYPE_CHECKING: + from .detect import Card + +STAGING_INDEX = "index.toml" + +MEASURED_KEYS = ("width_mm", "height_mm", "aspect", "skew_deg", "edge_strategy", "dpi") + + +def utc_now() -> str: + """An ISO-8601 UTC timestamp.""" + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def read_staging_index(project: Project) -> dict[str, Any]: + path = project.staging_dir / STAGING_INDEX + if not path.is_file(): + return {} + try: + with path.open("rb") as fh: + return tomllib.load(fh) + except (tomllib.TOMLDecodeError, OSError): + return {} + + +def write_staging_index(project: Project, index: dict[str, Any]) -> None: + project.staging_dir.mkdir(parents=True, exist_ok=True) + path = project.staging_dir / STAGING_INDEX + tmp = path.with_name(f".{STAGING_INDEX}.{os.getpid()}.tmp") + try: + tmp.write_text(tomlout.dumps(dict(sorted(index.items()))), encoding="utf-8") + os.replace(tmp, path) + finally: + tmp.unlink(missing_ok=True) + + +def record_rectified( + project: Project, + scan: Path, + cards: list[Card], + written: list[Path], + *, + strategy: str, + dpi: float, +) -> None: + if project.degraded: + return + + staging = project.staging_dir.resolve() + rows = { + path.name: { + "source_scan": scan.name, + "slot": f"r{card.window.row}c{card.window.col}", + "width_mm": round(card.width_mm, 4), + "height_mm": round(card.height_mm, 4), + "aspect": round(card.aspect, 5), + "skew_deg": round(card.skew_deg, 4), + "edge_strategy": str(strategy), + "dpi": dpi, + } + for path, card in zip(written, cards, strict=True) + if path.parent.resolve() == staging + } + + if not rows: + return + + index = read_staging_index(project) + index.update(rows) + write_staging_index(project, index) + + +def provenance_for(filename: str, index: dict[str, Any], *, rotate_180: bool) -> dict[str, Any]: + """A provenance row for a staged file.""" + row = dict(index.get(filename, {})) + stem = filename[: -len(".png")] if filename.endswith(".png") else filename + scan, _, slot = stem.rpartition("_") + prov: dict[str, Any] = {} + prov["source_scan"] = row.get("source_scan") or (f"{scan}.jpg" if scan else filename) + + if row.get("slot") or slot: + prov["slot"] = row.get("slot") or slot + + for key in MEASURED_KEYS: + if key in row: + prov[key] = row[key] + + prov["rotate_180"] = rotate_180 + prov["deckle_version"] = build_id() + prov["assigned_at"] = utc_now() + + return prov diff --git a/src/deckle/status.py b/src/deckle/status.py new file mode 100644 index 0000000..95a3ccd --- /dev/null +++ b/src/deckle/status.py @@ -0,0 +1,66 @@ +"""Status of scanned vs staged.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from .project import Master, Project, ProjectError + + +@dataclass(frozen=True) +class Status: + root: Path + degraded_reason: str | None + edge_strategy: str | None + config_problem: str | None + present: list[Master] + total: int | None + missing: list[str] + unexpected: list[str] + backs: list[str] + strays: list[Path] + + +def status(project: Project) -> Status: + masters = project.masters() + + strategy: str | None = None + config_problem: str | None = None + if not project.degraded: + try: + strategy = project.config.edge_strategy + except ProjectError as exc: + config_problem = str(exc) + + backs = [m.ref for m in masters if m.is_back] + fronts = [m for m in masters if not m.is_back] + present_ids = {str(m.card) for m in fronts} + + if project.degraded: + return Status( + root=project.root, + degraded_reason=project.degraded_reason, + edge_strategy=None, + config_problem=None, + present=fronts, + total=None, + missing=[], + unexpected=[], + backs=backs, + strays=project.strays(), + ) + + wanted = [str(c) for c in project.roster.cards()] + return Status( + root=project.root, + degraded_reason=None, + edge_strategy=strategy, + config_problem=config_problem, + present=fronts, + total=len(wanted), + missing=[c for c in wanted if c not in present_ids], + unexpected=sorted(present_ids - set(wanted)), + backs=backs, + strays=project.strays(), + ) diff --git a/tests/test_assign.py b/tests/test_assign.py new file mode 100644 index 0000000..93e00a8 --- /dev/null +++ b/tests/test_assign.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import cv2 +import numpy as np +import pytest + +from deckle.assign import AssignError, assign +from deckle.ids import IdError +from deckle.project import PROJECT_FILE, Project, init +from deckle.staging import read_staging_index, write_staging_index + + +def make_project(tmp_path): + return init(tmp_path / "proj", "Test Deck") + + +def stage(project, name: str): + path = project.staging_dir / name + path.write_bytes(b"pixels") + return path + + +def stage_png(project, name: str) -> tuple[object, np.ndarray]: + img = np.arange(4 * 6 * 3, dtype=np.uint8).reshape(4, 6, 3) + path = project.staging_dir / name + assert cv2.imwrite(str(path), img) + return path, img + + +def test_assign_moves_and_records_provenance(tmp_path): + project = make_project(tmp_path) + src = stage(project, "scan_r0c0.png") + write_staging_index( + project, + { + src.name: { + "source_scan": "scan.jpg", + "slot": "r0c0", + "width_mm": 70.2, + "height_mm": 120.3, + "skew_deg": -0.31, + "edge_strategy": "brightest", + } + }, + ) + + result = assign(project, src, "minor_arcana.pentacles.king") + + assert not src.exists() + assert result.dest == project.masters_dir / "minor_arcana/pentacles/king.png" + assert result.dest.is_file() + + row = Project.load(project.root).cards["minor_arcana.pentacles.king"] + assert row["source_scan"] == "scan.jpg" + assert row["slot"] == "r0c0" + assert row["width_mm"] == 70.2 + assert row["skew_deg"] == -0.31 + assert row["rotate_180"] is False + assert row["master"] == "masters/minor_arcana/pentacles/king.png" + assert row["deckle_version"] and row["assigned_at"].endswith("Z") + + +def test_assign_without_an_index_reads_scan_and_slot_from_the_name(tmp_path): + project = make_project(tmp_path) + assign(project, stage(project, "scan_001_r0c1.png"), "minor_arcana.cups.king") + row = project.cards["minor_arcana.cups.king"] + assert row["source_scan"] == "scan_001.jpg" + assert row["slot"] == "r0c1" + assert "width_mm" not in row + + +def test_assign_consumes_the_staging_index_row(tmp_path): + project = make_project(tmp_path) + src = stage(project, "s_r0c0.png") + write_staging_index(project, {src.name: {"source_scan": "s.jpg", "slot": "r0c0"}}) + assign(project, src, "major_arcana.03") + assert read_staging_index(project) == {} + + +def test_reassigning_an_id_replaces_its_master(tmp_path): + project = make_project(tmp_path) + assign(project, stage(project, "a_r0c0.png"), "major_arcana.03") + result = assign(project, stage(project, "b_r0c0.png"), "major_arcana.03") + + assert result.replaced is None + assert [m.ref for m in project.masters()] == ["major_arcana.03"] + assert Project.load(project.root).cards["major_arcana.03"]["source_scan"] == "b.jpg" + + +def test_reassigning_removes_a_previous_master_at_another_path(tmp_path): + project = make_project(tmp_path) + old = project.masters_dir / "major_arcana/old.png" + old.parent.mkdir(parents=True) + old.write_bytes(b"pixels") + project.record_card("major_arcana.03", {"master": "masters/major_arcana/old.png"}) + + result = assign(project, stage(project, "s_r0c0.png"), "major_arcana.03") + + assert result.replaced == old + assert not old.exists() + assert result.dest.is_file() + + +def test_assign_rotates_180(tmp_path): + project = make_project(tmp_path) + src, img = stage_png(project, "s_r0c0.png") + result = assign(project, src, "major_arcana.03", rotate_180=True) + + assert not src.exists() + out = cv2.imread(str(result.dest), cv2.IMREAD_UNCHANGED) + assert np.array_equal(out, img[::-1, ::-1]) + assert project.cards["major_arcana.03"]["rotate_180"] is True + + +@pytest.mark.parametrize("rotate_180", [True, False]) +def test_assigning_a_master_onto_itself_is_refused(tmp_path, rotate_180): + project = make_project(tmp_path) + src, img = stage_png(project, "s_r0c0.png") + master = assign(project, src, "major_arcana.12").dest + before = (project.root / PROJECT_FILE).read_bytes() + + with pytest.raises(AssignError, match="already the master"): + assign(project, master, "major_arcana.12", rotate_180=rotate_180) + + assert np.array_equal(cv2.imread(str(master), cv2.IMREAD_UNCHANGED), img) + assert (project.root / PROJECT_FILE).read_bytes() == before + + +def test_rejected_ids_leave_the_file_in_staging(tmp_path): + project = make_project(tmp_path) + for bad in ("major_arcana.6", "minor_arcana.cups.King", "major_arcana.2women"): + src = stage(project, "s_r0c0.png") + with pytest.raises(IdError): + assign(project, src, bad) + assert src.exists() + + +def test_assign_accepts_extended_and_custom_ids(tmp_path): + project = make_project(tmp_path) + ids = ["major_arcana.23", "major_arcana.happy_squirrel", "minor_arcana.stars.ace"] + for i, cid in enumerate(ids): + assign(project, stage(project, f"s_r0c{i}.png"), cid) + assert {m.ref for m in project.masters()} == set(ids) + + +def test_assign_variant(tmp_path): + project = make_project(tmp_path) + assign(project, stage(project, "s_r0c0.png"), "major_arcana.06:two_women") + assert (project.masters_dir / "major_arcana/06.two_women.png").is_file() + + +def test_assign_card_back(tmp_path): + project = make_project(tmp_path) + assign(project, stage(project, "s_r0c0.png"), None, card_back="classic") + assert (project.masters_dir / "card_backs/classic.png").is_file() + assert project.cards["card_backs.classic"]["slot"] == "r0c0" + + +def test_assign_needs_exactly_one_target(tmp_path): + project = make_project(tmp_path) + src = stage(project, "s_r0c0.png") + with pytest.raises(AssignError): + assign(project, src, None) + with pytest.raises(AssignError): + assign(project, src, "major_arcana.03", card_back="classic") + + +def test_a_degraded_project_refuses_to_assign(tmp_path): + project = make_project(tmp_path) + src = stage(project, "s_r0c0.png") + (project.root / PROJECT_FILE).write_text("broken [", encoding="utf-8") + with pytest.raises(AssignError, match="not valid TOML"): + assign(Project.load(project.root), src, "major_arcana.03") + assert src.exists() diff --git a/tests/test_staging.py b/tests/test_staging.py new file mode 100644 index 0000000..b7c29a4 --- /dev/null +++ b/tests/test_staging.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +from deckle.project import PROJECT_FILE, Project, init +from deckle.staging import ( + STAGING_INDEX, + provenance_for, + read_staging_index, + record_rectified, + write_staging_index, +) +from deckle.units import EdgeStrategy + + +def make_project(tmp_path): + return init(tmp_path / "proj", "Test Deck") + + +def fake_card(row: int, col: int, **measured): + """The attributes record_rectified reads from a detected card.""" + m = dict(width_mm=70.12345, height_mm=120.54321, aspect=0.5817234, skew_deg=-0.312345) + m.update(measured) + return SimpleNamespace(window=SimpleNamespace(row=row, col=col), **m) + + +def test_index_round_trips_sorted(tmp_path): + project = make_project(tmp_path) + write_staging_index(project, {"b.png": {"slot": "r0c1"}, "a.png": {"slot": "r0c0"}}) + assert list(read_staging_index(project)) == ["a.png", "b.png"] + + +def test_a_missing_or_corrupt_index_reads_empty(tmp_path): + project = make_project(tmp_path) + assert read_staging_index(project) == {} + (project.staging_dir / STAGING_INDEX).write_text("not [ toml", encoding="utf-8") + assert read_staging_index(project) == {} + + +def test_record_rectified_indexes_staged_masters(tmp_path): + project = make_project(tmp_path) + written = [project.staging_dir / "scan_r0c0.png", project.staging_dir / "scan_r1c1.png"] + cards = [fake_card(0, 0), fake_card(1, 1, skew_deg=0.5)] + + record_rectified( + project, Path("in/scan.jpg"), cards, written, strategy=EdgeStrategy.OUTERMOST, dpi=600.0 + ) + + index = read_staging_index(project) + assert index["scan_r0c0.png"] == { + "source_scan": "scan.jpg", + "slot": "r0c0", + "width_mm": 70.1235, + "height_mm": 120.5432, + "aspect": 0.58172, + "skew_deg": -0.3123, + "edge_strategy": "outermost", + "dpi": 600.0, + } + assert index["scan_r1c1.png"]["slot"] == "r1c1" + assert index["scan_r1c1.png"]["skew_deg"] == 0.5 + + +def test_record_rectified_keeps_earlier_rows(tmp_path): + project = make_project(tmp_path) + write_staging_index(project, {"old_r0c0.png": {"source_scan": "old.jpg"}}) + record_rectified( + project, + Path("new.jpg"), + [fake_card(0, 0)], + [project.staging_dir / "new_r0c0.png"], + strategy="brightest", + dpi=600.0, + ) + assert list(read_staging_index(project)) == ["new_r0c0.png", "old_r0c0.png"] + + +def test_record_rectified_ignores_masters_written_elsewhere(tmp_path): + project = make_project(tmp_path) + record_rectified( + project, + Path("scan.jpg"), + [fake_card(0, 0)], + [tmp_path / "elsewhere" / "scan_r0c0.png"], + strategy="brightest", + dpi=600.0, + ) + assert not (project.staging_dir / STAGING_INDEX).exists() + + +def test_record_rectified_does_nothing_in_a_degraded_project(tmp_path): + project = make_project(tmp_path) + (project.root / PROJECT_FILE).unlink() + degraded = Project.load(project.root) + record_rectified( + degraded, + Path("scan.jpg"), + [fake_card(0, 0)], + [degraded.staging_dir / "scan_r0c0.png"], + strategy="brightest", + dpi=600.0, + ) + assert not (project.staging_dir / STAGING_INDEX).exists() + + +def test_provenance_carries_the_index_row(): + index = { + "scan_r0c0.png": { + "source_scan": "scan.jpg", + "slot": "r0c0", + "width_mm": 70.2, + "edge_strategy": "brightest", + "unrelated": 1, + } + } + prov = provenance_for("scan_r0c0.png", index, rotate_180=True) + assert prov["source_scan"] == "scan.jpg" + assert prov["slot"] == "r0c0" + assert prov["width_mm"] == 70.2 + assert prov["edge_strategy"] == "brightest" + assert "unrelated" not in prov + assert prov["rotate_180"] is True + assert prov["deckle_version"] + assert prov["assigned_at"].endswith("Z") + + +def test_provenance_without_an_index_reads_the_filename(): + prov = provenance_for("scan_001_r0c1.png", {}, rotate_180=False) + assert prov["source_scan"] == "scan_001.jpg" + assert prov["slot"] == "r0c1" + assert "width_mm" not in prov diff --git a/tests/test_status.py b/tests/test_status.py new file mode 100644 index 0000000..cd86720 --- /dev/null +++ b/tests/test_status.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from deckle.project import PROJECT_FILE, Project, init +from deckle.status import status +from deckle.units import EdgeStrategy + + +def make_project(tmp_path, **kw): + return init(tmp_path / "proj", "Test Deck", **kw) + + +def touch_master(project, relpath: str) -> None: + path = project.masters_dir / relpath + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"not really a png") + + +def test_empty_project_is_zero_of_78(tmp_path): + st = status(make_project(tmp_path)) + assert (len(st.present), st.total) == (0, 78) + assert len(st.missing) == 78 + assert st.edge_strategy == EdgeStrategy.BRIGHTEST + assert st.config_problem is None + + +def test_variants_do_not_double_count(tmp_path): + project = make_project(tmp_path) + touch_master(project, "major_arcana/06.png") + touch_master(project, "major_arcana/06.two_women.png") + st = status(project) + assert len(st.present) == 2 + assert st.total == 78 + assert "major_arcana.06" not in st.missing + + +def test_a_deleted_state_file_reports_no_total(tmp_path): + project = make_project(tmp_path) + for rel in ("major_arcana/03.png", "minor_arcana/cups/king.png", "card_backs/classic.png"): + touch_master(project, rel) + (project.root / PROJECT_FILE).unlink() + + st = status(Project.load(project.root)) + assert "missing" in st.degraded_reason + assert st.total is None + assert [m.ref for m in st.present] == ["major_arcana.03", "minor_arcana.cups.king"] + assert st.backs == ["classic"] + assert st.missing == [] and st.unexpected == [] + + +def test_a_corrupt_state_file_still_lists_the_masters(tmp_path): + project = make_project(tmp_path) + touch_master(project, "major_arcana/03.png") + (project.root / PROJECT_FILE).write_text("this is not [ valid toml", encoding="utf-8") + + st = status(Project.load(project.root)) + assert "not valid TOML" in st.degraded_reason + assert [m.ref for m in st.present] == ["major_arcana.03"] + + +def test_an_unknown_edge_strategy_is_reported_not_raised(tmp_path): + project = make_project(tmp_path) + touch_master(project, "major_arcana/03.png") + project.doc["project"]["edge_strategy"] = "sharpest" + project.save() + + st = status(Project.load(project.root)) + assert st.edge_strategy is None + assert "'sharpest' is not a known strategy" in st.config_problem + assert (len(st.present), st.total) == (1, 78) + + +def test_masters_the_roster_does_not_mention_are_unexpected(tmp_path): + project = make_project(tmp_path) + touch_master(project, "major_arcana/23.png") + assert status(project).unexpected == ["major_arcana.23"] + + +def test_roster_excludes_are_removed_from_the_total(tmp_path): + project = make_project(tmp_path) + project.doc["roster"]["excluded"] = [ + "minor_arcana.pentacles.page", + "minor_arcana.pentacles.knight", + ] + project.save() + st = status(Project.load(project.root)) + assert st.total == 76 + assert "minor_arcana.pentacles.page" not in st.missing + + +def test_strays_are_reported(tmp_path): + project = make_project(tmp_path) + touch_master(project, "major_arcana/03.png") + touch_master(project, "minor_arcana/cups/King.png") + assert [p.name for p in status(project).strays] == ["King.png"]