Compare commits
3
Commits
2d0f7a219f
...
698ac4240c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
698ac4240c | ||
|
|
bd1964ffcd | ||
|
|
48602c24f1 |
No files matched your search
@@ -4,6 +4,7 @@ import argparse
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from collections.abc import Sequence
|
||||
@@ -175,6 +176,10 @@ def render_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=N
|
||||
final_img.alpha_composite(img)
|
||||
return final_img
|
||||
|
||||
class AnnotationLatexError(RuntimeError):
|
||||
"""A feedback comment failed to render; diagnostics have been retained."""
|
||||
|
||||
|
||||
def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_lines=None, fontsize=19):
|
||||
dpi = 100
|
||||
width_in = width_px / dpi
|
||||
@@ -194,16 +199,37 @@ def render_real_latex_text(text, width_px, bg_color=(255, 255, 255, 255), max_li
|
||||
f.write(latex_template)
|
||||
|
||||
# Compile to PDF
|
||||
subprocess.run(
|
||||
compilation = subprocess.run(
|
||||
['pdflatex', '-interaction=nonstopmode', 'text.tex'],
|
||||
cwd=temp_dir,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
check=False,
|
||||
)
|
||||
|
||||
if not os.path.exists(pdf_path):
|
||||
raise RuntimeError("LaTeX compilation failed. Check your LaTeX syntax.")
|
||||
diagnostics = Path(tempfile.mkdtemp(prefix="copienator-latex-error-"))
|
||||
shutil.copy2(tex_path, diagnostics / "text.tex")
|
||||
log_path = Path(temp_dir) / "text.log"
|
||||
output = (compilation.stdout or b"").decode("utf-8", errors="replace")
|
||||
(diagnostics / "compiler-output.txt").write_text(output, encoding="utf-8")
|
||||
if log_path.exists():
|
||||
shutil.copy2(log_path, diagnostics / "text.log")
|
||||
log = log_path.read_text(encoding="utf-8", errors="replace")
|
||||
else:
|
||||
log = output
|
||||
lines = log.splitlines()
|
||||
first_error = next((i for i, line in enumerate(lines) if line.startswith("!")), max(0, len(lines) - 20))
|
||||
excerpt = "\n".join(lines[first_error:first_error + 25])
|
||||
raise AnnotationLatexError(
|
||||
f"Échec de compilation LaTeX (code {compilation.returncode}).\n"
|
||||
f"Commentaire LaTeX fautif :\n{text}\n\n"
|
||||
f"Extrait du journal :\n{excerpt}\n\n"
|
||||
f"Source LaTeX complète : {diagnostics / 'text.tex'}\n"
|
||||
f"Diagnostics conservés dans : {diagnostics}\n"
|
||||
"Modifiez le commentaire concerné dans correction.json, enregistrez le fichier, "
|
||||
"puis relancez cette étape de génération des annotations."
|
||||
)
|
||||
|
||||
# Convert PDF to grayscale (ignoring pdf2image's broken transparency)
|
||||
images = convert_from_path(pdf_path, dpi=dpi)
|
||||
@@ -295,7 +321,7 @@ def compose_label_image(base_img, label, result, hmin,
|
||||
with_error=True,
|
||||
with_empty=False,
|
||||
more_right=False,
|
||||
with_id=None):
|
||||
with_id=None, source_pdf=None):
|
||||
"""
|
||||
Composes the final image with annotations.
|
||||
|
||||
@@ -309,6 +335,21 @@ def compose_label_image(base_img, label, result, hmin,
|
||||
called when elements are placed. Used for checkboxes.
|
||||
"""
|
||||
|
||||
original_render = render_fn
|
||||
|
||||
def render_fn(text, *args, **kwargs):
|
||||
try:
|
||||
return original_render(text, *args, **kwargs)
|
||||
except AnnotationLatexError as exc:
|
||||
context = f"Question : {label}"
|
||||
if source_pdf is not None:
|
||||
source = Path(source_pdf)
|
||||
context = f"Copie : {source.parent.name} — {context}\nPDF source : {source}"
|
||||
context += f"\nFichier à modifier : {source.parent.parent.parent / 'correction.json'}"
|
||||
elif with_id is not None:
|
||||
context = f"Copie : {with_id} — {context}"
|
||||
raise AnnotationLatexError(f"{context}\n{exc}") from exc
|
||||
|
||||
left_pad = 0
|
||||
if base_img.width < TARGET_MIN_WIDTH:
|
||||
total_missing = TARGET_MIN_WIDTH - base_img.width
|
||||
@@ -489,6 +530,7 @@ def process_student(student_id, labels_data, root_dir, all_labels, overwrite):
|
||||
coordinates[0],
|
||||
with_empty=True,
|
||||
render_fn=render_real_latex_text,
|
||||
source_pdf=pdf_full_path,
|
||||
)
|
||||
final_img.save(staging / f"{label}.jpg")
|
||||
if result.get('error', "") != "empty-answer":
|
||||
|
||||
@@ -47,6 +47,7 @@ def render_item(item):
|
||||
draw_callback=checkbox_renderer.callback,
|
||||
more_right=True,
|
||||
with_id=student_id,
|
||||
source_pdf=pdf_path,
|
||||
)
|
||||
if final_image is None:
|
||||
return None
|
||||
@@ -233,15 +234,17 @@ def _unique_prefixes(groups: list[list[str]]) -> list[tuple[str, list[str]]]:
|
||||
candidate = f"{safe_labels[0]}+"
|
||||
if natural_key(candidate) > natural_key(previous):
|
||||
base = candidate
|
||||
prefix = base.removesuffix("i")
|
||||
base = base.removesuffix("i")
|
||||
# Keep related groups under the same short stem (I), I)-2, ...).
|
||||
# Normalize before allocating a suffix so no later rewrite can collide.
|
||||
if previous and base.startswith(previous):
|
||||
base = previous
|
||||
prefix = base
|
||||
counter = 2
|
||||
while prefix in used:
|
||||
prefix = f"{base}-{counter}"
|
||||
counter += 1
|
||||
if counter == 2 and previous and previous in prefix:
|
||||
prefix = f"{previous}-{counter}"
|
||||
elif counter == 2:
|
||||
previous = prefix
|
||||
previous = base
|
||||
used.add(prefix)
|
||||
result.append((prefix, labels))
|
||||
return result
|
||||
@@ -249,14 +252,16 @@ def _unique_prefixes(groups: list[list[str]]) -> list[tuple[str, list[str]]]:
|
||||
|
||||
def _existing_group_state(
|
||||
output_root: Path,
|
||||
prefix: str,
|
||||
prefix: str | None = None,
|
||||
) -> tuple[set[tuple[str, str]], int]:
|
||||
existing_items: set[tuple[str, str]] = set()
|
||||
maximum_group = 0
|
||||
if not output_root.is_dir():
|
||||
return existing_items, maximum_group
|
||||
for directory in output_root.iterdir():
|
||||
if not directory.is_dir() or not directory.name.startswith(f"{prefix} G"):
|
||||
if not directory.is_dir():
|
||||
continue
|
||||
if prefix is not None and not directory.name.startswith(f"{prefix} G"):
|
||||
continue
|
||||
try:
|
||||
maximum_group = max(maximum_group, int(directory.name.split(" G")[-1]))
|
||||
@@ -317,8 +322,11 @@ def _generate_groups(
|
||||
) -> tuple[int, bool]:
|
||||
generated = 0
|
||||
problems = False
|
||||
# Resume by answer identity across all prefixes, including legacy names.
|
||||
# Changing the naming scheme must not regenerate already reviewed PDFs.
|
||||
existing_items = _existing_group_state(output_root)[0] if resume else set()
|
||||
for prefix, labels in _unique_prefixes(groups):
|
||||
existing_items, maximum_group = (
|
||||
_, maximum_group = (
|
||||
_existing_group_state(output_root, prefix) if resume else (set(), 0)
|
||||
)
|
||||
items = [
|
||||
|
||||
@@ -184,6 +184,7 @@ def _render_student(
|
||||
content["result"],
|
||||
content["coordinates"][0],
|
||||
draw_callback=checkbox_renderer.callback,
|
||||
source_pdf=pdf_path,
|
||||
)
|
||||
if final_image is None:
|
||||
problems = True
|
||||
|
||||
@@ -716,7 +716,7 @@ def process_exam(
|
||||
|
||||
with io.StringIO() as f, io.StringIO() as f_full:
|
||||
|
||||
header = "# Edit labels. Modify groups (---). Ensure label uniqueness (XX). Duplicate CONTEXT.\n\n"
|
||||
header = "# Edit labels. Modify groups (---). Ensure label uniqueness (XX). Duplicate/modify CONTEXT.\n\n"
|
||||
f.write(header)
|
||||
f_full.write(header)
|
||||
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Optional detector explanations and independent, manually annotated crop checks.
|
||||
|
||||
Annotations are evaluation data: they are never passed to the ink detector.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
|
||||
MASK_DESCRIPTIONS = {
|
||||
'isolated_edge_marks': 'Small border marks excluded by same-page edge position and isolation',
|
||||
'confirmed_ruling': 'Confirmed neutral-paper ruling pixels',
|
||||
'paper_suppressed': 'Foreground pixels suppressed by paper cleanup',
|
||||
'rejected_artifacts': 'Pixels rejected as specks, thin ruling remnants or repeated holes',
|
||||
'repeated_holes': 'Pixels rejected as repeated hole components during initial cleanup',
|
||||
'refinement_removed': 'Pixels removed by the large-blank refinement',
|
||||
'ruling_residuals': 'Thin seed-candidate remnants adjoining confirmed suppressed paper pixels',
|
||||
'strong_seeds': 'Strong seed pixels, including any surviving paper or pagination',
|
||||
'weak_candidates': 'Weak candidates before seed proximity and connectivity checks',
|
||||
'foreground': 'All retained pixels that actually determine the crop',
|
||||
'recovered_weak': 'Weak pixels recovered around strong seeds',
|
||||
'unretained_weak': 'Weak candidates not retained; these may be ink OR paper',
|
||||
}
|
||||
|
||||
|
||||
def check_protected_regions(regions, result, *, width_mm, height_mm, shape):
|
||||
"""Fail if either crop boundary crosses a reviewed content rectangle.
|
||||
|
||||
Rectangles are [left, top, right, bottom] in original displayed millimetres.
|
||||
They conservatively protect their entire area, including enclosed whitespace.
|
||||
"""
|
||||
h = shape[0]
|
||||
top = result['top_px'] / h * height_mm
|
||||
bottom = result['bottom_px'] / h * height_mm
|
||||
if not 0 <= result['top_px'] <= result['bottom_px'] <= h:
|
||||
raise ValueError('Invalid crop bounds')
|
||||
rows = []
|
||||
seen = set()
|
||||
for region in regions:
|
||||
name = region['id']
|
||||
if name in seen:
|
||||
raise ValueError(f'Duplicate protected region: {name}')
|
||||
seen.add(name)
|
||||
x0, y0, x1, y1 = region['bbox_mm']
|
||||
if not all(np.isfinite([x0,y0,x1,y1])) or not (
|
||||
0 <= x0 < x1 <= width_mm and 0 <= y0 < y1 <= height_mm):
|
||||
raise ValueError(f'Invalid protected rectangle: {name}')
|
||||
top_loss = max(0.0, min(y1, top)-y0)
|
||||
bottom_loss = max(0.0, y1-max(y0, bottom))
|
||||
rows.append(dict(id=name, label=region['label'], bbox_mm=region['bbox_mm'],
|
||||
passed=top_loss <= 1e-9 and bottom_loss <= 1e-9,
|
||||
clipped_edges=[edge for edge, loss in [('top',top_loss),('bottom',bottom_loss)] if loss > 1e-9],
|
||||
clipped_height_mm=round(top_loss+bottom_loss,4),
|
||||
clipped_area_fraction=round(min(1.0,(top_loss+bottom_loss)/(y1-y0)),4)))
|
||||
return rows
|
||||
|
||||
|
||||
def save_diagnostics(rgb, result, diagnostics, destination: Path, *, regions=(),
|
||||
width_mm, height_mm):
|
||||
"""Export exact binary masks, readable overlays and JSON boundary explanations."""
|
||||
destination.mkdir(parents=True, exist_ok=True)
|
||||
masks = diagnostics['masks']
|
||||
files = {}
|
||||
for name, mask in masks.items():
|
||||
if mask.shape != rgb.shape[:2] or mask.dtype != np.bool_:
|
||||
raise ValueError(f'Invalid diagnostic mask: {name}')
|
||||
path = destination / f'{name}.png'
|
||||
Image.fromarray(mask).save(path)
|
||||
files[name] = path.name
|
||||
views = {
|
||||
'strong-seeds': [('strong_seeds',(220,50,20))],
|
||||
'retained-ink': [('strong_seeds',(220,50,20)),('recovered_weak',(245,140,0))],
|
||||
'unretained-weak': [('unretained_weak',(155,40,200))],
|
||||
'paper-and-artifacts': [('paper_suppressed',(20,140,200)),
|
||||
('rejected_artifacts',(155,40,200)),('ruling_residuals',(155,40,200)),
|
||||
('isolated_edge_marks',(220,105,0)),
|
||||
('refinement_removed',(245,140,0))],
|
||||
'protected-content': [],
|
||||
}
|
||||
preview = Image.fromarray(rgb)
|
||||
preview.thumbnail((1000,1400))
|
||||
sy = preview.height / rgb.shape[0]
|
||||
overlays = {}
|
||||
for name, layers in views.items():
|
||||
canvas = np.array(preview).astype(np.float32)
|
||||
for mask_name, colour in layers:
|
||||
# Mark every source pixel represented by a preview pixel; thin strokes
|
||||
# must not disappear merely because a preview was downsampled.
|
||||
mask = masks[mask_name]
|
||||
coverage = cv2.resize(mask.astype(np.float32), preview.size, interpolation=cv2.INTER_AREA) > 0
|
||||
canvas[coverage] = .3*canvas[coverage] + .7*np.array(colour)
|
||||
image = Image.fromarray(np.clip(canvas,0,255).astype(np.uint8))
|
||||
draw = ImageDraw.Draw(image)
|
||||
for region in regions:
|
||||
x0,y0,x1,y1 = region['bbox_mm']
|
||||
box = (x0/width_mm*image.width,y0/height_mm*image.height,
|
||||
x1/width_mm*image.width,y1/height_mm*image.height)
|
||||
clipped = result['top_px']*sy > box[1] or result['bottom_px']*sy < box[3]
|
||||
colour = (200,0,160) if clipped else (0,150,170)
|
||||
draw.rectangle(box, outline=colour, width=2)
|
||||
for y in (result['top_px']*sy, result['bottom_px']*sy):
|
||||
draw.line((0,min(y,image.height-1),image.width,min(y,image.height-1)),fill=(215,20,20),width=2)
|
||||
path = destination / f'{name}.jpg'
|
||||
image.save(path, quality=90)
|
||||
overlays[name] = path.name
|
||||
data = {key:value for key,value in diagnostics.items() if key != 'masks'}
|
||||
data.update(detector=result, mask_files=files, mask_descriptions=MASK_DESCRIPTIONS,
|
||||
overlays=overlays, protected_regions=list(regions))
|
||||
report = destination/'explanation.json'
|
||||
report.write_text(json.dumps(data,indent=2)+'\n')
|
||||
return dict(report=report.name, overlays=overlays, masks=files)
|
||||
@@ -24,7 +24,6 @@ from copienator.ink_detection import detect_bounds
|
||||
|
||||
LINES_PER_PAGE = 36
|
||||
MINIMUM_HEIGHT_LINES = 10
|
||||
IGNORED_BOTTOM_LINES = 0.75
|
||||
MINIMUM_CROP_LINES = 4
|
||||
|
||||
|
||||
@@ -93,13 +92,8 @@ def process_exercise_pdf(
|
||||
pixmap.height, pixmap.width, 3
|
||||
)
|
||||
pixels_per_point = pixmap.height / visible_height
|
||||
ignored_pixels = min(
|
||||
pixmap.height - 1,
|
||||
round(IGNORED_BOTTOM_LINES * line_points * pixels_per_point),
|
||||
)
|
||||
analysis_bottom = pixmap.height - ignored_pixels
|
||||
detection = detect_bounds(
|
||||
rgb[:analysis_bottom], dpi=dpi, padding_mm=padding_mm, min_crop_mm=0
|
||||
rgb, dpi=dpi, padding_mm=padding_mm, min_crop_mm=0
|
||||
)
|
||||
proposed_bottom = detection["bottom_px"]
|
||||
removed_points = (pixmap.height - proposed_bottom) / pixels_per_point
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Group small isolated border marks using evidence from one displayed page.
|
||||
|
||||
This identifies dispensable border marks, without recognizing pagination or
|
||||
consulting other pages. Component IDs are local to the caller's foreground.
|
||||
Long paper-line candidates are excluded from grouping but remain in the crop.
|
||||
"""
|
||||
import numpy as np
|
||||
|
||||
|
||||
def classify_edge_marks(labels, stats, eligible, rgb, dpi):
|
||||
px = dpi/25.4
|
||||
height, width = labels.shape
|
||||
components = []
|
||||
for i, (x, y, w, h, area) in enumerate(stats[1:], 1):
|
||||
if not eligible[i] or area < .15*px*px:
|
||||
continue
|
||||
# Exclusion from candidate grouping ONLY. Pixels remain in the crop mask.
|
||||
if ((h > 25*px and w < 3*px) or (w > 25*px and h < .6*px)):
|
||||
continue
|
||||
components.append(i)
|
||||
parent = {i:i for i in components}
|
||||
def find(i):
|
||||
while parent[i] != i:
|
||||
parent[i] = parent[parent[i]]
|
||||
i = parent[i]
|
||||
return i
|
||||
# A y sweep bounds the search on noisy pages; uncertain large groups stay.
|
||||
edge_components = [i for i in components if stats[i,2] <= 40*px
|
||||
and stats[i,3] <= 18*px and
|
||||
(stats[i,1]+stats[i,3] <= 40*px
|
||||
or stats[i,1] >= height-40*px)]
|
||||
if len(edge_components) > 512:
|
||||
return [], []
|
||||
colours = {}
|
||||
if rgb is not None:
|
||||
for i in edge_components:
|
||||
x,y,w,h,_ = stats[i]
|
||||
values = rgb[y:y+h,x:x+w][labels[y:y+h,x:x+w]==i]
|
||||
colours[i] = float(np.median(values.max(axis=1)-values.min(axis=1)))
|
||||
ordered = sorted(edge_components, key=lambda i:stats[i,1])
|
||||
for a, i in enumerate(ordered):
|
||||
x,y,w,h,_ = stats[i]
|
||||
for j in ordered[a+1:]:
|
||||
xx,yy,ww,hh,_ = stats[j]
|
||||
if yy > y+h+1.5*px:
|
||||
break
|
||||
gap_x = max(0, xx-(x+w), x-(xx+ww))
|
||||
gap_y = max(0, yy-(y+h), y-(yy+hh))
|
||||
incompatible = (rgb is not None and
|
||||
((colours[i] < 35 and colours[j] > 60)
|
||||
or (colours[j] < 35 and colours[i] > 60)))
|
||||
if not incompatible and gap_x <= 3*px and gap_y <= 1.5*px:
|
||||
parent[find(j)] = find(i)
|
||||
groups = {}
|
||||
for i in edge_components:
|
||||
groups.setdefault(find(i), []).append(i)
|
||||
rows = []
|
||||
for ids in groups.values():
|
||||
s=stats[ids];x,y=s[:,:2].min(axis=0);right,bottom=(s[:,:2]+s[:,2:4]).max(axis=0)
|
||||
w,h=right-x,bottom-y
|
||||
if not (w <= 40*px and h <= 18*px and h >= 1.5*px):
|
||||
continue
|
||||
edge = 'top' if bottom <= 40*px else 'bottom' if y >= height-40*px else None
|
||||
if edge is None:
|
||||
continue
|
||||
underline_ids=[]
|
||||
for i in ids:
|
||||
xx,yy,ww,hh,area=stats[i]
|
||||
if ww > 3*hh and ww >= .65*w and yy+hh >= bottom-1.5*px:
|
||||
underline_ids.append(i)
|
||||
lateral = (x+w/2 < width*.25 or x+w/2 > width*.75)
|
||||
# Check separation from other compact strokes across the full width.
|
||||
# Confirmed long line candidates remain in the actual crop mask.
|
||||
body = [i for i in components if i not in ids and stats[i,3] < 18*px]
|
||||
edge_distance = (y if edge == 'top' else height-bottom)/px
|
||||
if body:
|
||||
gap = ((min(stats[i,1] for i in body)-(y+h))/px if edge=='top' else
|
||||
(y-max(stats[i,1]+stats[i,3] for i in body))/px)
|
||||
else:
|
||||
gap = None
|
||||
truncated = (x <= px or y <= px or right >= width-px or bottom >= height-px)
|
||||
reasons=[]
|
||||
if lateral:reasons.append('lateral-placement')
|
||||
if gap is not None and gap >= 20:reasons.append('all-column-stroke-gap>=20mm')
|
||||
if edge_distance <= 15:reasons.append('within-15mm-of-edge')
|
||||
if underline_ids:reasons.append('underline-protection')
|
||||
if truncated:reasons.append('scan-edge-truncated')
|
||||
proposed = (lateral and edge_distance <= 15 and gap is not None
|
||||
and gap >= 20 and not underline_ids and not truncated)
|
||||
rows.append(dict(bbox_mm=[round(float(v/px),2) for v in [x,y,x+w,y+h]],
|
||||
component_ids=ids,edge=edge,edge_distance_mm=round(edge_distance,2),gap_mm=None if gap is None else round(gap,2),
|
||||
evidence=reasons,proposed_pagination=bool(proposed)))
|
||||
removed = [i for row in rows if row['proposed_pagination'] for i in row['component_ids']]
|
||||
return removed, rows
|
||||
|
||||
+151
-13
@@ -10,6 +10,8 @@ from functools import lru_cache
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from copienator.edge_marks import classify_edge_marks
|
||||
from copienator.paper_ruling import ruling_suppression, _edge_connected_family
|
||||
from copienator.paper_background import _skew, _ruling, _foreground, _content_mask
|
||||
|
||||
|
||||
@@ -136,7 +138,8 @@ def _large_blank_ink(gray: np.ndarray, clean: np.ndarray, dpi: float):
|
||||
return result, len(holes), changed
|
||||
|
||||
|
||||
def _neutral_paper_foreground(gray: np.ndarray, chroma: np.ndarray, dpi: float):
|
||||
def _neutral_paper_foreground(gray: np.ndarray, chroma: np.ndarray, dpi: float,
|
||||
*, diagnostics: dict | None = None):
|
||||
"""Clean confirmed dark ruling before it can seed whole pages.
|
||||
|
||||
Returns None for ordinary ink components or unconfirmed paper geometry.
|
||||
@@ -167,15 +170,26 @@ def _neutral_paper_foreground(gray: np.ndarray, chroma: np.ndarray, dpi: float):
|
||||
cv2.THRESH_BINARY_INV,block,9)
|
||||
horizontal, nh = _ruling(binary,True)
|
||||
vertical, nv = _ruling(binary,False)
|
||||
if not (nh or nv):
|
||||
if not ((nh and _edge_connected_family(binary,horizontal,True))
|
||||
or (nv and _edge_connected_family(binary,vertical,False))):
|
||||
return None, dict(paper_cleanup=False)
|
||||
# The permissive mask retains faint, isolated marks.
|
||||
# Bands identify where ruling is expected, but only actual dark pixels
|
||||
# inside them may be suppressed. Erasing the complete band loses faint
|
||||
# writing alongside a dark grid line.
|
||||
paper_pixels = (horizontal|vertical) & ((normalized < 160).astype(np.uint8)*255)
|
||||
clean, holes = _content_mask(_foreground(normalized,dpi,9,paper_pixels),dpi,nh,nv)
|
||||
stages = {} if diagnostics is not None else None
|
||||
foreground = _foreground(normalized,dpi,9,paper_pixels, diagnostics=stages)
|
||||
clean, holes = _content_mask(foreground,dpi,nh,nv, diagnostics=stages)
|
||||
before_refinement = clean.copy() if diagnostics is not None else None
|
||||
clean, repeated_holes, refined = _large_blank_ink(deskewed, clean, dpi)
|
||||
if diagnostics is not None:
|
||||
stages['confirmed_ruling'] = paper_pixels > 0
|
||||
stages['refinement_removed'] = (before_refinement > 0) & (clean == 0)
|
||||
inverse = cv2.invertAffineTransform(matrix)
|
||||
for name, mask in stages.items():
|
||||
diagnostics[name] = cv2.warpAffine(mask.astype(np.uint8), inverse,
|
||||
(width,height), flags=cv2.INTER_NEAREST, borderValue=0) > 0
|
||||
restored = cv2.warpAffine(clean, cv2.invertAffineTransform(matrix),
|
||||
(width,height), flags=cv2.INTER_NEAREST, borderValue=0)
|
||||
return restored > 0, dict(paper_cleanup=True, angle_deg=round(angle,3),
|
||||
@@ -184,22 +198,45 @@ def _neutral_paper_foreground(gray: np.ndarray, chroma: np.ndarray, dpi: float):
|
||||
large_blank_refinement=refined)
|
||||
|
||||
|
||||
def _seed_mask(mask: np.ndarray, px: float) -> np.ndarray:
|
||||
def _seed_mask(mask: np.ndarray, px: float, *, paper_nearby=None,
|
||||
diagnostics=None) -> np.ndarray:
|
||||
n, labels, stats, _ = cv2.connectedComponentsWithStats(mask.astype(np.uint8), 8)
|
||||
keep = np.zeros(n, bool)
|
||||
keep[1:] = ((stats[1:, 4] >= max(4, .12*px*px))
|
||||
& (stats[1:, 2] >= max(2, round(.35*px)))
|
||||
& (stats[1:, 3] >= max(2, round(.35*px))))
|
||||
if paper_nearby is not None:
|
||||
w, h = stats[:,2], stats[:,3]
|
||||
thin = max(2, round(.45*px))
|
||||
aligned = ((np.minimum(w,h) <= thin)
|
||||
& (np.maximum(w,h) >= 3*np.minimum(w,h)))
|
||||
overlap = np.bincount(labels[paper_nearby], minlength=n)
|
||||
remnants = keep & aligned & (overlap >= .8*stats[:,4])
|
||||
keep[remnants] = False
|
||||
if diagnostics is not None:
|
||||
diagnostics['ruling_residuals'] |= remnants[labels]
|
||||
return keep[labels]
|
||||
|
||||
|
||||
def detect_bounds(rgb: np.ndarray, dpi: float = 200, padding_mm: float = 6,
|
||||
min_crop_mm: float = 5) -> dict:
|
||||
min_crop_mm: float = 5, *, diagnostics: dict | None = None,
|
||||
colour_cleanup: bool = False, edge_mark_cleanup: bool = True) -> dict:
|
||||
"""Locate strong ink, recover adjacent faint strokes, and retain padding.
|
||||
|
||||
Neutral punched-hole shadows usually have neither sufficient chroma nor
|
||||
sufficient darkness to seed a region. Nothing is discarded merely because
|
||||
it is in a side margin. Two seed thresholds expose unstable boundaries.
|
||||
|
||||
Small isolated side-margin marks near the physical top/bottom edge are
|
||||
excluded using same-page position and separation. Marks near text remain.
|
||||
edge_mark_cleanup=False is available for evaluation comparisons.
|
||||
|
||||
Experimental colour cleanup is opt-in because its full-resolution model
|
||||
is too expensive for the default PDF processing path.
|
||||
|
||||
Optional diagnostics are written to a separate caller-owned dictionary.
|
||||
Its masks use the original image coordinates and never influence decisions
|
||||
or enter the JSON-serializable return value.
|
||||
"""
|
||||
px = dpi/25.4
|
||||
h, w = rgb.shape[:2]
|
||||
@@ -214,14 +251,38 @@ def detect_bounds(rgb: np.ndarray, dpi: float = 200, padding_mm: float = 6,
|
||||
vertical = cv2.morphologyEx(darkness, cv2.MORPH_OPEN,
|
||||
np.ones((length, 1), np.uint8))
|
||||
residual = cv2.subtract(darkness, np.maximum(horizontal, vertical))
|
||||
# Strong strokes bypass the line-background estimate entirely: even a long
|
||||
# isolated black or coloured fraction bar must survive. Local contrast is
|
||||
# used only to recover weaker surrounding strokes.
|
||||
# The directional estimate alone cannot veto a strong fraction bar. Only
|
||||
# independently confirmed paper geometry may suppress a strong candidate.
|
||||
neutral_ink = gray < 95
|
||||
cleaned, paper_info = _neutral_paper_foreground(gray, chroma, dpi)
|
||||
if diagnostics is not None:
|
||||
diagnostics.clear()
|
||||
diagnostics['masks'] = {}
|
||||
diagnostics['thresholds'] = []
|
||||
masks = diagnostics['masks'] if diagnostics is not None else None
|
||||
cleaned, paper_info = _neutral_paper_foreground(gray, chroma, dpi, diagnostics=masks)
|
||||
if cleaned is not None:
|
||||
neutral_ink = cleaned
|
||||
# Adaptive neutral cleanup includes faint paper fragments. Such pixels
|
||||
# need local contrast or substantial darkness before they seed a crop.
|
||||
neutral_ink = cleaned & ((residual > 25) | (gray < 95))
|
||||
if colour_cleanup:
|
||||
suppressed, ruling_info = ruling_suppression(rgb, dpi)
|
||||
else:
|
||||
suppressed, ruling_info = np.zeros((h,w), bool), {}
|
||||
paper_info.update(ruling_info)
|
||||
neutral_ink &= ~suppressed
|
||||
paper_nearby = (cv2.dilate(suppressed.astype(np.uint8),
|
||||
np.ones((3,3),np.uint8)) > 0) if np.any(suppressed) else None
|
||||
if masks is not None:
|
||||
masks['paper_suppressed'] = masks.get('paper_suppressed', np.zeros((h,w), bool)) | suppressed
|
||||
masks['confirmed_ruling'] = masks.get('confirmed_ruling', np.zeros((h,w), bool)) | suppressed
|
||||
masks['ruling_residuals'] = np.zeros((h,w), bool)
|
||||
masks['isolated_edge_marks'] = np.zeros((h,w), bool)
|
||||
weak = ((chroma > 60) | ((gray < 175) & (residual > 25))).astype(np.uint8)
|
||||
weak[suppressed] = 0
|
||||
if masks is not None:
|
||||
masks['strong_seeds'] = np.zeros((h,w), bool)
|
||||
masks['foreground'] = np.zeros((h,w), bool)
|
||||
masks['weak_candidates'] = weak > 0
|
||||
radius = max(1, round(2*px))
|
||||
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2*radius+1, 2*radius+1))
|
||||
extents = []
|
||||
@@ -232,26 +293,67 @@ def detect_bounds(rgb: np.ndarray, dpi: float = 200, padding_mm: float = 6,
|
||||
counts.append(counts[0])
|
||||
if extents:
|
||||
extents.append(extents[0])
|
||||
if diagnostics is not None:
|
||||
diagnostics['thresholds'].append(dict(
|
||||
diagnostics['thresholds'][0], threshold=115, reused=True))
|
||||
continue
|
||||
seeds = _seed_mask((chroma > threshold) | neutral_ink, px)
|
||||
seeds = _seed_mask(((chroma > threshold) | neutral_ink) & ~suppressed, px,
|
||||
paper_nearby=paper_nearby, diagnostics=masks)
|
||||
counts.append(int(np.count_nonzero(seeds)))
|
||||
if not np.any(seeds):
|
||||
if diagnostics is not None:
|
||||
diagnostics['thresholds'].append(dict(threshold=threshold,
|
||||
seed_pixels=0, extent=None, limiting_components={'top': [], 'bottom': []}))
|
||||
continue
|
||||
# Limit weak recovery to a physical neighbourhood: faint grid lines
|
||||
# connected to a letter cannot grow into a full-page foreground mask.
|
||||
nearby = cv2.dilate(seeds.astype(np.uint8), kernel)
|
||||
candidate = (weak & nearby) | seeds.astype(np.uint8)
|
||||
n, labels = cv2.connectedComponents(candidate, 8)
|
||||
if edge_mark_cleanup:
|
||||
n, labels, stats, _ = cv2.connectedComponentsWithStats(candidate, 8)
|
||||
else:
|
||||
n, labels = cv2.connectedComponents(candidate, 8)
|
||||
seeded_labels = np.zeros(n, bool)
|
||||
seeded_labels[np.unique(labels[seeds])] = True
|
||||
seeded_labels[0] = False
|
||||
ys = np.flatnonzero(np.any(seeded_labels[labels], axis=1))
|
||||
removed, groups = (classify_edge_marks(labels, stats, seeded_labels, rgb, dpi)
|
||||
if edge_mark_cleanup else ([], []))
|
||||
if removed:
|
||||
remaining = seeded_labels.copy()
|
||||
remaining[removed] = False
|
||||
# Keep the original ink bounds if no foreground would remain.
|
||||
if np.any(remaining):
|
||||
if masks is not None:
|
||||
discarded = np.zeros(n,bool)
|
||||
discarded[removed] = True
|
||||
masks['isolated_edge_marks'] |= discarded[labels]
|
||||
seeded_labels = remaining
|
||||
retained = seeded_labels[labels]
|
||||
if masks is not None:
|
||||
masks['strong_seeds'] |= seeds & retained
|
||||
ys = np.flatnonzero(np.any(retained, axis=1))
|
||||
extents.append((int(ys.min()), int(ys.max())+1))
|
||||
if diagnostics is not None:
|
||||
masks['foreground'] |= retained
|
||||
diagnostics['thresholds'].append(_explain_extent(retained, seeds,
|
||||
chroma, neutral_ink, threshold, extents[-1], px))
|
||||
diagnostics['thresholds'][-1]['edge_mark_groups'] = groups
|
||||
result = dict(top_px=0, bottom_px=h, angle_deg=None,
|
||||
horizontal_lines=0, vertical_lines=0, edge_artifacts=0,
|
||||
detector="ink", seed_pixels=counts, status="review-no-ink-seeds")
|
||||
result.update(paper_info)
|
||||
if diagnostics is not None:
|
||||
masks['recovered_weak'] = masks['foreground'] & ~masks['strong_seeds']
|
||||
masks['unretained_weak'] = masks['weak_candidates'] & ~masks['foreground']
|
||||
for name in ('confirmed_ruling', 'paper_suppressed', 'rejected_artifacts',
|
||||
'repeated_holes', 'refinement_removed'):
|
||||
masks.setdefault(name, np.zeros((h,w), bool))
|
||||
diagnostics['shape'] = [h,w]
|
||||
diagnostics['mask_pixels'] = {name: int(mask.sum()) for name, mask in masks.items()}
|
||||
if not extents:
|
||||
if diagnostics is not None:
|
||||
diagnostics['edges'] = {edge: {'reason': 'no-reliable-ink-seeds',
|
||||
'boundary_px': 0 if edge == 'top' else h} for edge in ('top', 'bottom')}
|
||||
return result
|
||||
pad = padding_mm*px
|
||||
top = max(0, int(np.floor(min(e[0] for e in extents)-pad)))
|
||||
@@ -263,6 +365,7 @@ def detect_bounds(rgb: np.ndarray, dpi: float = 200, padding_mm: float = 6,
|
||||
for edge, name in ((0, 'top'), (1, 'bottom')):
|
||||
if abs(extents[0][edge]-extents[1][edge]) > max(pad, 3*px):
|
||||
uncertain.append(name)
|
||||
proposed_top, proposed_bottom = top, bottom
|
||||
if top < min_crop_mm*px:
|
||||
top = 0
|
||||
if h-bottom < min_crop_mm*px:
|
||||
@@ -270,4 +373,39 @@ def detect_bounds(rgb: np.ndarray, dpi: float = 200, padding_mm: float = 6,
|
||||
result.update(top_px=top, bottom_px=bottom,
|
||||
status=('review-'+'-'.join(uncertain) if uncertain else
|
||||
'cropped' if top or bottom < h else 'unchanged'))
|
||||
if diagnostics is not None:
|
||||
diagnostics['edges'] = {}
|
||||
for edge, boundary, proposed, extent in (
|
||||
('top', top, proposed_top, min(e[0] for e in extents)),
|
||||
('bottom', bottom, proposed_bottom, max(e[1] for e in extents))):
|
||||
removal = boundary if edge == 'top' else h-boundary
|
||||
reason = 'crop-to-retained-ink-with-padding' if removal else (
|
||||
'below-minimum-crop' if boundary != proposed else 'ink-or-padding-reaches-edge')
|
||||
diagnostics['edges'][edge] = dict(boundary_px=boundary,
|
||||
proposed_boundary_px=proposed, ink_extent_px=extent,
|
||||
reason=reason, threshold_disagreement=edge in uncertain,
|
||||
limiting_thresholds=[i['threshold'] for i in diagnostics['thresholds']
|
||||
if i['extent'] is not None and i['extent'][0 if edge == 'top' else 1] == extent])
|
||||
return result
|
||||
|
||||
|
||||
def _explain_extent(retained, seeds, chroma, neutral, threshold, extent, px):
|
||||
"""Describe actual boundary-setting regions; infer no handwriting semantics."""
|
||||
_, labels, stats, _ = cv2.connectedComponentsWithStats(retained.astype(np.uint8), 8)
|
||||
limiting = {}
|
||||
for edge, row in (('top', extent[0]), ('bottom', extent[1]-1)):
|
||||
ids = [int(i) for i in np.unique(labels[row]) if i]
|
||||
components = []
|
||||
for identifier in ids:
|
||||
x,y,w,h,area = map(int, stats[identifier])
|
||||
region = labels[y:y+h,x:x+w] == identifier
|
||||
strong = seeds[y:y+h,x:x+w] & region
|
||||
components.append(dict(component_id=identifier, bbox_px=[x,y,x+w,y+h],
|
||||
bbox_mm=[round(v/px,3) for v in (x,y,x+w,y+h)], area_px=area,
|
||||
seed_pixels=int(strong.sum()),
|
||||
coloured_seed_pixels=int((strong & (chroma[y:y+h,x:x+w] > threshold)).sum()),
|
||||
neutral_seed_pixels=int((strong & neutral[y:y+h,x:x+w]).sum()),
|
||||
reason='retained-component-touches-' + edge + '-ink-extent'))
|
||||
limiting[edge] = components
|
||||
return dict(threshold=threshold, seed_pixels=int(seeds.sum()),
|
||||
extent=list(extent), limiting_components=limiting)
|
||||
@@ -23,6 +23,14 @@ def read_json(path: str | Path, *, default: T | object = _MISSING) -> T:
|
||||
target = Path(path)
|
||||
try:
|
||||
with target.open("r", encoding="utf-8") as stream:
|
||||
if target.name == "refaire.json":
|
||||
# This user-edited selection supports whole-line // comments.
|
||||
# Preserve line numbers in JSON diagnostics and keep other JSON strict.
|
||||
text = "".join(
|
||||
"\n" if line.lstrip().startswith("//") else line
|
||||
for line in stream
|
||||
)
|
||||
return cast(T, json.loads(text))
|
||||
return cast(T, json.load(stream))
|
||||
except FileNotFoundError:
|
||||
if default is _MISSING:
|
||||
|
||||
@@ -82,7 +82,7 @@ def _ruling(binary: np.ndarray, horizontal: bool) -> tuple[np.ndarray, int]:
|
||||
|
||||
|
||||
def _foreground(gray: np.ndarray, dpi: float, threshold: int,
|
||||
ruling: np.ndarray) -> np.ndarray:
|
||||
ruling: np.ndarray, *, diagnostics: dict | None = None) -> np.ndarray:
|
||||
block = max(15, int(dpi / 5) | 1)
|
||||
binary = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
|
||||
cv2.THRESH_BINARY_INV, block, threshold)
|
||||
@@ -100,12 +100,14 @@ def _foreground(gray: np.ndarray, dpi: float, threshold: int,
|
||||
# locally thicker excess strokes inside the suppression mask.
|
||||
excess_ink = cv2.erode(excess_ink, np.ones((2, 2), np.uint8)) > 0
|
||||
mask &= ~excess_ink
|
||||
if diagnostics is not None:
|
||||
diagnostics['paper_suppressed'] = mask & (binary > 0)
|
||||
binary[mask] = 0
|
||||
return binary
|
||||
|
||||
|
||||
def _content_mask(binary: np.ndarray, dpi: float, nh: int = 0,
|
||||
nv: int = 0) -> tuple[np.ndarray, int]:
|
||||
nv: int = 0, *, diagnostics: dict | None = None) -> tuple[np.ndarray, int]:
|
||||
"""Filter only tiny speckles and repeated, matching edge-hole components."""
|
||||
px = dpi / 25.4
|
||||
# Small closing reconnects strokes interrupted by paper-line suppression.
|
||||
@@ -142,5 +144,10 @@ def _content_mask(binary: np.ndarray, dpi: float, nh: int = 0,
|
||||
holes.update(similar)
|
||||
if holes:
|
||||
keep[list(holes)] = False
|
||||
if diagnostics is not None:
|
||||
diagnostics['rejected_artifacts'] = ~keep[labels] & (binary > 0)
|
||||
hole_labels = np.zeros(n, bool)
|
||||
hole_labels[list(holes)] = True
|
||||
diagnostics['repeated_holes'] = hole_labels[labels] & (binary > 0)
|
||||
# Bound original residual ink, not the expanded/grouped mask.
|
||||
return (keep[labels] & (binary > 0)).astype(np.uint8), len(holes)
|
||||
@@ -0,0 +1,165 @@
|
||||
"""Colour-aware suppression of pixels supported by recurring paper geometry.
|
||||
|
||||
This only identifies paper ruling. It does not classify pagination or introduce
|
||||
an independent faint-writing detector.
|
||||
"""
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from copienator.paper_background import _ruling, _runs, _skew
|
||||
|
||||
|
||||
def _maximum(image):
|
||||
first, second, third = cv2.split(image)
|
||||
return cv2.max(cv2.max(first, second), third)
|
||||
|
||||
|
||||
def _minimum(image):
|
||||
first, second, third = cv2.split(image)
|
||||
return cv2.min(cv2.min(first, second), third)
|
||||
|
||||
|
||||
def _thick_strokes(binary, dpi):
|
||||
"""Protect substantial excess thickness without keeping grid junctions."""
|
||||
height, width = binary.shape
|
||||
widths = []
|
||||
for horizontal in (True, False):
|
||||
dimension = width if horizontal else height
|
||||
length = max(25, round(dimension*.12))
|
||||
kernel = np.ones((1,length) if horizontal else (length,1), np.uint8)
|
||||
lines = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)
|
||||
counts = np.count_nonzero(lines, axis=1 if horizontal else 0)
|
||||
widths.extend(b-a for a,b in _runs(counts > dimension*.3))
|
||||
thickness = max(3, int(np.ceil(np.quantile(widths,.95)))+2) if widths else 3
|
||||
kernel = np.ones((thickness,thickness), np.uint8)
|
||||
cores = cv2.erode(binary, kernel)
|
||||
count, labels, stats, _ = cv2.connectedComponentsWithStats(cores, 8)
|
||||
keep = np.zeros(count, bool)
|
||||
px = dpi/25.4
|
||||
x, _, w, h, area = stats[1:].T
|
||||
keep[1:] = ((area >= .3*px*px) & (np.maximum(w,h) >= 1.5*px)
|
||||
& (np.minimum(w,h) >= 2))
|
||||
# A slightly thicker full-height margin rule is still a paper candidate.
|
||||
# It will only be removed if its colour also passes the ruling model below.
|
||||
side_line = ((h > height*.7) & (w < 2*px)
|
||||
& ((x < width*.3) | (x > width*.7)))
|
||||
keep[1:] &= ~side_line
|
||||
return cv2.dilate(keep[labels].astype(np.uint8), kernel) > 0
|
||||
|
||||
|
||||
def _edge_connected_family(binary, bands, horizontal):
|
||||
"""Require several ruling bands reaching a physical edge of the scan.
|
||||
|
||||
Regularly spaced fraction bars inside an answer are not enough evidence.
|
||||
"""
|
||||
dimension = binary.shape[1] if horizontal else binary.shape[0]
|
||||
length = max(25, round(dimension*.12))
|
||||
lines = cv2.morphologyEx(binary, cv2.MORPH_OPEN,
|
||||
np.ones((1,length) if horizontal else (length,1), np.uint8))
|
||||
edge = max(3, round(dimension*.08))
|
||||
if horizontal:
|
||||
touching = np.any(lines[:,:edge] | lines[:,-edge:], axis=1)
|
||||
positions = np.any(bands, axis=1)
|
||||
else:
|
||||
touching = np.any(lines[:edge] | lines[-edge:], axis=0)
|
||||
positions = np.any(bands, axis=0)
|
||||
return sum(np.any(touching[a:b]) for a,b in _runs(positions)) >= 5
|
||||
|
||||
|
||||
def ruling_suppression(rgb: np.ndarray, dpi: float):
|
||||
"""Return a mask in original coordinates and evidence for paper cleanup.
|
||||
|
||||
Require a periodic horizontal or vertical family before modelling any line.
|
||||
Compare actual RGB pixels with locally persistent line colour, retain darker
|
||||
or different-colour strokes, and guard unusually thick strokes. A separate
|
||||
full-height margin rule is eligible only alongside that confirmed family.
|
||||
"""
|
||||
height, width = rgb.shape[:2]
|
||||
lowest = _minimum(rgb)
|
||||
chroma = cv2.subtract(_maximum(rgb), lowest)
|
||||
_, _, stats, _ = cv2.connectedComponentsWithStats(
|
||||
((lowest < 215) & ((chroma > 20) | (lowest < 160))).astype(np.uint8), 8)
|
||||
spans = np.maximum(stats[1:,2]/width, stats[1:,3]/height)
|
||||
if not (np.any(spans > .35) or np.count_nonzero(spans > .05) >= 3):
|
||||
return np.zeros((height,width), bool), {}
|
||||
|
||||
angle = _skew(lowest, angle_step=.5)
|
||||
matrix = cv2.getRotationMatrix2D((width/2,height/2), angle, 1)
|
||||
corners = cv2.transform(np.array([[[0,0],[width,0],[0,height],
|
||||
[width,height]]], dtype=float), matrix)[0]
|
||||
origin = np.floor(corners.min(axis=0))
|
||||
size = np.ceil(corners.max(axis=0)-origin).astype(int)
|
||||
matrix[:,2] -= origin
|
||||
image = cv2.warpAffine(rgb, matrix, tuple(size), borderValue=(255,255,255))
|
||||
dark = 255-image
|
||||
dark16 = dark.astype(np.int16)
|
||||
darkness = _maximum(dark)
|
||||
binary = cv2.adaptiveThreshold(_minimum(image), 255,
|
||||
cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV,
|
||||
max(15,int(dpi/5) | 1), 5)
|
||||
horizontal, nh = _ruling(binary, True)
|
||||
vertical, nv = _ruling(binary, False)
|
||||
if not ((nh and _edge_connected_family(binary,horizontal,True))
|
||||
or (nv and _edge_connected_family(binary,vertical,False))):
|
||||
return np.zeros((height,width), bool), {}
|
||||
|
||||
thick = _thick_strokes(binary, dpi)
|
||||
removed = np.zeros(binary.shape, bool)
|
||||
protected = np.zeros(binary.shape, bool)
|
||||
for is_horizontal, bands, count in ((True,horizontal,nh),(False,vertical,nv)):
|
||||
length = max(25, round(5.2*dpi/25.4))
|
||||
tolerance = max(3, round(.8*dpi/25.4) | 1)
|
||||
transverse = np.ones((tolerance,1) if is_horizontal else (1,tolerance), np.uint8)
|
||||
along = np.ones((1,length) if is_horizontal else (length,1), np.uint8)
|
||||
# Small transverse tolerance follows local scan warp. The RGB model is
|
||||
# still tested against actual pixels; it does not erase entire bands.
|
||||
broadened = cv2.dilate(dark, transverse)
|
||||
model = cv2.morphologyEx(broadened, cv2.MORPH_OPEN, along)
|
||||
support = _maximum(model) > 9
|
||||
if count:
|
||||
support &= bands > 0
|
||||
close = _maximum(np.abs(dark16-model.astype(np.int16))) < 40
|
||||
sample = support & (darkness > 9) & close & ~thick
|
||||
# Neutral, red, green and blue line families are modelled separately so
|
||||
# a red margin rule cannot be mistaken for unusually dark blue writing.
|
||||
first, second, third = cv2.split(model)
|
||||
classes = np.full(binary.shape, 2, np.uint8)
|
||||
classes[second <= third] = 1
|
||||
classes[(first <= second) & (first <= third)] = 0
|
||||
classes[cv2.subtract(_maximum(model),_minimum(model)) < 25] = 3
|
||||
credible = np.zeros(binary.shape, bool)
|
||||
for colour in range(4):
|
||||
selected = sample & (classes == colour)
|
||||
if selected.sum() < 100:
|
||||
continue
|
||||
yy, xx = np.nonzero(selected)
|
||||
spread = np.ptp(yy if is_horizontal else xx)
|
||||
axis_size = image.shape[0] if is_horizontal else image.shape[1]
|
||||
margin_rule = (not is_horizontal and np.ptp(yy) > image.shape[0]*.7
|
||||
and (np.median(xx) < image.shape[1]*.3 or np.median(xx) > image.shape[1]*.7))
|
||||
if spread < axis_size*.35 and not margin_rule:
|
||||
continue
|
||||
# An isolated long fraction bar must not define its own paper colour.
|
||||
# The upper quantile admits darker major rules within an established
|
||||
# family while rejecting excess ink on most of its pixels.
|
||||
ceiling = np.floor(np.quantile(dark[selected],.99,axis=0)+25).astype(np.int16)
|
||||
credible |= ((classes == colour)
|
||||
& (_maximum(dark16-ceiling) <= 0))
|
||||
match = close & credible & (darkness > 9) & ~thick
|
||||
removed |= support & match
|
||||
protected |= support & ~match & (binary > 0)
|
||||
|
||||
inverse = cv2.invertAffineTransform(matrix)
|
||||
result = cv2.warpAffine(removed.astype(np.uint8), inverse,
|
||||
(width,height), flags=cv2.INTER_LINEAR) > 0
|
||||
guard = cv2.warpAffine(protected.astype(np.uint8), inverse,
|
||||
(width,height), flags=cv2.INTER_LINEAR) > 0
|
||||
# Cover one-pixel deskew/interpolation remnants, except next to writing that
|
||||
# failed the paper test. Confirmed pixels of the other orientation remain
|
||||
# removable at line crossings.
|
||||
kernel = np.ones((3,3), np.uint8)
|
||||
result |= ((cv2.dilate(result.astype(np.uint8),kernel) > 0)
|
||||
& ~(cv2.dilate(guard.astype(np.uint8),kernel) > 0))
|
||||
return result, dict(ruling_cleanup=True, ruling_angle_deg=round(angle,3),
|
||||
ruling_horizontal_lines=nh, ruling_vertical_lines=nv)
|
||||
@@ -30,6 +30,7 @@ from .scores import ScoreEditorPanel
|
||||
from .notifications import notify_desktop
|
||||
from .syncthing import SyncthingPanel
|
||||
from .manual_resolution import ManualResolutionPanel
|
||||
from .recorrection import prepare_selection_file
|
||||
from .refaire import SECTION as REFAIRE_SECTION
|
||||
from .refaire import (
|
||||
RefaireSelection,
|
||||
@@ -930,6 +931,14 @@ class CopienatorApp(tk.Tk):
|
||||
self.arg_vars['target'].trace_add('write', lambda *_args: self._refresh_persp_inventory())
|
||||
self._refresh_persp_inventory()
|
||||
|
||||
if step.id == "prepare_recorrection":
|
||||
self.edit_recorrection_button = ttk.Button(
|
||||
self.form, text="Ouvrir refaire.json (avec exemples)",
|
||||
command=self._edit_recorrection,
|
||||
)
|
||||
self.edit_recorrection_button.grid(row=row, column=0, columnspan=2, sticky="w", pady=8)
|
||||
row += 1
|
||||
|
||||
if step.id == "manual_resolution":
|
||||
def manual_evaluation():
|
||||
target = self.arg_vars.get("target")
|
||||
@@ -1117,6 +1126,7 @@ class CopienatorApp(tk.Tk):
|
||||
"cutleft",
|
||||
"labels",
|
||||
"plotting",
|
||||
"splitting",
|
||||
}
|
||||
if step.id in copy_selection_steps:
|
||||
evaluation = self.evaluation
|
||||
@@ -1310,6 +1320,16 @@ class CopienatorApp(tk.Tk):
|
||||
except (OSError, RuntimeError) as exc:
|
||||
messagebox.showerror("Ouverture impossible", str(exc))
|
||||
|
||||
def _edit_recorrection(self) -> None:
|
||||
if self.evaluation is None:
|
||||
return
|
||||
try:
|
||||
path = prepare_selection_file(self.evaluation)
|
||||
except OSError as exc:
|
||||
messagebox.showerror("Préparation impossible", str(exc))
|
||||
return
|
||||
self._open_desktop_path(path, "fichier refaire.json")
|
||||
|
||||
def _select_variant(self, index: int) -> None:
|
||||
if not self.current_step or index < 0:
|
||||
return
|
||||
@@ -1546,6 +1566,15 @@ class CopienatorApp(tk.Tk):
|
||||
if step.id == "refaire_selection":
|
||||
self._save_refaire_selection()
|
||||
return
|
||||
if step.id == "prepare_recorrection":
|
||||
try:
|
||||
load_selection(evaluation)
|
||||
except (OSError, ValueError, TypeError) as exc:
|
||||
messagebox.showerror("Sélection à vérifier", str(exc))
|
||||
return
|
||||
self._mark_step("success")
|
||||
self._select_refaire_correction()
|
||||
return
|
||||
if step.section == REFAIRE_SECTION:
|
||||
try:
|
||||
self._refaire_selection()
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Editable examples for the optional correction redo step."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from copienator import atomic_write_text
|
||||
|
||||
|
||||
EXAMPLES = """// Exemples de sélection pour refaire la correction
|
||||
// Les lignes commençant par // sont des commentaires et sont ignorées.
|
||||
// Modifiez la liste JSON ci-dessous, puis enregistrez le fichier.
|
||||
// Une copie entière : [["Copie46", []]]
|
||||
// Certaines questions : [["Copie46", ["II)8)", "II)9)"]]
|
||||
// Plusieurs copies : [["Copie46", []], ["Copie03", ["Ex 1"]]]
|
||||
// Utilisez les noms exacts des copies et des labels de votre évaluation.
|
||||
// [] après une copie signifie toutes ses questions ; [] seul ne sélectionne rien.
|
||||
// Séparez les entrées par des virgules, sans virgule après la dernière.
|
||||
"""
|
||||
|
||||
|
||||
def prepare_selection_file(evaluation: Path) -> Path:
|
||||
path = evaluation / "refaire.json"
|
||||
existing = path.read_text(encoding="utf-8") if path.exists() else "[]\n"
|
||||
if not existing.startswith(EXAMPLES):
|
||||
atomic_write_text(path, EXAMPLES + existing)
|
||||
return path
|
||||
@@ -371,12 +371,12 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
|
||||
"splitting",
|
||||
"Labels et regroupement",
|
||||
"Découper les réponses par question",
|
||||
"Découpe les copies à partir des coordonnées de labels vérifiées.",
|
||||
"Découpe les copies à partir des coordonnées de labels vérifiées. "
|
||||
"Ciblez une copie avec le sélecteur ou tout le dossier, puis cliquez sur Exécuter.",
|
||||
(python("default", "Découpage", "split-answers"),),
|
||||
arguments=(arg_target(),),
|
||||
requires=("Copies",),
|
||||
artifacts=("Copies/Copie*/*",),
|
||||
auto_start_first_visit=True,
|
||||
),
|
||||
StepDefinition(
|
||||
"crop_exercise_bottoms",
|
||||
@@ -511,6 +511,19 @@ def build_workflow(show_personal_steps: bool) -> list[StepDefinition]:
|
||||
requires=("manual_resolutions.txt", "correction.json"),
|
||||
skip_without_manual_conflicts=True,
|
||||
),
|
||||
StepDefinition(
|
||||
"prepare_recorrection",
|
||||
"Correction",
|
||||
"Choisir des copies à recorriger",
|
||||
"Facultatif : ouvrez refaire.json et indiquez les copies entières ou les questions à recorriger. "
|
||||
"Les exemples commentés expliquent la syntaxe. Enregistrez le fichier, puis marquez cette étape terminée : "
|
||||
"vous reviendrez à Correction avec --refaire sélectionné, sans lancement automatique. "
|
||||
"Les corrections sélectionnées seront remplacées ; celles des autres copies seront conservées. "
|
||||
"Si vous avez modifié les découpes, régénérez les réponses avant la recorrection. "
|
||||
"Ignorez cette étape si aucune reprise n’est nécessaire.",
|
||||
(manual("default"),),
|
||||
optional=True,
|
||||
),
|
||||
StepDefinition(
|
||||
"post_correction",
|
||||
"Correction",
|
||||
|
||||
@@ -251,6 +251,31 @@ coupure, et quelle moitié doit être gardée. L'interface graphique vous
|
||||
fournira alors une instruction texte à copier et à mettre dans le
|
||||
fichier texte.
|
||||
|
||||
Vous pouvez aussi ajouter vos propres instructions dans
|
||||
`manual_resolutions.txt`, même si Gemini n’a pas signalé ces conflits.
|
||||
Les lignes sont appliquées **de haut en bas**, avec une distinction importante :
|
||||
|
||||
- La **source** est toujours le PDF tel qu’il existait au début de cette
|
||||
exécution, avant les modifications des lignes précédentes.
|
||||
- La **cible** est sa version courante, tenant compte des lignes précédentes.
|
||||
|
||||
Par exemple, pour une copie ayant les labels `A`, `B` et `C` :
|
||||
|
||||
```text
|
||||
Copie46 A c{40}1> B|
|
||||
Copie46 C -> A|
|
||||
```
|
||||
|
||||
La première ligne coupe A à 40 % de sa hauteur cumulée, conserve la première
|
||||
partie sous le label A et ajoute les 60 % restants après B. La seconde ligne
|
||||
ajoute C après **la partie conservée de A**, et non après A entier. C reste
|
||||
également disponible sous son propre label.
|
||||
|
||||
En revanche, si une ligne ultérieure utilise **A comme source**, elle utilise
|
||||
le PDF A entier du début de l’exécution, pas seulement sa partie conservée.
|
||||
Une seule coupe par label source et par copie est autorisée dans une même
|
||||
exécution. Remplacez `A`, `B` et `C` par les labels exacts de votre évaluation.
|
||||
|
||||
Enregistrez le fichier texte, éventuellement le **Recharger**, puis
|
||||
**Exécuter**.
|
||||
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from subprocess import CompletedProcess
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from copienator.commands import annotating
|
||||
|
||||
|
||||
class AnnotationLatexErrorTests(unittest.TestCase):
|
||||
def test_failure_preserves_source_and_log_and_explains_recovery(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
diagnostics = Path(directory) / "diagnostics"
|
||||
diagnostics.mkdir()
|
||||
|
||||
def compile_latex(command, **kwargs):
|
||||
Path(kwargs["cwd"], "text.log").write_bytes(
|
||||
b"preamble\n! Missing \\endcsname inserted.\nl.12 \\end{document}\n\xff"
|
||||
)
|
||||
return CompletedProcess(command, 1, stdout=b"compiler output\xff")
|
||||
|
||||
# Patch only the retained diagnostics directory, not TemporaryDirectory.
|
||||
original_mkdtemp = tempfile.mkdtemp
|
||||
|
||||
def make_directory(*args, **kwargs):
|
||||
if kwargs.get("prefix") == "copienator-latex-error-":
|
||||
return str(diagnostics)
|
||||
return original_mkdtemp(*args, **kwargs)
|
||||
|
||||
with patch.object(annotating.subprocess, "run", side_effect=compile_latex), patch.object(
|
||||
annotating.tempfile, "mkdtemp", side_effect=make_directory
|
||||
):
|
||||
with self.assertRaises(annotating.AnnotationLatexError) as caught:
|
||||
annotating.render_real_latex_text(r"Antislash \.", 400)
|
||||
message = str(caught.exception)
|
||||
for expected in ("code 1", r"Antislash \.", "Missing", "correction.json", "relancez cette étape", str(diagnostics)):
|
||||
self.assertIn(expected, message)
|
||||
self.assertIn(r"Antislash \.", (diagnostics / "text.tex").read_text())
|
||||
self.assertTrue((diagnostics / "text.log").exists())
|
||||
self.assertTrue((diagnostics / "compiler-output.txt").exists())
|
||||
|
||||
def test_comment_error_identifies_copy_label_and_correction_file(self):
|
||||
for box in (None, [0, 0, 20, 20]):
|
||||
with self.subTest(box=box), patch.object(
|
||||
annotating, "render_score_text", return_value=Image.new("RGBA", (200, 40))
|
||||
):
|
||||
with self.assertRaises(annotating.AnnotationLatexError) as caught:
|
||||
annotating.compose_label_image(
|
||||
Image.new("RGB", (800, 200)), "I)7)",
|
||||
{"feedback": [{"text": "Bad comment", "box_2d": box}]}, 0,
|
||||
render_fn=Mock(side_effect=annotating.AnnotationLatexError("Bad comment")),
|
||||
source_pdf=Path("/exam/Copies/Copie39/I)7).pdf"),
|
||||
)
|
||||
message = str(caught.exception)
|
||||
for expected in ("Copie39", "I)7)", "/exam/correction.json", "Bad comment"):
|
||||
self.assertIn(expected, message)
|
||||
@@ -0,0 +1,118 @@
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
from copienator.crop_diagnostics import check_protected_regions, save_diagnostics
|
||||
from copienator.ink_detection import detect_bounds
|
||||
|
||||
|
||||
class CropDiagnosticsTests(unittest.TestCase):
|
||||
def scan(self, *, dark=False, skew=False):
|
||||
rgb = np.full((900,650,3),255,np.uint8)
|
||||
if dark:
|
||||
for y in range(15,900,20):
|
||||
cv2.line(rgb,(0,y),(649,y),(65,65,65),1)
|
||||
for x in range(10,650,20):
|
||||
cv2.line(rgb,(x,0),(x,899),(65,65,65),1)
|
||||
cv2.putText(rgb,'answer',(100,400),cv2.FONT_HERSHEY_SIMPLEX,1,(20,50,180),2)
|
||||
cv2.putText(rgb,'more',(100,560),cv2.FONT_HERSHEY_SIMPLEX,.8,(30,30,30),2)
|
||||
if skew:
|
||||
rgb = cv2.warpAffine(rgb,cv2.getRotationMatrix2D((325,450),2,1),
|
||||
(650,900),borderValue=(255,255,255))
|
||||
return rgb
|
||||
|
||||
def test_diagnostics_leave_decisions_and_json_contract_unchanged(self):
|
||||
for dark,skew in [(False,False),(True,False),(True,True)]:
|
||||
with self.subTest(dark=dark,skew=skew):
|
||||
rgb = self.scan(dark=dark,skew=skew)
|
||||
before = rgb.copy()
|
||||
normal = detect_bounds(rgb,dpi=150)
|
||||
debug = {'stale': 'discard this'}
|
||||
observed = detect_bounds(rgb,dpi=150,diagnostics=debug)
|
||||
self.assertEqual(normal,observed)
|
||||
json.dumps(observed)
|
||||
np.testing.assert_array_equal(rgb,before)
|
||||
self.assertNotIn('stale',debug)
|
||||
for name,mask in debug['masks'].items():
|
||||
self.assertEqual(mask.shape,rgb.shape[:2],name)
|
||||
self.assertEqual(mask.dtype,np.bool_,name)
|
||||
masks = debug['masks']
|
||||
self.assertFalse(np.any(masks['strong_seeds'] & ~masks['foreground']))
|
||||
self.assertFalse(np.any(masks['recovered_weak'] & masks['strong_seeds']))
|
||||
for item in debug['thresholds']:
|
||||
for edge in ['top','bottom']:
|
||||
for component in item['limiting_components'][edge]:
|
||||
box = component['bbox_px']
|
||||
self.assertEqual(box[1] if edge=='top' else box[3],
|
||||
item['extent'][0 if edge=='top' else 1])
|
||||
self.assertGreater(component['seed_pixels'],0)
|
||||
|
||||
def test_blank_fallback_has_empty_foreground_and_explanation(self):
|
||||
rgb = np.full((400,300,3),255,np.uint8)
|
||||
debug = {}
|
||||
result = detect_bounds(rgb,diagnostics=debug)
|
||||
self.assertEqual((result['top_px'],result['bottom_px']),(0,400))
|
||||
self.assertFalse(debug['masks']['foreground'].any())
|
||||
self.assertEqual(debug['edges']['bottom']['reason'],'no-reliable-ink-seeds')
|
||||
|
||||
def test_faint_unseeded_candidate_is_distinguished_from_recovered_ink(self):
|
||||
rgb = self.scan()
|
||||
cv2.putText(rgb,'weak',(100,800),cv2.FONT_HERSHEY_SIMPLEX,1,(130,160,205),2)
|
||||
debug = {}
|
||||
detect_bounds(rgb,dpi=150,diagnostics=debug)
|
||||
self.assertTrue(debug['masks']['unretained_weak'][770:810,100:220].any())
|
||||
self.assertFalse(debug['masks']['foreground'][770:810,100:220].any())
|
||||
|
||||
def test_exported_masks_match_actual_pixels_and_report_is_serializable(self):
|
||||
rgb = self.scan(dark=True,skew=True)
|
||||
debug = {}
|
||||
result = detect_bounds(rgb,dpi=150,diagnostics=debug)
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
files = save_diagnostics(rgb,result,debug,Path(tmp),width_mm=110,height_mm=152)
|
||||
for name,path in files['masks'].items():
|
||||
with Image.open(Path(tmp)/path) as im:
|
||||
np.testing.assert_array_equal(np.array(im).astype(bool),debug['masks'][name])
|
||||
report = json.loads((Path(tmp)/files['report']).read_text())
|
||||
self.assertEqual(report['detector'],result)
|
||||
self.assertEqual(report['shape'],list(rgb.shape[:2]))
|
||||
for path in files['overlays'].values():
|
||||
self.assertTrue((Path(tmp)/path).is_file())
|
||||
|
||||
|
||||
class ProtectedContentTests(unittest.TestCase):
|
||||
def check(self,top,bottom,box):
|
||||
return check_protected_regions([dict(id='ink',label='Real ink',bbox_mm=box)],
|
||||
dict(top_px=top,bottom_px=bottom),width_mm=80,height_mm=100,shape=(1000,800))[0]
|
||||
|
||||
def test_both_boundaries_can_clip_a_protected_region(self):
|
||||
top = self.check(210,800,[10,20,50,30])
|
||||
self.assertFalse(top['passed'])
|
||||
self.assertEqual(top['clipped_edges'],['top'])
|
||||
self.assertAlmostEqual(top['clipped_area_fraction'],.1)
|
||||
bottom = self.check(100,790,[10,70,50,80])
|
||||
self.assertFalse(bottom['passed'])
|
||||
self.assertEqual(bottom['clipped_edges'],['bottom'])
|
||||
self.assertAlmostEqual(bottom['clipped_area_fraction'],.1)
|
||||
|
||||
def test_contact_at_boundary_and_short_page_physical_height(self):
|
||||
self.assertTrue(self.check(200,300,[10,20,50,30])['passed'])
|
||||
row = check_protected_regions([dict(id='end',label='End',bbox_mm=[2,17,20,19])],
|
||||
dict(top_px=0,bottom_px=180),width_mm=40,height_mm=20,shape=(200,400))[0]
|
||||
self.assertFalse(row['passed'])
|
||||
self.assertAlmostEqual(row['clipped_area_fraction'],.5)
|
||||
|
||||
def test_full_removal_and_invalid_ground_truth(self):
|
||||
self.assertEqual(self.check(400,800,[10,20,50,30])['clipped_area_fraction'],1)
|
||||
with self.assertRaises(ValueError):
|
||||
self.check(0,1000,[10,20,90,30])
|
||||
with self.assertRaises(ValueError):
|
||||
self.check(0,1000,[10,30,50,20])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -20,7 +20,7 @@ class CropExerciseBottomsTests(unittest.TestCase):
|
||||
page = document.new_page(width=600, height=height)
|
||||
page.insert_text((80, text_y), "student answer", fontsize=16)
|
||||
if footer:
|
||||
page.insert_text((20, height - 3), "Ex 2", fontsize=10)
|
||||
page.insert_text((20, height - 15), "9/9", fontsize=10)
|
||||
document.save(path)
|
||||
return path
|
||||
|
||||
@@ -33,7 +33,7 @@ class CropExerciseBottomsTests(unittest.TestCase):
|
||||
self.assertEqual(records[0]["status"], "skipped-short")
|
||||
self.assertFalse(destination.exists())
|
||||
|
||||
def test_footer_fragment_is_ignored_for_a_large_bottom_crop(self):
|
||||
def test_isolated_page_number_allows_a_large_bottom_crop(self):
|
||||
source = self.make_pdf("large.pdf", 400, 100)
|
||||
destination = self.root / "out" / source.name
|
||||
records = process_exercise_pdf(
|
||||
@@ -45,6 +45,18 @@ class CropExerciseBottomsTests(unittest.TestCase):
|
||||
self.assertLess(result[0].rect.height, 250)
|
||||
self.assertAlmostEqual(result[0].rect.width, 600)
|
||||
|
||||
def test_content_in_former_ignored_bottom_strip_is_retained(self):
|
||||
source = self.make_pdf("bottom-content.pdf", 400, 100, footer=False)
|
||||
with pymupdf.open(source) as doc:
|
||||
doc[0].insert_text((20, 399), "Ex 2", fontsize=10)
|
||||
doc.save(self.root / "bottom-label.pdf")
|
||||
records = process_exercise_pdf(
|
||||
self.root / "bottom-label.pdf", self.root / "retained.pdf",
|
||||
self.review, 800, dpi=150
|
||||
)
|
||||
self.assertEqual(records[0]["status"], "unchanged-small-crop")
|
||||
self.assertFalse((self.root / "retained.pdf").exists())
|
||||
|
||||
def test_crop_smaller_than_four_lines_is_not_written(self):
|
||||
source = self.make_pdf("small.pdf", 400, 330, footer=False)
|
||||
destination = self.root / "out" / source.name
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
import unittest
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from copienator.ink_detection import detect_bounds
|
||||
|
||||
|
||||
class EdgeMarksTests(unittest.TestCase):
|
||||
def page(self, last_answer=1000):
|
||||
rgb = np.full((2340,1654,3),255,np.uint8)
|
||||
cv2.putText(rgb,'answer',(500,600),cv2.FONT_HERSHEY_SIMPLEX,1.5,(20,60,180),2)
|
||||
cv2.putText(rgb,'more',(500,last_answer),cv2.FONT_HERSHEY_SIMPLEX,1.5,(20,60,180),2)
|
||||
return rgb
|
||||
|
||||
def test_isolated_footer_fraction_is_removed(self):
|
||||
rgb=self.page()
|
||||
cv2.putText(rgb,'9/9',(1400,2220),cv2.FONT_HERSHEY_SIMPLEX,1.2,(20,20,20),2)
|
||||
debug={}
|
||||
result=detect_bounds(rgb,diagnostics=debug)
|
||||
self.assertLess(result['bottom_px'],1100)
|
||||
self.assertTrue(debug['masks']['isolated_edge_marks'].any())
|
||||
self.assertFalse(debug['masks']['foreground'][2150:].any())
|
||||
self.assertFalse(debug['masks']['strong_seeds'][2150:].any())
|
||||
self.assertEqual(result,detect_bounds(rgb))
|
||||
|
||||
def test_close_footer_stays_with_answer(self):
|
||||
rgb=self.page(last_answer=2170)
|
||||
cv2.putText(rgb,'9/9',(1400,2220),cv2.FONT_HERSHEY_SIMPLEX,1.2,(20,20,20),2)
|
||||
self.assertGreater(detect_bounds(rgb)['bottom_px'],2220)
|
||||
|
||||
def test_central_sparse_ratio_stays(self):
|
||||
rgb=self.page()
|
||||
cv2.putText(rgb,'9/9',(760,2220),cv2.FONT_HERSHEY_SIMPLEX,1.2,(20,20,20),2)
|
||||
self.assertGreater(detect_bounds(rgb)['bottom_px'],2220)
|
||||
|
||||
def test_isolated_bare_border_digit_needs_no_recognition(self):
|
||||
rgb=self.page()
|
||||
cv2.putText(rgb,'7',(1400,2240),cv2.FONT_HERSHEY_SIMPLEX,1.2,(20,20,20),2)
|
||||
self.assertLess(detect_bounds(rgb)['bottom_px'],1100)
|
||||
|
||||
def test_underlined_question_at_upper_edge_stays(self):
|
||||
rgb=self.page()
|
||||
cv2.putText(rgb,'16',(310,85),cv2.FONT_HERSHEY_SIMPLEX,1.2,(20,20,20),2)
|
||||
cv2.line(rgb,(310,90),(355,90),(20,20,20),2)
|
||||
self.assertLess(detect_bounds(rgb)['top_px'],60)
|
||||
|
||||
def test_name_in_other_margin_blocks_header_removal(self):
|
||||
rgb=self.page()
|
||||
cv2.circle(rgb,(90,55),22,(20,20,20),2)
|
||||
cv2.putText(rgb,'1',(81,64),cv2.FONT_HERSHEY_SIMPLEX,.7,(20,20,20),1)
|
||||
cv2.putText(rgb,'Name',(1350,120),cv2.FONT_HERSHEY_SIMPLEX,1,(20,20,20),2)
|
||||
result=detect_bounds(rgb)
|
||||
self.assertEqual(result['top_px'],0)
|
||||
|
||||
|
||||
if __name__=='__main__':
|
||||
unittest.main()
|
||||
@@ -214,6 +214,49 @@ class GuiConvenienceTests(unittest.TestCase):
|
||||
self.assertNotIn("overwrite", self.app.arg_vars)
|
||||
self.assertIn("--refaire", self.app._make_command())
|
||||
|
||||
def test_optional_recorrection_opens_examples_and_returns_to_correction(self):
|
||||
from copienator import read_json
|
||||
ids = [step.id for step in self.app.steps]
|
||||
self.assertEqual(ids[ids.index("manual_resolution") + 1], "prepare_recorrection")
|
||||
copies = self.evaluation / "Copies"
|
||||
copies.mkdir()
|
||||
(copies / "Copie46.pdf").touch()
|
||||
(self.evaluation / "labels").write_text("II)9)\n")
|
||||
self.app.tree.selection_set("prepare_recorrection")
|
||||
self.app.update()
|
||||
with patch("copienator_gui.app.open_path") as opened:
|
||||
self.app.edit_recorrection_button.invoke()
|
||||
path = self.evaluation / "refaire.json"
|
||||
opened.assert_called_once_with(path)
|
||||
self.assertEqual(read_json(path), [])
|
||||
path.write_text(path.read_text().replace("\n[]\n", '\n[["Copie46", []]]\n'))
|
||||
with patch.object(self.app.runner, "start") as start:
|
||||
self.app.run_button.invoke()
|
||||
self.app.update()
|
||||
start.assert_not_called()
|
||||
self.assertEqual(self.app.state_store.step("prepare_recorrection")["status"], "success")
|
||||
self.assertEqual(self.app.current_step.id, "correction")
|
||||
self.assertEqual(self.app.variant_var.get(), "refaire")
|
||||
self.assertIn("--refaire", self.app._make_command())
|
||||
self.assertNotIn("--overwrite", self.app._make_command())
|
||||
|
||||
def test_ignoring_optional_recorrection_continues_without_creating_file(self):
|
||||
self.app.tree.selection_set("prepare_recorrection")
|
||||
self.app.update()
|
||||
self.app._skip_step()
|
||||
self.app.update()
|
||||
self.assertEqual(self.app.current_step.id, "post_correction")
|
||||
self.assertEqual(self.app.state_store.step("prepare_recorrection")["status"], "skipped")
|
||||
self.assertFalse((self.evaluation / "refaire.json").exists())
|
||||
|
||||
def test_optional_recorrection_requires_a_saved_selection(self):
|
||||
self.app.tree.selection_set("prepare_recorrection")
|
||||
self.app.update()
|
||||
with patch("copienator_gui.app.messagebox.showerror") as error:
|
||||
self.app.run_button.invoke()
|
||||
error.assert_called_once()
|
||||
self.assertEqual(self.app.current_step.id, "prepare_recorrection")
|
||||
|
||||
def test_giving_names_waits_for_explicit_manual_completion(self):
|
||||
return_dir = self.evaluation / "A Rendre" / "Student (01)"
|
||||
return_dir.mkdir(parents=True)
|
||||
@@ -560,6 +603,25 @@ class GuiConvenienceTests(unittest.TestCase):
|
||||
self.assertIn("Cibler la copie sélectionnée", controls)
|
||||
self.assertIn("Cibler tout le dossier", controls)
|
||||
|
||||
def test_answer_splitting_waits_for_target_and_can_select_one_copy(self):
|
||||
copies = self.evaluation / "Copies"
|
||||
copies.mkdir()
|
||||
for name in ("Copie01.pdf", "Copie46.pdf"):
|
||||
(copies / name).touch()
|
||||
with patch.object(self.app, "_run_current_step") as run:
|
||||
self.app.tree.selection_set("splitting")
|
||||
self.app.update()
|
||||
run.assert_not_called()
|
||||
self.assertIn("Copie46.pdf", self.app.copy_paths)
|
||||
self.app.copy_var.set("Copie46.pdf")
|
||||
self.app._target_selected_copy()
|
||||
command = self.app._make_command()
|
||||
self.assertIn("split-answers", command)
|
||||
self.assertEqual(Path(command[5]), copies / "Copie46.pdf")
|
||||
self.app._target_all_pages()
|
||||
command = self.app._make_command()
|
||||
self.assertEqual((self.repository / command[5]).resolve(), self.evaluation)
|
||||
|
||||
def test_label_review_can_show_and_target_one_copy(self):
|
||||
copies = self.evaluation / "Copies"
|
||||
copies.mkdir()
|
||||
|
||||
@@ -2378,7 +2378,7 @@ class WorkflowTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(
|
||||
auto_start,
|
||||
{"crop_blank_margins", "splitting", "crop_exercise_bottoms", "grouping"},
|
||||
{"crop_blank_margins", "crop_exercise_bottoms", "grouping"},
|
||||
)
|
||||
self.assertEqual(
|
||||
skip_for_live, {"submit_batches", "batch_status", "fetch_batches"}
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from copienator.ink_detection import detect_bounds
|
||||
|
||||
|
||||
class PaperRulingTests(unittest.TestCase):
|
||||
def scan(self, colour, *, warp=False):
|
||||
image = np.full((1200, 850, 3), 255, np.uint8)
|
||||
for y in range(20, 1200, 20):
|
||||
cv2.line(image, (0, y), (849, y), colour, 1)
|
||||
for x in range(10, 850, 20):
|
||||
cv2.line(image, (x, 0), (x, 1199), colour, 1)
|
||||
cv2.line(image, (80, 0), (80, 1199), (210, 70, 80), 2)
|
||||
cv2.putText(image, 'answer', (120, 410), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1, (20, 60, 180), 2)
|
||||
cv2.putText(image, 'more', (120, 710), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1, (20, 60, 180), 2)
|
||||
if warp:
|
||||
yy, xx = np.indices(image.shape[:2], dtype=np.float32)
|
||||
image = cv2.remap(image, xx+2*np.sin(yy/95), yy,
|
||||
cv2.INTER_LINEAR, borderValue=(255,255,255))
|
||||
return image
|
||||
|
||||
def test_default_path_does_not_run_experimental_colour_model(self):
|
||||
image = self.scan((70, 110, 225))
|
||||
with patch('copienator.ink_detection.ruling_suppression',
|
||||
side_effect=AssertionError('Expensive colour model invoked')):
|
||||
result = detect_bounds(image, dpi=150)
|
||||
self.assertLess(result['top_px'], 410)
|
||||
self.assertGreater(result['bottom_px'], 710)
|
||||
self.assertFalse(result.get('ruling_cleanup', False))
|
||||
|
||||
def test_saturated_coloured_grid_and_margin_do_not_seed_blank_edges(self):
|
||||
for colour in [(65, 65, 65), (70, 110, 225), (155, 170, 240)]:
|
||||
for warp in [False, True]:
|
||||
with self.subTest(colour=colour, warp=warp):
|
||||
image = self.scan(colour, warp=warp)
|
||||
debug = {}
|
||||
result = detect_bounds(image, dpi=150, colour_cleanup=True, diagnostics=debug)
|
||||
self.assertGreater(result['top_px'], 280)
|
||||
self.assertLess(result['bottom_px'], 810)
|
||||
self.assertTrue(result['ruling_cleanup'])
|
||||
self.assertTrue(debug['masks']['paper_suppressed'].any())
|
||||
self.assertFalse(debug['masks']['foreground'][:250].any())
|
||||
|
||||
def test_writing_crossing_ruling_keeps_both_outer_bounds(self):
|
||||
image = self.scan((70, 110, 225))
|
||||
cv2.putText(image, 'top', (120, 200), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.7, (190, 20, 20), 2)
|
||||
cv2.putText(image, 'bottom', (120, 1000), cv2.FONT_HERSHEY_SIMPLEX,
|
||||
.7, (20, 20, 20), 2)
|
||||
result = detect_bounds(image, dpi=150, colour_cleanup=True)
|
||||
self.assertLess(result['top_px'], 180)
|
||||
self.assertGreater(result['bottom_px'], 1000)
|
||||
|
||||
def test_isolated_bars_and_axes_are_not_a_ruling_colour_family(self):
|
||||
for colour in [(190, 20, 20), (20, 20, 20), (20, 90, 190)]:
|
||||
with self.subTest(colour=colour):
|
||||
image = self.scan((70, 110, 225))
|
||||
cv2.line(image, (100, 1020), (700, 1020), colour, 3)
|
||||
cv2.line(image, (425, 980), (425, 1120), colour, 3)
|
||||
result = detect_bounds(image, dpi=150, colour_cleanup=True)
|
||||
self.assertGreater(result['bottom_px'], 1120)
|
||||
|
||||
def test_grid_colour_with_a_thicker_answer_bar_is_preserved(self):
|
||||
colour = (70, 110, 225)
|
||||
image = self.scan(colour)
|
||||
cv2.line(image, (150, 1020), (650, 1020), colour, 5)
|
||||
self.assertGreater(detect_bounds(image, dpi=150, colour_cleanup=True)['bottom_px'], 1020)
|
||||
|
||||
def test_regular_fraction_bars_inside_an_answer_are_not_paper(self):
|
||||
image = np.full((1200,850,3),255,np.uint8)
|
||||
cv2.putText(image,'answer',(120,300),cv2.FONT_HERSHEY_SIMPLEX,
|
||||
1,(20,60,180),2)
|
||||
for y in (400,550,700,850,1000):
|
||||
cv2.line(image,(130,y),(700,y),(20,20,20),3)
|
||||
result = detect_bounds(image,dpi=150, colour_cleanup=True)
|
||||
self.assertGreater(result['bottom_px'],1000)
|
||||
self.assertFalse(result.get('ruling_cleanup',False))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,34 @@
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from copienator import read_json
|
||||
from copienator_gui.recorrection import prepare_selection_file
|
||||
|
||||
|
||||
class RecorrectionFileTests(unittest.TestCase):
|
||||
def test_examples_are_inactive_and_opening_preserves_selection(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
path = prepare_selection_file(root)
|
||||
self.assertEqual(read_json(path), [])
|
||||
self.assertIn('// Une copie entière', path.read_text())
|
||||
selection = [["Copie46", ["II)9)"]]]
|
||||
path.write_text(json.dumps(selection))
|
||||
prepare_selection_file(root)
|
||||
self.assertEqual(read_json(path), selection)
|
||||
before = path.read_bytes()
|
||||
prepare_selection_file(root)
|
||||
self.assertEqual(path.read_bytes(), before)
|
||||
|
||||
def test_comments_do_not_affect_strings_or_other_json_files(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory)
|
||||
path = root / "refaire.json"
|
||||
path.write_text('// Example\n[["Copie01", ["a//b"]]]\n')
|
||||
self.assertEqual(read_json(path), [["Copie01", ["a//b"]]])
|
||||
other = root / "correction.json"
|
||||
other.write_text(path.read_text())
|
||||
with self.assertRaises(json.JSONDecodeError):
|
||||
read_json(other)
|
||||
Reference in new issue
Block a user