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..851fdba5 --- /dev/null +++ b/open_mythos/dj/__init__.py @@ -0,0 +1,168 @@ +""" +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.identify import ( + AudDIdentifier, + AudioIdentifier, + CallableIdentifier, + ShazamIdentifier, + analyze_setlist_audio, + assemble_setlist, + build_identifier, + scan_mix, + setlist_to_dict, +) +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_folder, + 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_folder", + "setlist_from_json", + "learn_profile_from_setlists", + "setlists_to_corpus", + # audio recognition (mix -> tracklist) + "scan_mix", + "analyze_setlist_audio", + "assemble_setlist", + "build_identifier", + "AudioIdentifier", + "AudDIdentifier", + "ShazamIdentifier", + "CallableIdentifier", + "setlist_to_dict", + # 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..b88fd179 --- /dev/null +++ b/open_mythos/dj/__main__.py @@ -0,0 +1,278 @@ +""" +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_identify(args) -> int: + from .identify import build_identifier, scan_mix, setlist_to_dict + + identifier = build_identifier(args.provider) + + def _prog(offset, res): + mm, ss = divmod(int(offset), 60) + who = f"{res.get('artist','')} - {res.get('title','')}" if res else "(no match)" + print(f" {mm:02d}:{ss:02d} {who}") + + setlist = scan_mix( + args.audio, identifier, name=args.name, + segment_seconds=args.segment, hop_seconds=args.hop, + delay=args.delay, on_progress=_prog, + ) + print(f"\nIdentified {len(setlist.tracks)} distinct track(s).") + with open(args.out, "w", encoding="utf-8") as fh: + json.dump(setlist_to_dict(setlist), fh, indent=2) + print(f"Wrote setlist -> {args.out}\n" + f"Next: python -m open_mythos.dj learn {args.out} --name {args.name} " + f"--enrich musicbrainz") + 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: + import os + + from .setlist import ( + learn_profile_from_setlists, + setlist_from_folder, + 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 os.path.isdir(path): + setlists.append(setlist_from_folder(path, name=nm)) + elif 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) + + pi = sub.add_parser("identify", + help="recognize tracks in a mix's audio -> setlist JSON") + pi.add_argument("audio", help="path to the mix audio file (you supply it)") + pi.add_argument("--provider", required=True, + help="'shazam' (no key) | 'audd:API_TOKEN'") + pi.add_argument("--out", required=True, help="write the setlist JSON here") + pi.add_argument("--name", default="scanned_mix", help="name for the setlist") + pi.add_argument("--segment", type=float, default=20.0, + help="recognition clip length in seconds (default 20)") + pi.add_argument("--hop", type=float, default=60.0, + help="seconds between clips sampled (default 60)") + pi.add_argument("--delay", type=float, default=1.0, + help="seconds between API calls (default 1.0)") + pi.set_defaults(func=_cmd_identify) + + 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/.json) or folders of ordered audio") + 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..60206917 --- /dev/null +++ b/open_mythos/dj/analysis.py @@ -0,0 +1,259 @@ +""" +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_samples(y, sr: int) -> dict: + """Estimate ``{bpm, key, energy, loudness}`` from a mono sample array. + + Works on any audio window — a whole file or a segment sliced out of a + continuous mix — so it powers both file analysis and per-track analysis of + a scanned mix. Requires librosa. + """ + import librosa # type: ignore + import numpy as np # type: ignore + + # 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))) + # Absolute RMS saturates on loud masters, so it's a poor energy value alone. + # Keep a rough estimate but also expose raw loudness so a set can be + # normalized relative to itself (normalize_energy). + energy = float(min(1.0, rms * 8.0)) + chroma = librosa.feature.chroma_cens(y=y, sr=sr).mean(axis=1) + return {"bpm": tempo, "key": estimate_key(chroma), "energy": energy, + "loudness": rms} + + +def _analyze_signal(path: str) -> dict | None: + """Estimate bpm/key/energy from a whole file via librosa, if available.""" + try: + import librosa # type: ignore + except Exception: + return None + try: + y, sr = librosa.load(path, mono=True) + except Exception: + return None + + out = analyze_samples(y, sr) + out.update({ + "title": os.path.splitext(os.path.basename(path))[0], + "duration": float(librosa.get_duration(y=y, sr=sr)), + "path": path, + }) + return out + + +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}} + tr = tracks_from_dicts([merged])[0] + if signal.get("loudness") is not None: + tr.meta["loudness"] = signal["loudness"] + return tr + if data: + return tracks_from_dicts([data])[0] + return Track(title=os.path.splitext(os.path.basename(path))[0], path=path) + + +def normalize_energy(tracks: list[Track], lo: float = 0.15, hi: float = 1.0): + """Set each track's energy by min-max scaling raw loudness across the set. + + Absolute loudness is collection-relative, so an energy *curve* only means + something when tracks are compared to each other. Tracks without a + ``meta['loudness']`` value are left untouched. + """ + louds = [(t, t.meta.get("loudness")) for t in tracks] + vals = [v for _, v in louds if v is not None] + if len(vals) < 2: + return tracks + lo_v, hi_v = min(vals), max(vals) + span = (hi_v - lo_v) or 1.0 + for t, v in louds: + if v is not None: + t.energy = round(lo + (hi - lo) * (v - lo_v) / span, 3) + return tracks + + +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/identify.py b/open_mythos/dj/identify.py new file mode 100644 index 00000000..98314019 --- /dev/null +++ b/open_mythos/dj/identify.py @@ -0,0 +1,349 @@ +""" +Audio recognition — build a tracklist from an unlabeled DJ mix. + +For DJs who never publish tracklists, the only way to learn their selection is +to identify tracks from the *audio* — the same approach set79 / trackid.net use. +This module slices a mix into short windows, fingerprints each through a +pluggable :class:`AudioIdentifier` (AudD to start), collapses consecutive +duplicate hits into an ordered tracklist with timestamps, and returns a +:class:`~open_mythos.dj.setlist.Setlist` that flows straight into the existing +``enrich`` → ``learn`` pipeline. + +Recognition gives you *artist + title* per track; BPM/key/energy still come from +the enrichment step (see :mod:`open_mythos.dj.enrich`). + +Requirements / boundaries: +* You supply the mix audio file. Obtain it legally — do not rip streams you + aren't allowed to. This module never downloads audio. +* A recognition API key (AudD has a free tier: https://audd.io/). +* Slicing needs the ``[audio]`` extra (numpy + soundfile); the assembly logic + itself is dependency-free and unit-tested offline. +""" + +from __future__ import annotations + +import io +import json +import time +import urllib.request +import uuid + +from .setlist import Setlist +from .analysis import tracks_from_dicts + + +class AudioIdentifier: + """Identify the primary track in a short audio clip. + + Implementations return ``{"artist": ..., "title": ...}`` (plus any extra + fields) or ``None`` when nothing is recognized. + """ + + def identify_clip(self, wav_bytes: bytes) -> dict | None: # pragma: no cover + raise NotImplementedError + + +class CallableIdentifier(AudioIdentifier): + """Wrap ``fn(wav_bytes) -> dict | None`` — handy for tests and custom engines.""" + + def __init__(self, fn): + self.fn = fn + + def identify_clip(self, wav_bytes: bytes) -> dict | None: + return self.fn(wav_bytes) + + +class AudDIdentifier(AudioIdentifier): + """Recognize a clip via the AudD API (https://docs.audd.io/).""" + + endpoint = "https://api.audd.io/" + user_agent = "OpenMythos-DJ/0.1" + timeout = 20.0 + + def __init__(self, api_token: str): + self.api_token = api_token + + def _post(self, fields: dict[str, str], wav_bytes: bytes) -> dict: + """POST a multipart request; overridden in tests to avoid the network.""" + boundary = uuid.uuid4().hex + parts: list[bytes] = [] + for k, v in fields.items(): + parts.append( + f"--{boundary}\r\nContent-Disposition: form-data; name=\"{k}\"" + f"\r\n\r\n{v}\r\n".encode() + ) + parts.append( + f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; " + f"filename=\"clip.wav\"\r\nContent-Type: audio/wav\r\n\r\n".encode() + ) + parts.append(wav_bytes) + parts.append(f"\r\n--{boundary}--\r\n".encode()) + body = b"".join(parts) + req = urllib.request.Request( + self.endpoint, + data=body, + headers={ + "User-Agent": self.user_agent, + "Content-Type": f"multipart/form-data; boundary={boundary}", + }, + ) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + def identify_clip(self, wav_bytes: bytes) -> dict | None: + try: + data = self._post({"api_token": self.api_token}, wav_bytes) + except Exception: + return None + return parse_audd_result(data) + + +def parse_audd_result(data: dict) -> dict | None: + """Extract ``{artist, title}`` from an AudD JSON response, or ``None``.""" + if not data or data.get("status") != "success": + return None + result = data.get("result") + if not result: + return None + artist = (result.get("artist") or "").strip() + title = (result.get("title") or "").strip() + if not title: + return None + return {"artist": artist, "title": title} + + +class ShazamIdentifier(AudioIdentifier): + """Recognize a clip via Shazam — **no API key required**. + + Uses the third-party ``shazamio`` client, which talks to Shazam's endpoint + directly. It needs no signup or token and its algorithm is robust to the + tempo/EQ changes in a DJ mix. Trade-off: ``shazamio`` is *unofficial* (ToS + gray area) and may break if Shazam changes their protocol. + + pip install "open-mythos[shazam]" # installs shazamio + + The async recognition call is isolated in :meth:`_recognize` so parsing is + unit-testable without network or the shazamio dependency. + """ + + def __init__(self, delay: float = 0.0): + self._delay = delay + + def _recognize(self, wav_bytes: bytes) -> dict: + """Call shazamio and return its raw response dict. Overridden in tests.""" + import asyncio + + from shazamio import Shazam # type: ignore + + async def _go() -> dict: + shazam = Shazam() + recognize = getattr(shazam, "recognize", None) or shazam.recognize_song + return await recognize(wav_bytes) + + return asyncio.run(_go()) + + def identify_clip(self, wav_bytes: bytes) -> dict | None: + try: + data = self._recognize(wav_bytes) + except Exception: + return None + return parse_shazam_result(data) + + +def parse_shazam_result(data: dict) -> dict | None: + """Extract ``{artist, title}`` from a shazamio response, or ``None``.""" + if not data: + return None + track = data.get("track") + if not track: # no match -> empty 'matches', no 'track' + return None + title = (track.get("title") or "").strip() + artist = (track.get("subtitle") or "").strip() # shazam: subtitle == artist + if not title: + return None + return {"artist": artist, "title": title} + + +# -------------------------------------------------------------------------- +# Assembly (pure, offline-testable) +# -------------------------------------------------------------------------- + +def _same(a: dict | None, b: dict | None) -> bool: + if not a or not b: + return False + return ( + a.get("artist", "").lower() == b.get("artist", "").lower() + and a.get("title", "").lower() == b.get("title", "").lower() + ) + + +def assemble_setlist( + recognitions: list[tuple[float, dict | None]], + name: str, + total_duration: float | None = None, +) -> Setlist: + """Collapse per-window recognitions into an ordered :class:`Setlist`. + + Args: + recognitions: ``(offset_seconds, {artist,title} | None)`` per window, + in time order. + name: Setlist name. + total_duration: Mix length in seconds, used to time the final track. + + Consecutive windows that identify the same track are merged into one entry + whose start is the first window it appeared in; unrecognized windows are + dropped. Track durations are derived from the gaps between starts. + """ + merged: list[tuple[float, dict]] = [] + for offset, res in recognitions: + if res is None: + continue + if merged and _same(merged[-1][1], res): + continue # same track still playing + merged.append((offset, res)) + + rows = [] + for i, (offset, res) in enumerate(merged): + end = merged[i + 1][0] if i + 1 < len(merged) else total_duration + row = {"artist": res.get("artist", ""), "title": res.get("title", "")} + if end is not None and end > offset: + row["duration"] = end - offset + rows.append(row) + + tracks = tracks_from_dicts(rows) + for tr, (offset, _) in zip(tracks, merged): + tr.meta["offset"] = offset # position in the mix, for later audio analysis + transitions = ["long_blend"] * (len(tracks) - 1) if len(tracks) > 1 else [] + return Setlist(name=name, tracks=tracks, transitions=transitions) + + +def analyze_setlist_audio(setlist, audio_path: str, window: float = 30.0): + """Fill BPM / key / energy on a scanned setlist from the mix audio itself. + + Recognition (Shazam/AudD) gives only artist + title. This analyzes a + ``window``-second slice of the mix at each track's stored ``meta['offset']`` + to recover the harmonic and tempo signature, then normalizes energy across + the set. Needs the ``[audio]`` extra. Returns the same setlist, enriched. + """ + import numpy as np # noqa: F401 + import soundfile as sf # type: ignore + + from .analysis import analyze_samples, normalize_energy + from .harmonic import to_camelot + + info = sf.info(audio_path) + sr = int(info.samplerate) + total = float(info.duration) + for tr in setlist.tracks: + off = tr.meta.get("offset") + if off is None: + continue + start = int(off * sr) + stop = int(min(off + window, total) * sr) + y, _ = sf.read(audio_path, start=start, stop=stop, dtype="float32", + always_2d=True) + y = y.mean(axis=1) + res = analyze_samples(y, sr) + tr.bpm = float(res["bpm"]) + tr.key = to_camelot(res["key"]) or tr.key + tr.meta["loudness"] = res["loudness"] + normalize_energy(setlist.tracks) + return setlist + + +# -------------------------------------------------------------------------- +# Mix scanning (needs the [audio] extra) +# -------------------------------------------------------------------------- + +def _clip_to_wav_bytes(y, sr: int) -> bytes: + import soundfile as sf # type: ignore + + buf = io.BytesIO() + sf.write(buf, y, sr, format="WAV", subtype="PCM_16") + return buf.getvalue() + + +def scan_mix( + audio_path: str, + identifier: AudioIdentifier, + name: str = "scanned_mix", + segment_seconds: float = 20.0, + hop_seconds: float = 60.0, + delay: float = 0.0, + on_progress=None, +) -> Setlist: + """Identify tracks across a full mix and return an ordered :class:`Setlist`. + + Args: + audio_path: Path to the mix audio (you supply it, legally). + identifier: The recognition engine. + name: Name for the resulting setlist. + segment_seconds: Length of each clip sent for recognition. + hop_seconds: Spacing between clip starts (how often to sample the mix). + delay: Seconds to sleep between recognitions (API rate-limit politeness). + on_progress: Optional ``callable(offset, result)`` for logging. + """ + try: + import numpy as np # noqa: F401 + import soundfile as sf # type: ignore + except Exception as exc: + raise RuntimeError( + "scan_mix needs the audio extra: pip install 'open-mythos[audio]'" + ) from exc + + info = sf.info(audio_path) + total = float(info.duration) + sr = int(info.samplerate) + + recognitions: list[tuple[float, dict | None]] = [] + offset = 0.0 + while offset < total: + start = int(offset * sr) + stop = int(min(offset + segment_seconds, total) * sr) + y, _ = sf.read(audio_path, start=start, stop=stop, dtype="float32", + always_2d=True) + y = y.mean(axis=1) # mono downmix + res = identifier.identify_clip(_clip_to_wav_bytes(y, sr)) + recognitions.append((offset, res)) + if on_progress: + on_progress(offset, res) + if delay: + time.sleep(delay) + offset += hop_seconds + + return assemble_setlist(recognitions, name=name, total_duration=total) + + +def build_identifier(spec: str) -> AudioIdentifier: + """Build an identifier from a CLI spec. + + ``"shazam"`` (no key) | ``"audd:API_TOKEN"``. + """ + name, _, arg = spec.partition(":") + name = name.lower() + if name == "shazam": + return ShazamIdentifier() + if name == "audd": + if not arg: + raise ValueError("audd needs an API token: audd:TOKEN") + return AudDIdentifier(arg) + raise ValueError(f"unknown identifier {name!r}") + + +def setlist_to_dict(setlist: Setlist) -> dict: + """Serialize a :class:`Setlist` to a JSON-friendly dict (for `learn`).""" + return { + "name": setlist.name, + "tracks": [ + { + "artist": t.artist, + "title": t.title, + "bpm": t.bpm, + "key": t.key, + "energy": t.energy, + "duration": t.duration, + } + for t in setlist.tracks + ], + "transitions": setlist.transitions, + } 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..ada2a100 --- /dev/null +++ b/open_mythos/dj/render.py @@ -0,0 +1,336 @@ +""" +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 stereo output). Mono sources are upmixed to dual-mono + so every rendered mix is stereo, regardless of source channel count. +* ``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`` at ``sr``, preserving channels, as shape ``(channels, n)``. + + Prefers soundfile, then librosa. Mono sources come back as a single-row + array (``channels == 1``); :func:`_to_stereo` upmixes later so every + segment entering the mix has the same channel count. + """ + np = _require_numpy() + try: + import soundfile as sf # type: ignore + + data, file_sr = sf.read(path, dtype="float32", always_2d=True) # (n, ch) + y = data.T # (channels, n) + 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=False) + if y.ndim == 1: + y = y[None, :] + return y.astype(np.float32) + except Exception as exc: + raise RuntimeError(f"could not load audio: {path} ({exc})") from exc + + +def _to_stereo(y): + """Normalize a ``(channels, n)`` array to exactly 2 channels. + + Mono is duplicated to both channels (standard dual-mono); anything beyond + stereo is truncated to the first two channels. + """ + if y.ndim == 1: + y = y[None, :] + if y.shape[0] == 1: + _require_numpy() + import numpy as np # type: ignore + + y = np.repeat(y, 2, axis=0) + elif y.shape[0] > 2: + y = y[:2] + return y + + +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, axis=-1) + except Exception: + np = _require_numpy() + n_src = y.shape[-1] + n_dst = int(round(n_src * dst_sr / src_sr)) + x_old = np.linspace(0.0, 1.0, num=n_src, endpoint=False) + x_new = np.linspace(0.0, 1.0, num=n_dst, endpoint=False) + if y.ndim == 1: + return np.interp(x_new, x_old, y).astype(y.dtype) + return np.stack( + [np.interp(x_new, x_old, y[c]) for c in range(y.shape[0])] + ).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_1d(np, y, start_a: float, end_a: float): + 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 + + +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). + + Accepts a 1D mono buffer or a ``(channels, n)`` stereo buffer. + """ + if y.ndim == 1: + return _lowpass_sweep_1d(np, y, start_a, end_a) + return np.stack( + [_lowpass_sweep_1d(np, y[c], start_a, end_a) for c in range(y.shape[0])] + ) + + +@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 a ``(channels, n)`` float array to a 16-bit PCM WAV. + + Uses soundfile if available; falls back to the stdlib ``wave`` module, + interleaving channels manually (a mono 1D array is also accepted). + """ + np = _require_numpy() + y = np.clip(y, -1.0, 1.0) + channels = y.shape[0] if y.ndim == 2 else 1 + try: + import soundfile as sf # type: ignore + + data = y.T if y.ndim == 2 else y # soundfile wants (n, channels) + sf.write(path, data, sr, subtype="PCM_16") + return + except Exception: + pass + data = y.T if y.ndim == 2 else y[:, None] + pcm = (data * 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 = _to_stereo(_load_audio(path_for(step.track), sr)) + if beatmatch and target_bpm and step.track.bpm > 0: + y = _to_stereo(_time_stretch(y, rate=step.track.bpm / target_bpm)) + cap = max_segment_seconds or step.cue_seconds or (y.shape[-1] / sr) + y = y[:, : int(cap * sr)] + segments.append(y) + + # Assemble with per-transition crossfades on a running output buffer. + # Internal shape is always (2, n); written back out as interleaved stereo. + 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, out.shape[-1], seg.shape[-1] // 2 or seg.shape[-1]) + + start = out.shape[-1] - 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], axis=-1) + 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:]], axis=-1) + + 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=out.shape[-1] / 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..323cdf89 --- /dev/null +++ b/open_mythos/dj/setlist.py @@ -0,0 +1,398 @@ +""" +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 _parse_track_filename(basename: str) -> tuple[str, str]: + """From ``"01 01 Drake - Fancy.mp3"`` -> ``("Drake", "Fancy")`` (artist-first). + + Strips the extension and any repeated leading index numbers, then splits on + the first ``" - "``. Falls back to (``""``, whole-name) when there's no dash. + """ + import os + + stem = os.path.splitext(basename)[0] + prev = None + while prev != stem: # strip repeated leading "NN " index groups + prev = stem + stem = _INDEX_RE.sub("", stem).strip() + if " - " in stem: + artist, title = stem.split(" - ", 1) + return artist.strip(), title.strip() + return "", stem.strip() + + +def setlist_from_folder(path: str, name: str, analyze_audio: bool = True) -> Setlist: + """Treat an ordered folder of audio files as a DJ's setlist. + + A split mixtape (one file per track, numbered in play order) is itself a + setlist: the file order is the DJ's sequencing, and each file's tags / + filename give the artist and title. With ``analyze_audio`` and the + ``[audio]`` extra installed, each track is also analyzed for BPM / key / + energy — otherwise those are left for the enrichment step. + + Args: + path: Folder of audio files (sorted by name = play order). + name: Name for the resulting setlist. + analyze_audio: If ``True``, read tags + run signal analysis per file. + """ + import os + + from .analysis import Track, analyze_file, normalize_energy + + exts = (".mp3", ".wav", ".flac", ".aiff", ".aif", ".m4a", ".ogg") + files = sorted(f for f in os.listdir(path) if f.lower().endswith(exts)) + tracks: list[Track] = [] + for f in files: + full = os.path.join(path, f) + if analyze_audio: + tr = analyze_file(full) # keeps duration/bpm/key/energy + else: + tr = Track(title=os.path.splitext(f)[0], path=full) + # Filenames on split mixtapes carry the clean "Artist - Title" (tags + # often hold the DJ as artist + a numbered title), so prefer them. + fn_artist, fn_title = _parse_track_filename(f) + if fn_artist: + tr.artist, tr.title = fn_artist, fn_title + elif not tr.artist: + tr.title = fn_title or tr.title + tracks.append(tr) + + if analyze_audio: + normalize_energy(tracks) # relative energy curve across the set + 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/open_mythos/main.py b/open_mythos/main.py old mode 100644 new mode 100755 index 65b0fa82..adab6185 --- a/open_mythos/main.py +++ b/open_mythos/main.py @@ -722,7 +722,11 @@ def get_A(self) -> torch.Tensor: # Compute in log space to avoid 0 * inf = NaN when log_dt → -∞, log_A → +∞. # dt * A_c = -exp(log_dt) * exp(log_A) = -exp(log_dt + log_A) # Clamp keeps the product finite in float32 for any gradient step size. - return torch.exp(-torch.exp((self.log_dt + self.log_A).clamp(-20, 20))) + A = torch.exp(-torch.exp((self.log_dt + self.log_A).clamp(-20, 20))) + # When the exponent underflows toward 0 (e.g. after a large gradient + # step), exp(-tiny) rounds to exactly 1.0 in float32/float16, silently + # violating the ρ(A) < 1 guarantee. Clamp explicitly so it always holds. + return torch.clamp(A, max=1.0 - torch.finfo(A.dtype).eps) def forward( self, h: torch.Tensor, e: torch.Tensor, transformer_out: torch.Tensor diff --git a/pyproject.toml b/pyproject.toml old mode 100644 new mode 100755 index 8129e904..a5b5a62f --- a/pyproject.toml +++ b/pyproject.toml @@ -44,11 +44,29 @@ datasets = ">=2.18.0" [tool.poetry.extras] flash = ["flash-attn"] +audio = ["librosa", "mutagen", "soundfile"] +shazam = ["shazamio"] [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.dependencies.shazamio] +version = ">=0.4.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_mix_output_is_stereo(tmp_path): + paths = [] + for i, bpm in enumerate((120, 124)): + p = str(tmp_path / f"st{i}.wav") + _synth_wav(p, seconds=4.0, bpm=bpm, roots_hz=[220.0 + 10 * i]) + paths.append(p) + plan = _plan_from_paths(paths, (120, 124)) + out = str(tmp_path / "stereo.wav") + render_mix(plan, out, sr=SR, target_bpm=122, beatmatch=True) + y, file_sr = sf.read(out, always_2d=True) + assert y.shape[1] == 2 # stereo output, even from mono synth sources + + +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") + + +def test_scan_mix_over_synth_audio(tmp_path): + # Build a 3-minute "mix" and a fake identifier that reports a different + # track for each third — scan_mix must slice, recognize, and dedup them. + from open_mythos.dj.identify import CallableIdentifier, scan_mix + + mix = str(tmp_path / "mix.wav") + _synth_wav(mix, seconds=180.0, bpm=124, roots_hz=[220.0]) + + # Decode window offset from clip length is not available to the fake, so + # key on a shared counter: windows arrive in order at 60s hops. + calls = {"n": 0} + + def fake(wav_bytes): + i = calls["n"] + calls["n"] += 1 + track = i // 1 # one window per hop; 3 hops over 180s at hop=60 + names = [("A", "One"), ("A", "One"), ("B", "Two")] + a, t = names[min(track, len(names) - 1)] + return {"artist": a, "title": t} + + sl = scan_mix(mix, CallableIdentifier(fake), name="scan", + segment_seconds=10.0, hop_seconds=60.0) + # Windows at 0, 60, 120 -> [One, One, Two] -> dedup -> [One, Two] + assert [t.title for t in sl.tracks] == ["One", "Two"] + assert sl.tracks[0].duration > 0 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_identify.py b/tests/test_dj_identify.py new file mode 100644 index 00000000..c34bf6dc --- /dev/null +++ b/tests/test_dj_identify.py @@ -0,0 +1,126 @@ +"""Tests for audio-recognition ingestion (mix -> tracklist). Offline.""" + +from open_mythos.dj.identify import ( + AudDIdentifier, + CallableIdentifier, + ShazamIdentifier, + assemble_setlist, + build_identifier, + parse_audd_result, + parse_shazam_result, + setlist_to_dict, +) + + +# -- AudD response parsing -------------------------------------------------- + +def test_parse_audd_success_and_failure(): + ok = {"status": "success", "result": {"artist": "Deetron", "title": "Runnin'"}} + assert parse_audd_result(ok) == {"artist": "Deetron", "title": "Runnin'"} + assert parse_audd_result({"status": "success", "result": None}) is None + assert parse_audd_result({"status": "error"}) is None + assert parse_audd_result({}) is None + + +def test_audd_identifier_uses_post(monkeypatch): + ident = AudDIdentifier("TOKEN") + monkeypatch.setattr(ident, "_post", lambda fields, wav: { + "status": "success", "result": {"artist": "A", "title": "B"} + }) + assert ident.identify_clip(b"fakewav") == {"artist": "A", "title": "B"} + + +# -- Shazam (no-key) parsing ------------------------------------------------ + +def test_parse_shazam_success_and_no_match(): + ok = {"track": {"title": "Runnin'", "subtitle": "Deetron"}} + assert parse_shazam_result(ok) == {"artist": "Deetron", "title": "Runnin'"} + assert parse_shazam_result({"matches": []}) is None # no 'track' key + assert parse_shazam_result({}) is None + + +def test_shazam_identifier_uses_recognize(monkeypatch): + ident = ShazamIdentifier() + monkeypatch.setattr(ident, "_recognize", lambda wav: { + "track": {"title": "T", "subtitle": "Ar"} + }) + assert ident.identify_clip(b"fakewav") == {"artist": "Ar", "title": "T"} + + +def test_shazam_identifier_handles_failure(monkeypatch): + ident = ShazamIdentifier() + + def boom(wav): + raise RuntimeError("network down") + + monkeypatch.setattr(ident, "_recognize", boom) + assert ident.identify_clip(b"x") is None + + +# -- assembly (dedup + timing) ---------------------------------------------- + +def test_assemble_dedups_consecutive_and_times_tracks(): + T1 = {"artist": "A", "title": "One"} + T2 = {"artist": "B", "title": "Two"} + recs = [ + (0.0, T1), (60.0, T1), (120.0, None), (180.0, T2), (240.0, T2), + ] + sl = assemble_setlist(recs, name="mix", total_duration=300.0) + assert [t.title for t in sl.tracks] == ["One", "Two"] + # One starts at 0, next distinct (Two) starts at 180 -> duration 180. + assert sl.tracks[0].duration == 180.0 + # Two runs to the end (300). + assert sl.tracks[1].duration == 120.0 + assert sl.transitions == ["long_blend"] + + +def test_assemble_stores_offset_for_audio_analysis(): + recs = [(60.0, {"artist": "A", "title": "One"}), + (180.0, {"artist": "B", "title": "Two"})] + sl = assemble_setlist(recs, name="m", total_duration=300.0) + assert sl.tracks[0].meta["offset"] == 60.0 + assert sl.tracks[1].meta["offset"] == 180.0 + + +def test_assemble_drops_unrecognized_and_handles_empty(): + assert assemble_setlist([(0.0, None), (60.0, None)], name="m").tracks == [] + assert assemble_setlist([], name="m").tracks == [] + + +def test_assemble_same_track_case_insensitive(): + recs = [ + (0.0, {"artist": "A", "title": "One"}), + (60.0, {"artist": "a", "title": "ONE"}), # same track, different case + (120.0, {"artist": "A", "title": "Two"}), + ] + sl = assemble_setlist(recs, name="m", total_duration=180.0) + assert [t.title for t in sl.tracks] == ["One", "Two"] + + +# -- CallableIdentifier + scan orchestration (no real audio) ---------------- + +def test_callable_identifier(): + ident = CallableIdentifier(lambda wav: {"artist": "X", "title": "Y"}) + assert ident.identify_clip(b"") == {"artist": "X", "title": "Y"} + + +def test_setlist_to_dict_roundtrip_shape(): + recs = [(0.0, {"artist": "A", "title": "One"})] + sl = assemble_setlist(recs, name="m", total_duration=60.0) + d = setlist_to_dict(sl) + assert d["name"] == "m" + assert d["tracks"][0]["artist"] == "A" + assert "energy" in d["tracks"][0] + + +# -- build_identifier ------------------------------------------------------- + +def test_build_identifier_spec(): + assert isinstance(build_identifier("audd:TOKEN"), AudDIdentifier) + assert isinstance(build_identifier("shazam"), ShazamIdentifier) + import pytest + + with pytest.raises(ValueError): + build_identifier("audd") # missing token + with pytest.raises(ValueError): + build_identifier("bogus:x") # unknown provider diff --git a/tests/test_dj_setlist.py b/tests/test_dj_setlist.py new file mode 100644 index 00000000..2dcce077 --- /dev/null +++ b/tests/test_dj_setlist.py @@ -0,0 +1,175 @@ +"""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 + + +# -- folder ingestion (split mixtape) --------------------------------------- + +def test_setlist_from_folder_parses_names_and_order(tmp_path): + from open_mythos.dj.setlist import setlist_from_folder + + # A split mixtape: doubled index prefix, "Artist - Title" in the name. + names = [ + "01 01 Drake Feat. T.I. - Fancy.mp3", + "02 02 Usher - There Goes My Baby.mp3", + "03 03 Trey Songz - Flatline.mp3", + ] + for n in names: + (tmp_path / n).write_bytes(b"") # empty; analyze_audio=False + sl = setlist_from_folder(str(tmp_path), name="djx", analyze_audio=False) + assert [t.title for t in sl.tracks] == ["Fancy", "There Goes My Baby", "Flatline"] + assert sl.tracks[0].artist == "Drake Feat. T.I." + assert len(sl.transitions) == 2 + + +def test_normalize_energy_spreads_across_set(): + from open_mythos.dj.analysis import Track, normalize_energy + + tracks = [ + Track(title="a", meta={"loudness": 0.10}), + Track(title="b", meta={"loudness": 0.20}), + Track(title="c", meta={"loudness": 0.30}), + ] + normalize_energy(tracks, lo=0.15, hi=1.0) + assert tracks[0].energy == 0.15 # quietest -> floor + assert tracks[2].energy == 1.0 # loudest -> ceiling + assert 0.15 < tracks[1].energy < 1.0 + + +# -- 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 diff --git a/tests/test_main.py b/tests/test_main.py old mode 100644 new mode 100755 index c54c4626..c45754e1 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -266,24 +266,25 @@ def setup_method(self): def test_output_shape(self): x = torch.randn(B, T, self.cfg.dim) - out = self.attn(x, self.freqs) + out = self.attn(x, self.freqs[:T]) assert out.shape == (B, T, self.cfg.dim) def test_kv_cache_accumulates(self): cache = {} x = torch.randn(B, T, self.cfg.dim) - self.attn(x, self.freqs, kv_cache=cache, cache_key="layer0") + # positions 0..T-1, then T..2T-1 — freqs must track the cache offset + self.attn(x, self.freqs[:T], kv_cache=cache, cache_key="layer0") assert "layer0" in cache k_len = cache["layer0"]["k"].shape[1] # second call adds T more tokens - self.attn(x, self.freqs, kv_cache=cache, cache_key="layer0") + self.attn(x, self.freqs[T : 2 * T], kv_cache=cache, cache_key="layer0") assert cache["layer0"]["k"].shape[1] == k_len + T def test_with_causal_mask(self): x = torch.randn(B, T, self.cfg.dim) mask = torch.full((1, 1, T, T), float("-inf")) mask = torch.triu(mask, diagonal=1) - out = self.attn(x, self.freqs, mask=mask) + out = self.attn(x, self.freqs[:T], mask=mask) assert out.shape == (B, T, self.cfg.dim) @@ -302,13 +303,13 @@ def setup_method(self): def test_output_shape(self): x = torch.randn(B, T, self.cfg.dim) - out = self.attn(x, self.freqs) + out = self.attn(x, self.freqs[:T]) assert out.shape == (B, T, self.cfg.dim) def test_cache_stores_compressed_kv(self): cache = {} x = torch.randn(B, T, self.cfg.dim) - self.attn(x, self.freqs, kv_cache=cache, cache_key="mla0") + self.attn(x, self.freqs[:T], kv_cache=cache, cache_key="mla0") assert "c_kv" in cache["mla0"] assert "k_rope" in cache["mla0"] # c_kv should have kv_lora_rank as last dim, not full K/V @@ -317,15 +318,16 @@ def test_cache_stores_compressed_kv(self): def test_cache_accumulates_across_steps(self): cache = {} x = torch.randn(B, T, self.cfg.dim) - self.attn(x, self.freqs, kv_cache=cache, cache_key="mla0") + # positions 0..T-1, then T..2T-1 — freqs must track the cache offset + self.attn(x, self.freqs[:T], kv_cache=cache, cache_key="mla0") first_len = cache["mla0"]["c_kv"].shape[1] - self.attn(x, self.freqs, kv_cache=cache, cache_key="mla0") + self.attn(x, self.freqs[T : 2 * T], kv_cache=cache, cache_key="mla0") assert cache["mla0"]["c_kv"].shape[1] == first_len + T def test_with_causal_mask(self): x = torch.randn(B, T, self.cfg.dim) mask = torch.triu(torch.full((1, 1, T, T), float("-inf")), diagonal=1) - out = self.attn(x, self.freqs, mask=mask) + out = self.attn(x, self.freqs[:T], mask=mask) assert out.shape == (B, T, self.cfg.dim) @@ -432,21 +434,21 @@ def test_gqa_output_shape(self): block = TransformerBlock(cfg, use_moe=False) freqs = precompute_rope_freqs(cfg.dim // cfg.n_heads, cfg.max_seq_len) x = torch.randn(B, T, cfg.dim) - assert block(x, freqs).shape == (B, T, cfg.dim) + assert block(x, freqs[:T]).shape == (B, T, cfg.dim) def test_mla_output_shape(self): cfg = mla_cfg() block = TransformerBlock(cfg, use_moe=False) freqs = precompute_rope_freqs(cfg.qk_rope_head_dim, cfg.max_seq_len) x = torch.randn(B, T, cfg.dim) - assert block(x, freqs).shape == (B, T, cfg.dim) + assert block(x, freqs[:T]).shape == (B, T, cfg.dim) def test_moe_block_output_shape(self): cfg = gqa_cfg() block = TransformerBlock(cfg, use_moe=True) freqs = precompute_rope_freqs(cfg.dim // cfg.n_heads, cfg.max_seq_len) x = torch.randn(B, T, cfg.dim) - assert block(x, freqs).shape == (B, T, cfg.dim) + assert block(x, freqs[:T]).shape == (B, T, cfg.dim) def test_attn_type_selection(self): assert isinstance(TransformerBlock(gqa_cfg()).attn, GQAttention) @@ -526,20 +528,20 @@ def setup_method(self): def test_output_shape(self): h = torch.randn(B, T, self.cfg.dim) e = torch.randn(B, T, self.cfg.dim) - out = self.block(h, e, self.freqs) + out = self.block(h, e, self.freqs[:T]) assert out.shape == (B, T, self.cfg.dim) def test_more_loops_changes_output(self): h = torch.randn(B, T, self.cfg.dim) e = torch.randn(B, T, self.cfg.dim) - out1 = self.block(h.clone(), e.clone(), self.freqs, n_loops=1) - out3 = self.block(h.clone(), e.clone(), self.freqs, n_loops=3) + out1 = self.block(h.clone(), e.clone(), self.freqs[:T], n_loops=1) + out3 = self.block(h.clone(), e.clone(), self.freqs[:T], n_loops=3) assert not torch.allclose(out1, out3) def test_single_loop_runs(self): h = torch.randn(B, T, self.cfg.dim) e = torch.randn(B, T, self.cfg.dim) - out = self.block(h, e, self.freqs, n_loops=1) + out = self.block(h, e, self.freqs[:T], n_loops=1) assert out.shape == (B, T, self.cfg.dim)