sebastien/Copies
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Standardisation 6

Browse Source
This commit is contained in:
sebastien
2026-08-20 15:01:17 +02:00
parent 644e287586
commit aa40e58dd1
4 changed files with 419 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plotting.py
+117 -80
View File
@@ -1,24 +1,29 @@
import json
from __future__ import annotations
import argparse
import queue
import sys
import threading
import tkinter as tk
from collections.abc import Sequence
from pathlib import Path
from tkinter import messagebox
from PIL import Image, ImageDraw, ImageFont, ImageTk
from copienator import (
EvaluationWorkspace,
ExitCode,
atomic_write_json,
execute,
read_json,
target_parser,
workspace_from_target,
)
from platform_utils import open_path
print("o to open pdf, O original pdf, e to emacs part, p to go back, i to interro, click for coordinates")
from utils import natural_key, read_all_labels
# --- Configuration & Globals ---
padding = 60
valid_labels_set = None
# Queue payload: (pil_image, json_path, metadata)
# metadata is a dict: {'copie': str, 'part': int, 'schema': dict}
image_queue = queue.Queue(maxsize=5)
try:
font = ImageFont.truetype("DejaVuSans.ttf", size=30)
@@ -55,6 +60,14 @@ def convert_list(l, group_id, json_schema):
ll.append(ee)
return ll
def normalized_labels(entries):
return [
str(value["label"]).removeprefix("|").removesuffix("|")
for value in entries
if str(value["label"]) != "_"
]
def prepare_image(image_path: str, bounding_boxes, all_labels, nb_pages, last_label_index):
im = Image.open(image_path)
im.load()
@@ -94,7 +107,7 @@ def prepare_image(image_path: str, bounding_boxes, all_labels, nb_pages, last_la
# --- Processing Logic (Worker Thread) ---
def worker_thread(base_dir, files_to_process, all_labels):
def _worker_items(base_dir, files_to_process, all_labels, output_queue):
"""
Iterates through files, prepares VISUALS only, and puts metadata in queue.
Does NOT write final JSON files anymore.
@@ -111,9 +124,8 @@ def worker_thread(base_dir, files_to_process, all_labels):
json_schema_path = base_dir / 'Cutleft' / f"{copie}_schema.json"
try:
with open(json_schema_path, 'r') as f:
json_schema = json.load(f)
except:
json_schema = read_json(json_schema_path)
except (OSError, TypeError, ValueError):
print("No json_schema : ", json_schema_path)
continue
@@ -124,11 +136,10 @@ def worker_thread(base_dir, files_to_process, all_labels):
bb_list = []
json_name = ""
try:
with open(json_path, 'r') as f:
json_result = json.load(f)
json_result = read_json(json_path)
bb_list = json_result.get("list", [])
json_name = json_result.get("name", "")
except Exception as e:
except Exception as e: # noqa: BLE001 - malformed user-editable JSON
print(f"Warning: {json_path.name} is malformed! Loading blank. {e}")
# We do NOT skip; we continue so the user can fix it in the GUI
@@ -138,7 +149,7 @@ def worker_thread(base_dir, files_to_process, all_labels):
prepare_image(str(img_path), bb_list, all_labels, nb_pages, last_label_index)
error_msg = None
except Exception as e:
except Exception as e: # noqa: BLE001 - keep the item editable in the GUI
print(f"Error processing {img_path.name}: {e}")
pil_image = Image.open(str(img_path))
error_msg = str(e)
@@ -151,15 +162,24 @@ def worker_thread(base_dir, files_to_process, all_labels):
"error": error_msg
}
image_queue.put((pil_image, json_path, metadata))
output_queue.put((pil_image, json_path, metadata))
# Sentinel to indicate finished
image_queue.put((None, None, None))
def worker_thread(base_dir, files_to_process, all_labels, output_queue):
"""Prepare queue items and always terminate the GUI stream."""
failure = None
try:
_worker_items(base_dir, files_to_process, all_labels, output_queue)
except Exception as exc: # noqa: BLE001 - worker boundary
failure = str(exc)
print(f"Plotting worker failed: {exc}")
finally:
metadata = {"worker_error": failure} if failure else None
output_queue.put((None, None, metadata))
# --- GUI Logic (Main Thread) ---
class ImageViewer:
def __init__(self, root, base_dir):
def __init__(self, root, workspace, valid_labels, input_queue):
self.root = root
self.root.resizable(False, False) # If you resize, coordinates will be wrong
@@ -171,7 +191,10 @@ class ImageViewer:
root.geometry(f"+{x}+{y}")
self.base_dir = base_dir
self.workspace = workspace
self.base_dir = workspace.root
self.valid_labels = valid_labels
self.image_queue = input_queue
self.root.title("Bounding Box Viewer")
self.label = tk.Label(root, text="Waiting for images...")
self.label.pack(expand=True, fill="both")
@@ -192,6 +215,7 @@ class ImageViewer:
self.history = []
self.forward_stack = []
self.current_pil_image = None
self.failed = False
from config import PLOTTING_KB
@@ -202,7 +226,8 @@ class ImageViewer:
self.root.bind(PLOTTING_KB["open pdf"], self.on_open_pdf)
self.root.bind(PLOTTING_KB["open original pdf"], self.on_open_ori_pdf)
self.root.bind(PLOTTING_KB["open eval"], self.on_open_interro)
self.root.bind('<Escape>', lambda e: self.root.quit())
self.root.bind('<Escape>', lambda _event: self.close())
self.root.protocol("WM_DELETE_WINDOW", self.close)
self.label.bind('<Button-1>', self.on_click)
self.poll_queue()
@@ -214,10 +239,15 @@ class ImageViewer:
if self.forward_stack:
pil_image, json_path, metadata = self.forward_stack.pop()
else:
pil_image, json_path, metadata = image_queue.get_nowait()
pil_image, json_path, metadata = self.image_queue.get_nowait()
# Handle End of Stream
if pil_image is None:
if metadata and metadata.get("worker_error"):
self.failed = True
messagebox.showerror(
"Processing Error", metadata["worker_error"]
)
self.save_current_batch() # Save any remaining data
print("All images processed.")
self.root.quit()
@@ -241,10 +271,12 @@ class ImageViewer:
if self.active_copie_name and self.accumulated_results:
main_json_path = self.base_dir / "Copies" / f"{self.active_copie_name}.json"
print(f"Writing aggregated result to {main_json_path}")
with open(main_json_path, 'w') as f:
json.dump(self.accumulated_results, f)
atomic_write_json(main_json_path, self.accumulated_results)
self.accumulated_results = None
def close(self):
self.root.quit()
def on_previous(self, event):
if self.is_viewing and self.history:
@@ -288,8 +320,7 @@ class ImageViewer:
num_added = 0 # ADD THIS LINE
try:
with open(self.current_json_path, 'r') as f:
current_data = json.load(f)
current_data = read_json(self.current_json_path)
# Perform the conversion now, post-edit
converted_items = convert_list(
@@ -298,11 +329,10 @@ class ImageViewer:
self.current_meta["schema"]
)
labels = [v["label"] for v in current_data["list"]]
labels = [label for label in labels if label != "_"]
labels = [label[1:] for label in labels if label[0] == "|"]
labels = [label[:-1] for label in labels if label[-1] == "|"]
false_labels = [label for label in labels if label not in valid_labels_set]
labels = normalized_labels(current_data["list"])
false_labels = [
label for label in labels if label not in self.valid_labels
]
if false_labels:
msg = f"Wrong label in {self.current_json_path.name}: {false_labels}\n\n\tPlease press 'e' to fix it, then press Enter again."
@@ -318,7 +348,7 @@ class ImageViewer:
if "name" in current_data and current_data["name"] != "Continued":
self.accumulated_results["name"] = current_data["name"]
except Exception as e:
except Exception as e: # noqa: BLE001 - interactive validation boundary
# Warn user and STOP (do not advance to next image)
msg = f"Error reading {self.current_json_path.name}:\n\n{e}\n\nPlease press 'e' to fix it, then press Enter again."
print(msg)
@@ -387,51 +417,58 @@ class ImageViewer:
self.root.clipboard_clear()
self.root.clipboard_append(box_str)
from utils import natural_key, read_all_labels
def _selected_images(workspace: EvaluationWorkspace, target: Path) -> list[Path]:
workspace.require_directories("Cutleft", "Copies")
if target.is_file():
stem = target.stem
exact = workspace.cutleft_dir / f"{stem}.jpg"
if exact.is_file():
return [exact]
return sorted(
workspace.cutleft_dir.glob(f"{stem}_*.jpg"),
key=natural_key,
)
return sorted(workspace.cutleft_dir.glob("*.jpg"), key=natural_key)
def run(workspace: EvaluationWorkspace, target: Path) -> ExitCode:
workspace.require_files("labels")
all_labels = read_all_labels(workspace.root)
files_to_process = _selected_images(workspace, target)
if not files_to_process:
print(f"No Cutleft images found for {target}")
return ExitCode.PARTIAL
print(
"o to open pdf, O original pdf, e to edit part, p to go back, "
"i to open the statement, click for coordinates"
)
input_queue = queue.Queue(maxsize=5)
worker = threading.Thread(
target=worker_thread,
args=(workspace.root, files_to_process, all_labels, input_queue),
daemon=True,
)
worker.start()
root = tk.Tk()
application = ImageViewer(root, workspace, set(all_labels), input_queue)
root.mainloop()
return ExitCode.PARTIAL if application.failed else ExitCode.SUCCESS
def build_parser() -> argparse.ArgumentParser:
return target_parser("Interactively verify detected label coordinates")
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
def handle(args: argparse.Namespace) -> ExitCode:
workspace, target = workspace_from_target(args)
return run(workspace, target)
return execute(parser, argv, handle)
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python plotting.py <directory_or_file>")
sys.exit(1)
input_path = Path(sys.argv[1])
files_to_process = []
if input_path.is_file():
# Correctly identify base_dir if we are in 'Copies' or 'Cutleft'
if input_path.parent.name in ["Copies", "Cutleft"]:
base_dir = input_path.parent.parent
else:
base_dir = input_path.parent
stem = input_path.stem
cutleft_dir = base_dir / "Cutleft"
img_path = cutleft_dir / f"{stem}.jpg"
if img_path.exists():
files_to_process = [img_path]
else:
# We're given something like Copie01.pdf, look for its split image parts
files_to_process = sorted(list(cutleft_dir.glob(f"{stem}_*.jpg")), key=natural_key)
else:
base_dir = input_path
cutleft_dir = base_dir / "Cutleft"
if not cutleft_dir.exists():
print(f"Error: {cutleft_dir} does not exist.")
sys.exit(1)
files_to_process = sorted(cutleft_dir.glob("*.jpg"))
try:
all_labels = read_all_labels(base_dir)
except FileNotFoundError:
all_labels = []
valid_labels_set = set(all_labels)
t = threading.Thread(target=worker_thread, args=(base_dir, files_to_process, all_labels))
t.daemon = True
t.start()
root = tk.Tk()
app = ImageViewer(root, base_dir)
root.mainloop()
raise SystemExit(main())