Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 124 additions & 23 deletions core/app/layout_gt/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,13 +121,27 @@ def _sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()


def _dedupe_xs(xs: list[float], *, min_gap: float = 4.0) -> list[float]:
xs = sorted(float(x) for x in xs)
if not xs:
return []
out = [xs[0]]
for x in xs[1:]:
if x - out[-1] >= min_gap:
out.append(x)
else:
out[-1] = 0.5 * (out[-1] + x)
return out


def _interior_splits_from_edges(
xs: list[float],
*,
x_left: float | None = None,
x_right: float | None = None,
n_measures: int | None = None,
edge_eps: float = 2.0,
min_gap: float = 4.0,
) -> list[float]:
"""Convert barline x list to #85 **interior** splits.

Expand All @@ -136,11 +150,12 @@ def _interior_splits_from_edges(
- **#85 interiors**: ``n_splits = n_measures - 1``
- **#66 edges**: ``n_barlines = n_measures + 1`` (includes outer measure bounds)
- raw detector xs that may hug L2 left/right
- messy detector sets (count ≠ edges/interiors) — caller should fall back

We drop endpoints near L2 bounds and, when ``n_measures`` is known and
``len(xs) == n_measures + 1``, drop the first/last edge.
"""
xs = sorted(float(x) for x in xs)
xs = _dedupe_xs([float(x) for x in xs], min_gap=min_gap)
if not xs:
return []

