diff --git a/core/converter.py b/core/converter.py new file mode 100644 index 0000000..e64c62a --- /dev/null +++ b/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=" + 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()) diff --git a/tests/test_converter.py b/tests/test_converter.py new file mode 100644 index 0000000..4fee84a --- /dev/null +++ b/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)