| """Kaggle: демонстрация и прогон пакета PLUS на своей пачке треков. |
| |
| Что делает ячейка: |
| 1. ставит зависимости (pretty_midi, librosa, soundfile, scipy); |
| 2. скачивает пакет с Hugging Face (или берёт локальную копию); |
| 3. прогоняет самопроверку на синтетике: чистая правда -> три «движка» |
| с ошибками -> слияние -> F1, чтобы видеть, что пакет реально работает; |
| 4. обрабатывает папку AUDIO_DIR (если в ней есть MIDI) и сводит отчёт; |
| 5. по флагу заливает результаты на Hugging Face. |
| |
| Локальный запуск: python3 kaggle_plus.py --demo |
| Своя папка: python3 kaggle_plus.py --input-dir /kaggle/input/takes --out-dir /kaggle/working/plus |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import subprocess |
| import sys |
|
|
| PACK = "plus.zip" |
| REPO = "Nokia123543/muscriptor-large-fork" |
| WORK = "/kaggle/working" if os.path.isdir("/kaggle/working") else os.getcwd() |
|
|
|
|
| def step(number: int, total: int, text: str) -> None: |
| print("[%d/%d] %s" % (number, total, text), flush=True) |
|
|
|
|
| def ensure_deps() -> None: |
| need = [] |
| for module, package in (("pretty_midi", "pretty_midi"), ("librosa", "librosa"), |
| ("soundfile", "soundfile"), ("scipy", "scipy")): |
| try: |
| __import__(module) |
| except Exception: |
| need.append(package) |
| if need: |
| step(1, 5, "ставлю: " + ", ".join(need)) |
| subprocess.run([sys.executable, "-m", "pip", "install", "-q"] + need, check=True) |
|
|
|
|
| def ensure_pack(target: str) -> str: |
| here = os.path.dirname(os.path.abspath(__file__)) |
| if os.path.isfile(os.path.join(here, "plus_core.py")): |
| step(2, 5, "пакет найден рядом: " + here) |
| return here |
| bundle = os.path.join(target, "plus") |
| if not os.path.isdir(bundle): |
| step(2, 5, "скачиваю пакет с HF") |
| os.makedirs(target, exist_ok=True) |
| from huggingface_hub import hf_hub_download |
| archive = hf_hub_download(repo_id=REPO, filename=PACK, local_dir=target) |
| import zipfile |
| with zipfile.ZipFile(archive) as zf: |
| zf.extractall(target) |
| step(2, 5, "пакет распакован: " + bundle) |
| return bundle |
|
|
|
|
| def demo(pack_dir: str) -> dict: |
| """Синтетика: правда, три движка с ошибками, слияние, метрики.""" |
| sys.path.insert(0, pack_dir) |
| import numpy as np |
| from plus_core import json_default, Note, Piece, save_midi |
| from plus_ensemble import compare, merge |
| from plus_transcribe import run_pipeline |
|
|
| step(3, 5, "самопроверка: собираю правду и три «движка»") |
| rng = np.random.default_rng(11) |
| beat = 60 / 72 / 2 |
| notes = [] |
| melody = [72, 74, 76, 77, 79, 77, 76, 74] * 4 |
| for index, pitch in enumerate(melody): |
| notes.append(Note(max(0.0, index * beat + rng.normal(0, 0.010)), beat * 0.9, pitch, 90)) |
| for bar in range(32): |
| root, shape = [(48, (0, 4, 7)), (53, (0, 3, 7)), (55, (0, 4, 7)), (48, (0, 4, 7))][bar % 4] |
| start = max(0.0, bar * beat * 2 + rng.normal(0, 0.006)) |
| for offset in shape: |
| notes.append(Note(start, beat * 1.8, root + offset, 80)) |
| truth = Piece(sorted(notes, key=lambda n: n.start), [(0.0, 72.0)]) |
|
|
| work = os.path.join(WORK, "plus_demo") |
| os.makedirs(work, exist_ok=True) |
| truth_path = os.path.join(work, "truth.mid") |
| save_midi(truth, truth_path) |
|
|
| sources = [] |
| for tag, seed, weight in (("muscriptor", 1, 1.0), ("aria", 2, 1.0)): |
| r = np.random.default_rng(seed) |
| variant = [Note(**vars(n)) for n in truth.notes if r.random() > 0.10] |
| for _ in range(4): |
| variant.append(Note(float(r.uniform(0, 20)), 0.30, int(r.integers(50, 88)), 90)) |
| variant.sort(key=lambda n: n.start) |
| path = os.path.join(work, "engine_%s.mid" % tag) |
| save_midi(Piece(variant, [(0.0, 72.0)]), path) |
| sources.append((tag, variant, weight)) |
|
|
| step(4, 5, "слияние голосованием") |
| merged, stats = merge(sources, tolerance=0.06, min_votes=2.0, keep_strong=False, tempo=72.0) |
| merged_path = os.path.join(work, "merged.mid") |
| save_midi(merged, merged_path) |
|
|
| result = {"merged_f1": compare(truth.notes, merged.notes), |
| "sources": {name: compare(truth.notes, notes_) for name, notes_, _w in sources}, |
| "agreement": stats} |
|
|
| step(5, 5, "пост-обработка и рендер превью") |
| polished = merged.copy() |
| pipeline_report = run_pipeline(polished, ["cleanup", "quantize", "hands", "pedal", |
| "chords", "report"], {}) |
| save_midi(polished, os.path.join(work, "merged_plus.mid")) |
| subprocess.run([sys.executable, os.path.join(pack_dir, "plus_render.py"), |
| "--midi", os.path.join(work, "merged_plus.mid"), |
| "--out", os.path.join(work, "preview.wav"), "--stop", "12"], check=False) |
|
|
| best = max(result["sources"].items(), key=lambda kv: kv[1]["f1"]) |
| print("\nF1 слияния %.3f | лучший вход %.3f (%s) | согласие %.2f источника" |
| % (result["merged_f1"]["f1"], best[1]["f1"], best[0], |
| result["agreement"]["agreement_mean"]), flush=True) |
| print("файлы: " + work, flush=True) |
| result["after_pipeline_f1"] = compare(truth.notes, polished.notes) |
| result["pipeline"] = polished.stats() |
| result["pipeline_stages"] = {k: v for k, v in pipeline_report["stages"].items() |
| if k in ("cleanup", "quantize", "hands", "pedal", "chords")} |
| with open(os.path.join(work, "report.json"), "w", encoding="utf-8") as fh: |
| json.dump(result, fh, ensure_ascii=False, indent=1, default=str) |
| return result |
|
|
|
|
|
|
| def _hf_token() -> str | None: |
| """Токен из окружения или из Kaggle Secrets, если ячейка запущена в Kaggle.""" |
| token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_TOKEN") |
| if token: |
| return token |
| try: |
| from kaggle_secrets import UserSecretsClient |
| return UserSecretsClient().get_secret("HF_TOKEN") |
| except Exception: |
| return None |
|
|
|
|
| def _files(folder: str, with_audio: bool) -> list[str]: |
| if not os.path.isdir(folder): |
| return [] |
| skip = () if with_audio else (".wav", ".flac", ".mp3") |
| return [os.path.join(folder, n) for n in sorted(os.listdir(folder)) |
| if os.path.isfile(os.path.join(folder, n)) and not n.lower().endswith(skip)] |
|
|
|
|
| def upload_build(folders: dict[str, str], repo: str, stamp: str, |
| with_audio: bool = True, token: str | None = None) -> dict: |
| """Заливает сборку на HF в builds/<метка времени>/<имя папки>. |
| |
| Возвращает словарь с числом файлов, объёмом и ссылками. Ничего не делает, |
| если токена нет: сборка остаётся локально, и это печатается. |
| """ |
| token = token or _hf_token() |
| if not token: |
| print("заливка пропущена: нет HF_TOKEN (в Kaggle это Add-ons > Secrets)", flush=True) |
| return {"uploaded": 0, "reason": "no_token"} |
|
|
| from huggingface_hub import HfApi |
| api = HfApi(token=token) |
| uploaded, total_bytes = 0, 0 |
| links: list[str] = [] |
| for name, folder in folders.items(): |
| chosen = _files(folder, with_audio) |
| if not chosen: |
| continue |
| target = "builds/%s/%s" % (stamp, name) |
| print("[заливка] %s -> %s (%d файлов)" % (folder, target, len(chosen)), flush=True) |
| api.upload_folder(folder_path=folder, path_in_repo=target, repo_id=repo, |
| allow_patterns=[os.path.basename(f) for f in chosen], |
| commit_message="build %s: %s" % (stamp, name)) |
| for path in chosen: |
| uploaded += 1 |
| total_bytes += os.path.getsize(path) |
| links.append("https://huggingface.co/%s/resolve/main/%s/%s" |
| % (repo, target, os.path.basename(path))) |
| if uploaded: |
| print("[заливка] готово: %d файлов, %.1f МБ" % (uploaded, total_bytes / 2**20), flush=True) |
| for link in links[:6]: |
| print(" " + link, flush=True) |
| print(" каталог сборки: https://huggingface.co/%s/tree/main/builds/%s" |
| % (repo, stamp), flush=True) |
| return {"uploaded": uploaded, "bytes": total_bytes, "links": links, |
| "folder": "builds/" + stamp} |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| parser = argparse.ArgumentParser(description="Kaggle PLUS") |
| parser.add_argument("--demo", action="store_true", help="только самопроверка") |
| parser.add_argument("--input-dir", help="папка с MIDI и аудио") |
| parser.add_argument("--out-dir", default=os.path.join(WORK, "plus_out")) |
| parser.add_argument("--stages", default="cleanup,quantize,hands,pedal,dynamics,report") |
| parser.add_argument("--repo", default=REPO, help="репозиторий HF для сборки") |
| parser.add_argument("--no-upload", action="store_true", help="не заливать сборку") |
| parser.add_argument("--no-audio", action="store_true", help="не заливать wav превью") |
| args = parser.parse_args(argv) |
|
|
| ensure_deps() |
| pack_dir = ensure_pack(WORK) |
| report = demo(pack_dir) |
|
|
| if args.input_dir and os.path.isdir(args.input_dir): |
| command = [sys.executable, os.path.join(pack_dir, "plus_batch.py"), |
| "--input-dir", args.input_dir, "--out-dir", args.out_dir, |
| "--stages", args.stages] |
| print("пачка: " + " ".join(command[1:]), flush=True) |
| subprocess.run(command, check=False) |
| report["batch_out"] = args.out_dir |
|
|
| if args.no_upload: |
| print("заливка отключена флагом --no-upload", flush=True) |
| report["upload"] = {"uploaded": 0, "reason": "disabled"} |
| else: |
| stamp = __import__("datetime").datetime.utcnow().strftime("%Y%m%d-%H%M%S") |
| folders = {"demo": os.path.join(WORK, "plus_demo")} |
| if os.path.isdir(args.out_dir): |
| folders["batch"] = args.out_dir |
| report["upload"] = upload_build(folders, args.repo, stamp, |
| with_audio=not args.no_audio) |
|
|
| report_path = os.path.join(WORK, "plus_demo", "report.json") |
| with open(report_path, "w", encoding="utf-8") as fh: |
| json.dump(report, fh, ensure_ascii=False, indent=1, default=str) |
|
|
| |
| |
| uploaded = report.get("upload") or {} |
| if uploaded.get("folder"): |
| token = _hf_token() |
| if token: |
| from huggingface_hub import HfApi |
| HfApi(token=token).upload_file( |
| path_or_fileobj=report_path, |
| path_in_repo=uploaded["folder"] + "/demo/report.json", |
| repo_id=args.repo, commit_message="build report with upload links") |
| print("[заливка] report.json обновлён в сборке", flush=True) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
