commit - f9e7c66a868d68eae5ebe5b97123435b595a7559
commit + 3cd3ce9e524c330d17596563032a6baf23e0f70c
blob - 5814a3aa2d5347c5a386bc51f917417ee49a9e41
blob + 0c1d89aa383dc348fbcff9a31bebc86bc344e81b
--- README.md
+++ README.md
| [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
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
+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
+#!/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
fire==0.7.1
paramiko==5.0.0
requests==2.33.1
+mutagen==1.48.1
blob - d4d9bc97cbea5dc96926fd208cfc3f662d0bd278
blob + be74600755d24041c59dcd06f0c06fbe9fffd199
--- requirements.txt
+++ requirements.txt
--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
