Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions tests/tools/test_audio_energy_windows.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Fractional video durations must produce nonempty, fitting audio windows."""
import json
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from tools.analysis.audio_energy import AudioEnergy


def analyze(tmp_path, video_duration, loudness, audio_duration):
source = tmp_path / "audio.wav"
source.touch()
probe = SimpleNamespace(stdout=json.dumps({"format": {"duration": str(audio_duration)}}))
meter = SimpleNamespace(stderr="\n".join(f"t: {i + 0.1} M: {value}" for i, value in enumerate(loudness)))
with patch("tools.analysis.audio_energy.shutil.which", return_value="ffmpeg"), patch("tools.analysis.audio_energy.subprocess.run", side_effect=[probe, meter]):
return AudioEnergy().execute({"input_path": str(source), "video_duration_seconds": video_duration})


def test_subsecond_video_uses_one_analysis_bucket(tmp_path):
result = analyze(tmp_path, 0.5, [-30, -10, -20], 3)
assert result.success, result.error
assert result.data["recommended_offset_seconds"] == 1


def test_fractional_window_covers_the_full_video(tmp_path):
result = analyze(tmp_path, 1.5, [-10, -50, -20, -20], 4)
assert result.success, result.error
assert result.data["recommended_offset_seconds"] == 2
assert not result.data["needs_loop"]


def test_partial_final_bucket_cannot_force_unnecessary_loop(tmp_path):
result = analyze(tmp_path, 0.5, [-30, -20, -10], 2.2)
assert result.success, result.error
assert result.data["recommended_offset_seconds"] == 1
assert not result.data["needs_loop"]


@pytest.mark.parametrize("duration", [0, -1, float("nan"), float("inf")])
def test_invalid_duration_returns_tool_error(tmp_path, duration):
result = analyze(tmp_path, duration, [-30, -20], 2)
assert not result.success
assert "duration" in result.error.lower()
12 changes: 10 additions & 2 deletions tools/analysis/audio_energy.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from __future__ import annotations

import json
import math
import re
import shutil
import subprocess
Expand Down Expand Up @@ -76,6 +77,7 @@ class AudioEnergy(BaseTool):
},
"video_duration_seconds": {
"type": "number",
"exclusiveMinimum": 0,
"description": "Duration of the video this music will accompany. "
"Used to recommend looping and find the best offset window.",
},
Expand Down Expand Up @@ -115,6 +117,8 @@ def execute(self, inputs: dict[str, Any]) -> ToolResult:

threshold_lufs = inputs.get("energy_threshold_lufs", -40)
video_duration = inputs.get("video_duration_seconds")
if video_duration is not None and (not math.isfinite(video_duration) or video_duration <= 0):
return ToolResult(success=False, error="Video duration must be finite and greater than zero")

start = time.time()

Expand Down Expand Up @@ -229,7 +233,7 @@ def execute(self, inputs: dict[str, Any]) -> ToolResult:
)

if video_duration and video_duration < audio_duration:
window_size = int(video_duration)
window_size = math.ceil(video_duration)
loudness_values = [
s["loudness_lufs"] if s["loudness_lufs"] > -120 else -60
for s in energy_profile
Expand All @@ -239,7 +243,11 @@ def execute(self, inputs: dict[str, Any]) -> ToolResult:
best_avg = -999.0
best_start = 0

for i in range(len(loudness_values) - window_size + 1):
# A partial final analysis bucket may not contain enough
# audio for the requested window, even when it scores highest.
last_start = min(len(loudness_values) - window_size,
math.floor(audio_duration - video_duration))
for i in range(last_start + 1):
window = loudness_values[i : i + window_size]
avg = sum(window) / len(window)
if avg > best_avg:
Expand Down
Loading