Esa Kataja
cdf37302d7
A bundle without a source block cannot be round-tripped, but its PDF can still be cut from scratch. Offer that instead of refusing: confirm, run detection, and line the markers up by position when the slice counts match exactly.
388 lines
15 KiB
Python
388 lines
15 KiB
Python
"""Bundle export — the only channel to noteman (ADR 0001).
|
|
|
|
song.zip
|
|
song.json
|
|
original.pdf
|
|
001.webp 002.webp …
|
|
|
|
Array order in `song.json` *is* slice order: one ordering, not two. Markers
|
|
nest inside the slice they sit on, so an index appears in exactly one place —
|
|
a jump source's `destination`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from .pdf import Source
|
|
from .project import Project
|
|
from .render import render_song
|
|
|
|
FORMAT_VERSION = 1
|
|
METADATA_FIELDS = (
|
|
"title",
|
|
"subtitle",
|
|
"composer",
|
|
"original_artist",
|
|
"arranger",
|
|
"lyricist",
|
|
"translator",
|
|
"tempo",
|
|
"voices",
|
|
)
|
|
|
|
# Beats per minute, exported as a JSON number. A figure is worth more than a
|
|
# word here: "Andante" cannot drive a metronome and two people will not agree
|
|
# what it means.
|
|
NUMERIC_FIELDS = frozenset({"tempo"})
|
|
|
|
|
|
def filename(project: Project) -> str:
|
|
"""The bundle's name, from the song's title.
|
|
|
|
Spaces become dashes and anything that is not a letter, digit, dash, dot or
|
|
underscore goes. Letters keep their accents — ä and ö are not a filesystem's
|
|
problem — but a leading dot would make the bundle invisible.
|
|
"""
|
|
# Drop the unsafe characters before collapsing whitespace, not after, or
|
|
# "Sävel & Ääni" keeps the dash the ampersand left behind.
|
|
title = re.sub(r"[^\w\s.-]", "", (project.metadata.get("title") or ""))
|
|
return f"{re.sub(r'\s+', '-', title.strip()).lstrip('.-') or 'song'}.zip"
|
|
|
|
|
|
def _engraving(project: Project, page: int, slot: int) -> dict | None:
|
|
"""The notation behind a re-engraved slice, or None for a scanned one.
|
|
|
|
The slice image stays the presentation; this is the notation it was made
|
|
from, carried so the music can be edited again or turned into sound. Key
|
|
and time are resolved against the song defaults here — a consumer reading
|
|
one slice should not have to know what the rest of the song inherited.
|
|
"""
|
|
replacement = project.pages[page].replacements[slot]
|
|
if not replacement or not replacement.voices:
|
|
return None
|
|
return {
|
|
"lang": "lilypond",
|
|
"key": replacement.key or project.key,
|
|
"time": replacement.time or project.time,
|
|
"print_time": replacement.print_time,
|
|
"voices": [
|
|
{"clef": v.clef, "notes": v.notes.strip()}
|
|
| ({"lyrics": v.lyrics.strip()} if v.lyrics.strip() else {})
|
|
for v in replacement.voices
|
|
],
|
|
}
|
|
|
|
|
|
def _source(project: Project) -> dict:
|
|
"""How the slices were cut from the archived PDF.
|
|
|
|
Without this a bundle is a one-way trip: the images are output and the cuts
|
|
that made them live only in the producer's own project file, so reopening
|
|
someone else's bundle would mean cutting the score again from scratch. It
|
|
is geometry in normalised page coordinates, so it survives the PDF being
|
|
rendered at any resolution.
|
|
|
|
Only the pages are here. Which slot on which page a slice came from is on
|
|
the slice itself, so that one ordering — the slices array — stays the only
|
|
one, and a slot no slice claims is a slot that was discarded.
|
|
"""
|
|
return {
|
|
"file": "original.pdf",
|
|
"pages": [
|
|
{
|
|
"skew": round(page.skew, 2),
|
|
"content": [round(v, 5) for v in project.page_content_rect(i)],
|
|
"levels": list(project.page_levels(i)),
|
|
"cuts": [[[round(x, 5), round(y, 5)] for x, y in cut.points] for cut in page.cuts],
|
|
}
|
|
for i, page in enumerate(project.pages)
|
|
],
|
|
}
|
|
|
|
|
|
def song_json(project: Project, files: list[str]) -> dict:
|
|
payload: dict = {"v": FORMAT_VERSION}
|
|
for field in METADATA_FIELDS:
|
|
value = (project.metadata.get(field) or "").strip()
|
|
if not value:
|
|
continue
|
|
if field in NUMERIC_FIELDS:
|
|
try:
|
|
payload[field] = int(value)
|
|
except ValueError:
|
|
continue # not a number, so not worth exporting as one
|
|
else:
|
|
payload[field] = value
|
|
|
|
kept = project.kept_slices()
|
|
# Markers reference slices by (page, slot) while editing, because that is
|
|
# what survives adding and removing cuts. In the bundle they become the
|
|
# array index, which is the only cross-reference the format has.
|
|
index_of = {position: i for i, position in enumerate(kept)}
|
|
|
|
slices: list[dict] = []
|
|
for name, (page, slot) in zip(files, kept):
|
|
entry: dict = {"file": name, "page": page, "slot": slot}
|
|
bar = project.pages[page].bars[slot]
|
|
if bar:
|
|
entry["bar"] = bar
|
|
engraving = _engraving(project, page, slot)
|
|
if engraving:
|
|
entry["engraving"] = engraving
|
|
markers = []
|
|
for marker in project.pages[page].markers[slot]:
|
|
item: dict = {"type": marker.type}
|
|
if marker.label:
|
|
item["label"] = marker.label
|
|
if marker.destination is not None:
|
|
target = index_of.get(tuple(marker.destination))
|
|
# A jump whose target was discarded or re-cut away is dropped
|
|
# rather than exported dangling: noteman would have nothing to
|
|
# resolve it to.
|
|
if target is None:
|
|
continue
|
|
item["destination"] = target
|
|
markers.append(item)
|
|
if markers:
|
|
entry["markers"] = markers
|
|
slices.append(entry)
|
|
|
|
payload["slices"] = slices
|
|
if project.source.exists():
|
|
payload["source"] = _source(project)
|
|
return payload
|
|
|
|
|
|
def write(project: Project, source: Source, path: Path) -> Path:
|
|
"""Render the song and write the bundle. Returns the zip path.
|
|
|
|
A title is required; every other metadata field is optional. noteman's own
|
|
rule is that a song needs a title and at least one slice, and a bundle that
|
|
cannot become a song is not worth writing.
|
|
"""
|
|
if not project.metadata.get("title", "").strip():
|
|
raise ValueError("a title is required before a song can be exported")
|
|
|
|
images = render_song(project, source)
|
|
if not images:
|
|
raise ValueError("no slices to export — every slice is discarded")
|
|
names = [f"{i + 1:03}.webp" for i in range(len(images))]
|
|
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
# ZIP_STORED for the images: WebP is already compressed, so deflating it
|
|
# only costs time. The JSON is small enough not to care.
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr(
|
|
"song.json",
|
|
json.dumps(song_json(project, names), indent=2, ensure_ascii=False),
|
|
zipfile.ZIP_DEFLATED,
|
|
)
|
|
if project.source.exists():
|
|
pdf = project.source.read_bytes()
|
|
if project.optimise_pdf:
|
|
import pymupdf
|
|
|
|
from .pdfopt import optimise
|
|
|
|
shrunk, _ = optimise(pymupdf.open(project.source), len(pdf))
|
|
pdf = shrunk or pdf # empty means it found no saving
|
|
zf.writestr("original.pdf", pdf, zipfile.ZIP_STORED)
|
|
for name, data in zip(names, images):
|
|
zf.writestr(name, data, zipfile.ZIP_STORED)
|
|
|
|
# The project is spent once its song has been exported: the next edit
|
|
# session starts fresh from detection rather than resuming these decisions.
|
|
# Recorded here so no caller can forget it.
|
|
project.exported = True
|
|
project.save()
|
|
return path
|
|
|
|
|
|
class NoCuts(ValueError):
|
|
"""The bundle carries no `source` block, so its cuts cannot be restored."""
|
|
|
|
|
|
def _pages_from(geometry: dict, slices: list[dict]):
|
|
"""Rebuild the pages from a manifest's source geometry."""
|
|
from .project import Cut, Page
|
|
|
|
# Every slot on a page exists; the ones no slice claims were discarded.
|
|
# That is the one thing the bundle states by omission rather than directly,
|
|
# since shipping a discarded slice's image would defeat discarding it.
|
|
claimed = {(s["page"], s["slot"]) for s in slices}
|
|
pages = []
|
|
for i, page in enumerate(geometry["pages"]):
|
|
cuts = [Cut([tuple(p) for p in cut]) for cut in page["cuts"]]
|
|
count = len(cuts) + 1
|
|
pages.append(
|
|
Page(
|
|
skew=page.get("skew", 0.0),
|
|
cuts=cuts,
|
|
discards=[(i, slot) not in claimed for slot in range(count)],
|
|
markers=[[] for _ in range(count)],
|
|
replacements=[None] * count,
|
|
bars=[None] * count,
|
|
content_rect=tuple(page["content"]) if page.get("content") else None,
|
|
levels=tuple(page["levels"]) if page.get("levels") else None,
|
|
)
|
|
)
|
|
return pages
|
|
|
|
|
|
def _pages_from_detection(pdf: Path):
|
|
"""Cut the PDF again from scratch, for a bundle that recorded no geometry."""
|
|
from .detect import detect_page
|
|
from .pdf import open_source, page_raster
|
|
from .project import Project
|
|
|
|
source = open_source(pdf)
|
|
detections, heights = [], []
|
|
try:
|
|
for i in range(len(source)):
|
|
gray = page_raster(source, i)
|
|
detections.append(detect_page(gray))
|
|
heights.append(gray.shape[0])
|
|
finally:
|
|
source.close()
|
|
return Project.from_detection(pdf, detections, heights)
|
|
|
|
|
|
def _restore(project, slices: list[dict], positions: list[tuple[int, int]]) -> None:
|
|
"""Put each slice's bar number, markers and engraving back on its slot."""
|
|
from .project import Marker, Replacement, Voice
|
|
|
|
for entry, (page, slot) in zip(slices, positions):
|
|
project.pages[page].bars[slot] = entry.get("bar")
|
|
project.pages[page].markers[slot] = [
|
|
Marker(
|
|
type=marker["type"],
|
|
label=marker.get("label"),
|
|
# Back from an array index to the (page, slot) the editor works
|
|
# in — the inverse of what export does.
|
|
destination=(
|
|
positions[marker["destination"]]
|
|
if marker.get("destination") is not None
|
|
and marker["destination"] < len(positions)
|
|
else None
|
|
),
|
|
)
|
|
for marker in entry.get("markers", [])
|
|
]
|
|
engraving = entry.get("engraving")
|
|
if engraving and engraving.get("lang") == "lilypond":
|
|
project.pages[page].replacements[slot] = Replacement(
|
|
voices=[
|
|
Voice(
|
|
clef=v.get("clef", "treble"),
|
|
notes=v.get("notes", ""),
|
|
lyrics=v.get("lyrics", ""),
|
|
)
|
|
for v in engraving.get("voices", [])
|
|
],
|
|
print_time=engraving.get("print_time", False),
|
|
)
|
|
for entry in slices:
|
|
# Key and time are per slice in the bundle and per song here; the first
|
|
# engraving that states them is as good a song default as exists.
|
|
engraving = entry.get("engraving") or {}
|
|
if engraving.get("key"):
|
|
project.key = engraving["key"]
|
|
project.time = engraving.get("time", project.time)
|
|
break
|
|
|
|
|
|
def has_cuts(path: Path) -> bool:
|
|
"""Whether this bundle records the geometry its slices were cut with.
|
|
|
|
Worth asking before unpacking, since the answer decides whether reopening
|
|
is a round trip or a fresh session with the same PDF.
|
|
"""
|
|
with zipfile.ZipFile(Path(path)) as zf:
|
|
if "song.json" not in zf.namelist():
|
|
return False
|
|
return bool(json.loads(zf.read("song.json")).get("source"))
|
|
|
|
|
|
def read(
|
|
path: Path, into: Path | None = None, *, force: bool = False, detect: bool = False
|
|
) -> tuple[Project, Path]:
|
|
"""Unpack a bundle back into an editable project. Returns it and its PDF.
|
|
|
|
The archived PDF is written out beside the bundle and becomes the project's
|
|
source again, because the PDF is what the pipeline renders from — the slice
|
|
images in the zip are output, and are discarded rather than re-imported.
|
|
|
|
The cuts, skew, levels and content rectangles come from the manifest's
|
|
`source` block, so this is a real round trip rather than a re-detection
|
|
that happens to land nearby. A bundle written without one raises `NoCuts`;
|
|
`detect` says to cut the PDF from scratch instead, which is a different
|
|
thing and worth a caller asking about first.
|
|
"""
|
|
from .project import Project, default_path, hash_file
|
|
|
|
path = Path(path)
|
|
with zipfile.ZipFile(path) as zf:
|
|
names = set(zf.namelist())
|
|
if "song.json" not in names:
|
|
raise ValueError(f"{path.name} is not a bundle: no song.json")
|
|
manifest = json.loads(zf.read("song.json"))
|
|
if manifest.get("v") != FORMAT_VERSION:
|
|
raise ValueError(f"unsupported bundle version {manifest.get('v')!r}")
|
|
geometry = manifest.get("source")
|
|
if not geometry and not detect:
|
|
raise NoCuts(
|
|
f"{path.name} carries no cuts — it was written by a producer that "
|
|
"does not record them"
|
|
)
|
|
pdf_name = (geometry or {}).get("file", "original.pdf")
|
|
if pdf_name not in names:
|
|
raise ValueError(f"{path.name} names {pdf_name} but does not contain it")
|
|
pdf_bytes = zf.read(pdf_name)
|
|
|
|
# Unpacking writes two files. Refuse to land on either if it is already
|
|
# there: the obvious place to open a bundle is next to the score it came
|
|
# from, and that is exactly where someone's unfinished cuts live.
|
|
target = Path(into) if into else path.with_suffix(".pdf")
|
|
existing = [f for f in (target, default_path(target)) if f.exists()]
|
|
if existing and not force:
|
|
raise ValueError(
|
|
f"{', '.join(f.name for f in existing)} already exists — "
|
|
"open it with --pdf elsewhere, or --force to overwrite"
|
|
)
|
|
target.write_bytes(pdf_bytes)
|
|
|
|
slices = manifest["slices"]
|
|
if geometry:
|
|
pages_json = geometry["pages"]
|
|
first = pages_json[0] if pages_json else {}
|
|
project = Project(
|
|
source=target,
|
|
source_hash=hash_file(target),
|
|
pages=_pages_from(geometry, slices),
|
|
content_rect=tuple(first.get("content", (0.0, 0.0, 1.0, 1.0))),
|
|
levels=tuple(first.get("levels", (0, 255))),
|
|
path=default_path(target),
|
|
)
|
|
positions = [(s["page"], s["slot"]) for s in slices]
|
|
else:
|
|
project = _pages_from_detection(target)
|
|
project.path = default_path(target)
|
|
# Detection's slices are in reading order and so are the bundle's, so
|
|
# they can be lined up — but only if there are exactly as many. One
|
|
# system found or missed shifts every marker onto the wrong slice,
|
|
# which is worse than not placing them at all.
|
|
positions = project.kept_slices()
|
|
if len(positions) != len(slices):
|
|
positions = []
|
|
|
|
project.metadata = {
|
|
field: str(manifest[field]) for field in METADATA_FIELDS if manifest.get(field) is not None
|
|
}
|
|
_restore(project, slices[: len(positions)], positions)
|
|
return project, target
|