commit 3cd3ce9e524c330d17596563032a6baf23e0f70c
from: mtmn <miro@haravara.org>
date: Mon Jul 27 21:14:08 2026 UTC

add kundali

commit - f9e7c66a868d68eae5ebe5b97123435b595a7559
commit + 3cd3ce9e524c330d17596563032a6baf23e0f70c
blob - 5814a3aa2d5347c5a386bc51f917417ee49a9e41
blob + 0c1d89aa383dc348fbcff9a31bebc86bc344e81b
--- README.md
+++ README.md
@@ -21,6 +21,7 @@ Various tools I have been using throughout the years.
 | [wrd](wrd/) | full-text search and reader for an offline web archive |
 | [plants](plants/) | bluetooth battery monitoring for linux |
 | [reflink-snap](reflink-snap/) | reflink (copy-on-write) snapshot manager for XFS |
+| [kundali](kudali/) | scans track metadata and builds a sequence around an anchor track |
 
 ## Toolchains
 
blob - 65f1ef3536529b96035072ccb6eb9642cf157727
blob + f56f2198a67ba8444edb02005b53a3795a7fed34
--- all.do
+++ all.do
@@ -5,5 +5,5 @@ redo-ifchange \
 	bin/wrd \
 	bin/mpd_add_to_playlist bin/mpd_add_to_queue bin/mpd_edit_queue \
 	bin/mpd_update_library bin/mpd_update_queue \
-	bin/diggah bin/lazymaster bin/nts bin/shuffle \
+	bin/diggah bin/lazymaster bin/nts bin/shuffle bin/kundali \
 	bin/hue bin/reflink-snap
