73 lines
4.1 KiB
Python
73 lines
4.1 KiB
Python
"""Re-render saved handwriting interpretation with current scores and remarks."""
|
|
from copienator import EvaluationWorkspace, ExitCode, evaluation_parser, execute, workspace_from_args, utils
|
|
from copienator.review_snapshot import load, FILENAME
|
|
from copienator.commands.reading_annotations import apply_actions_and_regenerate
|
|
from copienator.commands.reading_grouped_annotations import apply_actions_and_regenerate_grouped
|
|
|
|
|
|
def run(workspace: EvaluationWorkspace, mode: str, *, incremental: bool = False):
|
|
directories = sorted((p for p in (workspace.root / mode).glob('Copie*')
|
|
if p.is_dir() and p.name[5:].isdigit() and (p / 'score.json').is_file()))
|
|
# Legacy Anot outputs predate saved rendering inputs; create them without
|
|
# replacing any existing score or annotated image.
|
|
missing = [d for d in directories if not (d / FILENAME).is_file()]
|
|
if mode == 'Anot' and missing:
|
|
from copienator.annotation_data import load_annotation_data
|
|
from copienator.annotation_layout import layout_for
|
|
from copienator.filesystem import staged_files
|
|
from copienator.review_snapshot import stage
|
|
loaded = load_annotation_data(workspace)
|
|
if loaded.warnings:
|
|
raise ValueError("Relancez la génération Anot : " + "; ".join(loaded.warnings))
|
|
for directory in missing:
|
|
cid = directory.name[5:]
|
|
if cid not in loaded.data:
|
|
raise ValueError(f"Données Anot absentes pour {directory.name}.")
|
|
data = loaded.data[cid]
|
|
notes = {label: {'img': None, 'old_header_h': 0, 'layout': layout_for(content['pdf_path'])}
|
|
for label, content in data.items()}
|
|
with staged_files(directory) as staging:
|
|
stage(directory, staging, data, notes, engine='grouped')
|
|
from copienator.reviewed_cache import ReviewedReadCache
|
|
cache = ReviewedReadCache(workspace, mode, directories) if incremental else None
|
|
if cache:
|
|
pending = []
|
|
for directory in directories:
|
|
if cache.unchanged(directory.name[5:]):
|
|
print(f"Skipping unchanged {directory.name}", flush=True)
|
|
else:
|
|
pending.append(directory)
|
|
directories = pending
|
|
# Validate every snapshot before starting to publish any copy.
|
|
for directory in directories:
|
|
if not (directory / FILENAME).is_file():
|
|
raise ValueError(f'{directory.name} : résultat de lecture sauvegardé absent ; relisez les annotations manuscrites.')
|
|
load(workspace.root, directory)
|
|
status = ExitCode.SUCCESS
|
|
for directory in directories:
|
|
cid = directory.name[5:]
|
|
saved, data, notes = load(workspace.root, directory)
|
|
if saved['engine'] == 'standard':
|
|
result = apply_actions_and_regenerate(workspace, {cid: data}, cid, [], None,
|
|
utils.read_all_labels(workspace.root), update_score=True,
|
|
label_notes={label: note['img'] for label, note in notes.items()},
|
|
review_metadata=saved['bnote'])
|
|
else:
|
|
result, log = apply_actions_and_regenerate_grouped(workspace, {cid: data}, cid, [], notes,
|
|
utils.read_all_labels(workspace.root), update_score=True, annotation_dir=mode,
|
|
force_filtered=set(saved.get('force_filtered', [])))
|
|
print(log, flush=True)
|
|
if result != ExitCode.SUCCESS:
|
|
status = ExitCode.PARTIAL
|
|
elif cache:
|
|
# Rendering can normalize the saved snapshot; fingerprint the published inputs.
|
|
ReviewedReadCache(workspace, mode, [directory]).remember([cid])
|
|
return status
|
|
|
|
|
|
def main(argv=None):
|
|
parser = evaluation_parser('Regenerate from saved reviewed inputs without rescanning handwriting')
|
|
parser.add_argument('--annotation-dir', choices=('Bnot', 'BGnot', 'Anot'), required=True)
|
|
parser.add_argument("--incremental", action="store_true", help="Skip unchanged copies")
|
|
return execute(parser, argv, lambda args: run(workspace_from_args(args), args.annotation_dir, incremental=args.incremental))
|