Cache for reading annotation

copienator/annotation_read_cache.py

@@ -0,0 +1,77 @@
+"""Small content fingerprints for incremental --update-score reads."""
+from hashlib import file_digest, sha256
+import json
+from pathlib import Path
+
+from copienator import atomic_write_json, configuration, read_json
+from copienator.annotation_documents import copy_documents
+from copienator.return_answers import load_return_answer_options
+
+STATE_FILE = ".annotation-read-state.json"
+DOCUMENT_INPUTS = ("bnote.json", "checkboxes.json", "Reference.jpg", "Concat_annotated.pdf")
+
+
+def _files_digest(paths) -> str:
+    digest = sha256()
+    for path in sorted(set(paths)):
+        digest.update(str(path).encode())
+        if path.is_file():
+            with path.open("rb") as stream:
+                digest.update(file_digest(stream, "sha256").digest())
+        else:
+            digest.update(b"missing")
+    return digest.hexdigest()
+
+
+class AnnotationReadCache:
+    def __init__(self, workspace, mode: str, data: dict, group_dirs=()):
+        self.workspace = workspace
+        self.mode = mode
+        self.signatures = {}
+        root = workspace.root
+        shared = [root / "labels", root / mode / "annotation_modifications_applied.txt"]
+        for folder in ("Text2", "Sol2"):
+            shared.extend((root / folder).glob("*.pdf"))
+        for directory in group_dirs:
+            shared.extend(directory / name for name in DOCUMENT_INPUTS)
+        common = (_files_digest(shared), load_return_answer_options(root),
+                  configuration.RETURN_ANSWERS_ENABLED,
+                  configuration.LATEX_ANOT_BEFORE, configuration.LATEX_ANOT_AFTER)
+        for copy_id, labels in data.items():
+            directory = root / mode / f"Copie{copy_id}"
+            paths = [directory / "score.json", directory / "bnote.json"]
+            for document in copy_documents(directory):
+                paths.extend(document / name for name in DOCUMENT_INPUTS)
+            for content in labels.values():
+                pdf = Path(content["pdf_path"])
+                paths.extend((pdf, pdf.with_suffix(".layout.json")))
+            payload = json.dumps((1, common, labels, _files_digest(paths)),
+                                 sort_keys=True, default=str).encode()
+            self.signatures[copy_id] = sha256(payload).hexdigest()
+
+    def unchanged(self, copy_id: str) -> bool:
+        directory = self.workspace.root / self.mode / f"Copie{copy_id}"
+        try:
+            state = read_json(directory / STATE_FILE)
+            return (state["signature"] == self.signatures[copy_id]
+                    and bool(state["outputs"])
+                    and all(Path(path).is_file() for path in state["outputs"]))
+        except (OSError, ValueError, TypeError, KeyError):
+            return False
+
+    def remember(self, copy_ids) -> None:
+        for copy_id in copy_ids:
+            directory = self.workspace.root / self.mode / f"Copie{copy_id}"
+            outputs = [directory / "info.json"]
+            info = read_json(directory / "info.json")
+            outputs.extend(directory / f"{label}.jpg" for label, entry in info.items()
+                           if entry["present"])
+            outputs.extend(directory.glob("Concat*.*"))
+            outputs.extend(directory.glob(".annotation-base-*.png"))
+            outputs.extend(directory.glob("annotation_baseline.json"))
+            for destination in self.workspace.return_dir.glob(f"* ({copy_id})"):
+                outputs.extend((destination / "answers").glob("*.jpg"))
+            atomic_write_json(directory / STATE_FILE, {
+                "signature": self.signatures[copy_id],
+                "outputs": [str(path) for path in outputs],
+            })