From acb000b9e6e4b09cb4bc0718344fdb2460c171b5 Mon Sep 17 00:00:00 2001
From: Artem Akymenko <akimenko.artem@gmail.com>
Date: Wed, 15 Jul 2026 20:19:49 +0300
Subject: [PATCH] refactor: extract voice loading logic to domain layer

- Add abogen/domain/voice_loader.py with:
  - VoiceCache class: unified cache for loaded voices
  - resolve_voice(): load voice with optional caching
  - load_voice_cached(): compatibility wrapper for PyQt

- Update abogen/pyqt/conversion.py:
  - Replace load_voice_cached method body with call to domain function
  - Maintain backward compatibility with existing interface

- Add tests/test_voice_loader.py with unit tests for VoiceCache and voice loading
---
 abogen/domain/voice_loader.py | 116 ++++++++++++++++++++++++++++
 abogen/pyqt/conversion.py     |  20 ++---
 tests/test_voice_loader.py    | 137 ++++++++++++++++++++++++++++++++++
 3 files changed, 260 insertions(+), 13 deletions(-)
 create mode 100644 abogen/domain/voice_loader.py
 create mode 100644 tests/test_voice_loader.py

diff --git a/abogen/domain/voice_loader.py b/abogen/domain/voice_loader.py
new file mode 100644
index 0000000..3f7c9a8
--- /dev/null
+++ b/abogen/domain/voice_loader.py
@@ -0,0 +1,116 @@
+"""Voice loading and caching utilities.
+
+This module provides unified voice loading with caching support for both
+PyQt and WebUI interfaces.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Tuple
+
+from abogen.voice_formulas import get_new_voice
+
+
+class VoiceCache:
+    """Thread-safe voice cache for loaded voice tensors."""
+    
+    def __init__(self):
+        self._cache: Dict[str, Any] = {}
+    
+    def get(self, voice_spec: str) -> Optional[Any]:
+        """Get cached voice by spec."""
+        return self._cache.get(voice_spec)
+    
+    def set(self, voice_spec: str, voice: Any) -> None:
+        """Cache a loaded voice."""
+        self._cache[voice_spec] = voice
+    
+    def contains(self, voice_spec: str) -> bool:
+        """Check if voice is in cache."""
+        return voice_spec in self._cache
+    
+    def clear(self) -> None:
+        """Clear all cached voices."""
+        self._cache.clear()
+    
+    def __contains__(self, voice_spec: str) -> bool:
+        return self.contains(voice_spec)
+
+
+def resolve_voice(
+    voice_spec: str,
+    pipeline: Any,
+    use_gpu: bool,
+    cache: Optional[VoiceCache] = None,
+) -> Any:
+    """Resolve voice spec to actual voice tensor or name.
+    
+    If voice_spec contains '*' (formula), loads the voice using get_new_voice.
+    Otherwise, returns the voice_spec as-is (it's a voice name).
+    
+    Uses optional cache to avoid reloading same voice multiple times.
+    
+    Args:
+        voice_spec: Voice specification (name or formula string with '*').
+        pipeline: TTS pipeline instance for loading formula voices.
+        use_gpu: Whether to use GPU for voice loading.
+        cache: Optional VoiceCache instance for caching loaded voices.
+        
+    Returns:
+        Loaded voice tensor (for formulas) or voice name string.
+    """
+    # Check cache first
+    if cache and cache.contains(voice_spec):
+        return cache.get(voice_spec)
+    
+    # Load voice
+    if "*" in voice_spec:
+        if pipeline is None:
+            return voice_spec
+        loaded_voice = get_new_voice(pipeline, voice_spec, use_gpu)
+    else:
+        loaded_voice = voice_spec
+    
+    # Cache it
+    if cache:
+        cache.set(voice_spec, loaded_voice)
+    
+    return loaded_voice
+
+
+def load_voice_cached(
+    voice_name: str,
+    pipeline: Any,
+    use_gpu: bool,
+    cache: Optional[Dict[str, Any]] = None,
+) -> Any:
+    """Load voice with caching (compatibility wrapper for PyQt).
+    
+    This function maintains backward compatibility with the PyQt interface
+    while using the unified voice loading logic.
+    
+    Args:
+        voice_name: Voice name or formula string.
+        pipeline: TTS pipeline instance.
+        use_gpu: Whether to use GPU.
+        cache: Optional dict to use as cache (instead of VoiceCache).
+        
+    Returns:
+        Loaded voice tensor or voice name string.
+    """
+    # Use dict cache if provided (for backward compatibility)
+    if cache is not None:
+        if voice_name in cache:
+            return cache[voice_name]
+    
+    # Load voice
+    if "*" in voice_name:
+        loaded_voice = get_new_voice(pipeline, voice_name, use_gpu)
+    else:
+        loaded_voice = voice_name
+    
+    # Cache it
+    if cache is not None:
+        cache[voice_name] = loaded_voice
+    
+    return loaded_voice
diff --git a/abogen/pyqt/conversion.py b/abogen/pyqt/conversion.py
index fcfdbbc..1ab7b99 100644
--- a/abogen/pyqt/conversion.py
+++ b/abogen/pyqt/conversion.py
@@ -36,6 +36,7 @@ from abogen.domain.audio_buffer import (
     SAMPLE_RATE,
 )
 from abogen.domain.subtitle_generation import process_subtitle_tokens
+from abogen.domain.voice_loader import load_voice_cached
 import abogen.hf_tracker as hf_tracker
 import static_ffmpeg
 import threading  # for efficient waiting
@@ -287,19 +288,12 @@ class ConversionThread(QThread):
         Returns:
             Loaded voice tensor or voice name string
         """
-        # Check cache first
-        if voice_name in self.voice_cache:
-            return self.voice_cache[voice_name]
-
-        # Load voice
-        if "*" in voice_name:
-            loaded_voice = get_new_voice(tts, voice_name, self.use_gpu)
-        else:
-            loaded_voice = voice_name
-
-        # Cache it
-        self.voice_cache[voice_name] = loaded_voice
-        return loaded_voice
+        return load_voice_cached(
+            voice_name=voice_name,
+            pipeline=tts,
+            use_gpu=self.use_gpu,
+            cache=self.voice_cache,
+        )
 
     def _stream_audio_in_chunks(
         self, segments, process_func, progress_prefix="Processing"
diff --git a/tests/test_voice_loader.py b/tests/test_voice_loader.py
new file mode 100644
index 0000000..79f8483
--- /dev/null
+++ b/tests/test_voice_loader.py
@@ -0,0 +1,137 @@
+"""Tests for abogen.domain.voice_loader module."""
+
+import pytest
+
+from abogen.domain.voice_loader import (
+    VoiceCache,
+    resolve_voice,
+    load_voice_cached,
+)
+
+
+class TestVoiceCache:
+    """Tests for VoiceCache class."""
+
+    def test_get_set(self):
+        """Test basic get/set operations."""
+        cache = VoiceCache()
+        cache.set("test_voice", "loaded_voice")
+        assert cache.get("test_voice") == "loaded_voice"
+
+    def test_get_missing(self):
+        """Test get returns None for missing voice."""
+        cache = VoiceCache()
+        assert cache.get("missing_voice") is None
+
+    def test_contains(self):
+        """Test contains method."""
+        cache = VoiceCache()
+        cache.set("test_voice", "loaded_voice")
+        assert cache.contains("test_voice")
+        assert not cache.contains("missing_voice")
+
+    def test_in_operator(self):
+        """Test __contains__ (in operator)."""
+        cache = VoiceCache()
+        cache.set("test_voice", "loaded_voice")
+        assert "test_voice" in cache
+        assert "missing_voice" not in cache
+
+    def test_clear(self):
+        """Test clear method."""
+        cache = VoiceCache()
+        cache.set("voice1", "loaded1")
+        cache.set("voice2", "loaded2")
+        cache.clear()
+        assert not cache.contains("voice1")
+        assert not cache.contains("voice2")
+
+
+class TestResolveVoice:
+    """Tests for resolve_voice function."""
+
+    def test_simple_voice_name(self):
+        """Test that simple voice names are returned as-is."""
+        result = resolve_voice(
+            voice_spec="test_voice",
+            pipeline=None,
+            use_gpu=False,
+        )
+        assert result == "test_voice"
+
+    def test_formula_voice_without_pipeline(self):
+        """Test formula voice returns spec when no pipeline."""
+        result = resolve_voice(
+            voice_spec="model*0.5+0.3*other",
+            pipeline=None,
+            use_gpu=False,
+        )
+        assert result == "model*0.5+0.3*other"
+
+    def test_caching(self):
+        """Test that voices are cached."""
+        cache = VoiceCache()
+        
+        # First call should load (we'll mock with simple name)
+        result1 = resolve_voice(
+            voice_spec="test_voice",
+            pipeline=None,
+            use_gpu=False,
+            cache=cache,
+        )
+        assert result1 == "test_voice"
+        assert cache.contains("test_voice")
+        
+        # Second call should use cache
+        result2 = resolve_voice(
+            voice_spec="test_voice",
+            pipeline=None,
+            use_gpu=False,
+            cache=cache,
+        )
+        assert result2 == "test_voice"
+
+
+class TestLoadVoiceCached:
+    """Tests for load_voice_cached function."""
+
+    def test_simple_voice_name(self):
+        """Test that simple voice names are returned as-is."""
+        result = load_voice_cached(
+            voice_name="test_voice",
+            pipeline=None,
+            use_gpu=False,
+        )
+        assert result == "test_voice"
+
+    def test_dict_cache(self):
+        """Test caching with dict."""
+        cache = {}
+        
+        result1 = load_voice_cached(
+            voice_name="test_voice",
+            pipeline=None,
+            use_gpu=False,
+            cache=cache,
+        )
+        assert result1 == "test_voice"
+        assert "test_voice" in cache
+        
+        result2 = load_voice_cached(
+            voice_name="test_voice",
+            pipeline=None,
+            use_gpu=False,
+            cache=cache,
+        )
+        assert result2 == "test_voice"
+        assert cache["test_voice"] == "test_voice"
+
+    def test_no_cache(self):
+        """Test without cache parameter."""
+        result = load_voice_cached(
+            voice_name="test_voice",
+            pipeline=None,
+            use_gpu=False,
+            cache=None,
+        )
+        assert result == "test_voice"