Automatic orientation issues detection and fix, experimental

copienator/orientation.py

@@ -0,0 +1,107 @@
+"""Source-page provenance and versioned orientation reports."""
+from collections import defaultdict
+from pathlib import Path
+import re
+
+from copienator.annotation_layout import fingerprint
+from copienator.json_io import atomic_write_json, read_json
+
+
+def provenance_path(workspace, stem):
+    return workspace.metadata_dir / 'page_origins' / f'{stem}.json'
+
+
+def report_path(workspace, stem):
+    return workspace.metadata_dir / 'orientation' / f'{stem}.json'
+
+
+def save_provenance(workspace, pdf, pages):
+    atomic_write_json(provenance_path(workspace, pdf.stem), {
+        'version': 1, 'sha256': fingerprint(pdf), 'pages': pages,
+    })
+
+
+def load_provenance(workspace, pdf, count):
+    try:
+        data = read_json(provenance_path(workspace, pdf.stem), default={})
+        pages = data.get('pages', [])
+        valid = (data.get('sha256') == fingerprint(pdf) and isinstance(pages, list)
+                 and len(pages) == count and all(isinstance(p, dict)
+                 and type(p.get('origin')) is int and p['origin'] > 0 for p in pages))
+    except (OSError, ValueError, TypeError, AttributeError):
+        return None
+    return pages if valid else None
+
+
+def reports(workspace):
+    result = {}
+    for path in sorted((workspace.metadata_dir / 'orientation').glob('*.json')):
+        data = read_json(path)
+        pdf = workspace.copies_dir / f'{path.stem}.pdf'
+        if pdf.is_file() and fingerprint(pdf) in {data.get('sha256'), data.get('previous_sha256')}:
+            if any(data.get('images', {}).values()) or data.get('pending') or data.get('warning'):
+                result[pdf.name] = data
+    return result
+
+
+def record_detection(workspace, image, columns):
+    match = re.fullmatch(r'(.+)_(\d+)', image.stem)
+    if not match:
+        if columns:
+            raise ValueError('Cannot map orientation columns to a copy')
+        return False
+    stem, part = match[1], int(match[2]) - 1
+    pdf = workspace.copies_dir / f'{stem}.pdf'
+    if not columns and not report_path(workspace, stem).exists():
+        return False
+    schema = read_json(workspace.cutleft_dir / f'{stem}_schema.json')
+    counts = schema['columns_per_file']
+    if not 0 <= part < len(counts) or any(type(c) is not int or not 1 <= c <= counts[part] for c in columns):
+        raise ValueError('Invalid upside-down column numbers')
+    digest = fingerprint(pdf)
+    if schema.get('source_sha256', digest) != digest:
+        raise ValueError('Cutleft images are stale; regenerate them before detecting orientation')
+    data = read_json(report_path(workspace, stem), default={})
+    if data.get('sha256') != digest:
+        data = {'version': 1, 'sha256': digest, 'images': {}}
+    pages = sorted({sum(counts[:part]) + c for c in columns})
+    data.setdefault('images', {})[image.name] = pages
+    atomic_write_json(report_path(workspace, stem), data)
+    if pages:
+        print(f'[Orientation] {pdf.name}: pages à l’envers : {pages}', flush=True)
+    return bool(pages)
+
+
+def origin_groups(provenance, count):
+    groups = defaultdict(list)
+    for index in range(count):
+        groups[provenance[index]['origin'] if provenance else index].append(index + 1)
+    return list(groups.values())
+
+
+def resolve_unknowns(verdict, groups):
+    """Only known votes count; ties/all-unknown remain unresolved."""
+    result = dict(verdict)
+    for pages in groups:
+        up = sum(result[p] == 'upright' for p in pages)
+        down = sum(result[p] == 'upside_down' for p in pages)
+        majority = 'upright' if up > down else 'upside_down' if down > up else 'unknown'
+        for page in pages:
+            if result[page] == 'unknown':
+                result[page] = majority
+    return result
+
+
+def incoherent(verdict, pages):
+    return {'upright', 'upside_down'} <= {verdict[p] for p in pages}
+
+
+def repair_plan(verdict, groups):
+    """Output slots refer to input pages; keep scale and retained content intact."""
+    order = list(range(1, len(verdict) + 1))
+    for pages in groups:
+        if all(verdict[p] == 'upside_down' for p in pages):
+            # A 180-degree turn reverses both A3 halves and secondary strips.
+            for slot, source in zip(pages, reversed(pages)):
+                order[slot - 1] = source
+    return order, {p for p, value in verdict.items() if value == 'upside_down'}