blob - /dev/null
blob + a25508311932daabb5b8c8ac83b94766e44230ef (mode 644)
--- /dev/null
+++ bin/kundali.do
@@ -0,0 +1,8 @@
+exec >&2
+redo-ifchange venv-stamp ../kundali/kundali.py
+root=$(cd .. && pwd)
+cat > "$3" <<EOF
+#!/bin/sh
+exec "$root/.venv/bin/python" "$root/kundali/kundali.py" "\$@"
+EOF
+chmod +x "$3"
blob - /dev/null
blob + e56a45363f5debd7a6a14ec67c96f9fe98281fa2 (mode 755)
--- /dev/null
+++ kundali/kundali.py
@@ -0,0 +1,683 @@
+#!/usr/bin/env python3
+"""
+kundali: scans track metadata and builds a sequence around an anchor track
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+import mutagen
+
+PITCH_CLASSES = {
+    "C": 0,
+    "G": 1,
+    "D": 2,
+    "A": 3,
+    "E": 4,
+    "B": 5,
+    "F#": 6,
+    "Gb": 6,
+    "C#": 7,
+    "Db": 7,
+    "G#": 8,
+    "Ab": 8,
+    "D#": 9,
+    "Eb": 9,
+    "A#": 10,
+    "Bb": 10,
+    "F": 11,
+}
+
+CAMLOT_MAJOR = {
+    0: 8,
+    1: 9,
+    2: 10,
+    3: 11,
+    4: 12,
+    5: 1,
+    6: 2,
+    7: 3,
+    8: 4,
+    9: 5,
+    10: 6,
+    11: 7,
+}
+
+BPM_TAGS = ("TBPM", "BPM", "FBPM")
+KEY_TAGS = ("TKEY", "INITIALKEY", "KEY", "FKEY")
+TITLE_TAGS = ("TITLE", "TIT2", "TIT1")
+
+CURVES = {"linear": "tri", "qsin": "qsin", "hsin": "hsin", "exp": "exp", "log": "log"}
+CODECS = {"wav": "pcm_f32le", "flac": "flac", "mp3": "libmp3lame", "ogg": "libvorbis"}
+
+SAMPLE_RATE = 48000
+LOGGER = logging.getLogger("kundali")
+
+_KEY_RE = re.compile(r"^(?P<note>.*?)\s*(?P<mode>MAJOR|MINOR|M|m)?$", re.IGNORECASE)
+
+
+def error(msg: str) -> None:
+    LOGGER.error(msg)
+    sys.exit(1)
+
+
+def parse_key(key_str: str) -> tuple[int, bool]:
+    original = key_str.strip()
+    match = _KEY_RE.match(original)
+    if not match:
+        raise ValueError(f"Unknown key: {original!r}")
+
+    note = match.group("note").strip().title()
+    mode = match.group("mode")
+    major = True
+    if mode is not None:
+        major = mode.upper() == "MAJOR" or mode == "M"
+
+    pitch = PITCH_CLASSES.get(note)
+    if pitch is None:
+        raise ValueError(f"Unknown key: {original!r}")
+    return pitch, major
+
+
+def camelot(pitch: int, major: bool) -> str:
+    return f"{CAMLOT_MAJOR[pitch]}{'A' if major else 'B'}"
+
+
+def read_tag(tags: dict[str, object], keys: tuple[str, ...]) -> str | None:
+    for key in keys:
+        if key not in tags:
+            continue
+        raw = tags[key]
+        if isinstance(raw, list):
+            raw = raw[0] if raw else ""
+        value = str(raw).strip()
+        if value:
+            return value
+    return None
+
+
+def run_ffmpeg(*args: str) -> None:
+    cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", *args]
+    try:
+        subprocess.run(cmd, check=True, capture_output=True, text=True)
+    except subprocess.CalledProcessError as exc:
+        stderr = exc.stderr.strip() if exc.stderr else "(no stderr)"
+        error(f"ffmpeg failed: {stderr}")
+    except FileNotFoundError:
+        error("ffmpeg not found in PATH")
+
+
+def format_filter(sample_rate: int, channels: int) -> str:
+    layout = "stereo" if channels == 2 else f"{channels}c"
+    return (
+        f"aformat=sample_fmts=fltp:sample_rates={sample_rate}:channel_layouts={layout}"
+    )
+
+
+def load_tracks(directory: Path, recursive: bool = False) -> list[dict]:
+    if not directory.is_dir():
+        error(f"Not a directory: {directory}")
+
+    files = directory.rglob("*") if recursive else directory.iterdir()
+    tracks = []
+
+    for entry in sorted(files):
+        if not entry.is_file():
+            continue
+        try:
+            audio = mutagen.File(str(entry))
+        except Exception:
+            continue
+        if audio is None:
+            continue
+
+        tags = {k.upper(): v for k, v in audio.items()}
+
+        bpm_raw = read_tag(tags, BPM_TAGS)
+        key_raw = read_tag(tags, KEY_TAGS)
+        if bpm_raw is None or key_raw is None:
+            continue
+
+        try:
+            bpm = float(bpm_raw)
+        except ValueError:
+            continue
+        if bpm <= 0:
+            continue
+
+        try:
+            pitch, major = parse_key(key_raw)
+        except ValueError as exc:
+            LOGGER.warning("Skipping %s: %s", entry, exc)
+            continue
+
+        duration = getattr(getattr(audio, "info", None), "length", 0.0)
+        title = read_tag(tags, TITLE_TAGS) or entry.stem
+
+        tracks.append(
+            {
+                "path": entry.resolve(),
+                "bpm": bpm,
+                "key": key_raw,
+                "major": major,
+                "pitch": pitch,
+                "duration": duration,
+                "title": title,
+                "rel_path": str(entry.relative_to(directory)),
+            }
+        )
+
+    if not tracks:
+        error(f"No tracks with BPM+key found in {directory}")
+
+    return tracks
+
+
+def key_distance(a: dict, b: dict) -> int:
+    diff = abs((a["pitch"] - b["pitch"]) % 12)
+    diff = min(diff, 12 - diff)
+
+    if a["major"] != b["major"] and diff == 3:
+        return 0
+
+    return diff
+
+
+def bpm_distance(a: dict, b: dict) -> float:
+    return abs(a["bpm"] - b["bpm"]) / max(a["bpm"], b["bpm"])
+
+
+def score_transition(a: dict, b: dict, bpm_weight: float = 10.0) -> float:
+    return key_distance(a, b) * 2.0 + bpm_distance(a, b) * bpm_weight
+
+
+def build_set(
+    tracks: list[dict],
+    *,
+    first_track: str | None,
+    start_idx: int,
+    target_duration: float,
+    play_duration: float,
+    fade_duration: float,
+    bpm_window: float,
+) -> list[dict]:
+    pool = list(tracks)
+
+    if first_track is not None:
+        query = first_track.lower()
+        matches = [
+            i
+            for i, t in enumerate(pool)
+            if query in t["title"].lower()
+            or query in t["path"].name.lower()
+            or query in t["rel_path"].lower()
+        ]
+        if not matches:
+            error(f"No track matching {first_track!r} found")
+        first_idx = matches[0]
+    else:
+        first_idx = start_idx % len(pool)
+
+    first = pool.pop(first_idx)
+    anchor_bpm = first["bpm"]
+
+    pool = [t for t in pool if abs(t["bpm"] - anchor_bpm) <= bpm_window]
+    if not pool:
+        error(
+            f"No tracks within {bpm_window:.1f} BPM of anchor track "
+            f"({first['title']}: {anchor_bpm:.1f} BPM). "
+            f"Widen with --bpm-window."
+        )
+
+    sequence = [first]
+    first_play = min(play_duration, first["duration"])
+    first_fade_out = max(0.0, first_play - fade_duration)
+    segments = [
+        {
+            "track": first,
+            "start": 0.0,
+            "end": first_play,
+            "fade_out_start": first_fade_out,
+        }
+    ]
+    elapsed = first_fade_out
+
+    while elapsed < target_duration and pool:
+        last = sequence[-1]
+        best = min(pool, key=lambda candidate: score_transition(last, candidate))
+        pool.remove(best)
+        sequence.append(best)
+
+        play = min(play_duration, best["duration"])
+        start = elapsed
+        end = start + play
+        fade_out = max(0.0, end - fade_duration)
+        segments.append(
+            {
+                "track": best,
+                "start": start,
+                "end": end,
+                "fade_out_start": fade_out,
+            }
+        )
+        elapsed = fade_out
+
+    return segments
+
+
+def build_atempo_chain(ratio: float) -> list[str]:
+    stages = []
+    remaining = ratio
+    while remaining > 2.0:
+        stages.append("atempo=2.0")
+        remaining /= 2.0
+    while remaining < 0.5:
+        stages.append("atempo=0.5")
+        remaining /= 0.5
+    stages.append(f"atempo={remaining:.6f}")
+    return stages
+
+
+def fade_filters(
+    duration: float, fade_in: float, fade_out: float, curve: str
+) -> list[str]:
+    c = CURVES[curve]
+
+    fade_in = max(0.0, min(fade_in, duration / 2))
+    fade_out = max(0.0, min(fade_out, duration / 2))
+    fade_out_start = max(0.0, duration - fade_out)
+
+    return [
+        f"afade=t=in:ss=0:d={fade_in:.3f}:curve={c}",
+        f"afade=t=out:st={fade_out_start:.3f}:d={fade_out:.3f}:curve={c}",
+    ]
+
+
+def extract_segment(
+    src: Path, out: Path, duration: float, sample_rate: int, channels: int
+) -> None:
+    run_ffmpeg(
+        "-t",
+        f"{duration:.3f}",
+        "-i",
+        str(src),
+        "-ar",
+        str(sample_rate),
+        "-ac",
+        str(channels),
+        "-c:a",
+        "pcm_f32le",
+        str(out),
+    )
+
+
+def stretch_with_rubberband(src: Path, out: Path, ratio: float) -> None:
+    try:
+        subprocess.run(
+            ["rubberband", "-q", "-T", f"{ratio:.6f}", str(src), str(out)],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+    except subprocess.CalledProcessError as exc:
+        stderr = exc.stderr.strip() if exc.stderr else "(no stderr)"
+        error(f"rubberband failed: {stderr}")
+
+
+def render_segment(
+    segment: dict,
+    target_bpm: float,
+    *,
+    curve: str,
+    stretch_engine: str,
+    output_path: Path,
+    sample_rate: int,
+    channels: int,
+) -> None:
+    track = segment["track"]
+    play_duration = segment["end"] - segment["start"]
+    ratio = target_bpm / track["bpm"]
+    stretched_duration = play_duration * ratio
+    fade = min(play_duration / 2.0, play_duration - segment["fade_out_start"]) * ratio
+
+    if stretch_engine == "rubberband" and shutil.which("rubberband"):
+        with tempfile.TemporaryDirectory(prefix="dj_rb_") as tmpdir:
+            trimmed = Path(tmpdir) / "trimmed.wav"
+            extract_segment(
+                track["path"], trimmed, play_duration, sample_rate, channels
+            )
+            stretched = Path(tmpdir) / "stretched.wav"
+            stretch_with_rubberband(trimmed, stretched, ratio)
+            filters = fade_filters(stretched_duration, fade, fade, curve)
+            filters.append(format_filter(sample_rate, channels))
+            run_ffmpeg(
+                "-i",
+                str(stretched),
+                "-af",
+                ",".join(filters),
+                "-c:a",
+                "pcm_f32le",
+                str(output_path),
+            )
+        return
+
+    if stretch_engine == "rubberband":
+        LOGGER.warning(
+            "rubberband not found, falling back to atempo for %s", track["title"]
+        )
+
+    filters = [f"atrim=start=0.0:duration={play_duration:.3f}"]
+    if ratio != 1.0:
+        filters.extend(build_atempo_chain(ratio))
+    filters.extend(fade_filters(stretched_duration, fade, fade, curve))
+    filters.append(format_filter(sample_rate, channels))
+
+    run_ffmpeg(
+        "-t",
+        f"{play_duration:.3f}",
+        "-i",
+        str(track["path"]),
+        "-af",
+        ",".join(filters),
+        "-c:a",
+        "pcm_f32le",
+        str(output_path),
+    )
+
+
+def _mix_pair(
+    a_path: Path,
+    b_path: Path,
+    b_delay: float,
+    out_path: Path,
+    sample_rate: int,
+    channels: int,
+) -> None:
+    delay_ms = max(0, int(b_delay * 1000))
+    filter_complex = (
+        f"[1:a]adelay=delays={delay_ms}|{delay_ms}:all=1[delayed];"
+        f"[0:a][delayed]amix=inputs=2:duration=longest:normalize=0[sum];"
+        f"[sum]{format_filter(sample_rate, channels)}[out]"
+    )
+    run_ffmpeg(
+        "-i",
+        str(a_path),
+        "-i",
+        str(b_path),
+        "-filter_complex",
+        filter_complex,
+        "-map",
+        "[out]",
+        "-c:a",
+        "pcm_f32le",
+        str(out_path),
+    )
+
+
+def mix_segments(
+    segments: list[tuple[Path, float]],
+    output_path: Path,
+    *,
+    format_ext: str,
+    sample_rate: int,
+    channels: int,
+) -> None:
+    if not segments:
+        error("No segments to mix")
+
+    codec = CODECS.get(format_ext, "pcm_f32le")
+    quality = []
+    if codec == "libmp3lame":
+        quality = ["-q:a", "2"]
+    elif codec == "libvorbis":
+        quality = ["-q:a", "6"]
+
+    with tempfile.TemporaryDirectory(prefix="kundali_mix_") as tmpdir:
+        acc_path = segments[0][0]
+        for i in range(1, len(segments)):
+            seg_path, start = segments[i]
+            mix_path = Path(tmpdir) / f"mix_{i:03d}.wav"
+            _mix_pair(acc_path, seg_path, start, mix_path, sample_rate, channels)
+            acc_path = mix_path
+
+        run_ffmpeg(
+            "-i",
+            str(acc_path),
+            "-c:a",
+            codec,
+            "-ar",
+            str(sample_rate),
+            "-ac",
+            str(channels),
+            *quality,
+            str(output_path),
+        )
+
+
+def positive_float(value: str) -> float:
+    f = float(value)
+    if f <= 0:
+        raise argparse.ArgumentTypeError(f"{value!r} must be positive")
+    return f
+
+
+def nonnegative_float(value: str) -> float:
+    f = float(value)
+    if f < 0:
+        raise argparse.ArgumentTypeError(f"{value!r} must be non-negative")
+    return f
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Match songs based on their key and BPM and merge them into an audio file",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument(
+        "--dir",
+        type=Path,
+        default=Path("."),
+        help="Directory with analyzed audio files",
+    )
+    parser.add_argument(
+        "--recursive", action="store_true", help="Walk into subdirectories"
+    )
+    parser.add_argument(
+        "--out",
+        type=Path,
+        default=Path("dj_set.wav"),
+        help="Output file (wav/flac/mp3/ogg)",
+    )
+    parser.add_argument(
+        "--duration",
+        type=positive_float,
+        default=1800.0,
+        help="Target set length in seconds",
+    )
+    parser.add_argument(
+        "--start",
+        type=int,
+        default=0,
+        help="Index of first track (when --first-track is not given)",
+    )
+    parser.add_argument(
+        "--first-track",
+        type=str,
+        default=None,
+        help="Title/filename substring of the anchor track",
+    )
+    parser.add_argument(
+        "--target-bpm",
+        type=positive_float,
+        default=None,
+        help="Override the common BPM",
+    )
+    parser.add_argument(
+        "--play-duration",
+        type=positive_float,
+        default=240.0,
+        help="Max seconds per track",
+    )
+    parser.add_argument(
+        "--fade",
+        type=nonnegative_float,
+        default=16.0,
+        help="Crossfade duration in seconds",
+    )
+    parser.add_argument(
+        "--bpm-window",
+        type=nonnegative_float,
+        default=5.0,
+        help="Max BPM distance from the anchor track",
+    )
+    parser.add_argument(
+        "--fade-curve", choices=list(CURVES), default="qsin", help="Crossfade curve"
+    )
+    parser.add_argument(
+        "--stretch-engine",
+        choices=("atempo", "rubberband"),
+        default="atempo",
+        help="Time-stretch engine",
+    )
+    parser.add_argument(
+        "--sample-rate", type=int, default=SAMPLE_RATE, help="Output sample rate in Hz"
+    )
+    parser.add_argument(
+        "--channels", type=int, default=2, help="Output channel count (1 or 2)"
+    )
+    parser.add_argument(
+        "--save-plan", type=Path, default=None, help="Write JSON tracklist/plan"
+    )
+    parser.add_argument(
+        "-v", "--verbose", action="store_true", help="Enable debug logging"
+    )
+
+    if argv is None:
+        argv = sys.argv[1:]
+    if not argv:
+        parser.print_help(sys.stderr)
+        sys.exit(0)
+
+    return parser.parse_args(argv)
+
+
+def configure_logging(verbose: bool) -> None:
+    logging.basicConfig(
+        level=logging.DEBUG if verbose else logging.INFO,
+        format="%(levelname)s: %(message)s",
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    configure_logging(args.verbose)
+
+    tracks = load_tracks(args.dir, recursive=args.recursive)
+    tracks = list({t["path"]: t for t in tracks}.values())
+
+    segments = build_set(
+        tracks,
+        first_track=args.first_track,
+        start_idx=args.start,
+        target_duration=args.duration,
+        play_duration=args.play_duration,
+        fade_duration=args.fade,
+        bpm_window=args.bpm_window,
+    )
+
+    target_bpm = (
+        args.target_bpm if args.target_bpm is not None else segments[0]["track"]["bpm"]
+    )
+    format_ext = args.out.suffix.lstrip(".").lower() or "wav"
+
+    LOGGER.info("Found %d unique track(s)", len(tracks))
+    LOGGER.info("Anchor BPM: %.1f (window: ±%.1f)", target_bpm, args.bpm_window)
+    LOGGER.info("Sample rate: %d Hz", args.sample_rate)
+    LOGGER.info("Channels: %d", args.channels)
+    LOGGER.info("Fade curve: %s", args.fade_curve)
+    LOGGER.info("Stretch:    %s", args.stretch_engine)
+    LOGGER.info(
+        "Planned %d track(s) in set, total length ~%.1fs",
+        len(segments),
+        segments[-1]["end"],
+    )
+
+    print("\nSequence:")
+    for i, seg in enumerate(segments):
+        tr = seg["track"]
+        display = (
+            f"{tr['title']}  ({tr['rel_path']})"
+            if tr["title"] != tr["rel_path"]
+            else tr["title"]
+        )
+        print(
+            f"[{i:02d}] {tr['key']:12s} ({camelot(tr['pitch'], tr['major']):4s}) "
+            f"{tr['bpm']:5.1f} -> {target_bpm:5.1f}  {display}"
+        )
+    print()
+
+    if args.save_plan:
+        plan = {
+            "target_bpm": target_bpm,
+            "sample_rate": args.sample_rate,
+            "channels": args.channels,
+            "fade_curve": args.fade_curve,
+            "stretch_engine": args.stretch_engine,
+            "segments": [
+                {
+                    "title": seg["track"]["title"],
+                    "path": str(seg["track"]["path"]),
+                    "rel_path": seg["track"]["rel_path"],
+                    "key": seg["track"]["key"],
+                    "camelot": camelot(seg["track"]["pitch"], seg["track"]["major"]),
+                    "bpm": seg["track"]["bpm"],
+                    "start": round(seg["start"], 3),
+                    "end": round(seg["end"], 3),
+                    "fade_out_start": round(seg["fade_out_start"], 3),
+                }
+                for seg in segments
+            ],
+        }
+        args.save_plan.write_text(json.dumps(plan, indent=2), encoding="utf-8")
+        LOGGER.info("Saved plan to %s", args.save_plan)
+
+    with tempfile.TemporaryDirectory(prefix="kundali_") as tmpdir:
+        rendered = []
+        for i, seg in enumerate(segments):
+            seg_path = Path(tmpdir) / f"seg_{i:03d}.wav"
+            LOGGER.info("Rendering [%02d] %s ...", i, seg["track"]["title"])
+            render_segment(
+                seg,
+                target_bpm,
+                curve=args.fade_curve,
+                stretch_engine=args.stretch_engine,
+                output_path=seg_path,
+                sample_rate=args.sample_rate,
+                channels=args.channels,
+            )
+            rendered.append((seg_path, seg["start"]))
+
+        LOGGER.info("Mixing %d segments into %s ...", len(rendered), args.out)
+        mix_segments(
+            rendered,
+            args.out,
+            format_ext=format_ext,
+            sample_rate=args.sample_rate,
+            channels=args.channels,
+        )
+
+    LOGGER.info("Done: %s", args.out)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
blob - 5c72b307a54da09988d08b0292862662a1fa2689
blob + ae453e22d6af1a582268e915426b8de85bd8ca03
--- requirements.in
+++ requirements.in
@@ -1,3 +1,4 @@
 fire==0.7.1
 paramiko==5.0.0
 requests==2.33.1
+mutagen==1.48.1
blob - d4d9bc97cbea5dc96926fd208cfc3f662d0bd278
blob + be74600755d24041c59dcd06f0c06fbe9fffd199
--- requirements.txt
+++ requirements.txt
@@ -332,6 +332,10 @@ invoke==3.0.3 \
     --hash=sha256:437b6a622223824380bfb4e64f612711a6b648c795f565efc8625af66fb57f0c \
     --hash=sha256:f11327165e5cbb89b2ad1d88d3292b5113332c43b8553b494da435d6ec6f5053
     # via paramiko
+mutagen==1.48.1 \
+    --hash=sha256:4f077fe87d3fc7fba259aa63d8c026b18382ca6a42ef37c61e16f1b1b5b82fe7 \
+    --hash=sha256:8f95637ab9f6f305cec6bd1294e197debe207998e3e068596563c74f86b0a173
+    # via -r requirements.in
 paramiko==5.0.0 \
     --hash=sha256:36763b5b95c2a0dcfdf1abc48e48156ee425b21efe2f0e787c2dd5a95c0e5e79 \
     --hash=sha256:b7044611c30140d9a75261653210e2002977b71a0497ff3ba0d98d7edbf62f7c