diff --git a/src/songtrace/edits.py b/src/songtrace/edits.py new file mode 100644 index 0000000..116502f --- /dev/null +++ b/src/songtrace/edits.py @@ -0,0 +1,173 @@ +"""Manual edits to a song, kept apart from what the pipeline generates. + +Edits live in work//edits.json, which no stage produces, so no rebuild +removes them; grid lays them over the freshly generated notes every time. An +edit is a whole lyric line as the user left it, timed in seconds so that a +rebuild with a different BPM re-snaps it rather than breaking it. +""" +import difflib +import json +import time +from pathlib import Path + +from .notes import MIN_NOTE_SEC, TimedNote, low_confidence_warnings + +EDITS_FILE = "edits.json" +FINAL_FILE = "notes_final.json" +KINDS = {":", "*", "F", "R", "G"} + + +def _empty(lyrics: list[list[str]]) -> dict: + return {"lyrics": lyrics, "lines": {}, "stale": []} + + +def load(work: Path, lyrics: list[list[str]]) -> dict: + """The song's edits, keyed to `lyrics`. + + Edits are stored with the lyrics they were made against. When those + differ, each edited line is matched to its line in the new lyrics by + text and order; an edit whose line's words changed is set aside as stale + rather than applied to words it was not made for. + """ + path = work / EDITS_FILE + if not path.exists(): + return _empty(lyrics) + data = json.loads(path.read_text()) + if data["lyrics"] != lyrics: + data = _remap(data, lyrics) + save(work, data) + return data + + +def save(work: Path, data: dict) -> None: + path = work / EDITS_FILE + if not data["lines"] and not data["stale"]: + path.unlink(missing_ok=True) + return + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(data)) + tmp.replace(path) + + +def _remap(data: dict, lyrics: list[list[str]]) -> dict: + old = [tuple(line) for line in data["lyrics"]] + new = [tuple(line) for line in lyrics] + moved = {} + for tag, i1, i2, j1, _ in difflib.SequenceMatcher( + a=old, b=new, autojunk=False).get_opcodes(): + if tag == "equal": + moved.update({i1 + k: j1 + k for k in range(i2 - i1)}) + lines, stale = {}, list(data["stale"]) + for key, edit in data["lines"].items(): + to = moved.get(int(key)) + if to is None: + stale.append(edit) + else: + lines[str(to)] = {**edit, "line_index": to} + return {"lyrics": lyrics, "lines": lines, "stale": stale} + + +def validate_notes(notes: list[dict]) -> list[dict]: + """The notes of a line edit, checked and reduced to what is stored.""" + clean = [] + for n in notes: + try: + note = {"start": float(n["start"]), "duration": float(n["duration"]), + "pitch": int(n["pitch"]), "text": str(n["text"]), + "kind": str(n.get("kind", ":"))} + except (KeyError, TypeError, ValueError): + raise ValueError(f"malformed note {n!r}") + if note["start"] < 0 or note["duration"] <= 0: + raise ValueError(f"a note needs start >= 0 and duration > 0: {n!r}") + if note["kind"] not in KINDS: + raise ValueError(f"unknown note kind {note['kind']!r}") + if not note["text"].strip(): + raise ValueError("a note needs text") + clean.append(note) + return sorted(clean, key=lambda n: n["start"]) + + +def set_line(data: dict, index: int, notes: list[dict]) -> None: + data["lines"][str(index)] = { + "line_index": index, "words": data["lyrics"][index], + "notes": validate_notes(notes), "saved_at": time.time()} + + +def clear_line(data: dict, index: int) -> bool: + return data["lines"].pop(str(index), None) is not None + + +def merge(notes: list[dict], breaks: list[float], data: dict + ) -> tuple[list[dict], list[float]]: + """Generated notes and breaks with every edited line laid over them. + + An edited line replaces the generated notes of that line, and the user's + timing wins: generated notes of earlier lines are cut back to end where it + starts, and those of later lines to start where it ends. Line breaks next + to an edited line move to the middle of the gap it leaves. + """ + edited = {int(k): e["notes"] for k, e in data["lines"].items()} + if not edited: + return notes, breaks + spans = {li: (ns[0]["start"], ns[-1]["start"] + ns[-1]["duration"]) + for li, ns in edited.items() if ns} + kept = [] + for n in notes: + li = n["line_index"] + if li in edited: + continue + start, end = n["start"], n["start"] + n["duration"] + for eli, (a, b) in spans.items(): + if li < eli: + end = min(end, a) + elif li > eli: + start = max(start, b) + if end - start >= MIN_NOTE_SEC: + kept.append({**n, "start": start, "duration": end - start}) + for li, ns in edited.items(): + kept += [{**n, "line_index": li, "low_confidence": False} for n in ns] + kept.sort(key=lambda n: (n["start"], n["line_index"])) + + def line_after(t: float) -> int | None: + return next((n["line_index"] for n in notes if n["start"] >= t), None) + + first = {} + last = {} + for n in kept: + first.setdefault(n["line_index"], n) + last[n["line_index"]] = n + order = sorted(first, key=lambda li: first[li]["start"]) + out_breaks = [] + for prev, nxt in zip(order, order[1:]): + if prev not in edited and nxt not in edited: + # the generated break between these two lines, untouched + gen = [t for t in breaks if line_after(t) == nxt] + if gen: + out_breaks.append(gen[0]) + continue + end = last[prev]["start"] + last[prev]["duration"] + out_breaks.append((end + first[nxt]["start"]) / 2) + return kept, out_breaks + + +def finalize(work: Path, lyrics: list[list[str]]) -> dict: + """The song as it is to be charted: generated notes with the edits laid + over them, warnings brought up to date, written to notes_final.json.""" + generated = json.loads((work / "notes_sec.json").read_text()) + data = load(work, lyrics) + notes, breaks = merge(generated["notes"], generated["breaks"], data) + warnings = generated.get("warnings", []) + if data["lines"]: + # low-confidence warnings must describe the merged notes, so a line + # the user fixed stops being reported; the others are about lyrics + warnings = (low_confidence_warnings([TimedNote(**n) for n in notes]) + + [w for w in warnings if "low-confidence" not in w]) + if data["stale"]: + n = len(data["stale"]) + warnings.append(f"{n} edited line{' no longer matches' if n == 1 else 's no longer match'}" + " the lyrics; set aside in the editor") + final = {"notes": notes, "breaks": breaks, "warnings": warnings, + "edited_lines": sorted(int(k) for k in data["lines"]), + "stale": data["stale"]} + (work / FINAL_FILE).write_text(json.dumps(final)) + return final diff --git a/src/songtrace/grid.py b/src/songtrace/grid.py index a04a6f5..b594f75 100644 --- a/src/songtrace/grid.py +++ b/src/songtrace/grid.py @@ -3,7 +3,8 @@ from pathlib import Path import numpy as np -from .ultrastar import LineBreak, Note, UltraStarSong, format_bpm +from .ultrastar import (LineBreak, Note, UltraStarSong, format_bpm, get_header, + parse_bpm, parse_ultrastar, serialize_ultrastar) def detect_bpm(instrumental_path: Path) -> float: @@ -49,10 +50,24 @@ def to_ultrastar(notes: list[dict], breaks: list[float], file_bpm: float, else: b = max(beat(t), cursor) length = max(1, beat(t + n["duration"]) - beat(t)) - body.append(Note(kind=":", beat=b, length=length, pitch=n["pitch"], - text=n["text"])) + body.append(Note(kind=n.get("kind", ":"), beat=b, length=length, + pitch=n["pitch"], text=n["text"])) cursor = b + length out_headers = list(headers) out_headers.append(("BPM", format_bpm(file_bpm))) out_headers.append(("GAP", str(gap_ms))) return UltraStarSong(headers=out_headers, body=body) + + +def rechart(txt_path: Path, notes: list[dict], breaks: list[float]) -> None: + """Replace a chart's notes, keeping its headers and its BPM. + + For an edit saved from the editor: the tempo was settled when the song + was built, and the packaged chart's headers may differ from the work + copy's (a re-encoded #VIDEO, a dropped #COVER), so each keeps its own. + """ + song = parse_ultrastar(txt_path.read_text(encoding="utf-8")) + bpm = parse_bpm(get_header(song, "BPM")) + headers = [(k, v) for k, v in song.headers if k.upper() not in ("BPM", "GAP")] + txt_path.write_text(serialize_ultrastar(to_ultrastar(notes, breaks, bpm, headers)), + encoding="utf-8") diff --git a/src/songtrace/notes.py b/src/songtrace/notes.py index 1af3553..20e166f 100644 --- a/src/songtrace/notes.py +++ b/src/songtrace/notes.py @@ -34,6 +34,7 @@ class TimedNote: text: str # display text incl. leading space / "~" low_confidence: bool line_index: int + kind: str = ":" # UltraStar note type; generated notes are all normal def hz_to_midi_float(hz: float) -> float: @@ -170,8 +171,7 @@ def _fmt(t: float) -> str: return f"{int(t // 60)}:{int(t % 60):02d}" -def _warnings(notes: list[TimedNote], dropped_lines=(), missing_spans=(), - ref_lines=()) -> list[str]: +def low_confidence_warnings(notes: list[TimedNote]) -> list[str]: runs: list[list[TimedNote]] = [] for n in notes: if not n.low_confidence: @@ -181,10 +181,15 @@ def _warnings(notes: list[TimedNote], dropped_lines=(), missing_spans=(), runs[-1].append(n) else: runs.append([n]) - warnings = [ + return [ f"{len(run)} low-confidence note{'' if len(run) == 1 else 's'} " f"near {_fmt(run[0].start)}–{_fmt(run[-1].start + run[-1].duration)}" for run in runs] + + +def _warnings(notes: list[TimedNote], dropped_lines=(), missing_spans=(), + ref_lines=()) -> list[str]: + warnings = low_confidence_warnings(notes) for group in _consecutive_runs(dropped_lines): n = len(group) text = " ".join(ref_lines[group[0]]) if group[0] < len(ref_lines) else "" diff --git a/src/songtrace/pipeline.py b/src/songtrace/pipeline.py index 53b856d..f3faf80 100644 --- a/src/songtrace/pipeline.py +++ b/src/songtrace/pipeline.py @@ -47,7 +47,11 @@ def song_slug(inp: SongInput) -> str: return name.replace("/", "").replace("\0", "") def read_warnings(work: Path) -> list[str]: - notes_file = work / "notes_sec.json" + # grid's copy accounts for the user's edits; notes' is all a song built + # before edits existed has + notes_file = work / "notes_final.json" + if not notes_file.exists(): + notes_file = work / "notes_sec.json" if not notes_file.exists(): return [] return json.loads(notes_file.read_text()).get("warnings", []) diff --git a/src/songtrace/stages/grid.py b/src/songtrace/stages/grid.py index b5795b5..17f2629 100644 --- a/src/songtrace/stages/grid.py +++ b/src/songtrace/stages/grid.py @@ -1,13 +1,16 @@ import json from pathlib import Path -from .. import grid +from .. import edits, grid +from ..lyrics import normalize_lyrics from ..pipeline import Ctx, PipelineError, Stage, song_slug from ..ultrastar import serialize_ultrastar def _run(ctx: Ctx) -> None: - data = json.loads((ctx.work / "notes_sec.json").read_text()) + data = edits.finalize(ctx.work, normalize_lyrics(ctx.inp.lyrics_text).lines) + if data["edited_lines"]: + ctx.log(f"[grid] kept {len(data['edited_lines'])} edited line(s)") meta = json.loads((ctx.work / "meta.json").read_text()) if ctx.inp.bpm_override is not None: bpm = ctx.inp.bpm_override @@ -36,5 +39,6 @@ def _run(ctx: Ctx) -> None: def grid_stage() -> Stage: return Stage(name="grid", - outputs=lambda ctx: [ctx.work / "song.txt"], + outputs=lambda ctx: [ctx.work / "song.txt", + ctx.work / edits.FINAL_FILE], run=_run) diff --git a/src/songtrace/web.py b/src/songtrace/web.py index 156394f..07890ec 100644 --- a/src/songtrace/web.py +++ b/src/songtrace/web.py @@ -9,11 +9,12 @@ import uuid from dataclasses import asdict, dataclass, field from pathlib import Path -from fastapi import FastAPI, File, Form, HTTPException, UploadFile +from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile from fastapi.concurrency import run_in_threadpool from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles +from . import edits, grid from . import rebuild as rebuild_policy from .config import Config from .lyrics import normalize_lyrics @@ -132,6 +133,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI: jobs: dict[str, Job] = {} order: list[str] = [] lock = threading.Lock() + edit_lock = threading.Lock() todo: queue.Queue[str] = queue.Queue() for job in _discover_jobs(cfg): @@ -466,7 +468,9 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI: if job.status != "done": raise HTTPException(404) work = cfg.work_dir / song_slug(job.inp) - notes_file = work / "notes_sec.json" + notes_file = work / edits.FINAL_FILE + if not notes_file.exists(): + notes_file = work / "notes_sec.json" if not notes_file.exists(): raise HTTPException(404) notes_data = json.loads(notes_file.read_text()) @@ -486,13 +490,96 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI: for n in notes_data["notes"]: by_line.setdefault(n["line_index"], []).append({ "start": n["start"], "end": n["start"] + n["duration"], - "pitch": n["pitch"], "low_confidence": n["low_confidence"]}) + "pitch": n["pitch"], "low_confidence": n["low_confidence"], + "text": n["text"], "kind": n.get("kind", ":")}) + edited = set(notes_data.get("edited_lines", [])) lines = [{"index": i, "text": " ".join(words), "dropped": i in dropped, + "edited": i in edited, "notes": by_line.get(i, []), "words": words_by_line.get(i, [])} for i, words in enumerate(ref.lines)] + stale = [{"text": " ".join(e["words"]), "notes": e["notes"]} + for e in notes_data.get("stale", [])] media = {kind: _resolve_media(job, header) is not None for kind, header in MEDIA_HEADERS.items()} - return {"lines": lines, "missing_spans": missing_spans, "media": media} + return {"lines": lines, "missing_spans": missing_spans, "media": media, + "stale": stale, "grid": _chart_grid(work)} + + def _chart_grid(work: Path) -> dict | None: + """The beat grid the chart was built on, for snapping edits to it.""" + txt = work / "song.txt" + if not txt.exists(): + return None + song = parse_ultrastar(txt.read_text(encoding="utf-8")) + bpm, gap = get_header(song, "BPM"), get_header(song, "GAP") + if bpm is None or gap is None: + return None + return {"bpm": float(bpm.replace(",", ".")), "gap_ms": float(gap.replace(",", "."))} + + def _editable(job_id: str) -> Job: + job = _finished_job(job_id) + slug = song_slug(job.inp) + with lock: + # a build of this song would overwrite the chart being edited + if any(j.status in ("queued", "running") and song_slug(j.inp) == slug + for j in jobs.values()): + raise HTTPException(409, "this song is being rebuilt") + if job.status != "done": + raise HTTPException(409, "only a finished song can be edited") + return job + + def _edit(job_id: str, change) -> dict: + """Apply `change` to the song's edits, then rewrite its charts.""" + job = _editable(job_id) + work = cfg.work_dir / song_slug(job.inp) + lyrics = normalize_lyrics(job.inp.lyrics_text).lines + with edit_lock: + data = edits.load(work, lyrics) + try: + change(data, lyrics) + except ValueError as e: + raise HTTPException(400, str(e)) + edits.save(work, data) + final = edits.finalize(work, lyrics) + charts = [work / "song.txt"] + if job.output_dir: + charts.append(Path(job.output_dir) / f"{song_slug(job.inp)}.txt") + for txt in charts: + if txt.exists(): + grid.rechart(txt, final["notes"], final["breaks"]) + with lock: + job.warnings = list(final["warnings"]) + return get_review(job_id) + + def _line(lyrics, index: int) -> int: + if not 0 <= index < len(lyrics): + raise HTTPException(404, f"no line {index}") + return index + + @app.put("/api/jobs/{job_id}/lines/{index}") + def save_line(job_id: str, index: int, body: dict = Body(...)): + """Save a line as edited: its full note list, in seconds.""" + notes = body.get("notes") + if not isinstance(notes, list): + raise HTTPException(400, "body must be {\"notes\": [...]}") + return _edit(job_id, lambda data, lyrics: edits.set_line( + data, _line(lyrics, index), notes)) + + @app.delete("/api/jobs/{job_id}/lines/{index}") + def revert_line(job_id: str, index: int): + """Drop a line's edit, so it follows what the pipeline generates.""" + def change(data, lyrics): + if not edits.clear_line(data, _line(lyrics, index)): + raise HTTPException(404, f"line {index} has no edit") + return _edit(job_id, change) + + @app.delete("/api/jobs/{job_id}/stale/{n}") + def discard_stale(job_id: str, n: int): + """Throw away an edit whose line's lyrics have since changed.""" + def change(data, lyrics): + if not 0 <= n < len(data["stale"]): + raise HTTPException(404, f"no stale edit {n}") + data["stale"].pop(n) + return _edit(job_id, change) return app diff --git a/tests/test_edits.py b/tests/test_edits.py new file mode 100644 index 0000000..7cb3b7b --- /dev/null +++ b/tests/test_edits.py @@ -0,0 +1,124 @@ +import json + +import pytest + +from songtrace import edits + +LYRICS = [["one", "two"], ["three", "four"], ["five", "six"]] + + +def gen(start, line, pitch=0, dur=0.4, low=False, text="la"): + return {"start": start, "duration": dur, "pitch": pitch, "text": text, + "low_confidence": low, "line_index": line} + + +def edited(*lines): + data = edits._empty(LYRICS) + for index, notes in lines: + edits.set_line(data, index, notes) + return data + + +def note(start, pitch=5, dur=0.4, text="la", kind=":"): + return {"start": start, "duration": dur, "pitch": pitch, "text": text, "kind": kind} + + +GENERATED = [gen(1.0, 0), gen(1.5, 0), gen(4.0, 1), gen(4.5, 1), gen(8.0, 2), gen(8.5, 2)] +BREAKS = [3.0, 6.0] + + +def test_no_edits_leaves_the_generated_song_alone(): + assert edits.merge(GENERATED, BREAKS, edits._empty(LYRICS)) == (GENERATED, BREAKS) + + +def test_an_edited_line_replaces_its_generated_notes(): + notes, _ = edits.merge(GENERATED, BREAKS, edited( + (1, [note(4.2, pitch=7, kind="*"), note(4.8, pitch=9)]))) + line1 = [n for n in notes if n["line_index"] == 1] + assert [(n["start"], n["pitch"], n["kind"]) for n in line1] == [ + (4.2, 7, "*"), (4.8, 9, ":")] + assert not any(n["low_confidence"] for n in line1) + assert [n["line_index"] for n in notes] == [0, 0, 1, 1, 2, 2] + + +def test_neighbours_give_way_to_an_edited_line(): + # line 1 edited to run from 1.7s to 8.6s, over its neighbours' notes + notes, _ = edits.merge(GENERATED, BREAKS, edited( + (1, [note(1.7), note(8.2, dur=0.4)]))) + by_line = {} + for n in notes: + by_line.setdefault(n["line_index"], []).append( + (round(n["start"], 2), round(n["start"] + n["duration"], 2))) + assert by_line[0] == [(1.0, 1.4), (1.5, 1.7)] # cut back to end at 1.7s + assert by_line[2] == [(8.6, 8.9)] # 8.0 note gone, 8.5 starts at 8.6 + + +def test_breaks_beside_an_edited_line_move_and_the_rest_stay(): + data = edited((2, [note(9.0), note(9.5)])) + _, breaks = edits.merge(GENERATED, BREAKS, data) + assert breaks[0] == 3.0 # lines 0-1: generated + assert breaks[1] == pytest.approx((4.9 + 9.0) / 2) # lines 1-2: re-centred + + +def test_an_edit_follows_its_line_when_lines_are_added_above(tmp_path): + data = edited((1, [note(4.0)])) + edits.save(tmp_path, data) + moved = edits.load(tmp_path, [["zero"]] + LYRICS) + assert list(moved["lines"]) == ["2"] and moved["lines"]["2"]["line_index"] == 2 + assert moved["stale"] == [] + assert json.loads((tmp_path / edits.EDITS_FILE).read_text())["lyrics"][0] == ["zero"] + + +def test_an_edit_whose_words_changed_is_set_aside(tmp_path): + edits.save(tmp_path, edited((1, [note(4.0)]), (2, [note(8.0)]))) + changed = [LYRICS[0], ["three", "more"], LYRICS[2]] + data = edits.load(tmp_path, changed) + assert list(data["lines"]) == ["2"] + assert [s["words"] for s in data["stale"]] == [["three", "four"]] + + +def test_validate_notes_checks_and_orders(): + assert [n["start"] for n in edits.validate_notes([note(2.0), note(1.0)])] == [1.0, 2.0] + for bad in (note(1.0, kind="X"), note(1.0, dur=0), note(-1.0), note(1.0, text=" "), + {"start": 1.0}): + with pytest.raises(ValueError): + edits.validate_notes([bad]) + + +def test_saving_no_edits_removes_the_file(tmp_path): + edits.save(tmp_path, edited((0, [note(1.0)]))) + data = edits.load(tmp_path, LYRICS) + edits.clear_line(data, 0) + edits.save(tmp_path, data) + assert not (tmp_path / edits.EDITS_FILE).exists() + + +def _write_generated(work, notes, warnings): + (work / "notes_sec.json").write_text(json.dumps( + {"notes": notes, "breaks": BREAKS, "warnings": warnings})) + + +def test_finalize_without_edits_passes_warnings_through(tmp_path): + _write_generated(tmp_path, GENERATED, ["anything at all"]) + final = edits.finalize(tmp_path, LYRICS) + assert final["warnings"] == ["anything at all"] + assert final["edited_lines"] == [] and final["stale"] == [] + assert json.loads((tmp_path / edits.FINAL_FILE).read_text()) == final + + +def test_finalize_stops_reporting_a_line_the_user_fixed(tmp_path): + notes = [gen(1.0, 0), gen(4.0, 1, low=True), gen(8.0, 2)] + _write_generated(tmp_path, notes, ["1 low-confidence note near 0:04–0:04", + "missing lyrics near 0:20–0:25"]) + edits.save(tmp_path, edited((1, [note(4.0)]))) + final = edits.finalize(tmp_path, LYRICS) + assert final["warnings"] == ["missing lyrics near 0:20–0:25"] + assert final["edited_lines"] == [1] + + +def test_finalize_reports_stale_edits(tmp_path): + _write_generated(tmp_path, GENERATED, []) + edits.save(tmp_path, edited((1, [note(4.0)]))) + final = edits.finalize(tmp_path, [LYRICS[0], ["other"], LYRICS[2]]) + assert final["warnings"] == ["1 edited line no longer matches the lyrics; " + "set aside in the editor"] diff --git a/tests/test_grid.py b/tests/test_grid.py index 88ab114..f73fe22 100644 --- a/tests/test_grid.py +++ b/tests/test_grid.py @@ -169,7 +169,7 @@ def test_grid_stage_writes_song_txt(tmp_path, monkeypatch): monkeypatch.setattr(grid_stage_mod.grid, "detect_bpm", lambda p: 120.0) ctx = make_ctx(work, language="English", genre="Pop", year=1999) stage = grid_stage_mod.grid_stage() - assert stage.outputs(ctx) == [work / "song.txt"] + assert stage.outputs(ctx) == [work / "song.txt", work / "notes_final.json"] stage.run(ctx) song = parse_ultrastar((work / "song.txt").read_text(encoding="utf-8")) assert [k for k, _ in song.headers] == [ @@ -257,3 +257,45 @@ def test_grid_stage_reports_no_notes_as_pipeline_error(tmp_path, monkeypatch): def test_grid_stage_registered(): names = [s.name for s in all_stages()] assert names[-1] == "package" and names[-2] == "grid" + + +def _regenerate(work, pitch): + """What a rebuild leaves: fresh notes for both lines, at `pitch`.""" + (work / "notes_sec.json").write_text(json.dumps({ + "notes": [{**note(1.0, 0.5, pitch, "one"), "line_index": 0}, + {**note(3.0, 0.5, pitch, "three"), "line_index": 1}], + "breaks": [2.0], "warnings": []})) + + +def test_an_edited_line_survives_a_rebuild(tmp_path, monkeypatch): + from songtrace import edits + work = tmp_path / "w" + write_work(work) + monkeypatch.setattr(grid_stage_mod.grid, "detect_bpm", lambda p: 120.0) + ctx = make_ctx(work) + ctx.inp.lyrics_text = "one\nthree" + data = edits.load(work, [["one"], ["three"]]) + edits.set_line(data, 1, [{"start": 3.2, "duration": 0.6, "pitch": 9, + "text": "three", "kind": "*"}]) + edits.save(work, data) + for regenerated_pitch in (0, 4): + _regenerate(work, regenerated_pitch) + grid_stage_mod.grid_stage().run(ctx) + song = parse_ultrastar((work / "song.txt").read_text(encoding="utf-8")) + notes = [n for n in song.body if isinstance(n, Note)] + assert [(n.kind, n.pitch, n.text) for n in notes] == [ + (":", regenerated_pitch, "one"), ("*", 9, "three")] + + +def test_rechart_replaces_the_notes_and_keeps_headers_and_tempo(tmp_path): + from songtrace.grid import rechart + from songtrace.ultrastar import serialize_ultrastar + txt = tmp_path / "song.txt" + txt.write_text(serialize_ultrastar(to_ultrastar( + [note(1.0, 0.5)], [], 300.0, + [("TITLE", "Song"), ("VIDEO", "Band - Song.mp4")])), encoding="utf-8") + rechart(txt, [{**note(2.0, 0.5, 7, "new"), "kind": "F"}], []) + song = parse_ultrastar(txt.read_text(encoding="utf-8")) + assert [k for k, _ in song.headers] == ["TITLE", "VIDEO", "BPM", "GAP"] + assert get_header(song, "BPM") == "300" and get_header(song, "GAP") == "2000" + assert [(n.kind, n.pitch, n.text) for n in song.body] == [("F", 7, "new")] diff --git a/tests/test_web.py b/tests/test_web.py index bcdcbd2..5896939 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -1412,13 +1412,16 @@ def test_review_endpoint_groups_notes_by_line(tmp_path): wait_status(client, job_id, "done") body = client.get(f"/api/jobs/{job_id}/review").json() assert body["lines"] == [ - {"index": 0, "text": "hi", "dropped": False, - "notes": [{"start": 1.0, "end": 1.4, "pitch": 0, "low_confidence": False}], + {"index": 0, "text": "hi", "dropped": False, "edited": False, + "notes": [{"start": 1.0, "end": 1.4, "pitch": 0, "low_confidence": False, + "text": "hi", "kind": ":"}], "words": []}, - {"index": 1, "text": "there", "dropped": False, - "notes": [{"start": 5.0, "end": 5.3, "pitch": 2, "low_confidence": True}], + {"index": 1, "text": "there", "dropped": False, "edited": False, + "notes": [{"start": 5.0, "end": 5.3, "pitch": 2, "low_confidence": True, + "text": "there", "kind": ":"}], + "words": []}, + {"index": 2, "text": "gone", "dropped": True, "edited": False, "notes": [], "words": []}, - {"index": 2, "text": "gone", "dropped": True, "notes": [], "words": []}, ] assert body["missing_spans"] == [{"start": 8.0, "end": 9.5}] @@ -1872,3 +1875,105 @@ def test_a_job_is_only_reported_finished_once_it_has_finished(tmp_path, monkeypa job = wait_status(client, job_id, "done") assert job["finished_at"] is not None assert not (cfg.work_dir / "uploads" / job_id / "v.mp4").exists() + + +def chart_runner(inp, cfg, log=print, force_from=None, stages=None): + """Leaves what a real build does for the editor: generated notes and a + chart in the work dir, and the packaged chart.""" + from songtrace.grid import to_ultrastar + from songtrace.ultrastar import serialize_ultrastar + work = cfg.work_dir / song_slug(inp) + work.mkdir(parents=True, exist_ok=True) + notes = [{"start": 1.0, "duration": 0.5, "pitch": 0, "text": "one", + "low_confidence": False, "line_index": 0}, + {"start": 3.0, "duration": 0.5, "pitch": 2, "text": "three", + "low_confidence": True, "line_index": 1}] + warnings = ["1 low-confidence note near 0:03–0:03"] + (work / "notes_sec.json").write_text(json.dumps( + {"notes": notes, "breaks": [2.0], "warnings": warnings})) + chart = serialize_ultrastar(to_ultrastar(notes, [2.0], 300.0, [("TITLE", inp.title)])) + (work / "song.txt").write_text(chart, encoding="utf-8") + out = cfg.output_dir / song_slug(inp) + out.mkdir(parents=True, exist_ok=True) + (out / f"{song_slug(inp)}.txt").write_text(chart, encoding="utf-8") + return PipelineResult(output_dir=out, warnings=warnings) + + +def make_charted_job(client, lyrics="one\nthree"): + job_id = client.post("/api/jobs", data={ + "title": "T", "artist": "A", "lyrics": lyrics, "youtube_url": "http://y"}).json()["id"] + wait_status(client, job_id, "done") + return job_id + + +EDIT = {"notes": [{"start": 3.1, "duration": 0.5, "pitch": 9, "text": "three", + "kind": "*"}]} + + +def test_saving_a_line_rewrites_both_charts(tmp_path): + from songtrace.ultrastar import Note, parse_ultrastar + client, cfg = make_client(tmp_path, chart_runner) + job_id = make_charted_job(client) + resp = client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT) + assert resp.status_code == 200 + line = resp.json()["lines"][1] + assert line["edited"] and line["notes"][0]["kind"] == "*" and line["notes"][0]["pitch"] == 9 + assert resp.json()["grid"] == {"bpm": 300.0, "gap_ms": 1000.0} + for txt in (cfg.work_dir / "A - T" / "song.txt", cfg.output_dir / "A - T" / "A - T.txt"): + notes = [n for n in parse_ultrastar(txt.read_text(encoding="utf-8")).body + if isinstance(n, Note)] + assert [(n.kind, n.pitch) for n in notes] == [(":", 0), ("*", 9)], txt + # the fixed line was the one being warned about + assert client.get(f"/api/jobs/{job_id}").json()["warnings"] == [] + + +def test_reverting_a_line_returns_it_to_the_generated_notes(tmp_path): + client, _ = make_client(tmp_path, chart_runner) + job_id = make_charted_job(client) + client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT) + resp = client.delete(f"/api/jobs/{job_id}/lines/1") + assert resp.status_code == 200 + line = resp.json()["lines"][1] + assert not line["edited"] and line["notes"][0]["pitch"] == 2 + assert client.delete(f"/api/jobs/{job_id}/lines/1").status_code == 404 + + +def test_line_edits_are_checked(tmp_path): + client, _ = make_client(tmp_path, chart_runner) + job_id = make_charted_job(client) + assert client.put(f"/api/jobs/{job_id}/lines/5", json=EDIT).status_code == 404 + bad = {"notes": [{**EDIT["notes"][0], "kind": "X"}]} + assert client.put(f"/api/jobs/{job_id}/lines/1", json=bad).status_code == 400 + assert client.put(f"/api/jobs/{job_id}/lines/1", json={"x": 1}).status_code == 400 + + +def test_a_song_cannot_be_edited_while_it_is_being_rebuilt(tmp_path): + gate = threading.Event() + + def runner(inp, cfg, log=print, force_from=None, stages=None): + if force_from: + gate.wait(timeout=5) + return chart_runner(inp, cfg) + + client, _ = make_client(tmp_path, runner) + try: + job_id = make_charted_job(client) + client.post(f"/api/jobs/{job_id}/rebuild-full") + assert client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT).status_code == 409 + finally: + gate.set() + + +def test_a_stale_edit_can_be_discarded(tmp_path): + from songtrace import edits + client, cfg = make_client(tmp_path, chart_runner) + first = make_charted_job(client) + client.put(f"/api/jobs/{first}/lines/1", json=EDIT) + # the same song again with that line's words changed + second = make_charted_job(client, lyrics="one\nthree more") + work = cfg.work_dir / "A - T" + assert len(edits.load(work, [["one"], ["three", "more"]])["stale"]) == 1 + resp = client.delete(f"/api/jobs/{second}/stale/0") + assert resp.status_code == 200 and resp.json()["stale"] == [] + assert not (work / edits.EDITS_FILE).exists() + assert client.delete(f"/api/jobs/{second}/stale/0").status_code == 404