From 543bc5dbb02810cd783d6fb43276c64d8fcae1e0 Mon Sep 17 00:00:00 2001 From: Home-Server Date: Sat, 1 Aug 2026 10:23:02 -0400 Subject: [PATCH 1/7] Add DJ style engine (open_mythos/dj) Learn a DJ's mixing personality and apply it to any music library, with a bridge to the OpenMythos RDT for the learned (deep-learning) path. Modules: - harmonic.py Camelot-wheel key parsing + compatibility scoring - profile.py DJStyleProfile (tunable personality) + built-in archetypes - analysis.py Track + library analysis (optional librosa/mutagen; KS key est.) - planner.py MixPlanner -> MixPlan (harmonic + tempo + energy-curve ordering) - render.py render_mix -> beatmatched, crossfaded audio + cue sheet - setlist.py ingest setlists (incl. timestamped tracklists) + learn a profile - enrich.py pluggable BPM/key/energy lookup (CSV/MusicBrainz/GetSongBPM) + cache - mixlang.py serialize mixes to a token string for RDT training - __main__.py CLI: plan / analyze / render / enrich / learn The parent open_mythos/__init__.py is made lazy (PEP 562) so the torch-free DJ engine imports without pulling in the model stack. pyproject gains an [audio] extra. 34 tests (28 torch/audio-free + 6 audio) cover all layers. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NYYoch4CutCodkPevPXHLq --- examples/dj_example.py | 65 +++++++ examples/dj_learn_example.py | 84 +++++++++ open_mythos/__init__.py | 113 +++++++++--- open_mythos/dj/__init__.py | 145 +++++++++++++++ open_mythos/dj/__main__.py | 233 ++++++++++++++++++++++++ open_mythos/dj/analysis.py | 226 +++++++++++++++++++++++ open_mythos/dj/enrich.py | 313 ++++++++++++++++++++++++++++++++ open_mythos/dj/harmonic.py | 137 ++++++++++++++ open_mythos/dj/mixlang.py | 114 ++++++++++++ open_mythos/dj/planner.py | 217 ++++++++++++++++++++++ open_mythos/dj/profile.py | 170 ++++++++++++++++++ open_mythos/dj/render.py | 287 +++++++++++++++++++++++++++++ open_mythos/dj/setlist.py | 337 +++++++++++++++++++++++++++++++++++ pyproject.toml | 13 ++ tests/test_dj.py | 121 +++++++++++++ tests/test_dj_audio.py | 116 ++++++++++++ tests/test_dj_enrich.py | 108 +++++++++++ tests/test_dj_setlist.py | 142 +++++++++++++++ 18 files changed, 2912 insertions(+), 29 deletions(-) create mode 100644 examples/dj_example.py create mode 100644 examples/dj_learn_example.py mode change 100644 => 100755 open_mythos/__init__.py create mode 100644 open_mythos/dj/__init__.py create mode 100644 open_mythos/dj/__main__.py create mode 100644 open_mythos/dj/analysis.py create mode 100644 open_mythos/dj/enrich.py create mode 100644 open_mythos/dj/harmonic.py create mode 100644 open_mythos/dj/mixlang.py create mode 100644 open_mythos/dj/planner.py create mode 100644 open_mythos/dj/profile.py create mode 100644 open_mythos/dj/render.py create mode 100644 open_mythos/dj/setlist.py mode change 100644 => 100755 pyproject.toml create mode 100644 tests/test_dj.py create mode 100644 tests/test_dj_audio.py create mode 100644 tests/test_dj_enrich.py create mode 100644 tests/test_dj_setlist.py diff --git a/examples/dj_example.py b/examples/dj_example.py new file mode 100644 index 00000000..a39f7173 --- /dev/null +++ b/examples/dj_example.py @@ -0,0 +1,65 @@ +""" +OpenMythos DJ engine — end-to-end example. + +Runs with no external dependencies and no audio files: it uses a small +synthetic library of track metadata so you can see the whole pipeline work, +then swap in your own library (a CSV, a folder of audio, or a list of dicts). + + python examples/dj_example.py +""" + +from open_mythos.dj import ( + BUILTIN_PROFILES, + analyze_library, + get_profile, + plan_mix, + plan_to_mixlang, +) + +# A synthetic library: title/artist/bpm/key(Camelot or name)/energy/genre. +LIBRARY = [ + {"title": "Opening Fog", "artist": "Aster", "bpm": 122, "key": "8A", "energy": 0.32, "genre": "melodic techno"}, + {"title": "First Light", "artist": "Vela", "bpm": 123, "key": "8A", "energy": 0.41, "genre": "melodic techno"}, + {"title": "Undertow", "artist": "Koa", "bpm": 124, "key": "9A", "energy": 0.52, "genre": "melodic house"}, + {"title": "Pulse Theory", "artist": "N-Six", "bpm": 124, "key": "9B", "energy": 0.6, "genre": "progressive"}, + {"title": "Glass Arc", "artist": "Mira", "bpm": 125, "key": "10A", "energy": 0.68, "genre": "melodic techno"}, + {"title": "Night Runner", "artist": "Kilo", "bpm": 126, "key": "10A", "energy": 0.75, "genre": "melodic techno"}, + {"title": "Ascend", "artist": "Orbit", "bpm": 126, "key": "11A", "energy": 0.82, "genre": "progressive"}, + {"title": "Peak Signal", "artist": "Volt", "bpm": 128, "key": "11A", "energy": 0.9, "genre": "melodic techno"}, + {"title": "Comedown", "artist": "Haze", "bpm": 122, "key": "7A", "energy": 0.45, "genre": "deep house"}, + {"title": "Afterglow", "artist": "Sol", "bpm": 120, "key": "7A", "energy": 0.3, "genre": "organic house"}, + {"title": "Detour", "artist": "Rue", "bpm": 127, "key": "3A", "energy": 0.7, "genre": "techno"}, + {"title": "Static Bloom", "artist": "Ohm", "bpm": 125, "key": "9A", "energy": 0.64, "genre": "melodic house"}, +] + + +def main() -> None: + library = analyze_library(LIBRARY) + print(f"Analyzed {len(library)} tracks.\n") + + print("Available style profiles:", ", ".join(sorted(BUILTIN_PROFILES)), "\n") + + profile = get_profile("melodic_journey") + plan = plan_mix(library, profile, length=10) + + print("=" * 68) + print(plan.to_text()) + print("=" * 68) + + print("\nTransitions used:", ", ".join(plan.transitions())) + + print("\n--- Same library, 'peak_time_techno' personality ---\n") + plan2 = plan_mix(library, get_profile("peak_time_techno"), length=8) + print(plan2.to_text()) + + print("\n--- OpenMythos bridge: mix serialized as training tokens ---\n") + mixlang = plan_to_mixlang(plan, dj_name="melodic_journey") + print(mixlang[:400] + (" ..." if len(mixlang) > 400 else "")) + print( + "\nFeed a corpus of real setlists in this format to the RDT " + "to learn a DJ's personality (Layer 2)." + ) + + +if __name__ == "__main__": + main() diff --git a/examples/dj_learn_example.py b/examples/dj_learn_example.py new file mode 100644 index 00000000..94096e9f --- /dev/null +++ b/examples/dj_learn_example.py @@ -0,0 +1,84 @@ +""" +OpenMythos DJ engine — learn a DJ's personality from their setlists, then mix +a *new* library in that learned style. + +Runs with no external dependencies. It parses two example setlists (inline +text), learns a DJStyleProfile from them, prints what it inferred, then uses +that learned profile to plan a mix over a separate library. + + python examples/dj_learn_example.py +""" + +from open_mythos.dj import ( + analyze_library, + learn_profile_from_setlists, + plan_mix, + setlist_from_text, + setlists_to_corpus, +) + +# Two real-ish setlists in the forgiving text format: +# Artist - Title | bpm key energy and >> transition +SET_ONE = """ +# Night 1 — a slow harmonic build +Aster - Opening Fog | 121 8A 0.30 +>> long_blend +Vela - First Light | 122 8A 0.42 +>> long_blend +Ohm - Static Bloom | 124 9A 0.58 +>> filter_sweep +Mira - Glass Arc | 125 10A 0.70 +>> long_blend +Volt - Peak Signal | 127 11A 0.88 +""" + +SET_TWO = """ +# Night 2 — same DJ, same shape +Sol - Afterglow | 120 7A 0.33 +>> long_blend +Koa - Undertow | 123 8A 0.50 +>> long_blend +Static - Rise | 125 9A 0.66 +>> filter_sweep +Orbit - Ascend | 126 10A 0.84 +""" + +NEW_LIBRARY = [ + {"title": "Cold Start", "artist": "Nyx", "bpm": 121, "key": "8A", "energy": 0.31, "genre": "melodic techno"}, + {"title": "Drift", "artist": "Lume", "bpm": 122, "key": "8A", "energy": 0.44, "genre": "melodic techno"}, + {"title": "Signal Path", "artist": "Ferro", "bpm": 124, "key": "9A", "energy": 0.6, "genre": "melodic house"}, + {"title": "Overpass", "artist": "Cane", "bpm": 125, "key": "10A", "energy": 0.72, "genre": "progressive"}, + {"title": "Skyline", "artist": "Ivo", "bpm": 126, "key": "11A", "energy": 0.85, "genre": "melodic techno"}, + {"title": "Zenith", "artist": "Rho", "bpm": 128, "key": "11A", "energy": 0.92, "genre": "melodic techno"}, + {"title": "Ebb", "artist": "Tal", "bpm": 121, "key": "7A", "energy": 0.4, "genre": "deep house"}, +] + + +def main() -> None: + setlists = [ + setlist_from_text(SET_ONE, name="night_1"), + setlist_from_text(SET_TWO, name="night_2"), + ] + print(f"Parsed {len(setlists)} setlists " + f"({sum(len(s.tracks) for s in setlists)} tracks total).\n") + + profile = learn_profile_from_setlists(setlists, name="learned_dj") + print("--- Learned personality ---") + print(profile.to_json()) + + print("\n--- Applying the learned style to a NEW library ---\n") + plan = plan_mix(analyze_library(NEW_LIBRARY), profile) + print(plan.to_text()) + + print("\n--- RDT training corpus (mix language) ---") + rows = setlists_to_corpus(setlists) + for r in rows: + print(r[:110] + " ...") + print( + f"\n{len(rows)} rows ready. Write many of these to a .txt corpus and " + "train the OpenMythos RDT to generate mixes in this DJ's style." + ) + + +if __name__ == "__main__": + main() diff --git a/open_mythos/__init__.py b/open_mythos/__init__.py old mode 100644 new mode 100755 index 73c2c046..bc8b6001 --- a/open_mythos/__init__.py +++ b/open_mythos/__init__.py @@ -1,30 +1,87 @@ -from open_mythos.main import ( - ACTHalting, - Expert, - GQAttention, - LoRAAdapter, - LTIInjection, - MLAttention, - MoEFFN, - MythosConfig, - OpenMythos, - RecurrentBlock, - RMSNorm, - TransformerBlock, - apply_rope, - loop_index_embedding, - precompute_rope_freqs, -) -from open_mythos.tokenizer import MythosTokenizer -from open_mythos.variants import ( - mythos_1b, - mythos_1t, - mythos_3b, - mythos_10b, - mythos_50b, - mythos_100b, - mythos_500b, -) +"""OpenMythos package root. + +Heavy, torch-dependent symbols (the model, variants) are imported *lazily* via +PEP 562 ``__getattr__`` so that torch-free subpackages — notably +``open_mythos.dj`` — can be imported and used without pulling in the full model +stack. ``from open_mythos import OpenMythos`` still works exactly as before; the +import just happens on first attribute access. +""" + +from importlib import import_module +from typing import TYPE_CHECKING + +# name -> submodule it lives in. Resolved lazily on first access. +_LAZY_EXPORTS = { + # open_mythos.main + "ACTHalting": "open_mythos.main", + "Expert": "open_mythos.main", + "GQAttention": "open_mythos.main", + "LoRAAdapter": "open_mythos.main", + "LTIInjection": "open_mythos.main", + "MLAttention": "open_mythos.main", + "MoEFFN": "open_mythos.main", + "MythosConfig": "open_mythos.main", + "OpenMythos": "open_mythos.main", + "RecurrentBlock": "open_mythos.main", + "RMSNorm": "open_mythos.main", + "TransformerBlock": "open_mythos.main", + "apply_rope": "open_mythos.main", + "loop_index_embedding": "open_mythos.main", + "precompute_rope_freqs": "open_mythos.main", + # open_mythos.tokenizer + "MythosTokenizer": "open_mythos.tokenizer", + # open_mythos.variants + "mythos_1b": "open_mythos.variants", + "mythos_3b": "open_mythos.variants", + "mythos_10b": "open_mythos.variants", + "mythos_50b": "open_mythos.variants", + "mythos_100b": "open_mythos.variants", + "mythos_500b": "open_mythos.variants", + "mythos_1t": "open_mythos.variants", +} + +if TYPE_CHECKING: # keep static analysis / IDEs happy + from open_mythos.main import ( # noqa: F401 + ACTHalting, + Expert, + GQAttention, + LoRAAdapter, + LTIInjection, + MLAttention, + MoEFFN, + MythosConfig, + OpenMythos, + RecurrentBlock, + RMSNorm, + TransformerBlock, + apply_rope, + loop_index_embedding, + precompute_rope_freqs, + ) + from open_mythos.tokenizer import MythosTokenizer # noqa: F401 + from open_mythos.variants import ( # noqa: F401 + mythos_1b, + mythos_1t, + mythos_3b, + mythos_10b, + mythos_50b, + mythos_100b, + mythos_500b, + ) + + +def __getattr__(name: str): + module = _LAZY_EXPORTS.get(name) + if module is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + value = getattr(import_module(module), name) + globals()[name] = value # cache so subsequent access is direct + return value + + +def __dir__(): + return sorted(list(globals().keys()) + list(_LAZY_EXPORTS.keys())) + __all__ = [ "MythosConfig", @@ -49,7 +106,5 @@ "mythos_100b", "mythos_500b", "mythos_1t", - "load_tokenizer", - "get_vocab_size", "MythosTokenizer", ] diff --git a/open_mythos/dj/__init__.py b/open_mythos/dj/__init__.py new file mode 100644 index 00000000..79ec0055 --- /dev/null +++ b/open_mythos/dj/__init__.py @@ -0,0 +1,145 @@ +""" +OpenMythos DJ engine — learn a DJ's mixing personality and apply it to any +music library. + +Two layers: + +* **Layer 1 (works today, no training):** analyze a library for BPM / key / + energy, then order tracks with harmonic (Camelot) mixing, an energy curve, + and transition annotations, all driven by a :class:`DJStyleProfile`. +* **Layer 2 (the OpenMythos bridge):** serialize mixes into a "mix language" + token string (:mod:`open_mythos.dj.mixlang`) so the Recurrent-Depth + Transformer can be trained on real setlists and *generate* mixes in a learned + personality. + +Quick start:: + + from open_mythos.dj import analyze_library, get_profile, plan_mix + + library = analyze_library("my_tracks.csv") # or a folder / list of dicts + profile = get_profile("melodic_journey") # or author your own + plan = plan_mix(library, profile, length=12) + print(plan.to_text()) +""" + +from open_mythos.dj.analysis import ( + Track, + analyze_file, + analyze_library, + estimate_key, + tracks_from_csv, + tracks_from_dicts, +) +from open_mythos.dj.enrich import ( + CallableProvider, + ChainProvider, + CsvProvider, + DictProvider, + GetSongBpmProvider, + JsonCache, + MetadataProvider, + MusicBrainzAcousticBrainzProvider, + build_provider, + enrich_tracks, +) +from open_mythos.dj.harmonic import ( + CAMELOT_TO_KEY, + compatibility, + compatible_codes, + to_camelot, +) +from open_mythos.dj.mixlang import ( + STRUCTURAL_TOKENS, + build_training_corpus, + encode_with_tokenizer, + plan_to_mixlang, + tracks_to_mixlang, +) +from open_mythos.dj.planner import ( + MixPlan, + MixPlanner, + MixStep, + plan_mix, + target_energy, +) +from open_mythos.dj.profile import ( + BUILTIN_PROFILES, + ENERGY_CURVES, + TRANSITION_TYPES, + DJStyleProfile, + get_profile, +) +from open_mythos.dj.setlist import ( + Setlist, + learn_profile_from_setlists, + setlist_from_json, + setlist_from_text, + setlist_from_tracklist, + setlists_to_corpus, +) + + +def __getattr__(name): + # Lazy access to the audio renderer so importing the DJ engine never + # requires numpy/soundfile/librosa unless you actually render audio. + if name in ("render_mix", "write_playlist", "RenderResult", "CuePoint"): + from open_mythos.dj import render as _render + + return getattr(_render, name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + +__all__ = [ + # analysis + "Track", + "analyze_library", + "analyze_file", + "estimate_key", + "tracks_from_csv", + "tracks_from_dicts", + # harmonic + "to_camelot", + "compatibility", + "compatible_codes", + "CAMELOT_TO_KEY", + # profile + "DJStyleProfile", + "get_profile", + "BUILTIN_PROFILES", + "ENERGY_CURVES", + "TRANSITION_TYPES", + # planner + "MixPlanner", + "MixPlan", + "MixStep", + "plan_mix", + "target_energy", + # mixlang / OpenMythos bridge + "plan_to_mixlang", + "tracks_to_mixlang", + "build_training_corpus", + "encode_with_tokenizer", + "STRUCTURAL_TOKENS", + # setlist ingestion + profile learning + "Setlist", + "setlist_from_text", + "setlist_from_tracklist", + "setlist_from_json", + "learn_profile_from_setlists", + "setlists_to_corpus", + # metadata enrichment + "enrich_tracks", + "build_provider", + "MetadataProvider", + "DictProvider", + "CsvProvider", + "CallableProvider", + "ChainProvider", + "GetSongBpmProvider", + "MusicBrainzAcousticBrainzProvider", + "JsonCache", + # audio rendering (lazy; needs the [audio] extra) + "render_mix", + "write_playlist", + "RenderResult", + "CuePoint", +] diff --git a/open_mythos/dj/__main__.py b/open_mythos/dj/__main__.py new file mode 100644 index 00000000..9efad210 --- /dev/null +++ b/open_mythos/dj/__main__.py @@ -0,0 +1,233 @@ +""" +Command-line interface for the OpenMythos DJ engine. + + # Plan a mix from a library in a given style + python -m open_mythos.dj plan ./my_tracks/ --profile melodic_journey --length 12 + python -m open_mythos.dj plan tracks.csv --profile peak_time_techno --mixlang + + # Analyze a library and dump metadata as CSV + python -m open_mythos.dj analyze ./my_tracks/ --out library.csv + + # Learn a DJ's personality from their setlists, save the profile + corpus + python -m open_mythos.dj learn set1.txt set2.txt --name my_dj \\ + --out my_dj.json --corpus my_dj_corpus.txt + +Audio folders need the optional deps: pip install "open-mythos[audio]" +""" + +from __future__ import annotations + +import argparse +import csv +import sys + +from .analysis import analyze_library +from .planner import plan_mix +from .profile import BUILTIN_PROFILES, DJStyleProfile, get_profile + + +def _load_profile(name: str) -> DJStyleProfile: + if name in BUILTIN_PROFILES: + return get_profile(name) + # Otherwise treat it as a path to a profile JSON. + return DJStyleProfile.from_json(name) + + +def _cmd_plan(args) -> int: + library = analyze_library(args.source) + profile = _load_profile(args.profile) + plan = plan_mix(library, profile, length=args.length) + if args.mixlang: + from .mixlang import plan_to_mixlang + + print(plan_to_mixlang(plan, dj_name=profile.name)) + else: + print(plan.to_text()) + return 0 + + +def _cmd_analyze(args) -> int: + library = analyze_library(args.source) + cols = ["title", "artist", "bpm", "key", "energy", "duration", "genre", "path"] + if args.out: + with open(args.out, "w", newline="", encoding="utf-8") as fh: + w = csv.DictWriter(fh, fieldnames=cols) + w.writeheader() + for t in library: + w.writerow({c: getattr(t, c) for c in cols}) + print(f"Wrote {len(library)} tracks to {args.out}") + else: + for t in library: + print(f"{t.label():40} {t.bpm:6.1f} BPM {t.key_name:5} e{t.energy:.2f}") + return 0 + + +def _cmd_render(args) -> int: + from .render import render_mix, write_playlist + + library = analyze_library(args.source) + profile = _load_profile(args.profile) + plan = plan_mix(library, profile, length=args.length) + + if args.playlist: + write_playlist(plan, args.playlist) + print(f"Wrote playlist -> {args.playlist}") + result = render_mix( + plan, + args.out, + target_bpm=args.bpm, + beatmatch=not args.no_beatmatch, + ) + bpm_note = f" (target {result.target_bpm:.0f} BPM)" if result.target_bpm else "" + print(f"Rendered {result.duration_seconds/60:.1f} min @ {result.sample_rate} Hz" + f"{bpm_note} -> {result.out_path}\n") + print(result.cue_sheet()) + return 0 + + +def _cmd_enrich(args) -> int: + from .enrich import JsonCache, build_provider, enrich_tracks + + library = analyze_library(args.source) + provider = build_provider(args.provider) + cache = JsonCache(args.cache) if args.cache else None + hits = 0 + + def _prog(tr, res): + nonlocal hits + mark = "ok " if res else "-- " + if res: + hits += 1 + print(f" {mark}{tr.label():40} {tr.bpm:6.1f} BPM {tr.key_name:5} e{tr.energy:.2f}") + + enrich_tracks(library, provider, cache=cache, delay=args.delay, on_progress=_prog) + print(f"\nEnriched {hits}/{len(library)} tracks.") + + cols = ["title", "artist", "bpm", "key", "energy", "duration", "genre", "path"] + with open(args.out, "w", newline="", encoding="utf-8") as fh: + w = csv.DictWriter(fh, fieldnames=cols) + w.writeheader() + for t in library: + w.writerow({c: getattr(t, c) for c in cols}) + print(f"Wrote enriched library -> {args.out}") + return 0 + + +def _cmd_learn(args) -> int: + from .setlist import ( + learn_profile_from_setlists, + setlist_from_json, + setlist_from_text, + setlist_from_tracklist, + setlists_to_corpus, + ) + + setlists = [] + for i, path in enumerate(args.setlists): + nm = f"{args.name}_{i + 1}" + if path.lower().endswith(".json"): + setlists.append(setlist_from_json(path, name=nm)) + else: + with open(path, encoding="utf-8") as fh: + text = fh.read() + if args.tracklist: + setlists.append(setlist_from_tracklist( + text, name=nm, title_first=not args.artist_first)) + else: + setlists.append(setlist_from_text(text, name=nm)) + + if args.enrich: + from .enrich import JsonCache, build_provider, enrich_tracks + + provider = build_provider(args.enrich) + cache = JsonCache(args.cache) if args.cache else None + filled = 0 + for sl in setlists: + def _prog(tr, res): + nonlocal filled + if res: + filled += 1 + enrich_tracks(sl.tracks, provider, cache=cache, + delay=args.delay, on_progress=_prog) + print(f"Enriched metadata for {filled} track(s).", file=sys.stderr) + + profile = learn_profile_from_setlists(setlists, name=args.name) + print(profile.to_json()) + + if args.out: + profile.to_json(args.out) + print(f"\nSaved profile -> {args.out}", file=sys.stderr) + if args.corpus: + rows = setlists_to_corpus(setlists, out_path=args.corpus) + print(f"Wrote {len(rows)} training row(s) -> {args.corpus}", file=sys.stderr) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="open_mythos.dj", description=__doc__) + sub = p.add_subparsers(dest="cmd", required=True) + + pp = sub.add_parser("plan", help="plan a mix from a library") + pp.add_argument("source", help="folder of audio, CSV, or JSON list of tracks") + pp.add_argument("--profile", default="melodic_journey", + help="built-in profile name or path to a profile JSON") + pp.add_argument("--length", type=int, default=None, help="number of tracks") + pp.add_argument("--mixlang", action="store_true", + help="emit the mix-language token string instead of text") + pp.set_defaults(func=_cmd_plan) + + pa = sub.add_parser("analyze", help="analyze a library's metadata") + pa.add_argument("source", help="folder of audio, CSV, or JSON list of tracks") + pa.add_argument("--out", help="write metadata to this CSV path") + pa.set_defaults(func=_cmd_analyze) + + pr = sub.add_parser("render", help="render a mix to an audio file") + pr.add_argument("source", help="folder of audio, CSV, or JSON list of tracks") + pr.add_argument("--out", required=True, help="output audio path (.wav)") + pr.add_argument("--profile", default="melodic_journey", + help="built-in profile name or path to a profile JSON") + pr.add_argument("--length", type=int, default=None, help="number of tracks") + pr.add_argument("--bpm", type=float, default=None, + help="beatmatch target BPM (default: median of tracks)") + pr.add_argument("--no-beatmatch", action="store_true", + help="do not time-stretch tracks to a common tempo") + pr.add_argument("--playlist", help="also write an M3U playlist here") + pr.set_defaults(func=_cmd_render) + + pe = sub.add_parser("enrich", help="fill missing bpm/key/energy metadata") + pe.add_argument("source", help="folder of audio, CSV, or JSON list of tracks") + pe.add_argument("--provider", required=True, + help="'musicbrainz' | 'getsongbpm:KEY' | 'csv:PATH' | 'a+b'") + pe.add_argument("--out", required=True, help="write the enriched library CSV here") + pe.add_argument("--cache", help="JSON cache path for lookups") + pe.add_argument("--delay", type=float, default=1.0, + help="seconds between network lookups (default 1.0)") + pe.set_defaults(func=_cmd_enrich) + + pl = sub.add_parser("learn", help="learn a DJ profile from setlists") + pl.add_argument("setlists", nargs="+", help="setlist files (.txt or .json)") + pl.add_argument("--name", required=True, help="name for the learned profile") + pl.add_argument("--out", help="write the learned profile JSON here") + pl.add_argument("--corpus", help="write an RDT training corpus here") + pl.add_argument("--tracklist", action="store_true", + help="parse inputs as pasted tracklists (indexes/timestamps)") + pl.add_argument("--artist-first", action="store_true", + help="with --tracklist: 'A - B' means Artist - Title") + pl.add_argument("--enrich", metavar="SPEC", + help="fill missing bpm/key/energy, e.g. 'musicbrainz', " + "'getsongbpm:KEY', 'csv:known.csv', 'csv:k.csv+musicbrainz'") + pl.add_argument("--cache", help="JSON cache path for enrichment lookups") + pl.add_argument("--delay", type=float, default=1.0, + help="seconds between network lookups (default 1.0)") + pl.set_defaults(func=_cmd_learn) + + return p + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/open_mythos/dj/analysis.py b/open_mythos/dj/analysis.py new file mode 100644 index 00000000..848766c2 --- /dev/null +++ b/open_mythos/dj/analysis.py @@ -0,0 +1,226 @@ +""" +Track representation and library analysis for the OpenMythos DJ engine. + +A :class:`Track` is the atom the planner mixes: it carries the musical metadata +that drives harmonic and tempo decisions (BPM, Camelot key, energy, duration). + +Analysis has three tiers, tried in order and degrading gracefully: + +1. **Provided metadata** — a list of dicts or a CSV with bpm/key/energy columns. +2. **Embedded tags** — read from audio files via ``mutagen`` if installed. +3. **Signal analysis** — estimate BPM/key/energy via ``librosa`` if installed. + +None of these are hard dependencies; the engine is fully usable with tier 1 +alone, which is why the runnable example ships a synthetic library. +""" + +from __future__ import annotations + +import csv +import os +from dataclasses import dataclass, field + +from .harmonic import CAMELOT_TO_KEY, to_camelot + + +@dataclass +class Track: + """A single track with the metadata the planner needs. + + Args: + title: Track title. + artist: Artist name. + bpm: Tempo in beats per minute. + key: Camelot code (normalized on init; e.g. ``"8A"``). + energy: Perceived intensity in ``[0, 1]``. + duration: Length in seconds. + genre: Optional genre tag (soft-matched against a profile). + path: Optional source file path. + """ + + title: str + artist: str = "" + bpm: float = 0.0 + key: str | None = None + energy: float = 0.5 + duration: float = 0.0 + genre: str = "" + path: str | None = None + meta: dict = field(default_factory=dict) + + def __post_init__(self) -> None: + self.key = to_camelot(self.key) + self.energy = float(min(1.0, max(0.0, self.energy))) + self.bpm = float(self.bpm) + + @property + def key_name(self) -> str: + return CAMELOT_TO_KEY.get(self.key or "", self.key or "?") + + def label(self) -> str: + who = f"{self.artist} — " if self.artist else "" + return f"{who}{self.title}" + + +def tracks_from_dicts(rows: list[dict]) -> list[Track]: + """Build tracks from a list of dicts (tier-1 metadata path).""" + out: list[Track] = [] + for r in rows: + out.append( + Track( + title=str(r.get("title", "Untitled")), + artist=str(r.get("artist", "")), + bpm=float(r.get("bpm", 0) or 0), + key=r.get("key"), + energy=float(r.get("energy", 0.5) or 0.5), + duration=float(r.get("duration", 0) or 0), + genre=str(r.get("genre", "")), + path=r.get("path"), + ) + ) + return out + + +def tracks_from_csv(path: str) -> list[Track]: + """Load tracks from a CSV with title/artist/bpm/key/energy/... columns.""" + with open(path, newline="", encoding="utf-8") as fh: + return tracks_from_dicts(list(csv.DictReader(fh))) + + +def _read_tags(path: str) -> dict | None: + """Read bpm/key/genre from embedded tags via mutagen, if available.""" + try: + from mutagen import File as MutagenFile # type: ignore + except Exception: + return None + try: + audio = MutagenFile(path, easy=True) + except Exception: + return None + if audio is None: + return None + + def first(*keys: str) -> str | None: + for k in keys: + v = audio.get(k) + if v: + return v[0] if isinstance(v, list) else str(v) + return None + + return { + "title": first("title") or os.path.splitext(os.path.basename(path))[0], + "artist": first("artist") or "", + "bpm": float(first("bpm") or 0) or 0.0, + "key": first("initialkey", "key"), + "genre": first("genre") or "", + "duration": float(getattr(getattr(audio, "info", None), "length", 0) or 0), + "path": path, + } + + +# Krumhansl-Schmuckler key profiles (major, minor). Correlating a track's mean +# chroma against all 24 rotations of these estimates both tonic and mode. +_KS_MAJOR = [6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88] +_KS_MINOR = [6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17] +_PITCH_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"] + + +def estimate_key(chroma_mean) -> str: + """Krumhansl-Schmuckler key estimate from a 12-bin mean chroma vector. + + Returns a key name like ``"A minor"`` / ``"C major"`` (feed to + :func:`~open_mythos.dj.harmonic.to_camelot`). Pure-python; no hard deps. + """ + def corr(a: list[float], b: list[float]) -> float: + n = len(a) + ma, mb = sum(a) / n, sum(b) / n + num = sum((a[i] - ma) * (b[i] - mb) for i in range(n)) + da = sum((x - ma) ** 2 for x in a) ** 0.5 + db = sum((x - mb) ** 2 for x in b) ** 0.5 + return num / (da * db) if da and db else 0.0 + + chroma = [float(x) for x in chroma_mean] + best = (-2.0, 0, "major") + for tonic in range(12): + rot = chroma[tonic:] + chroma[:tonic] + cmaj, cmin = corr(rot, _KS_MAJOR), corr(rot, _KS_MINOR) + if cmaj > best[0]: + best = (cmaj, tonic, "major") + if cmin > best[0]: + best = (cmin, tonic, "minor") + _, tonic, mode = best + return f"{_PITCH_NAMES[tonic]} {mode}" + + +def _analyze_signal(path: str) -> dict | None: + """Estimate bpm/key/energy from the waveform via librosa, if available.""" + try: + import librosa # type: ignore + import numpy as np # type: ignore + except Exception: + return None + try: + y, sr = librosa.load(path, mono=True) + except Exception: + return None + + # tempo moved to librosa.feature.rhythm.tempo in 0.10; fall back for older. + try: + tempo_fn = librosa.feature.rhythm.tempo + except AttributeError: + tempo_fn = librosa.beat.tempo + tempo = float(np.atleast_1d(tempo_fn(y=y, sr=sr))[0]) + rms = float(np.mean(librosa.feature.rms(y=y))) + # Normalize RMS to a rough 0..1 energy (heuristic; tune per collection). + energy = float(min(1.0, rms * 8.0)) + + chroma = librosa.feature.chroma_cens(y=y, sr=sr).mean(axis=1) + key_guess = estimate_key(chroma) + + return { + "title": os.path.splitext(os.path.basename(path))[0], + "bpm": tempo, + "key": key_guess, + "energy": energy, + "duration": float(librosa.get_duration(y=y, sr=sr)), + "path": path, + } + + +def analyze_file(path: str) -> Track: + """Analyze a single audio file, tags first then signal analysis.""" + data = _read_tags(path) + signal = _analyze_signal(path) + if signal: + # Fill any gaps from tags with the signal estimate. + merged = {**signal, **{k: v for k, v in (data or {}).items() if v}} + return tracks_from_dicts([merged])[0] + if data: + return tracks_from_dicts([data])[0] + return Track(title=os.path.splitext(os.path.basename(path))[0], path=path) + + +def analyze_library(source) -> list[Track]: + """Analyze a music library from any supported source. + + Accepts: + * a list of dicts (tier-1 metadata), + * a ``.csv`` path, + * a directory of audio files (tags/signal analysis), + * a list of audio file paths. + """ + if isinstance(source, list) and source and isinstance(source[0], dict): + return tracks_from_dicts(source) + if isinstance(source, list): + return [analyze_file(p) for p in source] + if isinstance(source, str) and source.lower().endswith(".csv"): + return tracks_from_csv(source) + if isinstance(source, str) and os.path.isdir(source): + exts = (".mp3", ".wav", ".flac", ".aiff", ".aif", ".m4a", ".ogg") + paths = [ + os.path.join(source, f) + for f in sorted(os.listdir(source)) + if f.lower().endswith(exts) + ] + return [analyze_file(p) for p in paths] + raise ValueError(f"unsupported library source: {source!r}") diff --git a/open_mythos/dj/enrich.py b/open_mythos/dj/enrich.py new file mode 100644 index 00000000..4ccef36b --- /dev/null +++ b/open_mythos/dj/enrich.py @@ -0,0 +1,313 @@ +""" +Metadata enrichment — fill in per-track BPM / key / energy for tracks that +arrive without it (e.g. from a title-only tracklist). + +The goal: a pasted tracklist gives track *selection* and *pacing*, but a DJ's +harmonic and tempo signature needs BPM and key per track. This module resolves +each ``Artist – Title`` to those values through a pluggable +:class:`MetadataProvider`, so you can mix free and paid sources or supply your +own: + +* :class:`CsvProvider` — your own ``artist,title,bpm,key,energy`` file (offline, exact). +* :class:`MusicBrainzAcousticBrainzProvider` — free; MusicBrainz lookup → AcousticBrainz features. +* :class:`GetSongBpmProvider` — getsongbpm.com API (needs a free API key); BPM + key. +* :class:`ChainProvider` — try several in order, first hit wins. + +:func:`enrich_tracks` fills only the missing fields, memoized through an +on-disk :class:`JsonCache` so repeated runs don't re-hit the network. + +Network access uses only the Python standard library (``urllib``); every remote +call fails soft (returns ``None``) so enrichment degrades gracefully offline. +""" + +from __future__ import annotations + +import json +import os +import time +import urllib.parse +import urllib.request +from dataclasses import dataclass + +from .analysis import Track +from .harmonic import to_camelot + +# One field-bundle a provider may return; any subset is allowed. +# {"bpm": float, "key": , "energy": float in [0,1]} + + +def _key(artist: str, title: str) -> str: + return f"{(artist or '').strip().lower()}\t{(title or '').strip().lower()}" + + +# -------------------------------------------------------------------------- +# Providers +# -------------------------------------------------------------------------- + +class MetadataProvider: + """Interface: return a dict of {bpm?, key?, energy?} or ``None``.""" + + def lookup(self, artist: str, title: str) -> dict | None: # pragma: no cover + raise NotImplementedError + + +@dataclass +class DictProvider(MetadataProvider): + """In-memory provider from a ``{(artist,title): {...}}`` style map. Testable.""" + + table: dict + + def lookup(self, artist: str, title: str) -> dict | None: + return self.table.get(_key(artist, title)) + + +class CsvProvider(MetadataProvider): + """User-supplied ``artist,title,bpm,key,energy`` CSV (offline, editable).""" + + def __init__(self, path: str): + import csv + + self.table: dict = {} + with open(path, newline="", encoding="utf-8") as fh: + for r in csv.DictReader(fh): + out = {} + if r.get("bpm"): + out["bpm"] = float(r["bpm"]) + if r.get("key"): + out["key"] = r["key"] + if r.get("energy"): + out["energy"] = float(r["energy"]) + self.table[_key(r.get("artist", ""), r.get("title", ""))] = out + + def lookup(self, artist: str, title: str) -> dict | None: + return self.table.get(_key(artist, title)) + + +class CallableProvider(MetadataProvider): + """Wrap any ``fn(artist, title) -> dict | None`` as a provider.""" + + def __init__(self, fn): + self.fn = fn + + def lookup(self, artist: str, title: str) -> dict | None: + return self.fn(artist, title) + + +class ChainProvider(MetadataProvider): + """Try providers in order; merge so earlier providers win per field.""" + + def __init__(self, providers: list[MetadataProvider]): + self.providers = providers + + def lookup(self, artist: str, title: str) -> dict | None: + merged: dict = {} + for p in self.providers: + try: + res = p.lookup(artist, title) + except Exception: + res = None + if res: + for k, v in res.items(): + merged.setdefault(k, v) + return merged or None + + +class _HttpProvider(MetadataProvider): + """Base with an overridable JSON GET (override in tests to avoid network).""" + + user_agent = "OpenMythos-DJ/0.1 (https://github.com/The-Swarm-Corporation/OpenMythos)" + timeout = 8.0 + + def _get_json(self, url: str, headers: dict | None = None): + req = urllib.request.Request(url, headers={"User-Agent": self.user_agent, + **(headers or {})}) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + +class GetSongBpmProvider(_HttpProvider): + """getsongbpm.com — free API key required. Returns BPM and key.""" + + def __init__(self, api_key: str): + self.api_key = api_key + + def lookup(self, artist: str, title: str) -> dict | None: + lookup = urllib.parse.quote(f"song:{title} artist:{artist}") + url = (f"https://api.getsong.co/search/?api_key={self.api_key}" + f"&type=both&lookup={lookup}") + try: + data = self._get_json(url) + except Exception: + return None + results = data.get("search") + if not results or isinstance(results, dict): # {"error": ...} + return None + top = results[0] + out: dict = {} + if top.get("tempo"): + try: + out["bpm"] = float(top["tempo"]) + except (TypeError, ValueError): + pass + key_of = top.get("key_of") + if key_of: + out["key"] = key_of + return out or None + + +class MusicBrainzAcousticBrainzProvider(_HttpProvider): + """Free: MusicBrainz recording search -> AcousticBrainz audio features. + + MusicBrainz asks for <=1 request/second and a descriptive User-Agent; the + ``delay`` is honored between lookups by :func:`enrich_tracks`. + """ + + def lookup(self, artist: str, title: str) -> dict | None: + q = urllib.parse.quote(f'artist:"{artist}" AND recording:"{title}"') + mb_url = f"https://musicbrainz.org/ws/2/recording/?query={q}&fmt=json&limit=1" + try: + mb = self._get_json(mb_url) + recs = mb.get("recordings") or [] + if not recs: + return None + mbid = recs[0]["id"] + except Exception: + return None + + out: dict = {} + try: + hl = self._get_json(f"https://acousticbrainz.org/api/v1/{mbid}/high-level") + rh = self._get_json(f"https://acousticbrainz.org/api/v1/{mbid}/low-level") + except Exception: + hl, rh = {}, {} + + rhythm = (rh.get("rhythm") or {}).get("bpm") + if rhythm: + out["bpm"] = float(rhythm) + tonal = rh.get("tonal") or {} + if tonal.get("key_key"): + scale = tonal.get("key_scale", "major") + out["key"] = f"{tonal['key_key']} {scale}" + # Rough energy proxy from danceability + loudness where present. + dance = ((hl.get("highlevel") or {}).get("danceability") or {}).get("all", {}) + loud = (rh.get("lowlevel") or {}).get("average_loudness") + if dance or loud is not None: + d = float(dance.get("danceable", 0.5)) if dance else 0.5 + l = float(loud) if loud is not None else 0.5 + out["energy"] = max(0.0, min(1.0, 0.5 * d + 0.5 * l)) + return out or None + + +# -------------------------------------------------------------------------- +# Cache + enrichment +# -------------------------------------------------------------------------- + +class JsonCache: + """Simple on-disk memo of provider results, keyed by artist+title.""" + + def __init__(self, path: str): + self.path = path + self.data: dict = {} + if path and os.path.exists(path): + try: + with open(path, encoding="utf-8") as fh: + self.data = json.load(fh) + except Exception: + self.data = {} + + def get(self, artist: str, title: str): + return self.data.get(_key(artist, title)) + + def put(self, artist: str, title: str, value) -> None: + self.data[_key(artist, title)] = value + + def save(self) -> None: + if not self.path: + return + with open(self.path, "w", encoding="utf-8") as fh: + json.dump(self.data, fh, indent=2) + + +def enrich_tracks( + tracks: list[Track], + provider: MetadataProvider, + only_missing: bool = True, + cache: JsonCache | None = None, + delay: float = 0.0, + on_progress=None, +) -> list[Track]: + """Fill missing BPM / key / energy on ``tracks`` in place via ``provider``. + + Args: + tracks: Tracks to enrich (mutated and returned). + provider: Where to look up metadata. + only_missing: If ``True``, never overwrite values a track already has. + cache: Optional :class:`JsonCache`; skips network on repeat lookups. + delay: Seconds to sleep between *network* lookups (rate-limit politeness). + on_progress: Optional ``callable(track, result)`` for logging. + + Returns: + The same ``tracks`` list, enriched. + """ + for tr in tracks: + needs = ( + (tr.bpm or 0) <= 0 + or tr.key is None + or (only_missing is False) + ) + if only_missing and not needs: + if on_progress: + on_progress(tr, None) + continue + + cached = cache.get(tr.artist, tr.title) if cache else None + if cached is not None: + res = cached + else: + res = provider.lookup(tr.artist, tr.title) or {} + if cache is not None: + cache.put(tr.artist, tr.title, res) + if delay: + time.sleep(delay) + + if res: + if res.get("bpm") and (not only_missing or (tr.bpm or 0) <= 0): + tr.bpm = float(res["bpm"]) + if res.get("key") and (not only_missing or tr.key is None): + tr.key = to_camelot(res["key"]) or tr.key + if res.get("energy") is not None and ( + not only_missing or tr.energy == 0.5 + ): + tr.energy = float(min(1.0, max(0.0, res["energy"]))) + if on_progress: + on_progress(tr, res or None) + + if cache is not None: + cache.save() + return tracks + + +def build_provider(spec: str) -> MetadataProvider: + """Build a provider from a CLI spec string. + + ``"musicbrainz"`` | ``"getsongbpm:API_KEY"`` | ``"csv:PATH"`` | + ``"a+b"`` for a chain (e.g. ``"csv:known.csv+musicbrainz"``). + """ + parts = [s.strip() for s in spec.split("+") if s.strip()] + providers: list[MetadataProvider] = [] + for part in parts: + name, _, arg = part.partition(":") + name = name.lower() + if name == "musicbrainz": + providers.append(MusicBrainzAcousticBrainzProvider()) + elif name == "getsongbpm": + if not arg: + raise ValueError("getsongbpm needs an API key: getsongbpm:KEY") + providers.append(GetSongBpmProvider(arg)) + elif name == "csv": + providers.append(CsvProvider(arg)) + else: + raise ValueError(f"unknown provider {name!r}") + if not providers: + raise ValueError(f"no provider in spec {spec!r}") + return providers[0] if len(providers) == 1 else ChainProvider(providers) diff --git a/open_mythos/dj/harmonic.py b/open_mythos/dj/harmonic.py new file mode 100644 index 00000000..530b438c --- /dev/null +++ b/open_mythos/dj/harmonic.py @@ -0,0 +1,137 @@ +""" +Harmonic mixing utilities for the OpenMythos DJ engine. + +Implements the Camelot wheel used by DJs for harmonic mixing. Every musical +key maps to a Camelot code such as ``8A`` (A minor) or ``8B`` (C major). Two +tracks mix harmonically when their codes are *compatible*: identical, adjacent +on the wheel (+/- 1 with the same letter), or the relative major/minor switch +(same number, different letter). + +The functions here are deliberately dependency-free so the engine runs anywhere. +""" + +from __future__ import annotations + +# Camelot code -> canonical key name (for display). +CAMELOT_TO_KEY: dict[str, str] = { + "1A": "Abm", "1B": "B", + "2A": "Ebm", "2B": "F#", + "3A": "Bbm", "3B": "Db", + "4A": "Fm", "4B": "Ab", + "5A": "Cm", "5B": "Eb", + "6A": "Gm", "6B": "Bb", + "7A": "Dm", "7B": "F", + "8A": "Am", "8B": "C", + "9A": "Em", "9B": "G", + "10A": "Bm", "10B": "D", + "11A": "F#m", "11B": "A", + "12A": "Dbm", "12B": "E", +} + +# Reverse + alias map: many spellings of the same key -> Camelot code. +_KEY_TO_CAMELOT: dict[str, str] = {} + + +def _register(code: str, *names: str) -> None: + for n in names: + _KEY_TO_CAMELOT[n.lower()] = code + + +# Minor keys (A) and their enharmonic spellings. +_register("1A", "abm", "g#m", "ab minor", "g# minor") +_register("2A", "ebm", "d#m", "eb minor", "d# minor") +_register("3A", "bbm", "a#m", "bb minor", "a# minor") +_register("4A", "fm", "f minor") +_register("5A", "cm", "c minor") +_register("6A", "gm", "g minor") +_register("7A", "dm", "d minor") +_register("8A", "am", "a minor") +_register("9A", "em", "e minor") +_register("10A", "bm", "b minor") +_register("11A", "f#m", "gbm", "f# minor", "gb minor") +_register("12A", "dbm", "c#m", "db minor", "c# minor") +# Major keys (B). +_register("1B", "b", "b major") +_register("2B", "f#", "gb", "f# major", "gb major") +_register("3B", "db", "c#", "db major", "c# major") +_register("4B", "ab", "g#", "ab major", "g# major") +_register("5B", "eb", "d#", "eb major", "d# major") +_register("6B", "bb", "a#", "bb major", "a# major") +_register("7B", "f", "f major") +_register("8B", "c", "c major") +_register("9B", "g", "g major") +_register("10B", "d", "d major") +_register("11B", "a", "a major") +_register("12B", "e", "e major") + + +def to_camelot(key: str | None) -> str | None: + """Normalize a key string to its Camelot code, or ``None`` if unknown. + + Accepts Camelot codes directly (``"8A"``, ``"8a"``) and common key + spellings (``"Am"``, ``"A minor"``, ``"C"``, ``"F#m"``). + """ + if not key: + return None + k = key.strip() + upper = k.upper() + if upper in CAMELOT_TO_KEY: + return upper + return _KEY_TO_CAMELOT.get(k.lower()) + + +def _parse(code: str) -> tuple[int, str] | None: + if not code or len(code) < 2: + return None + letter = code[-1].upper() + if letter not in ("A", "B"): + return None + try: + number = int(code[:-1]) + except ValueError: + return None + if not 1 <= number <= 12: + return None + return number, letter + + +def compatibility(a: str | None, b: str | None) -> float: + """Score harmonic compatibility of two keys in ``[0.0, 1.0]``. + + * ``1.0`` same key (perfect blend / energy hold) + * ``0.9`` relative major/minor (same number, flipped letter) + * ``0.85`` adjacent on the wheel (+/- 1, same letter) + * ``0.5`` two steps away (mixable with care) + * ``0.2`` everything else (clash) + * ``0.5`` if either key is unknown (neutral, don't over-penalize) + """ + ca, cb = to_camelot(a), to_camelot(b) + if ca is None or cb is None: + return 0.5 + pa, pb = _parse(ca), _parse(cb) + if pa is None or pb is None: + return 0.5 + (na, la), (nb, lb) = pa, pb + if na == nb and la == lb: + return 1.0 + if na == nb and la != lb: + return 0.9 + # Circular distance around the 12-hour wheel. + dist = min((na - nb) % 12, (nb - na) % 12) + if la == lb and dist == 1: + return 0.85 + if dist == 2: + return 0.5 + return 0.2 + + +def compatible_codes(code: str) -> list[str]: + """Return the Camelot codes that mix cleanly with ``code``.""" + p = _parse(to_camelot(code) or "") + if p is None: + return [] + n, letter = p + other = "B" if letter == "A" else "A" + up = (n % 12) + 1 + down = ((n - 2) % 12) + 1 + return [f"{n}{letter}", f"{n}{other}", f"{up}{letter}", f"{down}{letter}"] diff --git a/open_mythos/dj/mixlang.py b/open_mythos/dj/mixlang.py new file mode 100644 index 00000000..c259658c --- /dev/null +++ b/open_mythos/dj/mixlang.py @@ -0,0 +1,114 @@ +""" +The "mix language" — the bridge from DJ sets to the OpenMythos RDT. + +This is what makes the DJ engine genuinely *part of* OpenMythos: a DJ set is a +sequence, and OpenMythos generates sequences. Here we serialize a +:class:`~open_mythos.dj.planner.MixPlan` (or any list of tracks + transitions) +into a compact, regular token string that the :class:`MythosTokenizer` can +encode and the Recurrent-Depth Transformer can be trained on. + +Layer 1 (the rule-based planner) produces these strings from your library. +Feed a corpus of **real DJ setlists** rendered in this same language to the RDT +and it learns to *generate* mixes in that DJ's personality — Layer 2. At +inference you prime the model with a `` header plus your available +tracks and let it sequence them. + +Grammar (whitespace-separated tokens):: + + + + + ... + +""" + +from __future__ import annotations + +import re + +from .analysis import Track +from .planner import MixPlan, MixStep + +_SLUG = re.compile(r"[^a-z0-9]+") + + +def _slug(text: str) -> str: + """Collapse free text to a single lowercase token (keeps the vocab small).""" + s = _SLUG.sub("_", (text or "").strip().lower()).strip("_") + return s or "unknown" + + +def track_to_tokens(track: Track) -> str: + return ( + f" " + f" " + ) + + +def plan_to_mixlang(plan: MixPlan, dj_name: str | None = None) -> str: + """Serialize a :class:`MixPlan` into a mix-language string.""" + curve_hint = "" + parts = [ + "", + f"", + ] + for step in plan.steps: + if step.transition_in: + parts.append(f"") + parts.append(track_to_tokens(step.track)) + parts.append("") + return " ".join(parts) + + +def tracks_to_mixlang( + tracks: list[Track], transitions: list[str] | None, dj_name: str +) -> str: + """Serialize a raw track list (+ optional transitions) into mix language. + + Useful for turning **real, human-DJ setlists** into training rows without + going through the planner. + """ + parts = ["", f""] + transitions = transitions or [] + for i, tr in enumerate(tracks): + if i > 0 and i - 1 < len(transitions): + parts.append(f"") + parts.append(track_to_tokens(tr)) + parts.append("") + return " ".join(parts) + + +# The special tokens worth adding to the tokenizer as atomic units so the RDT +# treats structure as first-class rather than sub-word noise. +STRUCTURAL_TOKENS = ["", "", ""] + + +def build_training_corpus(plans_or_sets: list, dj_name: str | None = None) -> list[str]: + """Render many mixes into mix-language rows for RDT training. + + Each element may be a :class:`MixPlan` or a ``(tracks, transitions, name)`` + tuple (a human setlist). Returns one string per mix. + """ + rows: list[str] = [] + for item in plans_or_sets: + if isinstance(item, MixPlan): + rows.append(plan_to_mixlang(item, dj_name)) + elif isinstance(item, tuple) and len(item) == 3: + tracks, transitions, name = item + rows.append(tracks_to_mixlang(tracks, transitions, name)) + else: + raise TypeError(f"cannot serialize corpus item: {item!r}") + return rows + + +def encode_with_tokenizer(mixlang: str, tokenizer=None) -> list[int]: + """Encode a mix-language string with a :class:`MythosTokenizer`. + + Imported lazily so the rest of the DJ engine has no torch/transformers + dependency. Pass your own tokenizer to avoid re-loading it. + """ + if tokenizer is None: + from open_mythos.tokenizer import MythosTokenizer + + tokenizer = MythosTokenizer() + return tokenizer.encode(mixlang) diff --git a/open_mythos/dj/planner.py b/open_mythos/dj/planner.py new file mode 100644 index 00000000..3797008f --- /dev/null +++ b/open_mythos/dj/planner.py @@ -0,0 +1,217 @@ +""" +The mix planner: the heart of the OpenMythos DJ engine. + +Given a library of :class:`~open_mythos.dj.analysis.Track` and a +:class:`~open_mythos.dj.profile.DJStyleProfile`, the planner produces a +:class:`MixPlan` — an ordered set with a chosen transition and cue point +between each pair of tracks, following the DJ's tempo lane, harmonic +strictness, and energy arc. + +The algorithm is greedy with look-ahead scoring (fast, deterministic, and easy +to reason about). Each candidate next-track is scored on four axes weighted by +the profile: harmonic compatibility, BPM proximity, distance from the target +energy for that point in the set, and genre fit. This is Layer 1 — a strong +rule-based baseline that also produces the training data for Layer 2 (a learned +personality via the OpenMythos RDT; see :mod:`open_mythos.dj.mixlang`). +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +from .analysis import Track +from .harmonic import compatibility +from .profile import DJStyleProfile + + +def target_energy(curve: str, t: float) -> float: + """Target energy in ``[0, 1]`` at set-progress ``t`` for a given curve.""" + t = min(1.0, max(0.0, t)) + if curve == "build": + return 0.35 + 0.6 * t + if curve == "cooldown": + return 0.85 - 0.6 * t + if curve == "wave": + # Two gentle peaks across the set. + return 0.55 + 0.35 * math.sin(2 * math.pi * t - math.pi / 2) + if curve == "peak_time": + # Fast rise, sustained plateau, small dip at the very end. + return min(1.0, 0.5 + 1.1 * t) if t < 0.5 else max(0.7, 1.05 - 0.35 * t) + return 0.6 # flat + + +def _bpm_score(a: float, b: float, drift: int) -> float: + if a <= 0 or b <= 0: + return 0.5 + diff = abs(a - b) + if diff <= drift: + return 1.0 - (diff / (drift + 1e-9)) * 0.3 # small penalty within lane + # Allow half/double-time matching (e.g. 128 <-> 64/256). + for factor in (2.0, 0.5): + if abs(a - b * factor) <= drift: + return 0.6 + return max(0.0, 0.7 - (diff - drift) / 40.0) + + +def _genre_score(track: Track, profile: DJStyleProfile) -> float: + if not profile.genres: + return 0.5 + g = (track.genre or "").lower() + if not g: + return 0.5 + return 1.0 if any(pg.lower() in g or g in pg.lower() for pg in profile.genres) else 0.3 + + +def _choose_transition(prev: Track, nxt: Track, profile: DJStyleProfile) -> str: + """Pick a transition consistent with the profile and the track pair.""" + favored = profile.favored_transitions or ["long_blend"] + energy_jump = nxt.energy - prev.energy + bpm_jump = abs(nxt.bpm - prev.bpm) if prev.bpm and nxt.bpm else 0 + # Big upward energy move -> prefer a punchy move if the DJ has one. + if energy_jump > 0.25: + for t in ("quick_cut", "double_drop", "loop_roll"): + if t in favored: + return t + # Big tempo gap -> prefer an echo/filter escape if available. + if bpm_jump > profile.bpm_drift: + for t in ("echo_out", "filter_sweep"): + if t in favored: + return t + return favored[0] + + +@dataclass +class MixStep: + """One track in the plan plus how it was reached.""" + + track: Track + transition_in: str | None # transition used to arrive here (None for opener) + cue_seconds: float # where to start mixing this track out (approx) + score: float # planner score for this placement + target_energy: float # the curve's target at this position + + def describe(self, index: int) -> str: + head = f"{index:>2}. {self.track.label()}" + meta = f"[{self.track.bpm:.0f} BPM · {self.track.key_name} · e{self.track.energy:.2f}]" + if self.transition_in: + return f"{head} {meta}\n ↳ via {self.transition_in} (cue ~{self.cue_seconds:.0f}s)" + return f"{head} {meta} (opener)" + + +@dataclass +class MixPlan: + """An ordered DJ set produced by the planner.""" + + profile_name: str + steps: list[MixStep] = field(default_factory=list) + + @property + def tracks(self) -> list[Track]: + return [s.track for s in self.steps] + + @property + def total_seconds(self) -> float: + return sum(s.track.duration for s in self.steps) + + def transitions(self) -> list[str]: + return [s.transition_in for s in self.steps if s.transition_in] + + def to_text(self) -> str: + lines = [f"Mix plan — style: {self.profile_name} · {len(self.steps)} tracks"] + if self.total_seconds: + mins = self.total_seconds / 60.0 + lines.append(f"Approx runtime: {mins:.0f} min") + lines.append("") + for i, step in enumerate(self.steps, 1): + lines.append(step.describe(i)) + return "\n".join(lines) + + +class MixPlanner: + """Greedy, style-driven mix planner. + + Args: + profile: The DJ personality to mix in. + """ + + def __init__(self, profile: DJStyleProfile): + self.profile = profile + self.weights = profile.score_weights() + + def _candidate_score( + self, prev: Track, cand: Track, t: float + ) -> float: + w = self.weights + harmonic = compatibility(prev.key, cand.key) + bpm = _bpm_score(prev.bpm, cand.bpm, self.profile.bpm_drift) + tgt = target_energy(self.profile.energy_curve, t) + energy = 1.0 - abs(cand.energy - tgt) + genre = _genre_score(cand, self.profile) + return ( + w["harmonic"] * harmonic + + w["bpm"] * bpm + + w["energy"] * energy + + w["genre"] * genre + ) + + def _pick_opener(self, pool: list[Track]) -> Track: + """Open near the curve's starting energy and inside the tempo lane.""" + tgt = target_energy(self.profile.energy_curve, 0.0) + lo, hi = self.profile.bpm_range + + def opener_score(tr: Track) -> float: + # Soft tempo-lane preference so the energy target (curve start) + # still dominates when few tracks sit inside the lane. + in_lane = 1.0 if lo <= tr.bpm <= hi or tr.bpm == 0 else 0.7 + return in_lane * (1.0 - abs(tr.energy - tgt)) + + return max(pool, key=opener_score) + + def plan(self, tracks: list[Track], length: int | None = None) -> MixPlan: + """Order ``tracks`` into a :class:`MixPlan`. + + Args: + tracks: The available library. + length: How many tracks to include (defaults to all). + """ + pool = list(tracks) + if not pool: + return MixPlan(profile_name=self.profile.name) + n = min(length or len(pool), len(pool)) + + opener = self._pick_opener(pool) + pool.remove(opener) + steps = [ + MixStep( + track=opener, + transition_in=None, + cue_seconds=opener.duration or self.profile.avg_track_seconds, + score=0.0, + target_energy=target_energy(self.profile.energy_curve, 0.0), + ) + ] + + for i in range(1, n): + t = i / max(1, n - 1) + prev = steps[-1].track + best = max(pool, key=lambda c: self._candidate_score(prev, c, t)) + pool.remove(best) + hold = best.duration or self.profile.avg_track_seconds + steps.append( + MixStep( + track=best, + transition_in=_choose_transition(prev, best, self.profile), + cue_seconds=min(hold, self.profile.avg_track_seconds), + score=self._candidate_score(prev, best, t), + target_energy=target_energy(self.profile.energy_curve, t), + ) + ) + return MixPlan(profile_name=self.profile.name, steps=steps) + + +def plan_mix( + tracks: list[Track], profile: DJStyleProfile, length: int | None = None +) -> MixPlan: + """Convenience wrapper: build a :class:`MixPlan` in one call.""" + return MixPlanner(profile).plan(tracks, length=length) diff --git a/open_mythos/dj/profile.py b/open_mythos/dj/profile.py new file mode 100644 index 00000000..cfd35f25 --- /dev/null +++ b/open_mythos/dj/profile.py @@ -0,0 +1,170 @@ +""" +DJ style profiles for the OpenMythos DJ engine. + +A :class:`DJStyleProfile` captures a DJ's *personality* as structured, tunable +knowledge: which genres and tempos they live in, how much they respect harmonic +mixing, how long they ride a track, the shape of their energy arc across a set, +and their favored transition moves. + +Profiles can be authored by hand (from setlists / interviews / your own ear), +loaded from JSON, or eventually *learned* from real setlists via the OpenMythos +RDT (see :mod:`open_mythos.dj.mixlang`). +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field + +# Recognized energy-curve shapes. Each maps to a function of set-progress +# ``t in [0, 1]`` -> target energy in ``[0, 1]`` (see planner). +ENERGY_CURVES = ("build", "wave", "peak_time", "cooldown", "flat") + +# Transition vocabulary the planner can annotate. +TRANSITION_TYPES = ( + "long_blend", # 32-64 bar EQ blend, keeps both grooves alive + "quick_cut", # slam on the 1, high-energy + "echo_out", # delay/reverb tail out of the outgoing track + "loop_roll", # loop the incoming intro, roll into the drop + "bassline_swap", # swap lows on the phrase boundary + "filter_sweep", # high-pass sweep transition + "double_drop", # align two drops (advanced, risky) +) + + +@dataclass +class DJStyleProfile: + """Structured description of a DJ's mixing personality. + + Args: + name: Human-readable profile name. + genres: Genre lanes the DJ favors (used for soft genre matching). + bpm_range: (min, max) tempo the DJ operates in. + bpm_drift: Max BPM jump tolerated across a single transition. + energy_curve: One of :data:`ENERGY_CURVES` describing the set arc. + harmonic_strictness: 0..1 weight on Camelot compatibility. + avg_track_seconds: Typical time a track is held before mixing out. + favored_transitions: Preferred entries from :data:`TRANSITION_TYPES`. + signature_moves: Free-text signature tricks (for notes / prompts). + weights: Optional score-weight overrides for the planner. + """ + + name: str + genres: list[str] = field(default_factory=list) + bpm_range: tuple[int, int] = (120, 130) + bpm_drift: int = 6 + energy_curve: str = "build" + harmonic_strictness: float = 0.7 + avg_track_seconds: float = 210.0 + favored_transitions: list[str] = field(default_factory=lambda: ["long_blend"]) + signature_moves: list[str] = field(default_factory=list) + weights: dict[str, float] = field(default_factory=dict) + + def __post_init__(self) -> None: + if self.energy_curve not in ENERGY_CURVES: + raise ValueError( + f"energy_curve {self.energy_curve!r} not in {ENERGY_CURVES}" + ) + bad = [t for t in self.favored_transitions if t not in TRANSITION_TYPES] + if bad: + raise ValueError(f"unknown transitions {bad}; pick from {TRANSITION_TYPES}") + self.harmonic_strictness = float(min(1.0, max(0.0, self.harmonic_strictness))) + + # -- score weights ---------------------------------------------------- + def score_weights(self) -> dict[str, float]: + """Planner weights, blending defaults with per-profile overrides.""" + base = { + "harmonic": 1.0 * self.harmonic_strictness + 0.15, + "bpm": 1.0, + "energy": 0.9, + "genre": 0.5, + } + base.update(self.weights) + return base + + # -- serialization ---------------------------------------------------- + def to_json(self, path: str | None = None, indent: int = 2) -> str: + data = asdict(self) + data["bpm_range"] = list(self.bpm_range) + text = json.dumps(data, indent=indent) + if path: + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return text + + @classmethod + def from_json(cls, path_or_text: str) -> "DJStyleProfile": + text = path_or_text + if path_or_text.strip()[:1] not in "{[": + with open(path_or_text, encoding="utf-8") as fh: + text = fh.read() + data = json.loads(text) + if "bpm_range" in data: + data["bpm_range"] = tuple(data["bpm_range"]) + return cls(**data) + + +# -------------------------------------------------------------------------- +# Built-in archetype profiles. These are illustrative starting points, not +# claims about any specific real person. Clone one and tune it, or author your +# own from a DJ's actual setlists. +# -------------------------------------------------------------------------- + +def _profiles() -> dict[str, DJStyleProfile]: + return { + "peak_time_techno": DJStyleProfile( + name="peak_time_techno", + genres=["techno", "peak time", "driving techno"], + bpm_range=(128, 138), + bpm_drift=4, + energy_curve="build", + harmonic_strictness=0.55, + avg_track_seconds=180.0, + favored_transitions=["long_blend", "bassline_swap", "quick_cut"], + signature_moves=["relentless 16-bar EQ blends", "tool-loop bridges"], + ), + "melodic_journey": DJStyleProfile( + name="melodic_journey", + genres=["melodic techno", "melodic house", "progressive"], + bpm_range=(120, 126), + bpm_drift=3, + energy_curve="wave", + harmonic_strictness=0.95, + avg_track_seconds=300.0, + favored_transitions=["long_blend", "filter_sweep", "echo_out"], + signature_moves=["breakdown-to-breakdown key-locked blends"], + ), + "open_format": DJStyleProfile( + name="open_format", + genres=["hip hop", "house", "pop", "disco", "r&b"], + bpm_range=(95, 128), + bpm_drift=20, + energy_curve="peak_time", + harmonic_strictness=0.3, + avg_track_seconds=95.0, + favored_transitions=["quick_cut", "echo_out", "loop_roll"], + signature_moves=["double-time cuts", "acapella-over-instrumental"], + ), + "sunset_deep": DJStyleProfile( + name="sunset_deep", + genres=["deep house", "organic house", "downtempo"], + bpm_range=(110, 122), + bpm_drift=3, + energy_curve="cooldown", + harmonic_strictness=0.85, + avg_track_seconds=300.0, + favored_transitions=["long_blend", "filter_sweep"], + signature_moves=["long atmospheric intros", "never rushes the blend"], + ), + } + + +BUILTIN_PROFILES: dict[str, DJStyleProfile] = _profiles() + + +def get_profile(name: str) -> DJStyleProfile: + """Fetch a built-in profile by name (raises ``KeyError`` if unknown).""" + profiles = _profiles() + if name not in profiles: + raise KeyError(f"{name!r} not found; available: {sorted(profiles)}") + return profiles[name] diff --git a/open_mythos/dj/render.py b/open_mythos/dj/render.py new file mode 100644 index 00000000..4d708074 --- /dev/null +++ b/open_mythos/dj/render.py @@ -0,0 +1,287 @@ +""" +Audio rendering — turn a :class:`~open_mythos.dj.planner.MixPlan` into a single +continuous mixed audio file. + +This is the step that produces *actual sound*: it loads each track's audio, +optionally beatmatches (time-stretches every track to a common target BPM), +and crossfades between them using a curve chosen per transition type from the +plan. The result is one file plus a cue sheet of drop-in timestamps. + +Dependencies are optional and imported lazily: + +* ``numpy`` — required for any rendering (the DSP). +* ``soundfile`` — preferred audio I/O (falls back to the stdlib ``wave`` + module for 16-bit PCM mono output). +* ``librosa`` — needed only for beatmatching (time-stretch) and for decoding + compressed formats / resampling. + +Install everything with: ``pip install "open-mythos[audio]"``. + +The renderer is intentionally engine-agnostic. For a NLE-style handoff instead +of a rendered file, :func:`write_playlist` emits an M3U + cue sheet you can load +into Mixxx / rekordbox and perform live. +""" + +from __future__ import annotations + +import os +import wave +from dataclasses import dataclass, field + +from .planner import MixPlan + +# Crossfade length (seconds) and curve per transition type. "cut" => hard cut. +_TRANSITION_XFADE = { + "long_blend": (16.0, "equal_power"), + "quick_cut": (0.05, "equal_power"), + "echo_out": (6.0, "linear"), + "loop_roll": (2.0, "equal_power"), + "bassline_swap": (8.0, "equal_power"), + "filter_sweep": (10.0, "filter"), + "double_drop": (4.0, "equal_power"), +} + + +def _require_numpy(): + try: + import numpy as np # type: ignore + + return np + except Exception as exc: # pragma: no cover - env dependent + raise RuntimeError( + "rendering needs numpy. Install audio deps: pip install 'open-mythos[audio]'" + ) from exc + + +def _load_audio(path: str, sr: int): + """Load ``path`` as mono float32 at ``sr``. Prefers soundfile, then librosa.""" + np = _require_numpy() + try: + import soundfile as sf # type: ignore + + data, file_sr = sf.read(path, dtype="float32", always_2d=True) + y = data.mean(axis=1) # downmix to mono + if file_sr != sr: + y = _resample(y, file_sr, sr) + return y.astype(np.float32) + except Exception: + pass + try: + import librosa # type: ignore + + y, _ = librosa.load(path, sr=sr, mono=True) + return y.astype(np.float32) + except Exception as exc: + raise RuntimeError(f"could not load audio: {path} ({exc})") from exc + + +def _resample(y, src_sr: int, dst_sr: int): + if src_sr == dst_sr: + return y + try: + import librosa # type: ignore + + return librosa.resample(y, orig_sr=src_sr, target_sr=dst_sr) + except Exception: + np = _require_numpy() + # Linear-interpolation fallback (fine for our purposes). + n_dst = int(round(len(y) * dst_sr / src_sr)) + x_old = np.linspace(0.0, 1.0, num=len(y), endpoint=False) + x_new = np.linspace(0.0, 1.0, num=n_dst, endpoint=False) + return np.interp(x_new, x_old, y).astype(y.dtype) + + +def _time_stretch(y, rate: float): + """Time-stretch by ``rate`` (>1 = faster/shorter). No-op if librosa absent.""" + if abs(rate - 1.0) < 1e-3: + return y + try: + import librosa # type: ignore + + return librosa.effects.time_stretch(y, rate=rate) + except Exception: + return y # beatmatching unavailable -> leave tempo as-is + + +def _fade_pair(np, n: int, curve: str): + """Return (fade_out, fade_in) envelopes of length ``n`` for a crossfade.""" + t = np.linspace(0.0, 1.0, num=max(1, n), dtype="float32") + if curve == "equal_power": + return np.cos(t * np.pi / 2), np.sin(t * np.pi / 2) + if curve == "filter": + # Equal-power gain plus a lowpass sweep applied to the outgoing tail + # (handled by the caller); here just the gain envelopes. + return np.cos(t * np.pi / 2), np.sin(t * np.pi / 2) + return 1.0 - t, t # linear + + +def _lowpass_sweep(np, y, start_a: float = 1.0, end_a: float = 0.06): + """One-pole lowpass with a cutoff that closes across the buffer (sweep).""" + n = len(y) + if n == 0: + return y + a = np.linspace(start_a, end_a, num=n, dtype="float32") + out = np.empty(n, dtype="float32") + prev = 0.0 + for i in range(n): + prev = prev + a[i] * (y[i] - prev) + out[i] = prev + return out + + +@dataclass +class CuePoint: + index: int + label: str + start_seconds: float + transition_in: str | None + + +@dataclass +class RenderResult: + out_path: str + sample_rate: int + duration_seconds: float + target_bpm: float | None + cues: list[CuePoint] = field(default_factory=list) + + def cue_sheet(self) -> str: + lines = [f"# Mix cue sheet — {self.duration_seconds/60:.1f} min @ {self.sample_rate} Hz"] + for c in self.cues: + mm, ss = divmod(int(c.start_seconds), 60) + via = f" (via {c.transition_in})" if c.transition_in else " (opener)" + lines.append(f"{mm:02d}:{ss:02d} {c.index:>2}. {c.label}{via}") + return "\n".join(lines) + + +def _write_wav(path: str, y, sr: int) -> None: + """Write mono float array to a 16-bit PCM WAV (soundfile if available).""" + np = _require_numpy() + y = np.clip(y, -1.0, 1.0) + try: + import soundfile as sf # type: ignore + + sf.write(path, y, sr, subtype="PCM_16") + return + except Exception: + pass + pcm = (y * 32767.0).astype(" RenderResult: + """Render ``plan`` to a single mixed audio file. + + Args: + plan: The mix to render. Each step's ``track.path`` must point at audio + (or provide ``resolve_path``). + out_path: Destination file (``.wav``; other formats need soundfile). + sr: Working/output sample rate. + target_bpm: Common tempo to beatmatch to. Defaults to the median track + BPM in the plan. + beatmatch: Time-stretch each track to ``target_bpm`` (needs librosa). + max_segment_seconds: Cap how long each track plays before mixing out. + Defaults to each step's planned ``cue_seconds``. + resolve_path: Optional ``callable(track) -> path`` if tracks lack paths. + + Returns: + A :class:`RenderResult` with duration and cue points. + """ + np = _require_numpy() + steps = plan.steps + if not steps: + raise ValueError("empty plan") + + bpms = [s.track.bpm for s in steps if s.track.bpm > 0] + if target_bpm is None and bpms: + target_bpm = float(sorted(bpms)[len(bpms) // 2]) # median + + def path_for(track): + if resolve_path: + return resolve_path(track) + if not track.path: + raise ValueError(f"track {track.label()!r} has no audio path") + return track.path + + segments: list = [] + for step in steps: + y = _load_audio(path_for(step.track), sr) + if beatmatch and target_bpm and step.track.bpm > 0: + y = _time_stretch(y, rate=step.track.bpm / target_bpm) + cap = max_segment_seconds or step.cue_seconds or (len(y) / sr) + y = y[: int(cap * sr)] + segments.append(y) + + # Assemble with per-transition crossfades on a running output buffer. + out = segments[0].astype(np.float32).copy() + cues = [CuePoint(1, steps[0].track.label(), 0.0, None)] + + for i in range(1, len(segments)): + trans = steps[i].transition_in or "long_blend" + xfade_s, curve = _TRANSITION_XFADE.get(trans, (16.0, "equal_power")) + nx = int(xfade_s * sr) + seg = segments[i].astype(np.float32) + # Never let a crossfade swallow a whole track: cap it to half the + # incoming segment (and the available outgoing tail). + nx = min(nx, len(out), len(seg) // 2 or len(seg)) + + start = len(out) - nx # incoming track overlaps the last nx samples + cues.append(CuePoint(i + 1, steps[i].track.label(), start / sr, trans)) + + if nx <= 0: + out = np.concatenate([out, seg]) + continue + + fade_out, fade_in = _fade_pair(np, nx, curve) + tail = out[-nx:] * fade_out + if curve == "filter": + tail = _lowpass_sweep(np, tail) + head = seg[:nx] * fade_in + out = np.concatenate([out[:-nx], tail + head, seg[nx:]]) + + peak = float(np.max(np.abs(out))) or 1.0 + if peak > 1.0: + out = out / peak + + os.makedirs(os.path.dirname(os.path.abspath(out_path)) or ".", exist_ok=True) + _write_wav(out_path, out, sr) + + return RenderResult( + out_path=out_path, + sample_rate=sr, + duration_seconds=len(out) / sr, + target_bpm=target_bpm, + cues=cues, + ) + + +def write_playlist(plan: MixPlan, out_path: str, resolve_path=None) -> str: + """Write an M3U playlist + cue comments for handoff to Mixxx / rekordbox. + + A live-performance alternative to :func:`render_mix`: load this into your DJ + software and perform the transitions yourself, using the plan as the guide. + """ + lines = ["#EXTM3U", f"# OpenMythos mix — style: {plan.profile_name}"] + for i, step in enumerate(plan.steps, 1): + track = step.track + path = resolve_path(track) if resolve_path else (track.path or "") + via = f" | {step.transition_in}" if step.transition_in else " | opener" + secs = int(track.duration) + lines.append(f"#EXTINF:{secs},{track.label()}{via}") + lines.append(path) + text = "\n".join(lines) + "\n" + with open(out_path, "w", encoding="utf-8") as fh: + fh.write(text) + return out_path diff --git a/open_mythos/dj/setlist.py b/open_mythos/dj/setlist.py new file mode 100644 index 00000000..a586ee95 --- /dev/null +++ b/open_mythos/dj/setlist.py @@ -0,0 +1,337 @@ +""" +Setlist ingestion and profile learning — the "learn a real DJ" path. + +Two jobs: + +1. **Ingest** real setlists (a DJ's actual mixes) from JSON or plain text into + :class:`Setlist` objects. +2. **Learn** a :class:`~open_mythos.dj.profile.DJStyleProfile` from those + setlists *statistically* — deriving the tempo lane, tempo drift, energy-curve + shape, harmonic strictness, track hold time, and favored transitions from + what the DJ actually did. This is a rule-based (no-training) estimate of the + DJ's personality that the planner can immediately mix in. + +The same setlists also serialize to a mix-language training corpus (see +:func:`setlists_to_corpus`) for the OpenMythos RDT — the deep-learning path to +the *same* goal. + +Text setlist format (forgiving):: + + # comments and blank lines ignored + Artist - Title + Artist - Title | 128 8A 0.72 # optional: bpm key energy + >> quick_cut # optional transition into the next track + Another Artist - Another Title | 126 9A 0.8 +""" + +from __future__ import annotations + +import json +import re +import statistics +from dataclasses import dataclass, field + +from .analysis import Track, tracks_from_dicts +from .harmonic import compatibility +from .planner import target_energy +from .profile import ENERGY_CURVES, TRANSITION_TYPES, DJStyleProfile + + +@dataclass +class Setlist: + """One real mix: an ordered track list plus the transitions between them.""" + + name: str + tracks: list[Track] = field(default_factory=list) + transitions: list[str] = field(default_factory=list) # len == len(tracks)-1 + + def as_tuple(self) -> tuple[list[Track], list[str], str]: + return self.tracks, self.transitions, self.name + + +# -------------------------------------------------------------------------- +# Parsing +# -------------------------------------------------------------------------- + +def _parse_inline_meta(rest: str) -> dict: + """Parse ``"128 8A 0.72"`` -> {bpm, key, energy} (any subset, any order).""" + out: dict = {} + for tok in rest.replace(",", " ").split(): + t = tok.strip() + if not t: + continue + # energy: a float in [0,1] + try: + f = float(t) + if 0.0 <= f <= 1.0 and "." in t: + out["energy"] = f + continue + if f > 1.0: # treat as bpm + out["bpm"] = f + continue + except ValueError: + pass + out["key"] = t # otherwise assume it's a key/Camelot code + return out + + +def setlist_from_text(text: str, name: str) -> Setlist: + """Parse a plain-text setlist (see module docstring for the format).""" + tracks: list[Track] = [] + transitions: list[str] = [] + pending_transition: str | None = None + + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + if line.startswith(">>"): + pending_transition = line[2:].strip() or "long_blend" + continue + + meta: dict = {} + body = line + if "|" in line: + body, rest = line.split("|", 1) + meta = _parse_inline_meta(rest) + body = body.strip() + + artist, title = "", body + if " - " in body: + artist, title = body.split(" - ", 1) + row = {"artist": artist.strip(), "title": title.strip(), **meta} + tracks.append(tracks_from_dicts([row])[0]) + if len(tracks) > 1: + transitions.append(pending_transition or "long_blend") + pending_transition = None + + return Setlist(name=name, tracks=tracks, transitions=transitions) + + +# Leading index like "1." / "12)" / "01 -". +_INDEX_RE = re.compile(r"^\s*\d{1,3}\s*[\.\)\-]?\s+") +# A timestamp anywhere: [h:]mm:ss (also matches bare mm:ss). +_TS_RE = re.compile(r"\b(?:(\d{1,2}):)?(\d{1,2}):(\d{2})\b") + + +def _ts_to_seconds(m: re.Match) -> float: + h = int(m.group(1) or 0) + return h * 3600 + int(m.group(2)) * 60 + int(m.group(3)) + + +def setlist_from_tracklist( + text: str, name: str, title_first: bool = True +) -> Setlist: + """Parse a copy-pasted tracklist (e.g. from a mix description / 1001tracklists). + + Handles the messy real-world shape: leading index numbers, a timestamp + anywhere on the line, remix/edit tags, and an ``A - B`` split. Timestamps + are used to derive each track's **duration** (hold time) from the gap to the + next track — real pacing signal even when BPM/key are absent. + + Args: + text: The pasted tracklist, one track per line. + name: Name for the resulting setlist. + title_first: If ``True`` (common on tracklist sites), ``A - B`` means + ``Title - Artist``; if ``False`` it means ``Artist - Title``. + """ + entries: list[tuple[dict, float | None]] = [] + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + + # Extract a timestamp (start position) FIRST and remove it — otherwise + # a trailing "| 00:00:12" gets mistaken for a bpm/key meta block. + ts = None + m = _TS_RE.search(line) + if m: + ts = _ts_to_seconds(m) + line = (line[: m.start()] + line[m.end():]).strip(" \t-–|") + + # Pull an optional trailing "| bpm key energy" block. + meta: dict = {} + if "|" in line: + head, rest = line.rsplit("|", 1) + if _parse_inline_meta(rest): # only if it looks like real meta + meta = _parse_inline_meta(rest) + line = head.strip() + + line = _INDEX_RE.sub("", line).strip() + if not line: + continue + + if " - " in line: + a, b = line.split(" - ", 1) + title, artist = (a, b) if title_first else (b, a) + else: + title, artist = line, "" + + row = {"artist": artist.strip(), "title": title.strip(), **meta} + entries.append((row, ts)) + + # Derive durations from consecutive timestamps. + rows = [] + for i, (row, ts) in enumerate(entries): + if ts is not None and i + 1 < len(entries) and entries[i + 1][1] is not None: + dur = entries[i + 1][1] - ts + if dur > 0: + row["duration"] = dur + rows.append(row) + + tracks = tracks_from_dicts(rows) + transitions = ["long_blend"] * (len(tracks) - 1) if len(tracks) > 1 else [] + return Setlist(name=name, tracks=tracks, transitions=transitions) + + +def setlist_from_json(path_or_text: str, name: str | None = None) -> Setlist: + """Parse a JSON setlist. + + Accepts either a bare list of track dicts, or an object + ``{"name": ..., "tracks": [...], "transitions": [...]}``. + """ + text = path_or_text + if path_or_text.strip()[:1] not in "[{": + with open(path_or_text, encoding="utf-8") as fh: + text = fh.read() + data = json.loads(text) + if isinstance(data, list): + rows, transitions, nm = data, [], name or "unknown" + else: + rows = data.get("tracks", []) + transitions = data.get("transitions", []) + nm = name or data.get("name", "unknown") + tracks = tracks_from_dicts(rows) + if not transitions and len(tracks) > 1: + transitions = ["long_blend"] * (len(tracks) - 1) + return Setlist(name=nm, tracks=tracks, transitions=transitions) + + +# -------------------------------------------------------------------------- +# Profile learning +# -------------------------------------------------------------------------- + +def _fit_energy_curve(setlists: list[Setlist]) -> str: + """Pick the built-in curve whose shape best matches observed energy arcs.""" + trajectories = [] + for sl in setlists: + es = [t.energy for t in sl.tracks] + if len(es) >= 3: + trajectories.append(es) + if not trajectories: + return "build" + + best_curve, best_err = "build", float("inf") + for curve in ENERGY_CURVES: + if curve == "flat": + continue + err = 0.0 + for es in trajectories: + n = len(es) + for i, e in enumerate(es): + t = i / (n - 1) + err += (e - target_energy(curve, t)) ** 2 + if err < best_err: + best_curve, best_err = curve, err + return best_curve + + +def _percentile(values: list[float], p: float) -> float: + if not values: + return 0.0 + s = sorted(values) + k = (len(s) - 1) * p + lo, hi = int(k), min(int(k) + 1, len(s) - 1) + return s[lo] + (s[hi] - s[lo]) * (k - lo) + + +def learn_profile_from_setlists( + setlists: list[Setlist], name: str +) -> DJStyleProfile: + """Derive a :class:`DJStyleProfile` from a DJ's real setlists. + + Estimates, from what actually appears in the mixes: + + * ``bpm_range`` — 10th/90th percentile of track tempos + * ``bpm_drift`` — median absolute tempo change between adjacent tracks + * ``energy_curve`` — best-fitting built-in arc shape + * ``harmonic_strictness`` — fraction of adjacent pairs that mix harmonically + * ``avg_track_seconds`` — mean track duration (if present) + * ``favored_transitions`` — most frequent annotated transitions + * ``genres`` — most common genre tags + """ + all_tracks = [t for sl in setlists for t in sl.tracks] + if not all_tracks: + raise ValueError("no tracks in provided setlists") + + bpms = [t.bpm for t in all_tracks if t.bpm > 0] + if bpms: + lo = int(round(_percentile(bpms, 0.1))) + hi = int(round(_percentile(bpms, 0.9))) + bpm_range = (lo, hi if hi > lo else lo + 4) + else: + bpm_range = (120, 130) + + # Adjacent-pair statistics. + drifts, harmonic_hits, harmonic_total = [], 0, 0 + for sl in setlists: + for a, b in zip(sl.tracks, sl.tracks[1:]): + if a.bpm > 0 and b.bpm > 0: + drifts.append(abs(a.bpm - b.bpm)) + if a.key and b.key: + harmonic_total += 1 + if compatibility(a.key, b.key) >= 0.8: + harmonic_hits += 1 + bpm_drift = max(2, int(round(statistics.median(drifts)))) if drifts else 6 + harmonic_strictness = ( + harmonic_hits / harmonic_total if harmonic_total else 0.6 + ) + + durations = [t.duration for t in all_tracks if t.duration > 0] + avg_track_seconds = float(statistics.mean(durations)) if durations else 210.0 + + # Favored transitions from annotations; fall back to a sensible default. + counts: dict[str, int] = {} + for sl in setlists: + for tr in sl.transitions: + if tr in TRANSITION_TYPES: + counts[tr] = counts.get(tr, 0) + 1 + favored = [t for t, _ in sorted(counts.items(), key=lambda kv: -kv[1])][:3] + if not favored: + favored = ["long_blend"] + + genre_counts: dict[str, int] = {} + for t in all_tracks: + if t.genre: + genre_counts[t.genre] = genre_counts.get(t.genre, 0) + 1 + genres = [g for g, _ in sorted(genre_counts.items(), key=lambda kv: -kv[1])][:4] + + return DJStyleProfile( + name=name, + genres=genres, + bpm_range=bpm_range, + bpm_drift=bpm_drift, + energy_curve=_fit_energy_curve(setlists), + harmonic_strictness=round(harmonic_strictness, 3), + avg_track_seconds=round(avg_track_seconds, 1), + favored_transitions=favored, + signature_moves=[f"learned from {len(setlists)} setlist(s)"], + ) + + +# -------------------------------------------------------------------------- +# Training corpus for the OpenMythos RDT +# -------------------------------------------------------------------------- + +def setlists_to_corpus(setlists: list[Setlist], out_path: str | None = None) -> list[str]: + """Render setlists to newline-delimited mix-language rows for RDT training. + + Returns the rows; also writes them to ``out_path`` if given (one mix/line). + """ + from .mixlang import build_training_corpus + + rows = build_training_corpus([sl.as_tuple() for sl in setlists]) + if out_path: + with open(out_path, "w", encoding="utf-8") as fh: + fh.write("\n".join(rows) + "\n") + return rows diff --git a/pyproject.toml b/pyproject.toml old mode 100644 new mode 100755 index 8129e904..c14a79b6 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,11 +44,24 @@ datasets = ">=2.18.0" [tool.poetry.extras] flash = ["flash-attn"] +audio = ["librosa", "mutagen", "soundfile"] [tool.poetry.dependencies.flash-attn] version = ">=2.8.3" optional = true +[tool.poetry.dependencies.librosa] +version = ">=0.10.0" +optional = true + +[tool.poetry.dependencies.mutagen] +version = ">=1.47.0" +optional = true + +[tool.poetry.dependencies.soundfile] +version = ">=0.12.0" +optional = true + [tool.poetry.group.lint.dependencies] black = ">=23.1,<27.0" diff --git a/tests/test_dj.py b/tests/test_dj.py new file mode 100644 index 00000000..38e0b007 --- /dev/null +++ b/tests/test_dj.py @@ -0,0 +1,121 @@ +"""Tests for the OpenMythos DJ engine (torch-free Layer 1 + mix language).""" + +from open_mythos.dj import ( + DJStyleProfile, + Track, + analyze_library, + compatibility, + get_profile, + plan_mix, + plan_to_mixlang, + target_energy, + to_camelot, +) + +LIBRARY = [ + {"title": "A", "bpm": 122, "key": "8A", "energy": 0.3, "genre": "melodic techno"}, + {"title": "B", "bpm": 123, "key": "8A", "energy": 0.45, "genre": "melodic techno"}, + {"title": "C", "bpm": 124, "key": "9A", "energy": 0.6, "genre": "melodic house"}, + {"title": "D", "bpm": 125, "key": "10A", "energy": 0.75, "genre": "progressive"}, + {"title": "E", "bpm": 126, "key": "11A", "energy": 0.9, "genre": "melodic techno"}, +] + + +# -- harmonic --------------------------------------------------------------- + +def test_key_normalization(): + assert to_camelot("Am") == "8A" + assert to_camelot("A minor") == "8A" + assert to_camelot("8a") == "8A" + assert to_camelot("C") == "8B" + assert to_camelot("nonsense") is None + + +def test_compatibility_ordering(): + assert compatibility("8A", "8A") == 1.0 # same key + assert compatibility("8A", "8B") == 0.9 # relative major/minor + assert compatibility("8A", "9A") == 0.85 # adjacent + assert compatibility("8A", "2A") < 0.5 # clash + assert compatibility("8A", None) == 0.5 # unknown -> neutral + + +# -- profile ---------------------------------------------------------------- + +def test_builtin_profiles_load(): + p = get_profile("melodic_journey") + assert p.harmonic_strictness > 0.5 + assert "long_blend" in p.favored_transitions + + +def test_profile_json_roundtrip(): + p = get_profile("peak_time_techno") + p2 = DJStyleProfile.from_json(p.to_json()) + assert p2.name == p.name + assert p2.bpm_range == p.bpm_range + assert p2.energy_curve == p.energy_curve + + +def test_invalid_energy_curve_rejected(): + import pytest + + with pytest.raises(ValueError): + DJStyleProfile(name="bad", energy_curve="turbo") + + +# -- analysis --------------------------------------------------------------- + +def test_analyze_dicts(): + lib = analyze_library(LIBRARY) + assert len(lib) == 5 + assert all(isinstance(t, Track) for t in lib) + assert lib[0].key == "8A" + + +# -- planner ---------------------------------------------------------------- + +def test_target_energy_bounds(): + for curve in ("build", "wave", "peak_time", "cooldown", "flat"): + for t in (0.0, 0.25, 0.5, 0.75, 1.0): + e = target_energy(curve, t) + assert 0.0 <= e <= 1.0 + + +def test_plan_uses_all_and_is_ordered(): + lib = analyze_library(LIBRARY) + plan = plan_mix(lib, get_profile("melodic_journey")) + assert len(plan.steps) == len(lib) + # No track used twice. + titles = [s.track.title for s in plan.steps] + assert len(set(titles)) == len(titles) + # Opener has no incoming transition; the rest do. + assert plan.steps[0].transition_in is None + assert all(s.transition_in for s in plan.steps[1:]) + + +def test_build_curve_trends_upward(): + lib = analyze_library(LIBRARY) + plan = plan_mix(lib, get_profile("melodic_journey")) + energies = [s.track.energy for s in plan.steps] + # First half should on average be lower-energy than the second half. + mid = len(energies) // 2 + assert sum(energies[:mid]) / mid <= sum(energies[mid:]) / (len(energies) - mid) + + +def test_length_limit(): + lib = analyze_library(LIBRARY) + plan = plan_mix(lib, get_profile("melodic_journey"), length=3) + assert len(plan.steps) == 3 + + +# -- mix language / bridge -------------------------------------------------- + +def test_mixlang_structure(): + lib = analyze_library(LIBRARY) + plan = plan_mix(lib, get_profile("melodic_journey")) + s = plan_to_mixlang(plan, dj_name="melodic_journey") + assert s.startswith("") + assert s.endswith("") + assert s.count("") == len(plan.steps) + assert "" in s + # One transition token per non-opener track. + assert s.count(" None: + """Write a WAV: a click train at ``bpm`` plus sustained tones at roots_hz.""" + n = int(seconds * SR) + t = np.arange(n) / SR + y = np.zeros(n, dtype=np.float32) + for f in roots_hz: # a chord bed for key detection + y += 0.2 * np.sin(2 * np.pi * f * t) + # Percussive clicks on the beat for tempo detection. + period = 60.0 / bpm + for k in range(int(seconds / period) + 1): + i = int(k * period * SR) + if i < n: + env = np.exp(-np.arange(min(400, n - i)) / 60.0).astype(np.float32) + y[i : i + len(env)] += env + y = y / (np.max(np.abs(y)) or 1.0) + sf.write(path, y, SR) + + +def test_estimate_key_from_synth_chord(tmp_path): + # C major triad (C4 E4 G4) -> tonic should read as C (Camelot 8B / rel 8A). + p = str(tmp_path / "cmaj.wav") + _synth_wav(p, seconds=4.0, bpm=120, roots_hz=[261.63, 329.63, 392.00]) + tr = analyze_file(p) + assert tr.key is not None + assert to_camelot(tr.key) is not None + # Tonic family: C major and its relative A minor both start with '8'. + assert to_camelot(tr.key)[:-1] in {"8", "5", "3"} # C / Eb / Db neighbourhood + + +def test_analyze_file_bpm_and_duration(tmp_path): + p = str(tmp_path / "beat.wav") + _synth_wav(p, seconds=6.0, bpm=120, roots_hz=[220.0]) + tr = analyze_file(p) + assert 100 <= tr.bpm <= 140 # librosa tempo estimate near 120 (allow octave) + assert abs(tr.duration - 6.0) < 0.5 + assert 0.0 <= tr.energy <= 1.0 + + +def _plan_from_paths(paths, bpms): + steps = [] + for i, (path, bpm) in enumerate(zip(paths, bpms)): + steps.append( + MixStep( + track=Track(title=f"t{i}", bpm=bpm, key="8A", energy=0.5, + duration=4.0, path=path), + transition_in=None if i == 0 else "long_blend", + cue_seconds=4.0, + score=0.0, + target_energy=0.5, + ) + ) + return MixPlan(profile_name="test", steps=steps) + + +def test_render_mix_produces_audio(tmp_path): + paths = [] + for i, bpm in enumerate((120, 124, 128)): + p = str(tmp_path / f"trk{i}.wav") + _synth_wav(p, seconds=4.0, bpm=bpm, roots_hz=[220.0 + 20 * i]) + paths.append(p) + + plan = _plan_from_paths(paths, (120, 124, 128)) + out = str(tmp_path / "mix.wav") + result = render_mix(plan, out, sr=SR, target_bpm=124, beatmatch=True) + + assert os.path.exists(out) + assert result.duration_seconds > 4.0 # more than a single track + assert len(result.cues) == 3 + assert result.cues[0].transition_in is None + assert result.cues[1].transition_in == "long_blend" + # Rendered file is readable and non-silent. + y, file_sr = sf.read(out) + assert file_sr == SR + assert float(np.max(np.abs(y))) > 0.0 + + +def test_render_without_beatmatch(tmp_path): + paths = [] + for i in range(2): + p = str(tmp_path / f"nb{i}.wav") + _synth_wav(p, seconds=3.0, bpm=120, roots_hz=[220.0]) + paths.append(p) + plan = _plan_from_paths(paths, (120, 120)) + out = str(tmp_path / "nb.wav") + result = render_mix(plan, out, sr=SR, beatmatch=False) + assert os.path.exists(out) and result.duration_seconds > 3.0 + + +def test_estimate_key_pure_vectors(): + # Deterministic, dependency-light sanity on the key estimator. + cmaj = [10, 0, 4, 0, 6, 5, 0, 7, 0, 4, 0, 3] + assert to_camelot(estimate_key(cmaj)) == to_camelot("C") diff --git a/tests/test_dj_enrich.py b/tests/test_dj_enrich.py new file mode 100644 index 00000000..4ae05048 --- /dev/null +++ b/tests/test_dj_enrich.py @@ -0,0 +1,108 @@ +"""Tests for metadata enrichment. Fully offline: no network is ever touched.""" + +from open_mythos.dj import Track +from open_mythos.dj.enrich import ( + ChainProvider, + DictProvider, + GetSongBpmProvider, + JsonCache, + MusicBrainzAcousticBrainzProvider, + build_provider, + enrich_tracks, +) +from open_mythos.dj.enrich import _key + + +def test_dict_provider_and_fill_missing(): + table = { + _key("Deetron", "Runnin'"): {"bpm": 124, "key": "Am", "energy": 0.7}, + } + tracks = [Track(title="Runnin'", artist="Deetron")] # no bpm/key + enrich_tracks(tracks, DictProvider(table)) + assert tracks[0].bpm == 124 + assert tracks[0].key == "8A" # normalized to Camelot + assert abs(tracks[0].energy - 0.7) < 1e-6 + + +def test_only_missing_does_not_overwrite(): + table = {_key("A", "B"): {"bpm": 130, "key": "9A", "energy": 0.9}} + tracks = [Track(title="B", artist="A", bpm=122, key="8A", energy=0.4)] + enrich_tracks(tracks, DictProvider(table), only_missing=True) + # Existing values are preserved. + assert tracks[0].bpm == 122 + assert tracks[0].key == "8A" + assert abs(tracks[0].energy - 0.4) < 1e-6 + + +def test_chain_provider_first_hit_wins_per_field(): + a = DictProvider({_key("A", "B"): {"bpm": 120}}) + b = DictProvider({_key("A", "B"): {"bpm": 130, "key": "8A"}}) + merged = ChainProvider([a, b]).lookup("A", "B") + assert merged["bpm"] == 120 # earlier provider wins + assert merged["key"] == "8A" # filled from later provider + + +def test_cache_prevents_second_lookup(tmp_path): + calls = {"n": 0} + + class Counting(DictProvider): + def lookup(self, artist, title): + calls["n"] += 1 + return super().lookup(artist, title) + + prov = Counting({_key("A", "B"): {"bpm": 128}}) + cache = JsonCache(str(tmp_path / "c.json")) + enrich_tracks([Track(title="B", artist="A")], prov, cache=cache) + # New cache, new tracks -> one lookup, then persisted. + assert calls["n"] == 1 + cache2 = JsonCache(str(tmp_path / "c.json")) + enrich_tracks([Track(title="B", artist="A")], prov, cache=cache2) + assert calls["n"] == 1 # served from cache, no new lookup + + +def test_getsongbpm_response_parsing(monkeypatch): + prov = GetSongBpmProvider("FAKEKEY") + monkeypatch.setattr(prov, "_get_json", lambda url, headers=None: { + "search": [{"tempo": "126", "key_of": "F#m", "title": "x"}] + }) + res = prov.lookup("Some", "Track") + assert res == {"bpm": 126.0, "key": "F#m"} + + +def test_getsongbpm_handles_error_payload(monkeypatch): + prov = GetSongBpmProvider("FAKEKEY") + monkeypatch.setattr(prov, "_get_json", + lambda url, headers=None: {"search": {"error": "no results"}}) + assert prov.lookup("x", "y") is None + + +def test_musicbrainz_acousticbrainz_parsing(monkeypatch): + prov = MusicBrainzAcousticBrainzProvider() + payloads = { + "mb": {"recordings": [{"id": "mbid-123"}]}, + "ll": {"rhythm": {"bpm": 128.0}, + "tonal": {"key_key": "A", "key_scale": "minor"}, + "lowlevel": {"average_loudness": 0.8}}, + "hl": {"highlevel": {"danceability": {"all": {"danceable": 0.9}}}}, + } + + def fake_get(url, headers=None): + if "musicbrainz.org" in url: + return payloads["mb"] + if url.endswith("/high-level"): + return payloads["hl"] + return payloads["ll"] + + monkeypatch.setattr(prov, "_get_json", fake_get) + res = prov.lookup("Artist", "Title") + assert res["bpm"] == 128.0 + assert res["key"] == "A minor" + assert 0.0 <= res["energy"] <= 1.0 + + +def test_build_provider_specs(): + assert isinstance(build_provider("getsongbpm:KEY"), GetSongBpmProvider) + assert isinstance(build_provider("musicbrainz"), + MusicBrainzAcousticBrainzProvider) + chain = build_provider("getsongbpm:KEY+musicbrainz") + assert isinstance(chain, ChainProvider) diff --git a/tests/test_dj_setlist.py b/tests/test_dj_setlist.py new file mode 100644 index 00000000..20b4b8fa --- /dev/null +++ b/tests/test_dj_setlist.py @@ -0,0 +1,142 @@ +"""Tests for setlist ingestion, profile learning, key detection, and CLI.""" + +from open_mythos.dj import ( + estimate_key, + learn_profile_from_setlists, + setlist_from_json, + setlist_from_text, + setlist_from_tracklist, + setlists_to_corpus, +) +from open_mythos.dj.harmonic import to_camelot + +# A real, timestamped tracklist (title-first). Used to test the parser on +# messy real-world input — not any particular user's DJ. +REAL_TRACKLIST = """ +1. Hold Up (Mixed) - William Kiss & Luke Alessi | 00:00:12 +2. Future Primitive - Simone De Kunovich | 00:05:48 +3. Truly Jack - Dj Split | 00:06:36 +4. I Love the Bass - Krypz | 00:10:12 +5. Music (Piano Mix) - Jex Opolis | 00:15:12 +6. Mentira - Maria Karunna | 00:19:36 +7. When I Wake Up - Lxury | 00:23:24 +8. So Hot - Marc Brauner | 00:43:48 +9. Runnin' - Deetron | 00:49:00 +""" + +SET_TEXT = """ +# a harmonic build +Aster - Opening Fog | 121 8A 0.30 +>> long_blend +Vela - First Light | 122 8A 0.45 +>> filter_sweep +Mira - Glass Arc | 125 9A 0.70 +>> long_blend +Volt - Peak Signal | 127 10A 0.90 +""" + + +# -- parsing ---------------------------------------------------------------- + +def test_text_parse_tracks_and_transitions(): + sl = setlist_from_text(SET_TEXT, name="n1") + assert len(sl.tracks) == 4 + assert len(sl.transitions) == 3 + assert sl.tracks[0].artist == "Aster" + assert sl.tracks[0].title == "Opening Fog" + assert sl.tracks[0].bpm == 121 + assert sl.tracks[0].key == "8A" + assert abs(sl.tracks[0].energy - 0.30) < 1e-6 + assert "filter_sweep" in sl.transitions + + +def test_json_parse_list_and_object(): + a = setlist_from_json('[{"title": "X", "bpm": 128, "key": "8A"}]', name="j") + assert len(a.tracks) == 1 and a.tracks[0].key == "8A" + b = setlist_from_json( + '{"name": "set", "tracks": [{"title": "X"}, {"title": "Y"}],' + ' "transitions": ["quick_cut"]}' + ) + assert b.name == "set" + assert b.transitions == ["quick_cut"] + + +# -- real-world tracklist parsing ------------------------------------------- + +def test_tracklist_parses_index_timestamp_titlefirst(): + sl = setlist_from_tracklist(REAL_TRACKLIST, name="real") + assert len(sl.tracks) == 9 + # Index stripped, title-first split, timestamp removed from the name. + assert sl.tracks[0].title == "Hold Up (Mixed)" + assert sl.tracks[0].artist == "William Kiss & Luke Alessi" + assert "00:00" not in sl.tracks[0].title + # Duration derived from the gap to the next timestamp (12s -> 5:48). + assert abs(sl.tracks[0].duration - (5 * 60 + 48 - 12)) < 1.0 + # Last track has no following timestamp -> no derived duration. + assert sl.tracks[-1].duration == 0.0 + + +def test_tracklist_artist_first_flag(): + sl = setlist_from_tracklist( + "1. Deetron - Runnin' | 00:01:00\n2. Krypz - I Love the Bass | 00:04:00", + name="af", title_first=False, + ) + assert sl.tracks[0].artist == "Deetron" + assert sl.tracks[0].title == "Runnin'" + + +def test_tracklist_learns_pacing_from_timestamps(): + sl = setlist_from_tracklist(REAL_TRACKLIST, name="real") + p = learn_profile_from_setlists([sl], name="real_dj") + # Timestamps give real hold-times; avg should be a sensible track length. + assert 60 < p.avg_track_seconds < 2000 + + +# -- key detection ---------------------------------------------------------- + +def test_estimate_key_major_and_minor(): + # A chroma vector dominated by C -> should read as C major. + cmaj = [10, 0, 4, 0, 6, 5, 0, 7, 0, 4, 0, 3] + assert to_camelot(estimate_key(cmaj)) == to_camelot("C") # 8B + # An A-minor-shaped vector. + amin = [4, 0, 3, 0, 5, 3, 0, 6, 0, 10, 0, 4] + code = to_camelot(estimate_key(amin)) + assert code in ("8A", "8B") # tonic A; mode may be close, tonic must hold + assert code and code.startswith("8") + + +# -- learning --------------------------------------------------------------- + +def test_learn_profile_infers_reasonable_values(): + sets = [setlist_from_text(SET_TEXT, name="n1")] + p = learn_profile_from_setlists(sets, name="learned") + assert p.name == "learned" + assert 118 <= p.bpm_range[0] <= p.bpm_range[1] <= 130 + assert p.bpm_drift >= 2 + # Every adjacent pair here is harmonically compatible -> high strictness. + assert p.harmonic_strictness >= 0.8 + assert p.energy_curve in ("build", "wave", "peak_time", "cooldown") + # filter_sweep and long_blend were annotated -> should be favored. + assert set(p.favored_transitions) & {"long_blend", "filter_sweep"} + + +def test_learned_profile_roundtrips_and_is_usable(): + from open_mythos.dj import DJStyleProfile, analyze_library, plan_mix + + sets = [setlist_from_text(SET_TEXT, name="n1")] + p = learn_profile_from_setlists(sets, name="learned") + p2 = DJStyleProfile.from_json(p.to_json()) + lib = analyze_library([{"title": t.title, "bpm": t.bpm, "key": t.key, + "energy": t.energy} for t in sets[0].tracks]) + plan = plan_mix(lib, p2) + assert len(plan.steps) == len(lib) + + +# -- corpus ----------------------------------------------------------------- + +def test_corpus_rows_are_mixlang(): + sets = [setlist_from_text(SET_TEXT, name="n1")] + rows = setlists_to_corpus(sets) + assert len(rows) == 1 + assert rows[0].startswith("") and rows[0].endswith("") + assert rows[0].count("") == 4 From ee0479a8d8df7e0957e4448aaa3cb625f85debb7 Mon Sep 17 00:00:00 2001 From: Home-Server Date: Sat, 1 Aug 2026 21:31:38 -0400 Subject: [PATCH 2/7] Add audio-recognition ingestion (mix -> tracklist) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit For DJs who don't publish tracklists, identify tracks from the mix audio. - identify.py: pluggable AudioIdentifier (AudD API), scan_mix slices a mix into windows, recognizes each, and assemble_setlist dedups consecutive hits into an ordered Setlist with timestamp-derived durations -> flows into enrich/learn. - CLI: `identify