diff --git a/README.md b/README.md index 2ef2bf3..e6b4e29 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,8 @@ uv run ableton-cli device insert Utility --main uv run ableton-cli device parameter set-display --main --device-name "Glue Compressor" --parameter-name Threshold -- -10 uv run ableton-cli audio loudness analyze --path ./renders/remix.wav --engine ffmpeg uv run ableton-cli audio spectrum analyze --path ./renders/remix.wav --profile anime-club +uv run ableton-cli audio spectrum shares --stem Kick=./caps/kick.wav --stem Lead=./caps/lead.wav --focus Lead +uv run ableton-cli session capture-tracks --track-index 0 --track-index 9 --start 304 --bars 8 --bypass-main-devices --analyze uv run ableton-cli audio transient analyze --path ./loops/drum-break.wav --bpm 174 --max-slices 16 uv run ableton-cli remix mastering target set --project ./proj/remix_project.json --profile anime-club-demo uv run ableton-cli remix mastering analyze --project ./proj/remix_project.json --render ./renders/remix.wav diff --git a/docs/mastering.md b/docs/mastering.md index b7d3e88..6c69351 100644 --- a/docs/mastering.md +++ b/docs/mastering.md @@ -35,6 +35,42 @@ each band over the whole file: `sub_20_60_hz`, `bass_60_120_hz`, `low_mid_120_40 the command needs ffmpeg and fails with `CONFIG_INVALID` without it. To compare sections or tracks, capture each one (`session capture`) and analyze the captures. +### Per-track band diagnosis + +```bash +uv run ableton-cli --output json session capture-tracks --track-index 0 --track-index 4 --track-index 9 --start 304 --bars 8 --bypass-main-devices --analyze --resolution third-octave --focus Lead +uv run ableton-cli --output json audio spectrum shares --stem Kick=./caps/kick.wav --stem Bass=./caps/bass.wav --resolution bands +``` + +`session capture-tracks` solos each listed track in turn and records the main output into a +temporary audio track (input Resampling, volume 0), which it deletes afterwards. The recorded +files stay in the Set's `Samples/Recorded` folder and are listed in `captures`. With +`--bypass-main-devices` it turns off every active main-track device first, so the master +chain does not color the measurement. Solos, the temporary track, and bypassed devices are +restored even when a capture fails. It refuses to start while any track is soloed, because +that track would leak into every capture. Captures run in real time: one bar of pre-roll +plus `--bars` at the Set's tempo, per track. + +Both capture commands record from exactly `--start`. Live ignores a playhead moved while +stopped, and firing a clip from a stopped transport restarts the song at beat 0, so the +capture starts playback, moves the playhead one bar before `--start`, and fires the clip: +Live's launch quantization (1 Bar) begins the recording at `--start`. `--start` must +therefore be at least one bar. Each capture reports `recorded_start_beat` (the stop position +minus the file's length) and fails with `step: verify_recorded_start` when the recording +began more than half a beat away from `--start`, e.g. under another launch quantization. + +`audio spectrum shares` (and `--analyze`) measures every stem in each band and reports: + +- `shares_pct`: each stem's share of the band's summed energy over the whole window. A stem + that rests for part of the window reads low, so check its level while it plays. +- `dominant`: the stem with the largest share (`null` when the band is silent). +- `band_totals_dbfs`: the summed level, ignoring phase between stems. +- `focus.margin_db` with `--focus LABEL`: the focus stem's level minus the loudest other + stem's, so a negative value is the gap to close. It is `null` where either side is silent. + +`--resolution third-octave` uses the 29 ISO third-octave bands from 25 Hz to 16 kHz (named +like `1250_hz`) to place an EQ move; `bands` uses the seven analysis bands above. + Spectrum warnings judge tonal balance, not loudness: each band is compared, relative to the file's full-band level, with the profile's expected share. A band more than 6 dB above its share is `_hot` and one more than 8 dB below is `_low`; a file with no diff --git a/skills/ableton-cli/SKILL.md b/skills/ableton-cli/SKILL.md index 9fa6438..0ac1a4c 100644 --- a/skills/ableton-cli/SKILL.md +++ b/skills/ableton-cli/SKILL.md @@ -60,7 +60,8 @@ uv run ableton-cli session diff --from ./snapshot-before.json --to ./snapshot-af uv run ableton-cli session watch --interval-ms 500 --scope all --count 5 uv run ableton-cli session events --events tempo,is_playing --count 5 uv run ableton-cli session events --idle-timeout-ms 5000 -uv run ableton-cli session capture --track-index 0 --slot 0 --bars 8 --set-routing --analyze +uv run ableton-cli session capture --track-index 0 --slot 0 --bars 8 --start 32 --set-routing --analyze +uv run ableton-cli session capture-tracks --track-index 0 --track-index 3 --start 304 --bars 8 --bypass-main-devices --analyze --focus Lead uv run ableton-cli session stop-all-clips ``` @@ -137,6 +138,7 @@ uv run ableton-cli audio beatgrid import --project ./proj/remix_project.json --d uv run ableton-cli audio analyze --project ./proj/remix_project.json --detect bpm,key --manual-bpm 172 --manual-key "A minor" uv run ableton-cli audio loudness analyze --path ./renders/remix.wav --engine ffmpeg --report-out ./proj/reports/loudness.json uv run ableton-cli audio spectrum analyze --path ./renders/remix.wav --profile anime-club --report-out ./proj/reports/spectrum.json +uv run ableton-cli audio spectrum shares --stem Kick=./caps/kick.wav --stem Lead=./caps/lead.wav --resolution third-octave --focus Lead uv run ableton-cli audio reference compare --candidate ./renders/remix.wav --reference ./refs/reference.wav --metrics loudness,spectrum,stereo uv run ableton-cli remix set-target --project ./proj/remix_project.json --bpm 174 --key "F minor" uv run ableton-cli remix plan --project ./proj/remix_project.json --style anime-club @@ -174,6 +176,17 @@ Mastering workflow is measurement-first: - Run `audio reference compare` when a reference track is registered. - Do not chase LUFS by increasing limiter pressure when true peak is close to ceiling. - Treat large spectrum deltas as mix/stem issues before using master EQ compensation. +- To find which track owns a band, run `session capture-tracks` per section (it solos each + track into a temporary capture track and restores solos, the temporary track, and any + bypassed main devices even on failure), then read `shares.shares_pct` and + `shares.focus.margin_db`. A negative margin means another track is louder there. + Shares average the whole window, so a track that rests for part of it reads low; check + its level while it plays before raising it. +- `session capture` and `session capture-tracks` need `--start` (in beats) at least one bar + in: playback starts first, the playhead moves one bar before `--start`, and launch + quantization begins the recording at `--start`. The result reports `recorded_start_beat`; + a recording that began elsewhere (launch quantization other than 1 Bar) fails with + `step: verify_recorded_start`. - Run `remix mastering apply --dry-run` before `--yes`. - Pick device/preset targets from `browser search`; do not assume fixed URI availability. - If a standard effect wrapper fails on locale/device-variant parameter names, switch to generic parameter list/set commands. diff --git a/src/ableton_cli/audio_analysis/__init__.py b/src/ableton_cli/audio_analysis/__init__.py index ac7aa05..db855bc 100644 --- a/src/ableton_cli/audio_analysis/__init__.py +++ b/src/ableton_cli/audio_analysis/__init__.py @@ -1,5 +1,5 @@ from .compare import compare_reference from .loudness import analyze_loudness -from .spectrum import analyze_spectrum +from .spectrum import analyze_spectrum, spectrum_shares -__all__ = ["analyze_loudness", "analyze_spectrum", "compare_reference"] +__all__ = ["analyze_loudness", "analyze_spectrum", "compare_reference", "spectrum_shares"] diff --git a/src/ableton_cli/audio_analysis/io.py b/src/ableton_cli/audio_analysis/io.py index 6df23c9..d140537 100644 --- a/src/ableton_cli/audio_analysis/io.py +++ b/src/ableton_cli/audio_analysis/io.py @@ -66,6 +66,32 @@ def require_ffmpeg_engine(engine: str) -> str: return parsed +def wav_channels(path: str | Path) -> int: + """Read only the channel count of a PCM WAV file, without decoding its samples.""" + audio_path = require_existing_audio_path(path) + try: + with wave.open(str(audio_path), "rb") as wav: + return wav.getnchannels() + except wave.Error as exc: + raise audio_error( + message=f"unsupported audio format for builtin analysis: {audio_path}", + hint="Use PCM WAV for deterministic offline analysis.", + ) from exc + + +def wav_duration_sec(path: str | Path) -> float: + """Read a PCM WAV file's duration from its header, without decoding its samples.""" + audio_path = require_existing_audio_path(path) + try: + with wave.open(str(audio_path), "rb") as wav: + return wav.getnframes() / wav.getframerate() + except wave.Error as exc: + raise audio_error( + message=f"unsupported audio format for builtin analysis: {audio_path}", + hint="Use PCM WAV for deterministic offline analysis.", + ) from exc + + def read_wav(path: str | Path) -> AudioData: audio_path = require_existing_audio_path(path) try: diff --git a/src/ableton_cli/audio_analysis/spectrum.py b/src/ableton_cli/audio_analysis/spectrum.py index 162ca92..082d54b 100644 --- a/src/ableton_cli/audio_analysis/spectrum.py +++ b/src/ableton_cli/audio_analysis/spectrum.py @@ -6,9 +6,19 @@ from typing import Any from .ffmpeg import Runner, ffmpeg_output_error, run_ffmpeg -from .io import read_wav, require_ffmpeg_engine, rms_dbfs, write_report +from .io import ( + audio_error, + read_wav, + require_existing_audio_path, + require_ffmpeg_engine, + rms_dbfs, + wav_channels, + write_report, +) + +Band = tuple[str, float, float] -BANDS: tuple[tuple[str, float, float], ...] = ( +BANDS: tuple[Band, ...] = ( ("sub_20_60_hz", 20.0, 60.0), ("bass_60_120_hz", 60.0, 120.0), ("low_mid_120_400_hz", 120.0, 400.0), @@ -18,11 +28,34 @@ ("air_10000_18000_hz", 10000.0, 18000.0), ) +_THIRD_OCTAVE_NOMINAL_HZ = ( + "25", "31.5", "40", "50", "63", "80", "100", "125", "160", "200", + "250", "315", "400", "500", "630", "800", "1000", "1250", "1600", "2000", + "2500", "3150", "4000", "5000", "6300", "8000", "10000", "12500", "16000", +) # fmt: skip + + +def _third_octave_bands() -> tuple[Band, ...]: + # Base-2 ISO centers 1000 * 2^(k/3) named by their nominal frequency; adjacent bands + # share an exact edge. + first = _THIRD_OCTAVE_NOMINAL_HZ.index("1000") + edges = [ + 1000.0 * 2 ** ((index - first - 0.5) / 3) + for index in range(len(_THIRD_OCTAVE_NOMINAL_HZ) + 1) + ] + return tuple( + (f"{nominal}_hz", edges[index], edges[index + 1]) + for index, nominal in enumerate(_THIRD_OCTAVE_NOMINAL_HZ) + ) + + +THIRD_OCTAVE_BANDS: tuple[Band, ...] = _third_octave_bands() +RESOLUTIONS: dict[str, tuple[Band, ...]] = {"bands": BANDS, "third-octave": THIRD_OCTAVE_BANDS} + # Bands are cut brick-wall in the FFT domain: biquad slopes sag inside one-octave bands. -# 8192 points at 48 kHz gives 5.9 Hz bins, fine enough for the 20-60 Hz band. -_FFT_WINDOW = 8192 -_RMS_PATTERN = re.compile(r"RMS level dB:\s*(-?\d+(?:\.\d+)?|-inf)") -_HINT = "Check that ffmpeg provides the highpass, lowpass, and astats filters." +# 16384 points at 48 kHz gives 2.9 Hz bins, enough for the 22-28 Hz third-octave band. +_FFT_WINDOW = 16384 +_HINT = "Check that ffmpeg provides the asplit, afftfilt, astats, and amix filters." # Each band's expected share of the file's full-band (20 Hz-18 kHz) level, in dB. The # anime-club-balanced shape integrates a spectrum whose energy per octave falls 1.5 dB per @@ -105,34 +138,52 @@ def measure_band_levels( path: str | Path, *, channels: int, + bands: tuple[Band, ...] = BANDS, run: Runner | None = None, ) -> dict[str, float]: """Return each band's RMS level (dBFS) of the mono downmix over the whole file.""" + command = [ + "ffmpeg", + "-nostdin", + "-nostats", + "-hide_banner", + "-i", + str(path), + "-filter_complex", + _band_filter_graph(channels, bands), + "-map", + "[out]", + "-f", + "null", + "-", + ] + return _parse_band_levels(run_ffmpeg(command, run=run, hint=_HINT), bands) + + +def _band_filter_graph(channels: int, bands: tuple[Band, ...]) -> str: + """One pass: split the downmix into a branch per band, each read by ``astats@b``.""" downmix = "+".join(f"{1 / channels:g}*c{index}" for index in range(channels)) - levels: dict[str, float] = {} - for name, low, high in BANDS: + branches = "".join(f"[s{index}]" for index in range(len(bands))) + graph = [f"[0:a]pan=mono|c0={downmix},asplit={len(bands)}{branches}"] + for index, (_name, low, high) in enumerate(bands): in_band = f"between(b*sr/(2*(nb-1)),{low:g},{high:g})" - band_filter = ( - f"afftfilt=real='re*{in_band}':imag='im*{in_band}':win_size={_FFT_WINDOW}:overlap=0.75" + graph.append( + f"[s{index}]afftfilt=real='re*{in_band}':imag='im*{in_band}'" + f":win_size={_FFT_WINDOW}:overlap=0.75," + f"astats@b{index}=measure_perchannel=none:measure_overall=RMS_level[o{index}]" ) - graph = ( - f"pan=mono|c0={downmix},{band_filter}," - "astats=measure_perchannel=none:measure_overall=RMS_level" + outputs = "".join(f"[o{index}]" for index in range(len(bands))) + graph.append(f"{outputs}amix=inputs={len(bands)}[out]") + return ";".join(graph) + + +def _parse_band_levels(log: str, bands: tuple[Band, ...]) -> dict[str, float]: + # ffmpeg prints each named astats instance at uninit, in no guaranteed order. + levels: dict[str, float] = {} + for index, (name, _low, _high) in enumerate(bands): + match = re.search( + rf"\[astats@b{index} @ [^\]]*\] RMS level dB:\s*(-?\d+(?:\.\d+)?|-inf)", log ) - command = [ - "ffmpeg", - "-nostdin", - "-nostats", - "-hide_banner", - "-i", - str(path), - "-af", - graph, - "-f", - "null", - "-", - ] - match = _RMS_PATTERN.search(run_ffmpeg(command, run=run, hint=_HINT)) if match is None: raise ffmpeg_output_error( message=f"ffmpeg astats output has no RMS level for {name}", @@ -143,6 +194,126 @@ def measure_band_levels( return levels +def spectrum_shares( + stems: list[tuple[str, str | Path]], + *, + resolution: str = "bands", + focus: str | None = None, + report_out: str | Path | None = None, +) -> dict[str, Any]: + """Split each band's energy between stems, e.g. solo captures of a mix's tracks.""" + bands = _validate_shares_request(stems, resolution=resolution, focus=focus) + require_ffmpeg_engine("ffmpeg") + measured = [ + ( + label, + Path(path).expanduser(), + measure_band_levels(path, channels=wav_channels(path), bands=bands), + ) + for label, path in stems + ] + result = _shares_result(measured, bands, resolution=resolution) + if focus is not None: + result["focus"] = {"label": focus, "margin_db": _focus_margins(measured, focus, bands)} + write_report(report_out, result) + return result + + +def _shares_result( + measured: list[tuple[str, Path, dict[str, float]]], + bands: tuple[Band, ...], + *, + resolution: str, +) -> dict[str, Any]: + powers = { + name: {label: 10 ** (levels[name] / 10.0) for label, _path, levels in measured} + for name, _low, _high in bands + } + return { + "resolution": resolution, + "bands": [ + {"name": name, "low_hz": round(low, 1), "high_hz": round(high, 1)} + for name, low, high in bands + ], + "stems": [ + { + "label": label, + "path": str(path), + "levels": {name: round(level, 3) for name, level in levels.items()}, + } + for label, path, levels in measured + ], + "band_totals_dbfs": {name: _power_dbfs(sum(p.values())) for name, p in powers.items()}, + "shares_pct": {name: _shares_pct(by_label) for name, by_label in powers.items()}, + "dominant": {name: _dominant(by_label) for name, by_label in powers.items()}, + } + + +def _power_dbfs(power: float) -> float: + return round(10.0 * math.log10(power), 3) if power > 0 else float("-inf") + + +def _shares_pct(by_label: dict[str, float]) -> dict[str, float]: + total = sum(by_label.values()) + return { + label: round(100.0 * power / total, 1) if total > 0 else 0.0 + for label, power in by_label.items() + } + + +def _dominant(by_label: dict[str, float]) -> str | None: + """The stem with the most energy in the band, or None when the band is silent.""" + if sum(by_label.values()) <= 0: + return None + return max(by_label, key=by_label.__getitem__) + + +def _validate_shares_request( + stems: list[tuple[str, str | Path]], *, resolution: str, focus: str | None +) -> tuple[Band, ...]: + if resolution not in RESOLUTIONS: + raise audio_error( + message=f"unsupported spectrum resolution: {resolution}", + hint=f"Use one of: {', '.join(RESOLUTIONS)}.", + ) + if not stems: + raise audio_error( + message="spectrum shares needs at least one stem", + hint="Pass --stem LABEL=PATH once per stem.", + ) + labels = [label for label, _path in stems] + for label in labels: + if labels.count(label) > 1: + raise audio_error( + message=f"duplicate stem label: {label}", + hint="Give every --stem a unique LABEL.", + ) + for _label, path in stems: + require_existing_audio_path(path) + if focus is not None and focus not in labels: + raise audio_error( + message=f"focus label is not a stem: {focus}", + hint=f"Use one of: {', '.join(labels)}.", + ) + return RESOLUTIONS[resolution] + + +def _focus_margins( + measured: list[tuple[str, Path, dict[str, float]]], focus: str, bands: tuple[Band, ...] +) -> dict[str, float | None]: + """Focus stem level minus the loudest other stem; None when either side is silent.""" + margins: dict[str, float | None] = {} + for name, _low, _high in bands: + focus_level = next(levels[name] for label, _path, levels in measured if label == focus) + others = [levels[name] for label, _path, levels in measured if label != focus] + loudest_other = max(others, default=float("-inf")) + if math.isinf(focus_level) or math.isinf(loudest_other): + margins[name] = None + else: + margins[name] = round(focus_level - loudest_other, 3) + return margins + + def _stereo_metrics(data: Any) -> dict[str, Any]: if data.channels < 2: return { diff --git a/src/ableton_cli/capture.py b/src/ableton_cli/capture.py index f8f6245..e9210df 100644 --- a/src/ableton_cli/capture.py +++ b/src/ableton_cli/capture.py @@ -1,6 +1,11 @@ """Session capture workflow: compose existing primitives (routing, arm, record, transport, analysis) into one "hear what you composed" command. +``capture_tracks`` repeats that capture once per track with the track soloed, +into a temporary audio track, so each track's contribution to a section can be +measured on its own (``spectrum_shares``). Every solo, temporary track, and +bypassed main device it touches is restored even when a capture fails. + ``capture_session`` takes an injected ``client`` so it is unit-testable with fakes (see ``tests/test_capture.py``), per the dependency-injection style used by ``warp_conform.conform_session_clip_warp``. Each step is a @@ -11,15 +16,20 @@ from __future__ import annotations import time -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass from typing import Any -from .audio_analysis import analyze_loudness, analyze_spectrum +from .audio_analysis import analyze_loudness, analyze_spectrum, spectrum_shares +from .audio_analysis.io import wav_duration_sec +from .audio_analysis.spectrum import RESOLUTIONS from .errors import AppError, ErrorCode, ExitCode from .remix.mastering import run_mastering_qa _WAIT_PAD_SECONDS = 1.0 -_BEATS_PER_BAR = 4.0 +# Commands reach Live a fraction of a beat apart, so the measured start is never exact. +_RECORDED_START_TOLERANCE_BEATS = 0.5 def _invalid_argument(message: str, hint: str, *, step: str, **details: Any) -> AppError: @@ -38,14 +48,17 @@ def capture_session( track: int, slot: int, bars: float, - start: float = 0.0, + start: float, set_routing: bool = False, analyze: bool = False, qa_project: str | None = None, wait_fn: Callable[[float], None] | None = None, + recorded_seconds_fn: Callable[[str], float] | None = None, ) -> dict[str, Any]: if wait_fn is None: wait_fn = time.sleep + if recorded_seconds_fn is None: + recorded_seconds_fn = wav_duration_sec if bars <= 0: raise _invalid_argument( message=f"bars must be > 0, got {bars}", @@ -88,11 +101,6 @@ def capture_session( # let Live auto-select its default for Resampling. client.track_routing_input_set(track_ref, "Resampling", None) - client.track_arm_set(track_ref, True) - client.transport_position_set(start) - client.fire_clip(track, slot) - client.transport_play() - song = client.song_info() tempo = float(song.get("tempo", 0.0)) if tempo <= 0: @@ -101,11 +109,32 @@ def capture_session( hint="Ensure Ableton Live reports a valid tempo before capturing.", step="read_tempo", ) - wait_seconds = bars * _BEATS_PER_BAR * 60.0 / tempo - wait_fn(wait_seconds + _WAIT_PAD_SECONDS) + signature = song.get("signature", {}) + beats_per_bar = ( + float(signature.get("numerator", 4)) * 4.0 / float(signature.get("denominator", 4)) + ) + if start < beats_per_bar: + raise _invalid_argument( + message=f"start must leave one bar of pre-roll, got {start} beats", + hint=f"Use --start >= {beats_per_bar:g} (one bar); recording begins exactly there.", + step="validate_start", + ) + + # Live ignores a playhead moved while stopped, and firing a clip from a stopped + # transport restarts the song at 0. So start playback, then move the playhead one + # bar before start and fire: launch quantization begins the recording at start. + client.track_arm_set(track_ref, True) + client.transport_play() + client.transport_position_set(start - beats_per_bar) + client.fire_clip(track, slot) + + seconds_per_bar = beats_per_bar * 60.0 / tempo + wait_seconds = bars * seconds_per_bar + wait_fn(seconds_per_bar + wait_seconds + _WAIT_PAD_SECONDS) client.stop_clip(track, slot) client.transport_stop() + stop_beat = float(client.song_info().get("current_time", 0.0)) file_info = client.clip_file_path_get(track, slot) file_path = file_info.get("file_path") @@ -116,12 +145,24 @@ def capture_session( step="read_file_path", ) + recorded_start = round(stop_beat - recorded_seconds_fn(file_path) * tempo / 60.0, 3) + if abs(recorded_start - start) > _RECORDED_START_TOLERANCE_BEATS: + raise _invalid_argument( + message=f"recording began at beat {recorded_start}, not at start beat {start}", + hint="Set Live's global launch quantization to 1 Bar, then capture again.", + step="verify_recorded_start", + recorded_start_beat=recorded_start, + requested_start_beat=start, + file_path=file_path, + ) + result: dict[str, Any] = { "track": track, "slot": slot, "bars": bars, "tempo": tempo, "wait_seconds": wait_seconds, + "recorded_start_beat": recorded_start, "file_path": file_path, } @@ -133,3 +174,166 @@ def capture_session( result["qa"] = run_mastering_qa(qa_project, render=file_path) return result + + +_MAIN_TRACK = {"mode": "main"} + + +@dataclass(frozen=True, slots=True) +class ShareAnalysis: + """Band-share analysis to run on the captures (``spectrum_shares`` arguments).""" + + resolution: str = "bands" + focus: str | None = None + + +def capture_tracks( + client: Any, + *, + tracks: list[int], + bars: float, + start: float, + bypass_main_devices: bool = False, + analysis: ShareAnalysis | None = None, + wait_fn: Callable[[float], None] | None = None, +) -> dict[str, Any]: + _validate_analysis(analysis) + names = _validate_capture_tracks(client, tracks) + labels = _track_labels(tracks, names) + + with _main_devices_bypassed(client, enabled=bypass_main_devices) as bypassed: + captures = [ + { + "track_index": track, + "track_name": names[track], + "label": labels[track], + "file_path": _capture_soloed( + client, track=track, bars=bars, start=start, wait_fn=wait_fn + ), + } + for track in tracks + ] + + result: dict[str, Any] = { + "start": start, + "bars": bars, + "bypassed_main_devices": bypassed, + "captures": captures, + } + if analysis is not None: + result["shares"] = spectrum_shares( + [(capture["label"], capture["file_path"]) for capture in captures], + resolution=analysis.resolution, + focus=analysis.focus, + ) + return result + + +@contextmanager +def _main_devices_bypassed(client: Any, *, enabled: bool) -> Iterator[list[int]]: + """Turn off each active main device for the block, then turn them back on. + + A device is recorded as soon as it is off, so a failure part-way still restores it. + """ + bypassed: list[int] = [] + try: + if enabled: + _turn_off_active_main_devices(client, bypassed) + yield bypassed + finally: + for index in bypassed: + client.device_active_set(_MAIN_TRACK, {"mode": "index", "index": index}, True) + + +def _turn_off_active_main_devices(client: Any, bypassed: list[int]) -> None: + for device in client.master_devices_list().get("devices", []): + device_ref = {"mode": "index", "index": device["index"]} + if client.device_active_get(_MAIN_TRACK, device_ref).get("active"): + client.device_active_set(_MAIN_TRACK, device_ref, False) + bypassed.append(device["index"]) + + +def _validate_analysis(analysis: ShareAnalysis | None) -> None: + if analysis is not None and analysis.resolution not in RESOLUTIONS: + raise _invalid_argument( + message=f"unsupported spectrum resolution: {analysis.resolution}", + hint=f"Use one of: {', '.join(RESOLUTIONS)}.", + step="validate_resolution", + ) + + +def _validate_capture_tracks(client: Any, tracks: list[int]) -> dict[int, str]: + _require_unique_tracks(tracks) + listing = client.tracks_list().get("tracks", []) + names = {int(item["index"]): str(item["name"]) for item in listing} + for track in tracks: + if track not in names: + raise _invalid_argument( + message=f"track index {track} does not exist", + hint="Use 'tracks list' to find track indexes.", + step="validate_tracks", + ) + soloed = [str(item["name"]) for item in listing if item.get("solo")] + if soloed: + raise _invalid_argument( + message="tracks are already soloed; a solo capture would include them", + hint="Unsolo every track, then run capture-tracks again.", + step="check_solo", + soloed_tracks=soloed, + ) + return names + + +def _require_unique_tracks(tracks: list[int]) -> None: + if not tracks: + raise _invalid_argument( + message="capture-tracks needs at least one --track-index", + hint="Pass --track-index once per track to capture.", + step="validate_tracks", + ) + for track in tracks: + if tracks.count(track) > 1: + raise _invalid_argument( + message=f"duplicate track index: {track}", + hint="Pass each --track-index once.", + step="validate_tracks", + ) + + +def _track_labels(tracks: list[int], names: dict[int, str]) -> dict[int, str]: + chosen = [names[track] for track in tracks] + return { + track: names[track] if chosen.count(names[track]) == 1 else f"{track}:{names[track]}" + for track in tracks + } + + +def _capture_soloed( + client: Any, + *, + track: int, + bars: float, + start: float, + wait_fn: Callable[[float], None] | None, +) -> str: + track_ref = {"mode": "index", "index": track} + client.track_solo_set(track_ref, True) + try: + temporary = int(client.create_audio_track(-1)["index"]) + try: + # The capture track's own output stays silent so it never feeds the resample. + client.track_volume_set({"mode": "index", "index": temporary}, 0.0) + capture = capture_session( + client, + track=temporary, + slot=0, + bars=bars, + start=start, + set_routing=True, + wait_fn=wait_fn, + ) + finally: + client.tracks_delete(temporary) + finally: + client.track_solo_set(track_ref, False) + return str(capture["file_path"]) diff --git a/src/ableton_cli/command_registry.py b/src/ableton_cli/command_registry.py index b080f1c..9cd706f 100644 --- a/src/ableton_cli/command_registry.py +++ b/src/ableton_cli/command_registry.py @@ -108,6 +108,7 @@ def _d(command_name: str, remote_command: str | None, side_effect: SideEffectSpe _d("audio reference compare", None, _WRITE), _d("audio sections import", None, _WRITE), _d("audio spectrum analyze", None, _WRITE), + _d("audio spectrum shares", None, _WRITE), _d("audio stems list", None, _READ), _d("audio stems split", None, _WRITE), _d("audio transient analyze", None, _WRITE), @@ -284,6 +285,7 @@ def _d(command_name: str, remote_command: str | None, side_effect: SideEffectSpe _d("scenes move", "scenes_move", _WRITE), _d("scenes name set", "set_scene_name", _WRITE), _d("session capture", None, _WRITE), + _d("session capture-tracks", None, _WRITE), _d("session diff", None, _READ), _d("session events", None, _WRITE), _d("session info", "get_session_info", _READ), diff --git a/src/ableton_cli/commands/_audio_spectrum_commands.py b/src/ableton_cli/commands/_audio_spectrum_commands.py index d681a69..0d4b764 100644 --- a/src/ableton_cli/commands/_audio_spectrum_commands.py +++ b/src/ableton_cli/commands/_audio_spectrum_commands.py @@ -5,8 +5,9 @@ import typer -from ..audio_analysis import analyze_spectrum +from ..audio_analysis import analyze_spectrum, spectrum_shares from ..runtime import execute_command +from ._validation import invalid_argument def register(spectrum_app: typer.Typer) -> None: @@ -27,6 +28,70 @@ def audio_spectrum_analyze( action=lambda: analyze_spectrum(path, profile=profile, report_out=report_out), ) + _register_shares(spectrum_app) + + +def _register_shares(spectrum_app: typer.Typer) -> None: + @spectrum_app.command("shares") + def audio_spectrum_shares( + ctx: typer.Context, + stems: Annotated[ + list[str] | None, + typer.Option( + "--stem", + help="Stem as LABEL=PATH (repeatable), e.g. a solo capture of one track", + ), + ] = None, + resolution: Annotated[ + str, + typer.Option("--resolution", help="Band set: bands or third-octave"), + ] = "bands", + focus: Annotated[ + str | None, + typer.Option( + "--focus", help="Stem label to compare against the loudest other stem per band" + ), + ] = None, + report_out: Annotated[ + Path | None, + typer.Option("--report-out", help="Write JSON report to this path"), + ] = None, + ) -> None: + values = stems or [] + execute_command( + ctx, + command="audio spectrum shares", + args={ + "stems": [_describe_stem(value) for value in values], + "resolution": resolution, + "focus": focus, + "report_out": _path_or_none(report_out), + }, + action=lambda: spectrum_shares( + [_parse_stem(value) for value in values], + resolution=resolution, + focus=focus, + report_out=report_out, + ), + ) + + +def _describe_stem(value: str) -> dict[str, str | None]: + label, separator, path = value.partition("=") + if not separator: + return {"label": None, "path": value} + return {"label": label, "path": path} + + +def _parse_stem(value: str) -> tuple[str, str]: + label, separator, path = value.partition("=") + if not separator or not label or not path: + raise invalid_argument( + message=f"--stem must be LABEL=PATH, got {value!r}", + hint="Pass each stem as --stem LABEL=PATH, e.g. --stem Kick=./kick.wav.", + ) + return label, path + def _path_or_none(path: Path | None) -> str | None: return None if path is None else str(path) diff --git a/src/ableton_cli/commands/session.py b/src/ableton_cli/commands/session.py index 75ec798..5bb80e5 100644 --- a/src/ableton_cli/commands/session.py +++ b/src/ableton_cli/commands/session.py @@ -7,8 +7,9 @@ import typer -from ..capture import capture_session +from ..capture import ShareAnalysis, capture_session, capture_tracks from ..client.events import EventStream +from ..errors import read_only_violation from ..runtime import execute_command, get_client, get_runtime from ..session_diff import compute_session_diff from ..watch import WATCH_SCOPES, run_watch_loop @@ -154,8 +155,12 @@ def session_capture( ], bars: Annotated[float, typer.Option("--bars", help="Number of bars to capture")], start: Annotated[ - float, typer.Option("--start", help="Transport start position in beats") - ] = 0.0, + float, + typer.Option( + "--start", + help="Beat where recording begins; needs one bar of pre-roll (>= one bar)", + ), + ], set_routing: Annotated[ bool, typer.Option( @@ -204,6 +209,76 @@ def _run() -> dict[str, object]: ) +@session_app.command("capture-tracks") +def session_capture_tracks( + ctx: typer.Context, + start: Annotated[ + float, + typer.Option( + "--start", + help="Beat where recording begins; needs one bar of pre-roll (>= one bar)", + ), + ], + track_indexes: Annotated[ + list[int] | None, + typer.Option("--track-index", help="Track index (0-based) to solo and capture; repeatable"), + ] = None, + bars: Annotated[float, typer.Option("--bars", help="Number of bars to capture")] = 8.0, + bypass_main_devices: Annotated[ + bool, + typer.Option( + "--bypass-main-devices", + help="Turn off active main-track devices while capturing, then restore them", + ), + ] = False, + analyze: Annotated[ + bool, typer.Option("--analyze", help="Report each band's share per captured track") + ] = False, + resolution: Annotated[ + str, typer.Option("--resolution", help="Band set for --analyze: bands or third-octave") + ] = "bands", + focus: Annotated[ + str | None, + typer.Option("--focus", help="Track label to compare against the loudest other track"), + ] = None, +) -> None: + tracks = track_indexes or [] + + def _run() -> dict[str, object]: + if get_runtime(ctx).read_only: + # Checked up front: the first client call is a read, and the capture writes. + raise read_only_violation("session capture-tracks") + if focus is not None and not analyze: + raise invalid_argument( + message="--focus requires --analyze", hint="Add --analyze, or drop --focus." + ) + return capture_tracks( + get_client(ctx), + tracks=[require_track_index(track) for track in tracks], + bars=require_positive_float("bars", bars, hint="Use a positive --bars value."), + start=require_non_negative_float( + "start", start, hint="Use a non-negative --start value in beats." + ), + bypass_main_devices=bypass_main_devices, + analysis=ShareAnalysis(resolution=resolution, focus=focus) if analyze else None, + ) + + execute_command( + ctx, + command="session capture-tracks", + args={ + "track_indexes": tracks, + "bars": bars, + "start": start, + "bypass_main_devices": bypass_main_devices, + "analyze": analyze, + "resolution": resolution, + "focus": focus, + }, + action=_run, + ) + + def _validate_watch_interval(interval_ms: int) -> None: if interval_ms < 50: raise invalid_argument( diff --git a/tests/conftest.py b/tests/conftest.py index c95523e..c0fb15a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -43,8 +43,8 @@ def measure_loudness(path: str | Path) -> Ebur128Measurement: short_term_lufs_max=round(peak_dbfs - 3.0, 1), ) - def measure_bands(path: str | Path, *, channels: int) -> dict[str, float]: - return {name: -20.0 for name, _low, _high in BANDS} + def measure_bands(path: str | Path, *, channels: int, bands: tuple = BANDS) -> dict[str, float]: + return {name: -20.0 for name, _low, _high in bands} for module in ("loudness", "spectrum"): monkeypatch.setattr( diff --git a/tests/snapshots/command_surface_snapshot.json b/tests/snapshots/command_surface_snapshot.json index 9e8b092..632c552 100644 --- a/tests/snapshots/command_surface_snapshot.json +++ b/tests/snapshots/command_surface_snapshot.json @@ -3760,6 +3760,7 @@ "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f", "\u256d\u2500 Commands \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256e", "\u2502 analyze \u2502", + "\u2502 shares \u2502", "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f" ], "hidden": false, @@ -3772,7 +3773,8 @@ ], "short_help": null, "subcommands": [ - "analyze" + "analyze", + "shares" ] }, "audio spectrum analyze": { @@ -3879,6 +3881,137 @@ ], "short_help": null }, + "audio spectrum shares": { + "deprecated": false, + "help": null, + "help_text": [ + "", + " Usage: shares [OPTIONS]", + "", + "\u256d\u2500 Options \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256e", + "\u2502 --stem TEXT Stem as LABEL=PATH (repeatable), e.g. a solo capture of one track \u2502", + "\u2502 --resolution TEXT Band set: bands or third-octave [default: bands] \u2502", + "\u2502 --focus TEXT Stem label to compare against the loudest other stem per band \u2502", + "\u2502 --report-out PATH Write JSON report to this path \u2502", + "\u2502 --help Show this message and exit. \u2502", + "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f" + ], + "hidden": false, + "kind": "command", + "no_args_is_help": false, + "params": [ + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Stem as LABEL=PATH (repeatable), e.g. a solo capture of one track", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": true, + "name": "stems", + "nargs": 1, + "opts": [ + "--stem" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "StringParamType", + "name": "text" + } + }, + { + "count": false, + "default": "bands", + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Band set: bands or third-octave", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "resolution", + "nargs": 1, + "opts": [ + "--resolution" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "StringParamType", + "name": "text" + } + }, + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Stem label to compare against the loudest other stem per band", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "focus", + "nargs": 1, + "opts": [ + "--focus" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "StringParamType", + "name": "text" + } + }, + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Write JSON report to this path", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "report_out", + "nargs": 1, + "opts": [ + "--report-out" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "TyperPath", + "name": "path" + } + } + ], + "path": [ + "audio", + "spectrum", + "shares" + ], + "short_help": null + }, "audio stems": { "deprecated": false, "help": "Audio stem commands", @@ -33517,6 +33650,7 @@ "\u2502 snapshot \u2502", "\u2502 diff \u2502", "\u2502 capture \u2502", + "\u2502 capture-tracks \u2502", "\u2502 watch \u2502", "\u2502 events Stream events pushed by Live instead of polling for diffs. \u2502", "\u2502 stop-all-clips \u2502", @@ -33535,6 +33669,7 @@ "snapshot", "diff", "capture", + "capture-tracks", "watch", "events", "stop-all-clips" @@ -33552,7 +33687,9 @@ "\u2502 * --slot INTEGER Empty session clip slot index (0-based) to record into \u2502", "\u2502 [required] \u2502", "\u2502 * --bars FLOAT Number of bars to capture [required] \u2502", - "\u2502 --start FLOAT Transport start position in beats [default: 0.0] \u2502", + "\u2502 * --start FLOAT Beat where recording begins; needs one bar of pre-roll (>= one \u2502", + "\u2502 bar) \u2502", + "\u2502 [required] \u2502", "\u2502 --set-routing Set the track's input routing to Resampling automatically \u2502", "\u2502 --analyze Run loudness/spectrum analysis on the captured file \u2502", "\u2502 --qa-project TEXT Run remix mastering QA against this project file \u2502", @@ -33643,11 +33780,11 @@ }, { "count": false, - "default": 0.0, + "default": null, "envvar": null, "expose_value": true, "flag_value": null, - "help": "Transport start position in beats", + "help": "Beat where recording begins; needs one bar of pre-roll (>= one bar)", "hidden": false, "is_flag": false, "kind": "option", @@ -33659,7 +33796,7 @@ "--start" ], "prompt": null, - "required": false, + "required": true, "secondary_opts": [], "show_default": true, "type": { @@ -33752,6 +33889,221 @@ ], "short_help": null }, + "session capture-tracks": { + "deprecated": false, + "help": null, + "help_text": [ + "", + " Usage: capture-tracks [OPTIONS]", + "", + "\u256d\u2500 Options \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256e", + "\u2502 * --start FLOAT Beat where recording begins; needs one bar of pre-roll \u2502", + "\u2502 (>= one bar) \u2502", + "\u2502 [required] \u2502", + "\u2502 --track-index INTEGER Track index (0-based) to solo and capture; repeatable \u2502", + "\u2502 --bars FLOAT Number of bars to capture [default: 8.0] \u2502", + "\u2502 --bypass-main-devices Turn off active main-track devices while capturing, \u2502", + "\u2502 then restore them \u2502", + "\u2502 --analyze Report each band's share per captured track \u2502", + "\u2502 --resolution TEXT Band set for --analyze: bands or third-octave \u2502", + "\u2502 [default: bands] \u2502", + "\u2502 --focus TEXT Track label to compare against the loudest other track \u2502", + "\u2502 --help Show this message and exit. \u2502", + "\u2570\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u256f" + ], + "hidden": false, + "kind": "command", + "no_args_is_help": false, + "params": [ + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Beat where recording begins; needs one bar of pre-roll (>= one bar)", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "start", + "nargs": 1, + "opts": [ + "--start" + ], + "prompt": null, + "required": true, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "FloatParamType", + "name": "float" + } + }, + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Track index (0-based) to solo and capture; repeatable", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": true, + "name": "track_indexes", + "nargs": 1, + "opts": [ + "--track-index" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "IntParamType", + "name": "integer" + } + }, + { + "count": false, + "default": 8.0, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Number of bars to capture", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "bars", + "nargs": 1, + "opts": [ + "--bars" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "FloatParamType", + "name": "float" + } + }, + { + "count": false, + "default": false, + "envvar": null, + "expose_value": true, + "flag_value": true, + "help": "Turn off active main-track devices while capturing, then restore them", + "hidden": false, + "is_flag": true, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "bypass_main_devices", + "nargs": 1, + "opts": [ + "--bypass-main-devices" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "BoolParamType", + "name": "boolean" + } + }, + { + "count": false, + "default": false, + "envvar": null, + "expose_value": true, + "flag_value": true, + "help": "Report each band's share per captured track", + "hidden": false, + "is_flag": true, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "analyze", + "nargs": 1, + "opts": [ + "--analyze" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "BoolParamType", + "name": "boolean" + } + }, + { + "count": false, + "default": "bands", + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Band set for --analyze: bands or third-octave", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "resolution", + "nargs": 1, + "opts": [ + "--resolution" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "StringParamType", + "name": "text" + } + }, + { + "count": false, + "default": null, + "envvar": null, + "expose_value": true, + "flag_value": null, + "help": "Track label to compare against the loudest other track", + "hidden": false, + "is_flag": false, + "kind": "option", + "metavar": null, + "multiple": false, + "name": "focus", + "nargs": 1, + "opts": [ + "--focus" + ], + "prompt": null, + "required": false, + "secondary_opts": [], + "show_default": true, + "type": { + "class": "StringParamType", + "name": "text" + } + } + ], + "path": [ + "session", + "capture-tracks" + ], + "short_help": null + }, "session diff": { "deprecated": false, "help": null, diff --git a/tests/snapshots/public_contract_snapshot.json b/tests/snapshots/public_contract_snapshot.json index 06eb6a7..145e021 100644 --- a/tests/snapshots/public_contract_snapshot.json +++ b/tests/snapshots/public_contract_snapshot.json @@ -1276,6 +1276,50 @@ "requires_confirmation": false } }, + "audio spectrum shares": { + "args": { + "type": "object" + }, + "errors": { + "codes": [ + "ABLETON_NOT_REACHABLE", + "BATCH_ASSERT_FAILED", + "BATCH_PREFLIGHT_FAILED", + "BATCH_RETRY_EXHAUSTED", + "BATCH_STEP_FAILED", + "CONFIG_INVALID", + "CONFIRMATION_REQUIRED", + "INSTALL_TARGET_NOT_FOUND", + "INTERNAL_ERROR", + "INVALID_ARGUMENT", + "NOT_IMPLEMENTED", + "PROTOCOL_CONNECTION_CLOSED", + "PROTOCOL_INVALID_RESPONSE", + "PROTOCOL_MALFORMED_JSON", + "PROTOCOL_REQUEST_ID_MISMATCH", + "PROTOCOL_VERSION_MISMATCH", + "READ_ONLY_VIOLATION", + "REMOTE_BUSY", + "REMOTE_SCRIPT_INCOMPATIBLE", + "REMOTE_SCRIPT_NOT_INSTALLED", + "SET_FILE_INVALID", + "SET_INTEGRITY_FAILED", + "SET_SCHEMA_UNSUPPORTED", + "SKILL_SOURCE_NOT_FOUND", + "TIMEOUT", + "UNAUTHORIZED", + "UNSUPPORTED_OS" + ] + }, + "result": { + "type": "object" + }, + "side_effect": { + "idempotent": false, + "kind": "write", + "requires_confirmation": false + } + }, "audio stems list": { "args": { "type": "object" @@ -14407,6 +14451,50 @@ "requires_confirmation": false } }, + "session capture-tracks": { + "args": { + "type": "object" + }, + "errors": { + "codes": [ + "ABLETON_NOT_REACHABLE", + "BATCH_ASSERT_FAILED", + "BATCH_PREFLIGHT_FAILED", + "BATCH_RETRY_EXHAUSTED", + "BATCH_STEP_FAILED", + "CONFIG_INVALID", + "CONFIRMATION_REQUIRED", + "INSTALL_TARGET_NOT_FOUND", + "INTERNAL_ERROR", + "INVALID_ARGUMENT", + "NOT_IMPLEMENTED", + "PROTOCOL_CONNECTION_CLOSED", + "PROTOCOL_INVALID_RESPONSE", + "PROTOCOL_MALFORMED_JSON", + "PROTOCOL_REQUEST_ID_MISMATCH", + "PROTOCOL_VERSION_MISMATCH", + "READ_ONLY_VIOLATION", + "REMOTE_BUSY", + "REMOTE_SCRIPT_INCOMPATIBLE", + "REMOTE_SCRIPT_NOT_INSTALLED", + "SET_FILE_INVALID", + "SET_INTEGRITY_FAILED", + "SET_SCHEMA_UNSUPPORTED", + "SKILL_SOURCE_NOT_FOUND", + "TIMEOUT", + "UNAUTHORIZED", + "UNSUPPORTED_OS" + ] + }, + "result": { + "type": "object" + }, + "side_effect": { + "idempotent": false, + "kind": "write", + "requires_confirmation": false + } + }, "session diff": { "args": { "properties": { diff --git a/tests/test_audio_spectrum_bands.py b/tests/test_audio_spectrum_bands.py index adc4522..6a2e507 100644 --- a/tests/test_audio_spectrum_bands.py +++ b/tests/test_audio_spectrum_bands.py @@ -43,35 +43,62 @@ def test_bands_cover_20_hz_to_18_khz_without_gaps() -> None: assert "brilliance_6000_10000_hz" in [name for name, _low, _high in BANDS] -def test_measure_band_levels_filters_each_band_over_the_whole_file(tmp_path: Path) -> None: +def _astats_log(levels: list[str]) -> str: + # ffmpeg prints each named astats instance at uninit, in no guaranteed order. + return "".join( + f"[astats@b{index} @ 0x1] Overall\n[astats@b{index} @ 0x1] RMS level dB: {level}\n" + for index, level in reversed(list(enumerate(levels))) + ) + + +def test_measure_band_levels_measures_every_band_in_one_ffmpeg_pass(tmp_path: Path) -> None: from ableton_cli.audio_analysis.spectrum import BANDS, measure_band_levels commands: list[list[str]] = [] + readings = [f"-{20 + index}.5" for index in range(len(BANDS))] def run(command: list[str]) -> subprocess.CompletedProcess[str]: commands.append(command) - stderr = "[Parsed_astats_6 @ 0x1] Overall\n[Parsed_astats_6 @ 0x1] RMS level dB: -21.5\n" - return subprocess.CompletedProcess(args=command, returncode=0, stdout="", stderr=stderr) + return subprocess.CompletedProcess( + args=command, returncode=0, stdout="", stderr=_astats_log(readings) + ) audio_path = tmp_path / "mix.wav" levels = measure_band_levels(audio_path, channels=2, run=run) - assert levels == {name: -21.5 for name, _low, _high in BANDS} - assert len(commands) == len(BANDS) - for command, (_name, low, high) in zip(commands, BANDS, strict=True): - assert str(audio_path) in command - assert "-ss" not in command and "-t" not in command - graph = command[command.index("-af") + 1] - assert graph.startswith("pan=mono|c0=0.5*c0+0.5*c1,") + assert levels == {name: float(readings[i]) for i, (name, _low, _high) in enumerate(BANDS)} + [command] = commands + assert str(audio_path) in command + assert "-ss" not in command and "-t" not in command + graph = command[command.index("-filter_complex") + 1] + assert graph.startswith(f"[0:a]pan=mono|c0=0.5*c0+0.5*c1,asplit={len(BANDS)}") + for index, (_name, low, high) in enumerate(BANDS): assert graph.count(f"between(b*sr/(2*(nb-1)),{low:g},{high:g})") == 2 + assert f"astats@b{index}=" in graph -def test_measure_band_levels_reads_silent_band_as_negative_infinity(tmp_path: Path) -> None: +def test_measure_band_levels_measures_custom_bands(tmp_path: Path) -> None: from ableton_cli.audio_analysis.spectrum import measure_band_levels + bands = (("low", 50.0, 100.0), ("high", 1000.0, 2000.0)) + def run(command: list[str]) -> subprocess.CompletedProcess[str]: - stderr = "[Parsed_astats_4 @ 0x1] RMS level dB: -inf\n" - return subprocess.CompletedProcess(args=command, returncode=0, stdout="", stderr=stderr) + return subprocess.CompletedProcess( + args=command, returncode=0, stdout="", stderr=_astats_log(["-30.0", "-40.0"]) + ) + + levels = measure_band_levels(tmp_path / "mix.wav", channels=1, bands=bands, run=run) + + assert levels == {"low": -30.0, "high": -40.0} + + +def test_measure_band_levels_reads_silent_band_as_negative_infinity(tmp_path: Path) -> None: + from ableton_cli.audio_analysis.spectrum import BANDS, measure_band_levels + + def run(command: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + args=command, returncode=0, stdout="", stderr=_astats_log(["-inf"] * len(BANDS)) + ) levels = measure_band_levels(tmp_path / "mix.wav", channels=1, run=run) @@ -79,10 +106,13 @@ def run(command: list[str]) -> subprocess.CompletedProcess[str]: def test_measure_band_levels_fails_without_rms_reading(tmp_path: Path) -> None: - from ableton_cli.audio_analysis.spectrum import measure_band_levels + from ableton_cli.audio_analysis.spectrum import BANDS, measure_band_levels def run(command: list[str]) -> subprocess.CompletedProcess[str]: - return subprocess.CompletedProcess(args=command, returncode=0, stdout="", stderr="") + log = _astats_log(["-20.0"] * len(BANDS)).replace( + "[astats@b0 @ 0x1] RMS level dB: -20.0\n", "" + ) + return subprocess.CompletedProcess(args=command, returncode=0, stdout="", stderr=log) with pytest.raises(AppError) as exc_info: measure_band_levels(tmp_path / "mix.wav", channels=2, run=run) diff --git a/tests/test_capture.py b/tests/test_capture.py index 947d520..a43addc 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -17,7 +17,9 @@ def __init__( available_routing_types: list[str] | None = None, available_routing_channels: list[str] | None = None, tempo: float = 120.0, + signature: tuple[int, int] = (4, 4), file_path: str | None = "/tmp/capture/Track 1.wav", + stop_beat: float = 0.0, ) -> None: self.calls: list[tuple[str, Any]] = [] self._is_audio_track = is_audio_track @@ -25,7 +27,10 @@ def __init__( self._available_routing_types = available_routing_types or ["Resampling", "Ext. In"] self._available_routing_channels = available_routing_channels or ["1/2"] self._tempo = tempo + self._signature = signature self._file_path = file_path + self._stop_beat = stop_beat + self._stopped = False def get_track_info(self, track_ref): # noqa: ANN001, ANN201 self.calls.append(("get_track_info", track_ref)) @@ -64,7 +69,13 @@ def transport_play(self): # noqa: ANN201 def song_info(self): # noqa: ANN201 self.calls.append(("song_info",)) - return {"tempo": self._tempo, "is_playing": True} + numerator, denominator = self._signature + return { + "tempo": self._tempo, + "is_playing": not self._stopped, + "current_time": self._stop_beat if self._stopped else 0.0, + "signature": {"numerator": numerator, "denominator": denominator}, + } def stop_clip(self, track, clip): # noqa: ANN001, ANN201 self.calls.append(("stop_clip", track, clip)) @@ -72,6 +83,7 @@ def stop_clip(self, track, clip): # noqa: ANN001, ANN201 def transport_stop(self): # noqa: ANN201 self.calls.append(("transport_stop",)) + self._stopped = True return {"playing": False} def clip_file_path_get(self, track, clip): # noqa: ANN001, ANN201 @@ -79,6 +91,18 @@ def clip_file_path_get(self, track, clip): # noqa: ANN001, ANN201 return {"track": track, "clip": clip, "file_path": self._file_path} +def _recorded_from(start: float, *, stop_beat: float, tempo: float = 120.0): # noqa: ANN202 + """A file-duration reader whose recording began at ``start`` and ended at ``stop_beat``.""" + return lambda _path: (stop_beat - start) * 60.0 / tempo + + +def _client_recording_from(start: float, **kwargs): # noqa: ANN003, ANN202 + stop_beat = start + 64.0 + return _FakeCaptureClient(stop_beat=stop_beat, **kwargs), _recorded_from( + start, stop_beat=stop_beat, tempo=kwargs.get("tempo", 120.0) + ) + + def _waits(recorder: list[float]): # noqa: ANN201 def _wait(seconds: float) -> None: recorder.append(seconds) @@ -87,7 +111,7 @@ def _wait(seconds: float) -> None: def test_capture_session_call_sequence_and_wait_duration() -> None: - client = _FakeCaptureClient(tempo=120.0) + client, recorded_seconds = _client_recording_from(32.0, tempo=120.0) waits: list[float] = [] result = capture_session( @@ -95,38 +119,100 @@ def test_capture_session_call_sequence_and_wait_duration() -> None: track=0, slot=0, bars=8, + start=32.0, wait_fn=_waits(waits), + recorded_seconds_fn=recorded_seconds, ) call_names = [call[0] for call in client.calls] assert call_names == [ "get_track_info", "track_routing_input_get", + "song_info", "track_arm_set", + "transport_play", "transport_position_set", "fire_clip", - "transport_play", - "song_info", "stop_clip", "transport_stop", + "song_info", "clip_file_path_get", ] - assert waits == [16.0 + 1.0] + # One bar of pre-roll plus the 8 captured bars, at 2 s per bar. + assert waits == [2.0 + 16.0 + 1.0] assert result["wait_seconds"] == 16.0 assert result["file_path"] == "/tmp/capture/Track 1.wav" assert result["tempo"] == 120.0 + assert result["recorded_start_beat"] == 32.0 + + +@pytest.mark.parametrize(("signature", "pre_roll_beat"), [((4, 4), 28.0), ((3, 4), 29.0)]) +def test_capture_session_moves_the_playhead_one_bar_before_start_while_playing( + signature: tuple[int, int], pre_roll_beat: float +) -> None: + # Live ignores a playhead moved while stopped, and firing a clip from a stopped + # transport restarts the song at 0, so the playhead moves after playback starts. + # The clip fires one bar early so launch quantization starts recording at start. + client, recorded_seconds = _client_recording_from(32.0, signature=signature) + + capture_session( + client, + track=0, + slot=0, + bars=1, + start=32.0, + wait_fn=lambda _seconds: None, + recorded_seconds_fn=recorded_seconds, + ) + + names = [call[0] for call in client.calls] + assert names.index("transport_play") < names.index("transport_position_set") + assert names.index("transport_position_set") < names.index("fire_clip") + assert ("transport_position_set", pre_roll_beat) in client.calls + + +def test_capture_session_needs_one_bar_of_pre_roll_before_start() -> None: + client = _FakeCaptureClient() + + with pytest.raises(AppError) as excinfo: + capture_session(client, track=0, slot=0, bars=1, start=2.0, wait_fn=lambda _s: None) + + assert excinfo.value.details["step"] == "validate_start" + assert "transport_play" not in [call[0] for call in client.calls] + + +def test_capture_session_fails_when_recording_did_not_begin_at_start() -> None: + # Recording began a bar late (e.g. launch quantization set to 2 bars). + client = _FakeCaptureClient(stop_beat=100.0) + + with pytest.raises(AppError) as excinfo: + capture_session( + client, + track=0, + slot=0, + bars=1, + start=32.0, + wait_fn=lambda _s: None, + recorded_seconds_fn=_recorded_from(36.0, stop_beat=100.0), + ) + + assert excinfo.value.details["step"] == "verify_recorded_start" + assert excinfo.value.details["recorded_start_beat"] == 36.0 + assert excinfo.value.details["requested_start_beat"] == 32.0 def test_capture_session_sets_routing_when_flag_passed() -> None: - client = _FakeCaptureClient(current_routing_type="Ext. In") + client, recorded_seconds = _client_recording_from(4.0, current_routing_type="Ext. In") capture_session( client, track=0, slot=0, bars=1, + start=4.0, set_routing=True, wait_fn=lambda _seconds: None, + recorded_seconds_fn=recorded_seconds, ) set_calls = [call for call in client.calls if call[0] == "track_routing_input_set"] @@ -141,7 +227,7 @@ def test_capture_session_rejects_non_audio_track() -> None: client = _FakeCaptureClient(is_audio_track=False) with pytest.raises(AppError) as excinfo: - capture_session(client, track=0, slot=0, bars=1, wait_fn=lambda _seconds: None) + capture_session(client, track=0, slot=0, bars=1, start=4.0, wait_fn=lambda _seconds: None) assert excinfo.value.details["step"] == "validate_track_is_audio" @@ -150,7 +236,7 @@ def test_capture_session_rejects_wrong_routing_without_set_routing_flag() -> Non client = _FakeCaptureClient(current_routing_type="Ext. In") with pytest.raises(AppError) as excinfo: - capture_session(client, track=0, slot=0, bars=1, wait_fn=lambda _seconds: None) + capture_session(client, track=0, slot=0, bars=1, start=4.0, wait_fn=lambda _seconds: None) assert excinfo.value.details["step"] == "check_routing" @@ -163,7 +249,13 @@ def test_capture_session_rejects_resampling_unavailable() -> None: with pytest.raises(AppError) as excinfo: capture_session( - client, track=0, slot=0, bars=1, set_routing=True, wait_fn=lambda _seconds: None + client, + track=0, + slot=0, + bars=1, + start=4.0, + set_routing=True, + wait_fn=lambda _seconds: None, ) assert excinfo.value.details["step"] == "set_routing" @@ -173,7 +265,7 @@ def test_capture_session_rejects_no_recorded_file_path() -> None: client = _FakeCaptureClient(file_path=None) with pytest.raises(AppError) as excinfo: - capture_session(client, track=0, slot=0, bars=1, wait_fn=lambda _seconds: None) + capture_session(client, track=0, slot=0, bars=1, start=4.0, wait_fn=lambda _seconds: None) assert excinfo.value.details["step"] == "read_file_path" @@ -182,7 +274,7 @@ def test_capture_session_rejects_non_positive_bars() -> None: client = _FakeCaptureClient() with pytest.raises(AppError) as excinfo: - capture_session(client, track=0, slot=0, bars=0, wait_fn=lambda _seconds: None) + capture_session(client, track=0, slot=0, bars=0, start=4.0, wait_fn=lambda _seconds: None) assert excinfo.value.details["step"] == "validate_bars" @@ -202,10 +294,17 @@ def test_capture_session_analyze_invokes_analyzer_entry_points(monkeypatch) -> N "analyze_spectrum", lambda path: spectrum_calls.append(path) or {"bands": {}}, ) - client = _FakeCaptureClient(file_path="/tmp/capture/Track 1.wav") + client, recorded_seconds = _client_recording_from(4.0) result = capture_session( - client, track=0, slot=0, bars=1, analyze=True, wait_fn=lambda _seconds: None + client, + track=0, + slot=0, + bars=1, + start=4.0, + analyze=True, + wait_fn=lambda _seconds: None, + recorded_seconds_fn=recorded_seconds, ) assert loudness_calls == ["/tmp/capture/Track 1.wav"] @@ -223,15 +322,17 @@ def test_capture_session_qa_project_invokes_mastering_qa(monkeypatch) -> None: "run_mastering_qa", lambda project, *, render: qa_calls.append((project, render)) or {"ok": True}, ) - client = _FakeCaptureClient(file_path="/tmp/capture/Track 1.wav") + client, recorded_seconds = _client_recording_from(4.0) result = capture_session( client, track=0, slot=0, bars=1, + start=4.0, qa_project="./proj/x.json", wait_fn=lambda _seconds: None, + recorded_seconds_fn=recorded_seconds, ) assert qa_calls == [("./proj/x.json", "/tmp/capture/Track 1.wav")] diff --git a/tests/test_capture_tracks.py b/tests/test_capture_tracks.py new file mode 100644 index 0000000..3403cf1 --- /dev/null +++ b/tests/test_capture_tracks.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from ableton_cli.capture import ShareAnalysis, capture_tracks +from ableton_cli.errors import AppError + +MAIN = {"mode": "main"} + + +class _FakeLive: + """Track solo, temporary track, and main-device state for capture_tracks.""" + + def __init__( + self, + *, + tracks: list[tuple[str, bool]], + main_devices_active: list[bool] | None = None, + ) -> None: + self.calls: list[tuple[Any, ...]] = [] + self.tracks = [{"index": i, "name": n, "solo": s} for i, (n, s) in enumerate(tracks)] + self.main_devices_active = list(main_devices_active or []) + + def tracks_list(self): # noqa: ANN201 + return {"tracks": [dict(track) for track in self.tracks]} + + def track_solo_set(self, track_ref, value): # noqa: ANN001, ANN201 + self.calls.append(("solo", track_ref["index"], value)) + self.tracks[track_ref["index"]]["solo"] = value + return {"track": track_ref["index"], "solo": value} + + def create_audio_track(self, index=-1): # noqa: ANN001, ANN201 + self.calls.append(("create_audio_track", index)) + self.tracks.append({"index": len(self.tracks), "name": "Audio", "solo": False}) + return {"index": len(self.tracks) - 1, "name": "Audio", "kind": "audio"} + + def track_volume_set(self, track_ref, value): # noqa: ANN001, ANN201 + self.calls.append(("volume", track_ref["index"], value)) + return {"track": track_ref["index"], "volume": value} + + def tracks_delete(self, track): # noqa: ANN001, ANN201 + self.calls.append(("tracks_delete", track)) + self.tracks.pop(track) + return {"deleted": track} + + def master_devices_list(self): # noqa: ANN201 + return { + "devices": [ + {"index": i, "name": f"Device {i}"} for i in range(len(self.main_devices_active)) + ] + } + + def device_active_get(self, track_ref, device_ref): # noqa: ANN001, ANN201 + assert track_ref == MAIN + return { + "device": device_ref["index"], + "active": self.main_devices_active[device_ref["index"]], + } + + def device_active_set(self, track_ref, device_ref, value): # noqa: ANN001, ANN201 + assert track_ref == MAIN + self.calls.append(("main_device_active", device_ref["index"], value)) + self.main_devices_active[device_ref["index"]] = value + return {"device": device_ref["index"], "active": value} + + +def _fake_capture(monkeypatch, live: _FakeLive, *, fail_on_track: int | None = None) -> None: + def capture(client, *, track, slot, bars, start, set_routing, wait_fn): # noqa: ANN001, ANN202 + soloed = [t["name"] for t in live.tracks if t["solo"]] + live.calls.append(("capture", track, slot, bars, start, set_routing, tuple(soloed))) + if fail_on_track is not None and soloed == [live.tracks[fail_on_track]["name"]]: + raise AppError(error_code="INVALID_ARGUMENT", message="capture failed") + return {"file_path": f"/rec/{soloed[0]}.wav"} + + monkeypatch.setattr("ableton_cli.capture.capture_session", capture) + + +def test_capture_tracks_solos_each_track_into_a_temporary_capture_track(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False), ("Bass", False), ("Lead", False)]) + _fake_capture(monkeypatch, live) + + result = capture_tracks(live, tracks=[0, 2], bars=8.0, start=304.0) + + assert live.calls == [ + ("solo", 0, True), + ("create_audio_track", -1), + ("volume", 3, 0.0), + ("capture", 3, 0, 8.0, 304.0, True, ("Kick",)), + ("tracks_delete", 3), + ("solo", 0, False), + ("solo", 2, True), + ("create_audio_track", -1), + ("volume", 3, 0.0), + ("capture", 3, 0, 8.0, 304.0, True, ("Lead",)), + ("tracks_delete", 3), + ("solo", 2, False), + ] + assert result == { + "start": 304.0, + "bars": 8.0, + "bypassed_main_devices": [], + "captures": [ + {"track_index": 0, "track_name": "Kick", "label": "Kick", "file_path": "/rec/Kick.wav"}, + {"track_index": 2, "track_name": "Lead", "label": "Lead", "file_path": "/rec/Lead.wav"}, + ], + } + assert [track["name"] for track in live.tracks] == ["Kick", "Bass", "Lead"] + + +def test_capture_tracks_bypasses_only_active_main_devices_and_restores_them(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False)], main_devices_active=[True, False, True]) + _fake_capture(monkeypatch, live) + + result = capture_tracks(live, tracks=[0], bars=4.0, start=0.0, bypass_main_devices=True) + + device_calls = [call for call in live.calls if call[0] == "main_device_active"] + assert device_calls == [ + ("main_device_active", 0, False), + ("main_device_active", 2, False), + ("main_device_active", 0, True), + ("main_device_active", 2, True), + ] + assert live.calls.index(("main_device_active", 2, False)) < live.calls.index(("solo", 0, True)) + assert result["bypassed_main_devices"] == [0, 2] + assert live.main_devices_active == [True, False, True] + + +def test_capture_tracks_restores_solo_devices_and_track_when_capture_fails(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False), ("Lead", False)], main_devices_active=[True]) + _fake_capture(monkeypatch, live, fail_on_track=1) + + with pytest.raises(AppError) as exc_info: + capture_tracks(live, tracks=[0, 1], bars=4.0, start=0.0, bypass_main_devices=True) + + assert exc_info.value.message == "capture failed" + + assert [track["name"] for track in live.tracks] == ["Kick", "Lead"] + assert [track["solo"] for track in live.tracks] == [False, False] + assert live.main_devices_active == [True] + + +def test_capture_tracks_labels_duplicate_track_names_by_index(monkeypatch) -> None: + live = _FakeLive(tracks=[("Synth", False), ("Synth", False)]) + _fake_capture(monkeypatch, live) + + result = capture_tracks(live, tracks=[0, 1], bars=4.0, start=0.0) + + assert [capture["label"] for capture in result["captures"]] == ["0:Synth", "1:Synth"] + + +def test_capture_tracks_analyze_reports_band_shares_by_label(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False), ("Lead", False)]) + _fake_capture(monkeypatch, live) + calls: list[tuple[Any, ...]] = [] + + def shares(stems, *, resolution, focus): # noqa: ANN001, ANN202 + calls.append((stems, resolution, focus)) + return {"resolution": resolution} + + monkeypatch.setattr("ableton_cli.capture.spectrum_shares", shares) + + result = capture_tracks( + live, + tracks=[0, 1], + bars=4.0, + start=0.0, + analysis=ShareAnalysis(resolution="third-octave", focus="Lead"), + ) + + assert calls == [ + ([("Kick", "/rec/Kick.wav"), ("Lead", "/rec/Lead.wav")], "third-octave", "Lead") + ] + assert result["shares"] == {"resolution": "third-octave"} + + +@pytest.mark.parametrize( + ("tracks", "message"), + [ + ([], "at least one --track-index"), + ([0, 0], "duplicate track index: 0"), + ([5], "track index 5 does not exist"), + ], +) +def test_capture_tracks_rejects_invalid_track_lists( + monkeypatch, tracks: list[int], message: str +) -> None: + live = _FakeLive(tracks=[("Kick", False)]) + _fake_capture(monkeypatch, live) + + with pytest.raises(AppError) as exc_info: + capture_tracks(live, tracks=tracks, bars=4.0, start=0.0) + + assert exc_info.value.error_code == "INVALID_ARGUMENT" + assert message in exc_info.value.message + assert live.calls == [] + + +def test_capture_tracks_refuses_when_a_track_is_already_soloed(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False), ("Pad", True)]) + _fake_capture(monkeypatch, live) + + with pytest.raises(AppError) as exc_info: + capture_tracks(live, tracks=[0], bars=4.0, start=0.0) + + assert exc_info.value.error_code == "INVALID_ARGUMENT" + assert exc_info.value.details["soloed_tracks"] == ["Pad"] + assert live.calls == [] + + +def test_session_capture_tracks_rejects_focus_without_analyze(runner, cli_app) -> None: + result = runner.invoke( + cli_app, + [ + "--output", + "json", + "session", + "capture-tracks", + "--track-index", + "0", + "--start", + "8", + "--focus", + "Kick", + ], + ) + + assert result.exit_code == 2, result.stdout + assert "--focus requires --analyze" in result.stdout + + +def test_session_capture_tracks_is_blocked_in_read_only_mode(runner, cli_app) -> None: + result = runner.invoke( + cli_app, + [ + "--output", + "json", + "--read-only", + "session", + "capture-tracks", + "--track-index", + "0", + "--bars", + "4", + "--start", + "8", + ], + ) + + assert result.exit_code == 20, result.stdout + assert '"code": "READ_ONLY_VIOLATION"' in result.stdout + + +def test_capture_tracks_rejects_unknown_resolution_before_capturing(monkeypatch) -> None: + live = _FakeLive(tracks=[("Kick", False)]) + _fake_capture(monkeypatch, live) + + with pytest.raises(AppError) as exc_info: + capture_tracks( + live, tracks=[0], bars=4.0, start=0.0, analysis=ShareAnalysis(resolution="octave") + ) + + assert "unsupported spectrum resolution: octave" in exc_info.value.message + assert live.calls == [] diff --git a/tests/test_session_capture_cli.py b/tests/test_session_capture_cli.py index 41af082..9ea7d88 100644 --- a/tests/test_session_capture_cli.py +++ b/tests/test_session_capture_cli.py @@ -1,9 +1,12 @@ from __future__ import annotations import json +import re class _FakeCaptureClient: + stopped = False + def get_track_info(self, track_ref): # noqa: ANN001, ANN201 return {"is_audio_track": True, "is_midi_track": False} @@ -29,12 +32,18 @@ def transport_play(self): # noqa: ANN201 return {} def song_info(self): # noqa: ANN201 - return {"tempo": 120.0} + # Stopped two bars after start (beat 8) once the transport stops. + return { + "tempo": 120.0, + "current_time": 16.0 if self.stopped else 0.0, + "signature": {"numerator": 4, "denominator": 4}, + } def stop_clip(self, track, clip): # noqa: ANN001, ANN201 return {} def transport_stop(self): # noqa: ANN201 + self.stopped = True return {} def clip_file_path_get(self, track, clip): # noqa: ANN001, ANN201 @@ -46,6 +55,8 @@ def test_session_capture_calls_through_and_returns_file_path(runner, cli_app, mo monkeypatch.setattr(session, "get_client", lambda ctx: _FakeCaptureClient()) monkeypatch.setattr("ableton_cli.capture.time.sleep", lambda _seconds: None) + # 8 beats at 120 BPM: the recording began at beat 8. + monkeypatch.setattr("ableton_cli.capture.wav_duration_sec", lambda _path: 4.0) result = runner.invoke( cli_app, @@ -60,6 +71,8 @@ def test_session_capture_calls_through_and_returns_file_path(runner, cli_app, mo "0", "--bars", "4", + "--start", + "8", ], ) @@ -68,6 +81,29 @@ def test_session_capture_calls_through_and_returns_file_path(runner, cli_app, mo assert payload["ok"] is True assert payload["result"]["file_path"] == "/tmp/capture/Track 1.wav" assert payload["result"]["wait_seconds"] == 8.0 + assert payload["result"]["recorded_start_beat"] == 8.0 + + +def test_session_capture_requires_start(runner, cli_app) -> None: + result = runner.invoke( + cli_app, + [ + "--output", + "json", + "session", + "capture", + "--track-index", + "0", + "--slot", + "0", + "--bars", + "4", + ], + ) + + assert result.exit_code == 2 + # CI renders the usage error with ANSI colors that split the option name. + assert "--start" in re.sub(r"\x1b\[[0-9;]*m", "", result.output) def test_session_capture_rejects_non_positive_bars(runner, cli_app, monkeypatch) -> None: @@ -88,6 +124,8 @@ def test_session_capture_rejects_non_positive_bars(runner, cli_app, monkeypatch) "0", "--bars", "0", + "--start", + "8", ], ) diff --git a/tests/test_spectrum_shares.py b/tests/test_spectrum_shares.py new file mode 100644 index 0000000..2a40da5 --- /dev/null +++ b/tests/test_spectrum_shares.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +import json +import math +import struct +import wave +from pathlib import Path + +import pytest + +from ableton_cli.errors import AppError + +SAMPLE_RATE = 48000 + + +def _write_stereo(path: Path, frames: list[float]) -> None: + with wave.open(str(path), "wb") as wav: + wav.setnchannels(2) + wav.setsampwidth(2) + wav.setframerate(SAMPLE_RATE) + payload = bytearray() + for value in frames: + sample = int(value * 32767) + payload.extend(struct.pack(" list[float]: + return [ + amp * math.sin(2 * math.pi * freq * index / SAMPLE_RATE) + for index in range(int(SAMPLE_RATE * seconds)) + ] + + +def _stem(tmp_path: Path, name: str) -> Path: + path = tmp_path / f"{name}.wav" + _write_stereo(path, _sine(freq=80.0, amp=0.1, seconds=0.1)) + return path + + +def _fake_levels(monkeypatch, levels_by_stem: dict[str, dict[str, float]]) -> None: + """Measure each stem as the levels given for its file stem; unlisted bands are silent.""" + + def measure(path, *, channels, bands): # noqa: ANN001, ANN202 + levels = levels_by_stem[Path(path).stem] + return {name: levels.get(name, float("-inf")) for name, _low, _high in bands} + + monkeypatch.setattr( + "ableton_cli.audio_analysis.spectrum.require_ffmpeg_engine", lambda engine: engine + ) + monkeypatch.setattr("ableton_cli.audio_analysis.spectrum.measure_band_levels", measure) + + +def test_third_octave_bands_are_contiguous_iso_bands() -> None: + from ableton_cli.audio_analysis.spectrum import THIRD_OCTAVE_BANDS + + names = [name for name, _low, _high in THIRD_OCTAVE_BANDS] + assert len(THIRD_OCTAVE_BANDS) == 29 + assert names[0] == "25_hz" + assert names[names.index("1000_hz") - 1] == "800_hz" + assert "31.5_hz" in names + assert names[-1] == "16000_hz" + for (_n, _low, high), (_next, next_low, _h) in zip( + THIRD_OCTAVE_BANDS, THIRD_OCTAVE_BANDS[1:], strict=False + ): + assert high == next_low + _name, low, high = THIRD_OCTAVE_BANDS[names.index("1000_hz")] + assert math.isclose(low, 890.9, abs_tol=0.1) + assert math.isclose(high, 1122.5, abs_tol=0.1) + + +def test_spectrum_shares_splits_each_band_by_stem_energy(monkeypatch, tmp_path: Path) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels( + monkeypatch, + { + "kick": {"sub_20_60_hz": -20.0, "bass_60_120_hz": -20.0}, + "bass": {"sub_20_60_hz": -26.0206, "bass_60_120_hz": -20.0}, + }, + ) + + result = spectrum_shares([("Kick", _stem(tmp_path, "kick")), ("Bass", _stem(tmp_path, "bass"))]) + + assert result["resolution"] == "bands" + assert result["bands"][0] == {"name": "sub_20_60_hz", "low_hz": 20.0, "high_hz": 60.0} + assert [stem["label"] for stem in result["stems"]] == ["Kick", "Bass"] + assert result["stems"][0]["levels"]["sub_20_60_hz"] == -20.0 + assert result["shares_pct"]["sub_20_60_hz"] == {"Kick": 80.0, "Bass": 20.0} + assert result["shares_pct"]["bass_60_120_hz"] == {"Kick": 50.0, "Bass": 50.0} + assert result["dominant"]["sub_20_60_hz"] == "Kick" + assert math.isclose(result["band_totals_dbfs"]["bass_60_120_hz"], -16.99, abs_tol=0.01) + assert "focus" not in result + + +def test_spectrum_shares_leave_silent_bands_unowned(monkeypatch, tmp_path: Path) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels(monkeypatch, {"a": {"sub_20_60_hz": -20.0}, "b": {}}) + + result = spectrum_shares([("A", _stem(tmp_path, "a")), ("B", _stem(tmp_path, "b"))]) + + assert result["shares_pct"]["air_10000_18000_hz"] == {"A": 0.0, "B": 0.0} + assert result["dominant"]["air_10000_18000_hz"] is None + assert result["band_totals_dbfs"]["air_10000_18000_hz"] == float("-inf") + + +def test_spectrum_shares_focus_reports_margin_over_loudest_other_stem( + monkeypatch, tmp_path: Path +) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels( + monkeypatch, + { + "lead": {"mid_400_2000_hz": -30.0, "presence_2000_6000_hz": -30.0}, + "pluck": {"mid_400_2000_hz": -27.0, "presence_2000_6000_hz": -38.0}, + "perc": {"mid_400_2000_hz": -35.0, "presence_2000_6000_hz": -34.0}, + }, + ) + stems = [(label, _stem(tmp_path, label.lower())) for label in ("Lead", "Pluck", "Perc")] + + result = spectrum_shares(stems, focus="Lead") + + assert result["focus"]["label"] == "Lead" + assert result["focus"]["margin_db"]["mid_400_2000_hz"] == -3.0 + assert result["focus"]["margin_db"]["presence_2000_6000_hz"] == 4.0 + assert result["focus"]["margin_db"]["air_10000_18000_hz"] is None + + +def test_spectrum_shares_third_octave_measures_third_octave_bands( + monkeypatch, tmp_path: Path +) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels(monkeypatch, {"a": {"1000_hz": -20.0}}) + + result = spectrum_shares([("A", _stem(tmp_path, "a"))], resolution="third-octave") + + assert result["resolution"] == "third-octave" + assert len(result["bands"]) == 29 + assert result["shares_pct"]["1000_hz"] == {"A": 100.0} + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"stems": []}, "at least one stem"), + ({"focus": "Missing"}, "focus label"), + ({"resolution": "octave"}, "resolution"), + ], +) +def test_spectrum_shares_rejects_invalid_requests( + monkeypatch, tmp_path: Path, kwargs: dict[str, object], message: str +) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels(monkeypatch, {"a": {}}) + request: dict[str, object] = {"stems": [("A", _stem(tmp_path, "a"))], **kwargs} + + with pytest.raises(AppError) as exc_info: + spectrum_shares(**request) # type: ignore[arg-type] + + assert exc_info.value.error_code == "INVALID_ARGUMENT" + assert message in exc_info.value.message + + +def test_spectrum_shares_rejects_duplicate_labels(monkeypatch, tmp_path: Path) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + _fake_levels(monkeypatch, {"a": {}}) + path = _stem(tmp_path, "a") + + with pytest.raises(AppError) as exc_info: + spectrum_shares([("A", path), ("A", path)]) + + assert exc_info.value.error_code == "INVALID_ARGUMENT" + assert "duplicate stem label: A" in exc_info.value.message + + +def test_audio_spectrum_shares_cli_parses_stems_and_writes_report( + runner, cli_app, tmp_path: Path, monkeypatch +) -> None: + _fake_levels(monkeypatch, {"kick": {"sub_20_60_hz": -20.0}, "sub": {"sub_20_60_hz": -20.0}}) + report = tmp_path / "shares.json" + + result = runner.invoke( + cli_app, + [ + "--output", + "json", + "audio", + "spectrum", + "shares", + "--stem", + f"Kick={_stem(tmp_path, 'kick')}", + "--stem", + f"Sub={_stem(tmp_path, 'sub')}", + "--focus", + "Sub", + "--report-out", + str(report), + ], + ) + + assert result.exit_code == 0, result.stdout + payload = json.loads(result.stdout) + assert payload["args"]["stems"] == [ + {"label": "Kick", "path": str(tmp_path / "kick.wav")}, + {"label": "Sub", "path": str(tmp_path / "sub.wav")}, + ] + assert payload["result"]["shares_pct"]["sub_20_60_hz"] == {"Kick": 50.0, "Sub": 50.0} + assert payload["result"]["focus"]["margin_db"]["sub_20_60_hz"] == 0.0 + assert json.loads(report.read_text(encoding="utf-8"))["resolution"] == "bands" + + +@pytest.mark.parametrize("stem", ["no-separator.wav", "=missing-label.wav", "Label="]) +def test_audio_spectrum_shares_cli_rejects_malformed_stem( + runner, cli_app, tmp_path: Path, stem: str +) -> None: + result = runner.invoke( + cli_app, ["--output", "json", "audio", "spectrum", "shares", "--stem", stem] + ) + + assert result.exit_code == 2, result.stdout + assert json.loads(result.stdout)["error"]["code"] == "INVALID_ARGUMENT" + + +def test_audio_spectrum_shares_cli_rejects_missing_file(runner, cli_app, tmp_path: Path) -> None: + result = runner.invoke( + cli_app, + ["--output", "json", "audio", "spectrum", "shares", "--stem", f"A={tmp_path / 'x.wav'}"], + ) + + assert result.exit_code == 2, result.stdout + assert "audio file not found" in json.loads(result.stdout)["error"]["message"] + + +def test_spectrum_shares_measure_real_stems(tmp_path: Path, real_ffmpeg) -> None: + from ableton_cli.audio_analysis.spectrum import spectrum_shares + + loud = tmp_path / "loud.wav" + quiet = tmp_path / "quiet.wav" + _write_stereo(loud, _sine(freq=80.0, amp=0.5, seconds=2.0)) + _write_stereo(quiet, _sine(freq=80.0, amp=0.25, seconds=2.0)) + + result = spectrum_shares([("Loud", loud), ("Quiet", quiet)], focus="Quiet") + + # Half the amplitude is a quarter of the power: an 80/20 split of the band. + assert result["shares_pct"]["bass_60_120_hz"] == {"Loud": 80.0, "Quiet": 20.0} + assert math.isclose(result["focus"]["margin_db"]["bass_60_120_hz"], -6.02, abs_tol=0.1)