Expand All @@ -152,6 +167,21 @@ def _interior_splits_from_edges(
# Full edge chain → interiors
return xs[1:-1] if len(xs) >= 2 else []

# n_barlines == n_measures: often interiors plus one outer edge
if (
n_measures is not None
and n_measures >= 2
and len(xs) == n_measures
and x_left is not None
and x_right is not None
):
d_left = abs(xs[0] - x_left)
d_right = abs(xs[-1] - x_right)
if d_left <= d_right and d_left < 40:
xs = xs[1:]
elif d_right < 40:
xs = xs[:-1]

out = list(xs)
if x_left is not None:
out = [x for x in out if x > x_left + edge_eps]
Expand All @@ -178,6 +208,79 @@ def _measures_to_interior_xs(
return xs


def _derive_measures_from_splits(
*,
x_left: float,
x_right: float,
y1: float,
y2: float,
split_xs: list[float],
system_index: int = 0,
) -> list[dict[str, Any]]:
"""Rebuild measure boxes so ``n_measures == n_splits + 1`` always holds."""
interiors = _dedupe_xs(
[x for x in split_xs if x_left + 1e-6 < x < x_right - 1e-6],
min_gap=2.0,
)
edges = [float(x_left), *interiors, float(x_right)]
out: list[dict[str, Any]] = []
for i in range(len(edges) - 1):
out.append(
{
"id": f"l3-m-sys{system_index}-{i}",
"label": f"m{i + 1}",
"box": {
"x1": edges[i],
"y1": float(y1),
"x2": edges[i + 1],
"y2": float(y2),
},
"kind": "measure_derived",
}
)
return out


def _choose_interior_splits(
raw_xs: list[float],
*,
x_left: float,
x_right: float,
measure_boxes: list[dict[str, float]],
) -> tuple[list[float], str]:
"""Pick interior splits consistent with #85 when barlines and boxes disagree.

Preference:
1. barlines that already match ``n_measures - 1``
2. interiors derived from measure boxes (stable for messy detectors)
3. raw barline interiors even if count mismatches
"""
n_meas = len(measure_boxes)
from_bar = _interior_splits_from_edges(
raw_xs,
x_left=x_left,
x_right=x_right,
n_measures=n_meas if n_meas > 0 else None,
)
from_meas = (
_measures_to_interior_xs(measure_boxes) if n_meas >= 2 else []
)
expected = n_meas - 1 if n_meas >= 1 else None

if expected is not None:
if len(from_bar) == expected:
return from_bar, "barlines"
if from_meas and len(from_meas) == expected:
return from_meas, "measures"
if from_meas:
return from_meas, "measures_fallback"
return from_bar, "barlines_unmatched"

if from_bar:
return from_bar, "barlines"
return from_meas, "measures"


# ---------------------------------------------------------------------------
# Core conversion
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -342,23 +445,18 @@ def _assign_system(box: dict[str, float]) -> int | None:
candidates = list(raw_by_sys.get(si, []))

raw_xs = [float(b["x"]) for b in candidates]
n_meas = len(measures_by_sys.get(si) or [])
interiors = _interior_splits_from_edges(
measure_boxes = [m["box"] for m in measures_by_sys.get(si) or []]
interiors, split_src = _choose_interior_splits(
raw_xs,
x_left=s["bbox"]["x1"],
x_right=s["bbox"]["x2"],
n_measures=n_meas if n_meas > 0 else None,
x_left=float(s["bbox"]["x1"]),
x_right=float(s["bbox"]["x2"]),
measure_boxes=measure_boxes,
)
# Fallback: derive from measure boxes
if not interiors and n_meas >= 2:
interiors = _measures_to_interior_xs(
[m["box"] for m in measures_by_sys[si]]
)

# Attach metadata from nearest raw barline when possible
splits: list[dict[str, Any]] = []
for i, x in enumerate(interiors):
src = "migrate"
src = split_src if split_src.startswith("bar") else "migrate"
sid = f"s{si}-{i}"
conf = None
y1 = s["bbox"]["y1"]
Expand All @@ -370,8 +468,8 @@ def _assign_system(box: dict[str, float]) -> int | None:
if d < best_d:
best_d = d
best = b
if best is not None and best_d <= 4.0:
src = str(best.get("source") or "detect")
if best is not None and best_d <= 8.0:
src = str(best.get("source") or src or "detect")
if best.get("id"):
sid = str(best["id"])
if best.get("confidence") is not None:
Expand All @@ -395,16 +493,19 @@ def _assign_system(box: dict[str, float]) -> int | None:
"system_id": s["id"],
"system_index": si,
"splits": splits,
"split_source": split_src,
}
if include_derived_measures and measures_by_sys.get(si):
row["measures"] = [
{
"id": m.get("id"),
"label": m.get("label"),
"box": m["box"],
}
for m in measures_by_sys[si]
]
# Always re-derive measures from final splits so n_m == n_s+1
# (structure L3 boxes can disagree with messy barlines on real projects).
if include_derived_measures:
row["measures"] = _derive_measures_from_splits(
x_left=float(s["bbox"]["x1"]),
x_right=float(s["bbox"]["x2"]),
y1=float(s["bbox"]["y1"]),
y2=float(s["bbox"]["y2"]),
split_xs=[float(sp["x"]) for sp in splits],
system_index=si,
)
rows.append(row)

sample: dict[str, Any] = {
Expand Down
24 changes: 24 additions & 0 deletions core/tests/test_layout_gt.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,30 @@ def test_validate_rejects_measure_split_count_mismatch() -> None:
assert any("n_measures" in e for e in r.errors)


def test_messy_barlines_prefer_measures_and_rederive() -> None:
"""Real projects often have detector barlines that don't match L3 boxes."""
structure = _minimal_structure()
# 6 barlines + 3 measures → old code failed n_m == n_s+1
structure["barlines"] = [
{"system": 0, "x": 50, "y1": 80, "y2": 140},
{"system": 0, "x": 90, "y1": 80, "y2": 140},
{"system": 0, "x": 140, "y1": 80, "y2": 140},
{"system": 0, "x": 200, "y1": 80, "y2": 140},
{"system": 0, "x": 240, "y1": 80, "y2": 140},
{"system": 0, "x": 300, "y1": 80, "y2": 140},
]
sample = layout_sample_from_structure(
structure,
image={"path": "image.png", "width": 400, "height": 300},
)
row = sample["l3"]["rows"][0]
# Prefer measure-derived interiors: 3 measures → 2 splits
assert len(row["splits"]) == 2
assert len(row["measures"]) == 3
r = validate_layout_sample(sample)
assert r.ok, r.errors


def test_repo_sample_if_present() -> None:
"""Optional: samples/layout/*/layout.json committed for #93."""
root = Path(__file__).resolve().parents[2] / "samples" / "layout"
Expand Down
Loading