feat: add staging

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 <design>")
+    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)

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

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(),
+    )