converter: modulo ffmpeg WAV -> MP3 con VBR/CBR + stop

core/converter.py

@@ -0,0 +1,176 @@
+"""Conversione WAV -> MP3 via ffmpeg subprocess."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import threading
+from pathlib import Path
+from typing import Callable, Optional
+
+from core.paths import find_ffmpeg_dir, subprocess_flags
+
+
+_stop_event = threading.Event()
+_current_process: Optional[subprocess.Popen] = None
+_process_lock = threading.Lock()
+
+
+def request_stop() -> None:
+    _stop_event.set()
+    with _process_lock:
+        if _current_process and _current_process.poll() is None:
+            _current_process.terminate()
+
+
+def reset_stop() -> None:
+    _stop_event.clear()
+
+
+def is_stopped() -> bool:
+    return _stop_event.is_set()
+
+
+# VBR quality mapping da bitrate label a -q:a (libmp3lame). Piu basso = migliore qualita.
+# Vedi https://trac.ffmpeg.org/wiki/Encode/MP3
+_VBR_QUALITY = {
+    128: 5,
+    192: 2,
+    256: 0,
+    320: 0,  # V0 e' ~245k medio; per >V0 c'e' solo CBR 320
+}
+
+
+def _find_ffmpeg() -> str:
+    """Trova il binario ffmpeg dentro bundle_bin o PATH.
+
+    Solleva RuntimeError se non trovato.
+    """
+    ffmpeg_dir = find_ffmpeg_dir()
+    if ffmpeg_dir:
+        for name in ("ffmpeg", "ffmpeg.exe"):
+            p = Path(ffmpeg_dir) / name
+            if p.exists():
+                return str(p)
+    # Fallback PATH
+    import shutil
+    ff = shutil.which("ffmpeg")
+    if ff:
+        return ff
+    raise RuntimeError("ffmpeg non trovato nel bundle o nel PATH")
+
+
+def convert_wav_to_mp3(
+    input_path: str,
+    output_path: str,
+    bitrate: int = 320,
+    vbr: bool = False,
+    progress_callback: Optional[Callable] = None,
+) -> None:
+    """Converte un file WAV in MP3.
+
+    Args:
+        input_path: file .wav sorgente
+        output_path: file .mp3 destinazione
+        bitrate: 128 / 192 / 256 / 320
+        vbr: True per Variable Bit Rate (-q:a), False per Constant (-b:a)
+        progress_callback: opzionale, chiamato con (percent 0..100)
+
+    Raises:
+        RuntimeError: se ffmpeg non trovato o exit non-zero.
+        FileNotFoundError: se input_path non esiste.
+    """
+    src = Path(input_path)
+    if not src.exists():
+        raise FileNotFoundError(f"File sorgente non trovato: {input_path}")
+
+    dst = Path(output_path)
+    dst.parent.mkdir(parents=True, exist_ok=True)
+
+    ffmpeg = _find_ffmpeg()
+
+    cmd = [ffmpeg, "-y", "-i", str(src)]
+    if vbr:
+        quality = _VBR_QUALITY.get(bitrate, 2)
+        cmd.extend(["-c:a", "libmp3lame", "-q:a", str(quality)])
+    else:
+        cmd.extend(["-c:a", "libmp3lame", "-b:a", f"{bitrate}k"])
+    cmd.extend([
+        "-map_metadata", "0",     # copia metadata WAV se presenti
+        "-id3v2_version", "3",
+        "-progress", "pipe:2",     # progress su stderr in formato key=value
+        str(dst),
+    ])
+
+    # Prova a ottenere la durata totale per calcolare % — leggendo lo stream INFO
+    duration_us = _probe_duration_us(ffmpeg, str(src))
+
+    global _current_process
+    with _process_lock:
+        _current_process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1,
+            **subprocess_flags(),
+        )
+        proc = _current_process
+
+    try:
+        for line in proc.stderr:
+            if is_stopped():
+                proc.terminate()
+                if progress_callback:
+                    progress_callback(-1)  # segnale di interruzione
+                return
+            if not progress_callback or not duration_us:
+                continue
+            # ffmpeg -progress emette "out_time_us=<microsecondi>"
+            m = re.match(r"out_time_us=(\d+)", line.strip())
+            if m:
+                elapsed = int(m.group(1))
+                pct = min(100, int(elapsed * 100 / duration_us))
+                progress_callback(pct)
+
+        rc = proc.wait()
+
+        if rc != 0:
+            raise RuntimeError(f"ffmpeg exit {rc}")
+
+        if progress_callback:
+            progress_callback(100)
+    finally:
+        with _process_lock:
+            _current_process = None
+
+
+def _probe_duration_us(ffmpeg: str, input_path: str) -> int:
+    """Estrae la durata in microsecondi via ffmpeg -f null. Ritorna 0 se non riesce."""
+    try:
+        # ffprobe potrebbe non essere sempre bundlato; usa ffmpeg
+        result = subprocess.run(
+            [ffmpeg, "-i", input_path, "-f", "null", "-"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            **subprocess_flags(),
+        )
+        # Cerca "Duration: HH:MM:SS.ms"
+        m = re.search(r"Duration:\s+(\d+):(\d+):(\d+)\.(\d+)", result.stderr)
+        if m:
+            h, mi, s, ms = map(int, m.groups())
+            total_sec = h * 3600 + mi * 60 + s + ms / 100
+            return int(total_sec * 1_000_000)
+    except Exception:
+        pass
+    return 0
+
+
+def list_wav_files(directory: str, recursive: bool = True) -> list:
+    """Scan cartella per file .wav. Ritorna list di path stringa ordinati."""
+    root = Path(directory)
+    if not root.is_dir():
+        return []
+    pattern = "**/*.wav" if recursive else "*.wav"
+    return sorted(str(p) for p in root.glob(pattern) if p.is_file())

tests/test_converter.py

@@ -0,0 +1,104 @@
+"""Test per core.converter — file scanning + stop event.
+
+I test evitano di invocare ffmpeg reale (che potrebbe non essere in PATH
+in CI). La funzione di conversione vera e propria e' coperta manualmente.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from core import converter
+
+
+class TestListWavFiles:
+    def test_scans_recursively(self, tmp_path: Path):
+        # Struttura:
+        #   root/a.wav
+        #   root/b.WAV  (case insensitive: glob su POSIX distingue,
+        #                quindi controllo esplicito che almeno i .wav
+        #                minuscoli siano restituiti)
+        #   root/c.mp3
+        #   root/sub/d.wav
+        (tmp_path / "a.wav").write_bytes(b"RIFF")
+        (tmp_path / "c.mp3").write_bytes(b"ID3")
+        sub = tmp_path / "sub"
+        sub.mkdir()
+        (sub / "d.wav").write_bytes(b"RIFF")
+
+        got = converter.list_wav_files(str(tmp_path), recursive=True)
+
+        assert len(got) == 2
+        names = {Path(p).name for p in got}
+        assert names == {"a.wav", "d.wav"}
+        # Ordinamento stabile
+        assert got == sorted(got)
+
+    def test_non_recursive_ignores_subfolders(self, tmp_path: Path):
+        (tmp_path / "a.wav").write_bytes(b"RIFF")
+        sub = tmp_path / "sub"
+        sub.mkdir()
+        (sub / "b.wav").write_bytes(b"RIFF")
+
+        got = converter.list_wav_files(str(tmp_path), recursive=False)
+
+        assert len(got) == 1
+        assert Path(got[0]).name == "a.wav"
+
+    def test_empty_dir_returns_empty(self, tmp_path: Path):
+        assert converter.list_wav_files(str(tmp_path)) == []
+
+    def test_missing_dir_returns_empty(self, tmp_path: Path):
+        missing = tmp_path / "does-not-exist"
+        assert converter.list_wav_files(str(missing)) == []
+
+    def test_ignores_non_wav_files(self, tmp_path: Path):
+        (tmp_path / "song.mp3").write_bytes(b"ID3")
+        (tmp_path / "song.flac").write_bytes(b"fLaC")
+        (tmp_path / "notes.txt").write_text("hello")
+
+        assert converter.list_wav_files(str(tmp_path)) == []
+
+
+class TestStopEvent:
+    def setup_method(self):
+        # Ogni test parte con lo stop event pulito.
+        converter.reset_stop()
+
+    def teardown_method(self):
+        converter.reset_stop()
+
+    def test_reset_initially_not_stopped(self):
+        converter.reset_stop()
+        assert converter.is_stopped() is False
+
+    def test_request_stop_sets_flag(self):
+        converter.request_stop()
+        assert converter.is_stopped() is True
+
+    def test_reset_after_stop_clears_flag(self):
+        converter.request_stop()
+        assert converter.is_stopped() is True
+        converter.reset_stop()
+        assert converter.is_stopped() is False
+
+
+class TestConvertWavToMp3Errors:
+    def test_missing_input_raises(self, tmp_path: Path):
+        missing = tmp_path / "nope.wav"
+        out = tmp_path / "out.mp3"
+        with pytest.raises(FileNotFoundError):
+            converter.convert_wav_to_mp3(str(missing), str(out))
+
+
+class TestVbrQualityMap:
+    def test_all_bitrates_have_mapping(self):
+        for br in (128, 192, 256, 320):
+            assert br in converter._VBR_QUALITY
+
+    def test_higher_bitrate_lower_or_equal_quality_number(self):
+        # -q:a: piu' basso = migliore. Coerenza monotona sulle chiavi.
+        vals = [converter._VBR_QUALITY[br] for br in (128, 192, 256, 320)]
+        assert vals == sorted(vals, reverse=True)