sound_loader.py

import threading
from typing import Callable, Optional
import uuid

from sound import Sound
from sound_file_util import SoundFileUtil

class SoundLoader:
    """
    Loads a sound file in a separate thread.
    """

    def __init__(self):
        self._load_id: Optional[str] = None  # Track current load
        self._lock = threading.Lock()
    
    def load_sound(self, path: str, callback: Callable[[Sound | str], None]) -> Optional[str]:
        """
        Load sound only if no load is currently in progress.
        This operation is atomic and free of race conditions.
        Returns load_id if started, None if already loading.
        """
        load_id = str(uuid.uuid4())
        
        with self._lock:
            if self._load_id is not None:
                return None
            self._load_id = load_id
        
        def _load_in_thread():
            load_result = SoundFileUtil.load(path)
                
            with self._lock:
                if self._load_id != load_id:
                    return
                                
            with self._lock:
                if self._load_id == load_id:
                    self._load_id = None
                    callback(load_result)
        
        thread = threading.Thread(target=_load_in_thread, daemon=True)
        thread.start()
        return load_id

    def cancel_load(self):
        """Cancel any pending load."""
        with self._lock:
            self._load_id = None

    @property
    def is_loading(self) -> bool:
        """Check if a load operation is currently in progress."""
        with self._lock:
            return self._load_id is not None