207 lines
7.3 KiB
Python
207 lines
7.3 KiB
Python
import sys
|
|
import os
|
|
import json
|
|
import collections
|
|
from pathlib import Path
|
|
from PIL import Image
|
|
|
|
import annotating
|
|
from annotating_with_checks import natural_key
|
|
from reading_annotations import detect_checks_and_notes, has_significant_notes
|
|
|
|
def apply_actions_and_regenerate_grouped(root_dir, data, student_id, actions, label_notes, all_labels):
|
|
"""
|
|
Modifies data based on actions, pastes label-specific note crops,
|
|
regenerates label images for consistency, saves dirty ones,
|
|
and generates Concat.jpg in the BGnot/Copie{id} directory.
|
|
"""
|
|
output_dir = os.path.join(root_dir, "BGnot", f"Copie{student_id}")
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
score_path = os.path.join(output_dir, "score.json")
|
|
labels_data = data.get(student_id, {})
|
|
|
|
# --- 1. Apply Actions to Data (Update scores / Flags for deletion) ---
|
|
actions_by_label = collections.defaultdict(list)
|
|
for a in actions:
|
|
actions_by_label[a['label']].append(a)
|
|
|
|
dirty_labels = set()
|
|
|
|
for label, acts in actions_by_label.items():
|
|
if label not in labels_data: continue
|
|
|
|
content = labels_data[label]
|
|
result = content['result']
|
|
feedbacks = result.get('feedback', [])
|
|
|
|
# Helpers to find objects by index
|
|
global_fb = [f for f in feedbacks if not f.get('box_2d')]
|
|
local_fb = [f for f in feedbacks if f.get('box_2d')]
|
|
local_fb.sort(key=lambda x: x['box_2d'][0])
|
|
|
|
for act in acts:
|
|
if act['type'] == 'score':
|
|
result['score'] = act['value']
|
|
dirty_labels.add(label)
|
|
print(f" > Updated score for {label} to {act['value']}")
|
|
|
|
elif act['type'] == 'del_global':
|
|
if act['index'] < len(global_fb):
|
|
global_fb[act['index']]["to_delete"] = True
|
|
dirty_labels.add(label)
|
|
print(f" > Deleted global feedback in {label}")
|
|
|
|
elif act['type'] in ('del_local', 'del_local_rect'):
|
|
if act['index'] < len(local_fb):
|
|
target = local_fb[act['index']]
|
|
if act['type'] == 'del_local':
|
|
target["to_delete"] = True
|
|
print(f" > Deleted local feedback in {label}")
|
|
else:
|
|
target["norectangle"] = True
|
|
print(f" > Deleted rect in {label}")
|
|
dirty_labels.add(label)
|
|
|
|
# --- 2. Process Images (Regenerate & Concatenate) ---
|
|
concat_list = []
|
|
d_notes = dict.fromkeys(all_labels, "")
|
|
|
|
# Iterate over all labels naturally to assemble a complete student profile
|
|
sorted_labels = sorted(labels_data.items(), key=lambda x: natural_key(x[0]))
|
|
|
|
for label, content in sorted_labels:
|
|
d_notes[label] = str(content['result'].get('score', 0))
|
|
|
|
pdf_path = os.path.join(root_dir, f"Copie{student_id}", f"{label}.pdf")
|
|
if not os.path.exists(pdf_path): continue
|
|
|
|
(base_img, _, _) = annotating.make_base_image(pdf_path)
|
|
|
|
# Compose uses the result object we modified in step 1
|
|
final_img, _ = annotating.compose_label_image(
|
|
base_img, label, content['result'], content['coordinates'][0],
|
|
with_error=False
|
|
)
|
|
if final_img is None:
|
|
continue
|
|
|
|
# Overlay manual notes specific to this label
|
|
has_notes = False
|
|
if label in label_notes:
|
|
sub_note = label_notes[label]
|
|
if has_significant_notes(sub_note):
|
|
has_notes = True
|
|
final_img.paste(sub_note, (0, 0), mask=sub_note)
|
|
|
|
# Save individual file if Modified (Dirty logic or visual notes)
|
|
if (label in dirty_labels) or has_notes:
|
|
save_path = os.path.join(output_dir, f"{label}.jpg")
|
|
final_img.save(save_path)
|
|
print(f" Saved dirty image: {label}.jpg")
|
|
|
|
concat_list.append(final_img)
|
|
|
|
# --- 3. Save Final Outputs ---
|
|
with open(score_path, "w") as f:
|
|
json.dump(d_notes, f, indent=4)
|
|
print(f" Saved {score_path}")
|
|
|
|
if concat_list:
|
|
max_w = max(i.width for i in concat_list)
|
|
total_h = sum(i.height for i in concat_list)
|
|
full_img = Image.new("RGB", (max_w, total_h), "white")
|
|
|
|
y = 0
|
|
for img in concat_list:
|
|
full_img.paste(img, (0, y))
|
|
y += img.height
|
|
|
|
full_img.save(os.path.join(output_dir, "Concat.jpg"))
|
|
print(f" Saved regenerated Concat.jpg")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python reading_grouped_annotations.py <Dir>")
|
|
sys.exit(1)
|
|
|
|
root_dir = sys.argv[1]
|
|
bgnot_dir = os.path.join(root_dir, "BGnot")
|
|
|
|
if not os.path.exists(bgnot_dir):
|
|
print(f"Directory {bgnot_dir} does not exist. Run annotating_by_label.py first.")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
all_labels = sorted(list(filter(None,
|
|
(Path(root_dir) / "labels")
|
|
.read_text().splitlines())),
|
|
key=natural_key)
|
|
except FileNotFoundError:
|
|
all_labels = []
|
|
|
|
# Load original data
|
|
original_data = annotating.make_dictionary(root_dir)
|
|
|
|
actions_by_student = collections.defaultdict(list)
|
|
notes_by_student = collections.defaultdict(dict)
|
|
|
|
# --- 1. Scan BGnot grouped directories and extract all checks & notes ---
|
|
for entry in os.listdir(bgnot_dir):
|
|
gdir = os.path.join(bgnot_dir, entry)
|
|
|
|
if not os.path.isdir(gdir) or entry.startswith("Copie"):
|
|
continue # Ignore files and already compiled student folders
|
|
|
|
print(f"\nScanning grouped annotations in {entry}")
|
|
actions, notes_img = detect_checks_and_notes(gdir)
|
|
|
|
bnote_path = os.path.join(gdir, "bnote.json")
|
|
if not os.path.exists(bnote_path) or notes_img is None:
|
|
continue
|
|
|
|
with open(bnote_path, "r") as f:
|
|
bnote_data = json.load(f)
|
|
|
|
# Route actions to specific students
|
|
for act in actions:
|
|
sid = str(act.get("student_id"))
|
|
if sid:
|
|
actions_by_student[sid].append(act)
|
|
|
|
# Route manual note crops to specific students and labels
|
|
for img_info in bnote_data.get("images", []):
|
|
sid = str(img_info.get("id"))
|
|
lbl = img_info.get("label")
|
|
hmin = img_info.get("hmin", 0)
|
|
hmax = img_info.get("hmax", 0)
|
|
|
|
if hmax > hmin:
|
|
crop = notes_img.crop((0, hmin, notes_img.width, hmax))
|
|
# Store it if there are pen marks on it
|
|
if has_significant_notes(crop):
|
|
notes_by_student[sid][lbl] = crop
|
|
|
|
# --- 2. Dispatch data back to students and regenerate ---
|
|
# affected_students = set(actions_by_student.keys()).union(set(notes_by_student.keys()))
|
|
|
|
# if not affected_students:
|
|
# print("\nNo changes detected in any grouped annotations.")
|
|
# sys.exit(0)
|
|
|
|
# for sid in sorted(affected_students, key=natural_key):
|
|
for sid in sorted(original_data.keys(), key=natural_key):
|
|
if sid not in original_data:
|
|
continue
|
|
|
|
print(f"\nProcessing compilation for: Copie{sid}")
|
|
apply_actions_and_regenerate_grouped(
|
|
root_dir,
|
|
original_data,
|
|
sid,
|
|
actions_by_student[sid],
|
|
notes_by_student[sid],
|
|
all_labels
|
|
)
|