# player.py - AudioLibrary class only (SDL2 imported from _sdl2.py) import os import sys import time import atexit import warnings from collections import defaultdict from typing import * # ============================================================ # Import SDL2 from _sdl2.py (local file, NOT pysdl2 package) # ============================================================ try: from ._sdl2 import * from ._sdl2 import _mix_lib, _sdl_lib except ImportError: try: from _sdl2 import * from _sdl2 import _mix_lib, _sdl_lib except ImportError: raise ImportError( "SDL2 module (_sdl2) not found. " "Please ensure _sdl2.py is in the ap_ds package." ) # ============================================================ # Python 3.15+ lazy import support # ============================================================ # Check if _IS_PYTHON_315_PLUS already exists (imported from other module) try: # Try to get from current module first _IS_PYTHON_315_PLUS = _IS_PYTHON_315_PLUS print(f"ℹ️ _IS_PYTHON_315_PLUS already exists: {_IS_PYTHON_315_PLUS} (from other module)") except NameError: # Not defined yet, get it now _IS_PYTHON_315_PLUS = sys.version_info >= (3, 15) print(f"ℹ️ _IS_PYTHON_315_PLUS defined now: {_IS_PYTHON_315_PLUS} (Python {sys.version_info.major}.{sys.version_info.minor})") if _IS_PYTHON_315_PLUS: try: # Try to use lazy import (Python 3.15+) exec(""" lazy import ctypes lazy import urllib.request lazy import struct lazy import json lazy import typing lazy import shutil lazy import tempfile lazy import subprocess lazy import ssl lazy import hashlib """) print("✅ Using lazy imports (Python 3.15+)") except (SyntaxError, TypeError, NameError) as e: # Fallback to regular imports if lazy import fails print(f"⚠️ Lazy import failed ({e}), falling back to regular imports") import ctypes import urllib.request import struct import json import typing import shutil import tempfile import subprocess import ssl import hashlib else: import ctypes import urllib.request import struct import json import typing import shutil import tempfile import subprocess import ssl import hashlib # ctypes content still usable (already imported) from ctypes import * from typing import * # ============================================================ # Error Codes # ============================================================ AP_DS_SUCCESS = 0 AP_DS_ERR_FILE_NOT_FOUND = 1001 AP_DS_ERR_INVALID_AID = 1002 AP_DS_ERR_AUDIO_LOAD_FAILED = 1003 AP_DS_ERR_PLAYBACK_FAILED = 1004 AP_DS_ERR_SDL_INIT_FAILED = 1005 AP_DS_ERR_MIXER_INIT_FAILED = 1006 AP_DS_ERR_UNSUPPORTED_FORMAT = 1007 AP_DS_ERR_NOT_MUSIC_FILE = 1008 AP_DS_ERR_DAP_INVALID_EXT = 1009 AP_DS_ERR_DAP_SAVE_FAILED = 1010 AP_DS_ERR_METADATA_PARSE_FAILED = 1011 AP_DS_ERR_FADE_NOT_SUPPORTED = 1012 AP_DS_ERR_AUDIO_NOT_LOADED = 1013 AP_DS_ERR_INVALID_SOURCE = 1014 AP_DS_ERR_INVALID_VOLUME = 1015 AP_DS_ERR_SEEK_NOT_SUPPORTED = 1016 AP_DS_ERR_UNKNOWN = 1999 # ============================================================ # WAV Threshold # ============================================================ WAV_THRESHOLD = int(os.environ.get('AP_DS_WAV_THRESHOLD', '6')) if WAV_THRESHOLD >= 30: print(f"Warning: WAV threshold {WAV_THRESHOLD}s is too large. Using default 6s to prevent memory leaks.") WAV_THRESHOLD = 6 elif WAV_THRESHOLD < 0: print(f"Warning: WAV threshold {WAV_THRESHOLD}s is negative. Using default 6s.") WAV_THRESHOLD = 6 print(f"🎵 WAV playback mode threshold: {WAV_THRESHOLD}s (Files >= {WAV_THRESHOLD}s use music mode, < {WAV_THRESHOLD}s use sound effect mode)") # ============================================================ # audio_parser import # ============================================================ AUDIO_PARSER_AVAILABLE = False try: from .audio_parser import * AUDIO_PARSER_AVAILABLE = True except ImportError: try: from audio_parser import * AUDIO_PARSER_AVAILABLE = True except ImportError: print("Warning: audio_parser module not available, using fallback duration methods") # ============================================================ # Opus Support Integration # ============================================================ OPUS_PLAYER_AVAILABLE = False try: from .opusplayer import OpusAudio as _OpusAudio OPUS_PLAYER_AVAILABLE = True except ImportError: try: from opusplayer import OpusAudio as _OpusAudio OPUS_PLAYER_AVAILABLE = True except ImportError: _OpusAudio = None print("Warning: opusplayer module not available, Opus playback disabled") def _is_opus_file(file_path): """Check if a file is an Opus audio file.""" ext = os.path.splitext(str(file_path))[1].lower() return ext == '.opus' # ============================================================ # AudioLibrary Class # ============================================================ class AudioLibrary: def __init__(self, frequency: int = 44100, format: int = MIX_DEFAULT_FORMAT, channels: int = 2, chunksize: int = 2048): """Initialize the audio library""" # Initialize SDL audio if SDL_Init(SDL_INIT_AUDIO) != 0: raise RuntimeError(f"SDL initialization failed") if Mix_OpenAudio(frequency, format, channels, chunksize) != 0: raise RuntimeError(f"Mixer initialization failed") atexit.register(self.cleanup_function) self.MUS_NO_FADING = 0 self.MUS_FADING_IN = 1 self.MUS_FADING_OUT = 2 # Audio state tracking self._audio_cache = {} # File path -> Mix_Chunk self._music_cache = {} # File path -> Mix_Music self._channel_info = {} # Channel ID -> Playback info self._aid_to_filepath = {} # Store AID to file mapping self._aid_counter = 0 self._sample_rate = frequency self._format = format self._channels = channels # Initialize DAP recordings list self._dap_recordings = [] # List of DAP format recordings self._dap_records_set = set() # O(1) deduplication set # Opus playback support self._opus_audio = None # OpusAudio instance (lazy init) self._aid_to_opus_aid = {} # Main AID -> Opus sub-AID mapping def _get_opus_player(self): """Get or create the OpusAudio sub-player instance.""" if self._opus_audio is None: if not OPUS_PLAYER_AVAILABLE: return None self._opus_audio = _OpusAudio() return self._opus_audio def _map_opus_aid(self, main_aid, opus_aid): """Map a main library AID to an Opus sub-library AID.""" self._aid_to_opus_aid[main_aid] = opus_aid return main_aid def _is_opus_aid(self, aid): """Check if an AID is an Opus AID (has a mapped Opus sub-AID).""" return aid in self._aid_to_opus_aid def _get_opus_aid(self, aid): """Get the Opus sub-AID for a main AID.""" return self._aid_to_opus_aid.get(aid) def Delay(self, ms): _sdl_lib.SDL_Delay(ms) # ============================================================ # Batch APIs (passthrough to audio_parser) # ============================================================ def batch_get_metadata(self, file_paths, max_workers=None, show_progress=False): # Expand paths paths = file_paths if isinstance(file_paths, list) else [file_paths] opus_files = [p for p in paths if _is_opus_file(p)] non_opus_files = [p for p in paths if not _is_opus_file(p)] results = [] # Opus files -> use opusplayer (ensure player created) opus_player = self._get_opus_player() if opus_files and OPUS_PLAYER_AVAILABLE and opus_player is not None: results.extend(opus_player.batch_get_metadata(opus_files, max_workers, show_progress)) # Non-Opus files -> use audio_parser if non_opus_files: if not AUDIO_PARSER_AVAILABLE: raise RuntimeError("audio_parser not available") results.extend(batch_get_metadata(non_opus_files, max_workers, show_progress)) return results def batch_get_duration(self, file_paths, max_workers=None): # Expand paths paths = file_paths if isinstance(file_paths, list) else [file_paths] opus_files = [p for p in paths if _is_opus_file(p)] non_opus_files = [p for p in paths if not _is_opus_file(p)] result = {} # Opus files -> use opusplayer (ensure player created) opus_player = self._get_opus_player() if opus_files and OPUS_PLAYER_AVAILABLE and opus_player is not None: result.update(opus_player.batch_get_duration(opus_files, max_workers)) # Non-Opus files -> use audio_parser if non_opus_files: if not AUDIO_PARSER_AVAILABLE: raise RuntimeError("audio_parser not available") result.update(batch_get_duration(non_opus_files, max_workers)) return result def batch_get_metadata_by_type(self, file_paths, file_type, max_workers=None): # If Opus files requested, use opusplayer (ensure player created) opus_player = self._get_opus_player() if OPUS_PLAYER_AVAILABLE and opus_player is not None and file_type.lower() in ('opus', '.opus'): return opus_player.batch_get_metadata_by_type(file_paths, file_type, max_workers) if not AUDIO_PARSER_AVAILABLE: raise RuntimeError("audio_parser not available") return batch_get_metadata_by_type(file_paths, file_type, max_workers) # Core playback functionality ====================================================== def play_audio(self, aid: int) -> Tuple[int, str, str]: """ Play/resume audio with specified AID Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.play_audio(self._get_opus_aid(aid)) channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") info = self._channel_info[channel] if info['paused']: if info['is_music']: Mix_ResumeMusic() else: Mix_Resume(channel) info['paused'] = False info['start_time'] = time.time() - info['paused_position'] return (AP_DS_SUCCESS, "", "") # DAP recording functionality ====================================================== def play_from_file(self, file_path: str, loops: int = 0, start_pos: float = 0.0) -> Union[int, Tuple[int, str, str]]: """ Play audio directly from file, return AID. Note: .ap-ds-dap files are DAP export records (output format) and are NOT supported as playback input. Passing one will return AP_DS_ERR_AUDIO_LOAD_FAILED. Args: file_path: Path to audio file loops: Number of loops start_pos: Starting position Returns: int: Audio ID (AID) on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure """ if not isinstance(file_path, (str, bytes, os.PathLike)): return (AP_DS_ERR_FILE_NOT_FOUND, f"Invalid file path type: {type(file_path).__name__}", "file_path must be a string or os.PathLike") if not isinstance(loops, int) or isinstance(loops, bool): return (AP_DS_ERR_UNKNOWN, f"Invalid loops type: {type(loops).__name__}. Expected int.", "loops must be an integer (-1=infinite, 0=once, >0=count)") if not os.path.exists(file_path): return (AP_DS_ERR_FILE_NOT_FOUND, f"Audio file not found: {file_path}", "Verify the file path exists and is accessible") # Generate AID self._aid_counter += 1 aid = self._aid_counter # Record to DAP list without playing self._add_to_dap_recordings(file_path) self._aid_to_filepath[aid] = file_path # Opus playback: detect and delegate to OpusAudio sub-player if _is_opus_file(file_path): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Opus support not available for: {file_path}", "Install opusplayer module") opus_result = opus.play_from_file(file_path, loops, start_pos) if isinstance(opus_result, int): # Map main AID to Opus sub-AID self._map_opus_aid(aid, opus_result) return aid else: # Opus playback failed, return the Opus error return opus_result # Normal audio file playback (MP3, OGG, FLAC, WAV) # Music file handling if self._is_music_file(file_path): music = Mix_LoadMUS(file_path.encode()) if not music: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load music file: {file_path}", "Check file format and integrity") if Mix_PlayMusic(music, loops) != 0: Mix_FreeMusic(music) return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to play music: {file_path}", "Check audio device and file format") channel = -1 self._music_cache[file_path] = music # Sound effect handling else: audio = Mix_LoadWAV(file_path.encode()) if not audio: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load audio file: {file_path}", "Check file format and integrity") channel = Mix_PlayChannel(-1, audio, loops) if channel == -1: Mix_FreeChunk(audio) return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to play audio: {file_path}", "No available audio channels") self._audio_cache[file_path] = audio # Record playback information self._channel_info[channel] = { 'aid': aid, 'start_time': time.time() - start_pos, 'paused': False, 'file_path': file_path, 'is_music': channel == -1, 'loops': loops } # Seek to specified position if start_pos > 0: self._seek_audio(channel, start_pos) self._aid_to_filepath[aid] = file_path return aid def play_from_memory(self, file_path: str, loops: int = 0, start_pos: float = 0.0) -> Union[int, Tuple[int, str, str]]: """ Play audio from memory cache, return AID. Note: .ap-ds-dap files are DAP export records (output format) and are NOT supported as playback input. Args: file_path: Path to audio file loops: Number of loops start_pos: Starting position Returns: int: Audio ID (AID) on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure """ if not isinstance(file_path, (str, bytes, os.PathLike)): return (AP_DS_ERR_AUDIO_NOT_LOADED, f"Invalid file path type: {type(file_path).__name__}", "file_path must be a string or os.PathLike") if not isinstance(loops, int) or isinstance(loops, bool): return (AP_DS_ERR_UNKNOWN, f"Invalid loops type: {type(loops).__name__}. Expected int.", "loops must be an integer (-1=infinite, 0=once, >0=count)") # Generate AID self._aid_counter += 1 aid = self._aid_counter # Record to DAP list without playing self._add_to_dap_recordings(file_path) self._aid_to_filepath[aid] = file_path # Opus playback: detect and delegate if _is_opus_file(file_path): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Opus support not available for: {file_path}", "Install opusplayer module") opus_result = opus.play_from_file(file_path, loops, start_pos) if isinstance(opus_result, int): self._map_opus_aid(aid, opus_result) return aid else: return opus_result # Normal audio file playback if file_path in self._music_cache: if Mix_PlayMusic(self._music_cache[file_path], loops) != 0: return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to play music from cache: {file_path}", "Check audio device") channel = -1 elif file_path in self._audio_cache: channel = Mix_PlayChannel(-1, self._audio_cache[file_path], loops) if channel == -1: return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to play audio from cache: {file_path}", "No available audio channels") else: return (AP_DS_ERR_AUDIO_NOT_LOADED, f"Audio not loaded in memory: {file_path}", "Call new_aid() first to load the file") # Record playback information self._channel_info[channel] = { 'aid': aid, 'start_time': time.time() - start_pos, 'paused': False, 'file_path': file_path, 'is_music': channel == -1, 'loops': loops } if start_pos > 0: self._seek_audio(channel, start_pos) self._aid_to_filepath[aid] = file_path return aid def new_aid(self, file_path: str) -> Union[int, Tuple[int, str, str]]: """ Generate AID for file. Args: file_path: Path to audio file Returns: int: Audio ID (AID) on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure """ if not isinstance(file_path, (str, bytes, os.PathLike)): return (AP_DS_ERR_FILE_NOT_FOUND, f"Invalid file path type: {type(file_path).__name__}", "file_path must be a string or os.PathLike") if not os.path.exists(file_path): return (AP_DS_ERR_FILE_NOT_FOUND, f"Audio file not found: {file_path}", "Verify the file path exists and is accessible") # Generate AID self._aid_counter += 1 aid = self._aid_counter # Record to DAP list without loading self._add_to_dap_recordings(file_path) self._aid_to_filepath[aid] = file_path # Opus: register in opusplayer if _is_opus_file(file_path): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Opus support not available for: {file_path}", "Install opusplayer module") opus_aid = opus.new_aid(file_path) if isinstance(opus_aid, int): self._map_opus_aid(aid, opus_aid) return aid # Normal audio file loading if self._is_music_file(file_path): if file_path not in self._music_cache: music = Mix_LoadMUS(file_path.encode()) if not music: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load music file: {file_path}", "Check file format and integrity") self._music_cache[file_path] = music else: if file_path not in self._audio_cache: audio = Mix_LoadWAV(file_path.encode()) if not audio: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load audio file: {file_path}", "Check file format and integrity") self._audio_cache[file_path] = audio self._aid_to_filepath[aid] = file_path return aid # Music fade control functionality ====================================================== def fadein_music(self, aid: int, loops: int = -1, ms: int = 0) -> Tuple[int, str, str]: """ Fade in music (basic fade in) Args: aid: AID of the music file loops: Number of loops, -1 for infinite, 0 for no loop, >0 for loop count ms: Fade in time in milliseconds Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.fadein_music(self._get_opus_aid(aid), loops, ms) # Find AID corresponding music info for channel, info in self._channel_info.items(): if info['aid'] == aid and info['is_music']: file_path = info['file_path'] # Record to DAP (important!) self._add_to_dap_recordings(file_path) # Load music file music = self._music_cache.get(file_path) if not music: music = Mix_LoadMUS(file_path.encode()) if not music: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load music for fadein: {file_path}", "Check file format and integrity") self._music_cache[file_path] = music # Stop current playback Mix_HaltMusic() # Validate fade parameters if not isinstance(ms, int) or not isinstance(loops, int): return (AP_DS_ERR_UNKNOWN, "Invalid fade parameters: ms and loops must be integers", "Check fade parameters (ms: int milliseconds, loops: int)") # Start fade in result = Mix_FadeInMusic(music, loops, ms) if result == 0: # Update playback info info['start_time'] = time.time() info['paused'] = False info['loops'] = loops return (AP_DS_SUCCESS, "", "") else: error_msg = SDL_GetError().decode() if SDL_GetError() else 'Unknown error' return (AP_DS_ERR_PLAYBACK_FAILED, f"Mix_FadeInMusic failed: {error_msg}", "Check SDL_mixer compatibility") return (AP_DS_ERR_INVALID_AID, f"AID {aid} not found or not a music file", "Verify the AID corresponds to a music file") def fadein_music_pos(self, aid: int, loops: int = -1, ms: int = 0, position: float = 0.0) -> Tuple[int, str, str]: """ Fade in music from specified position Args: aid: AID of the music file loops: Number of loops, -1 for infinite ms: Fade in time in milliseconds position: Starting position in seconds Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Check if function exists if not hasattr(_mix_lib, 'Mix_FadeInMusicPos'): return (AP_DS_ERR_FADE_NOT_SUPPORTED, "Mix_FadeInMusicPos not supported in this SDL_mixer version", "Update SDL_mixer or use fadein_music()") # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.fadein_music(self._get_opus_aid(aid), loops, ms) # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.fadein_music_pos(self._get_opus_aid(aid), loops, ms, position) # Find AID corresponding music info for channel, info in self._channel_info.items(): if info['aid'] == aid and info['is_music']: file_path = info['file_path'] # Record to DAP (important!) self._add_to_dap_recordings(file_path) # Load music file music = self._music_cache.get(file_path) if not music: music = Mix_LoadMUS(file_path.encode()) if not music: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load music for fadein: {file_path}", "Check file format and integrity") self._music_cache[file_path] = music # Stop current playback Mix_HaltMusic() # Validate fade parameters if not isinstance(ms, int) or not isinstance(loops, int) or not isinstance(position, (int, float)): return (AP_DS_ERR_UNKNOWN, "Invalid fade parameters: ms/loops must be int, position must be a number", "Check fade parameters") # Start fade in from specified position result = Mix_FadeInMusicPos(music, loops, ms, position) if result == 0: # Update playback info info['start_time'] = time.time() - position info['paused'] = False info['loops'] = loops return (AP_DS_SUCCESS, "", "") else: error_msg = SDL_GetError().decode() if SDL_GetError() else 'Unknown error' return (AP_DS_ERR_PLAYBACK_FAILED, f"Mix_FadeInMusicPos failed: {error_msg}", "Check SDL_mixer compatibility") return (AP_DS_ERR_INVALID_AID, f"AID {aid} not found or not a music file", "Verify the AID corresponds to a music file") def fadeout_music(self, ms: int = 0) -> Tuple[int, str, str]: """ Fade out currently playing music Args: ms: Fade out time in milliseconds Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if Opus is playing if self._opus_audio is not None and self._opus_audio.is_music_playing(): return self._opus_audio.fadeout_music(ms) result = Mix_FadeOutMusic(ms) if result == 1: return (AP_DS_SUCCESS, "", "") else: return (AP_DS_ERR_PLAYBACK_FAILED, "Mix_FadeOutMusic failed or no music playing", "Ensure music is playing before calling fadeout") def is_music_playing(self) -> bool: """ Check if music is currently playing Returns: bool: True if playing, False otherwise """ # Check Opus sub-player first if self._opus_audio is not None and self._opus_audio.is_music_playing(): return True return Mix_PlayingMusic() == 1 def is_music_paused(self) -> bool: """ Check if music is paused Returns: bool: True if paused, False otherwise """ # Check Opus sub-player first if self._opus_audio is not None and self._opus_audio.is_music_paused(): return True return Mix_PausedMusic() == 1 def get_music_fading(self) -> int: """ Get fade state of music Returns: int: State code - 0 (MUS_NO_FADING): No fade - 1 (MUS_FADING_IN): Fading in - 2 (MUS_FADING_OUT): Fading out """ # Check Opus sub-player first if self._opus_audio is not None and self._opus_audio.get_music_fading() != 0: return self._opus_audio.get_music_fading() return Mix_FadingMusic() # ============================================================ # DAP Recording System (v3.1.0+ with O(1) deduplication + fallback) # ============================================================ def _add_to_dap_recordings(self, file_path: str) -> None: """ Add audio file metadata to DAP (Dvs Audio Playlist) records. Primary: O(1) set-based deduplication. Fallback: O(n) linear scan if set fails. Args: file_path: Path to the audio file """ try: # Get metadata using audio_parser metadata = self.get_audio_metadata_by_path(file_path) if not metadata or not isinstance(metadata, dict): return # Create DAP record record = { 'path': file_path, 'duration': metadata.get('duration', 0), 'bitrate': metadata.get('bitrate', 0), 'channels': metadata.get('channels', 2) } # Primary: O(1) set-based deduplication if file_path not in self._dap_records_set: self._dap_records_set.add(file_path) self._dap_recordings.append(record) print(f"📝 Recorded DAP file: {file_path}") return except Exception as e: # Fallback: O(n) linear scan if set fails print(f"⚠️ DAP set deduplication failed: {e}, falling back to O(n) scan") # ============================================================ # Fallback: O(n) linear scan (redundant backup) # ============================================================ try: metadata = self.get_audio_metadata_by_path(file_path) if not metadata or not isinstance(metadata, dict): return record = { 'path': file_path, 'duration': metadata.get('duration', 0), 'bitrate': metadata.get('bitrate', 0), 'channels': metadata.get('channels', 2) } # O(n) linear deduplication if not any(r.get('path') == file_path for r in self._dap_recordings): self._dap_recordings.append(record) print(f"📝 Recorded DAP file (fallback): {file_path}") except Exception as e: print(f"⚠️ Failed to record DAP (fallback also failed): {e}") def save_dap_to_json(self, save_path: str) -> Tuple[int, str, str]: """ Save DAP recordings to JSON file. ONLY WHEN USER CALLS THIS FUNCTION. Args: save_path: Path to save JSON file Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ try: # Validate file extension if not save_path.lower().endswith('.ap-ds-dap'): return (AP_DS_ERR_DAP_INVALID_EXT, f"Invalid file extension. Expected '.ap-ds-dap' but got '{os.path.splitext(save_path)[1]}'", "Use .ap-ds-dap extension for DAP files") # Save to JSON with UTF-8 encoding with open(save_path, 'w', encoding='utf-8') as f: json.dump(self._dap_recordings, f, ensure_ascii=False, indent=2) print(f"✅ Saved {len(self._dap_recordings)} DAP records to: {save_path}") return (AP_DS_SUCCESS, "", "") except Exception as e: return (AP_DS_ERR_DAP_SAVE_FAILED, f"Error saving DAP to JSON: {str(e)}", "Check write permissions and disk space") def get_dap_recordings(self) -> List[Dict]: """ Get current DAP recordings from memory. Returns: List[Dict]: List of DAP records """ return self._dap_recordings.copy() # ============================================================ # clear_dap_recordings() - Clear both list and set # ============================================================ def clear_dap_recordings(self) -> None: """Clear all DAP recordings from memory.""" self._dap_recordings.clear() self._dap_records_set.clear() print("🗑️ Cleared all DAP recordings") def _is_music_file(self, file_path: str) -> bool: """ Check if file should be treated as music file. For WAV files: - Duration < threshold (default 6s) → sound effect mode - Duration >= threshold → music mode For other formats: MP3, OGG, FLAC → music mode Others (non-WAV) → sound effect mode Args: file_path: Path to audio file Returns: bool: True if music mode, False if sound effect mode """ ext = os.path.splitext(file_path)[1].lower() # Non-WAV formats if ext in ['.mp3', '.ogg', '.flac']: return True # WAV files - check duration using existing metadata API if ext == '.wav': try: # Use existing metadata method to get duration metadata = self.get_audio_metadata_by_path(file_path) if metadata and 'duration' in metadata: duration = metadata['duration'] # Use global WAV_THRESHOLD return duration >= WAV_THRESHOLD else: # If metadata not available, default to music mode for safety print(f"Warning: Could not get duration for WAV file: {file_path}, defaulting to music mode") return True except Exception as e: print(f"Warning: Error getting WAV metadata for {file_path}: {e}") # Default to music mode for safety if metadata fetch fails return True # Other formats (AIF, AU, etc.) - sound effect mode return False # Control functionality ====================================================== def pause_audio(self, aid: int) -> Tuple[int, str, str]: """ Pause audio with specified AID Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.pause_audio(self._get_opus_aid(aid)) channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") info = self._channel_info[channel] if not info['paused']: if info['is_music']: Mix_PauseMusic() else: Mix_Pause(channel) info['paused'] = True info['paused_position'] = time.time() - info['start_time'] return (AP_DS_SUCCESS, "", "") def stop_audio(self, aid: int) -> Union[float, Tuple[int, str, str]]: """ Stop playback and return played duration Returns: float: Played duration in seconds on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") result = opus.stop_audio(self._get_opus_aid(aid)) # Remove AID mapping after stop self._aid_to_opus_aid.pop(aid, None) return result channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") info = self._channel_info[channel] played_time = time.time() - info['start_time'] if not info['paused'] else info['paused_position'] if info['is_music']: Mix_HaltMusic() else: Mix_HaltChannel(channel) del self._channel_info[channel] return played_time def seek_audio(self, aid: int, position: float) -> Tuple[int, str, str]: """ Seek to specified position (seconds) Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.seek_audio(self._get_opus_aid(aid), position) channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") if not isinstance(position, (int, float)): return (AP_DS_ERR_UNKNOWN, f"Invalid position type: {type(position).__name__}. Expected int or float.", "Position must be a number (seconds)") return self._seek_audio(channel, position) def _seek_audio(self, channel: int, position: float) -> Tuple[int, str, str]: """Internal method: seek audio""" info = self._channel_info.get(channel) if not info: return (AP_DS_ERR_INVALID_AID, f"Channel {channel} not found", "Internal error - channel not in tracking") # Music seeking if info['is_music']: Mix_HaltMusic() # Reuse the cached music object to avoid re-decoding the whole # file and leaking the previous Mix_Music instance on every seek. music = self._music_cache.get(info['file_path']) if not music: music = Mix_LoadMUS(info['file_path'].encode()) if not music: return (AP_DS_ERR_AUDIO_LOAD_FAILED, f"Failed to load music for seeking: {info['file_path']}", "Check file format and integrity") self._music_cache[info['file_path']] = music if Mix_PlayMusic(music, info['loops']) != 0: return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to reload music for seeking: {info['file_path']}", "Check file integrity") if hasattr(Mix_SetMusicPosition, '__call__'): Mix_SetMusicPosition(position) self._channel_info[channel] = { **info, 'start_time': time.time() - position, 'paused': False } return (AP_DS_SUCCESS, "", "") # Sound effect seeking - not supported else: Mix_HaltChannel(channel) audio = self._audio_cache.get(info['file_path']) if audio is None: return (AP_DS_ERR_AUDIO_NOT_LOADED, f"Audio not in cache: {info['file_path']}", "Call new_aid() to reload the file") new_channel = Mix_PlayChannel(-1, audio, info['loops']) if new_channel == -1: return (AP_DS_ERR_PLAYBACK_FAILED, "Failed to replay audio after seek", "No available audio channels") self._channel_info[new_channel] = { **info, 'start_time': time.time() - position, 'paused': False } if new_channel != channel: del self._channel_info[channel] return (AP_DS_SUCCESS, "", "") def set_volume(self, aid: int, volume: int) -> Tuple[int, str, str]: """ Set audio volume Args: aid: Audio ID volume: Volume value (0-128) Returns: Tuple[int, str, str]: (AP_DS_SUCCESS, "", "") on success, (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.set_volume(self._get_opus_aid(aid), volume) channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") if not isinstance(volume, int): return (AP_DS_ERR_INVALID_VOLUME, f"Invalid volume type: {type(volume).__name__} (must be integer 0-128)", "Volume must be an integer between 0 and 128") if volume < 0 or volume > 128: return (AP_DS_ERR_INVALID_VOLUME, f"Invalid volume: {volume} (must be 0-128)", "Volume range is 0-128") info = self._channel_info[channel] if info['is_music']: result = Mix_VolumeMusic(volume) if result != -1: return (AP_DS_SUCCESS, "", "") else: return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to set music volume", "Check audio device") else: result = Mix_Volume(channel, volume) if result != -1: return (AP_DS_SUCCESS, "", "") else: return (AP_DS_ERR_PLAYBACK_FAILED, f"Failed to set channel volume", "Check audio device") def get_volume(self, aid: int) -> Union[int, Tuple[int, str, str]]: """ Get current audio volume Args: aid: Audio ID Returns: int: Current volume value (0-128) on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure """ # Opus playback: delegate to OpusAudio sub-player if self._is_opus_aid(aid): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_PLAYBACK_FAILED, "Opus player not available", "Install opusplayer module") return opus.get_volume(self._get_opus_aid(aid)) channel = self._find_channel_by_aid(aid) if channel is None: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid and the audio is loaded") info = self._channel_info[channel] if info['is_music']: return Mix_VolumeMusic(-1) # -1 means get without setting else: return Mix_Volume(channel, -1) # -1 means get without setting # Helper methods ====================================================== def _find_channel_by_aid(self, aid: int) -> Optional[int]: """Find channel by AID""" for channel, info in self._channel_info.items(): if info['aid'] == aid: return channel return None def _get_file_path_by_aid(self, aid: int) -> Union[str, Tuple[int, str, str]]: """Get file path by AID""" for info in self._channel_info.values(): if info['aid'] == aid: return info['file_path'] return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid") def get_audio_duration(self, source: Union[str, int], is_file: bool = False) -> Union[int, Tuple[int, str, str]]: """ Get the duration of an audio file in seconds This method supports both file paths and AID (Audio ID) as input sources. It automatically detects the file format and uses the appropriate parser to calculate the duration accurately. Args: source: Either a file path string or an integer AID is_file: If True, treats source as file path; if False, as AID Returns: int: Duration in seconds (rounded) on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure Examples: >>> # Get duration by file path >>> duration = lib.get_audio_duration("audio/song.mp3", is_file=True) >>> # Get duration by AID >>> duration = lib.get_audio_duration(123, is_file=False) """ try: # Case 1: Get duration by AID (Audio ID) if not is_file and isinstance(source, int): if source not in self._aid_to_filepath: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {source}", "Check that the AID is valid") file_path = self._aid_to_filepath[source] return self._get_duration_by_filepath(file_path) # Case 2: Get duration by file path file_path = str(source) return self._get_duration_by_filepath(file_path) except Exception as e: return (AP_DS_ERR_UNKNOWN, f"Error getting audio duration: {str(e)}", "Check file integrity and try again") def _get_duration_by_filepath(self, file_path: str) -> Union[int, Tuple[int, str, str]]: try: if not os.path.exists(file_path): return (AP_DS_ERR_FILE_NOT_FOUND, f"File not found: {file_path}", "Verify the file path exists") # Opus duration: use opusplayer if _is_opus_file(file_path): opus = self._get_opus_player() if opus is not None: return opus.get_audio_duration(file_path, is_file=True) if AUDIO_PARSER_AVAILABLE: try: from .audio_parser import get_audio_duration duration = get_audio_duration(file_path) if duration > 0: return duration except Exception as e: print(f"Audio parser error: {e}") # Fallback: estimate from file size try: return int(self.simple_mp3_duration_estimation(file_path)) except: return (AP_DS_ERR_METADATA_PARSE_FAILED, f"Could not determine duration for {file_path}", "File may be corrupted or unsupported format") except Exception as e: return (AP_DS_ERR_UNKNOWN, f"Error calculating audio duration: {str(e)}", "Check file integrity and try again") def simple_mp3_duration_estimation(self, filename: str) -> float: """ Estimate MP3 duration based on file size and common bitrates This provides a fallback when frame-by-frame parsing fails. Args: filename: Path to the MP3 file Returns: float: Estimated duration in seconds, 0 on error """ try: file_size = os.path.getsize(filename) # Estimate audio data size (subtract possible ID3 tag) audio_data_size = max(file_size - 2048, file_size * 0.98) # Estimate bitrate based on file size if file_size < 2 * 1024 * 1024: # < 2MB bitrate = 128 elif file_size < 5 * 1024 * 1024: # < 5MB bitrate = 192 elif file_size < 10 * 1024 * 1024: # < 10MB bitrate = 256 else: bitrate = 320 # Calculate duration: (file_size_bytes * 8) / (bitrate_bps) duration = (audio_data_size * 8) / (bitrate * 1000) return duration except Exception as e: print(f"MP3 duration estimation error: {e}") return 0 def get_audio_metadata_by_aid(self, aid: int) -> Union[Dict, Tuple[int, str, str]]: """ Obtain complete metadata of audio files based on AID Args: aid: Audio ID Returns: Dict: Dictionary containing complete audio metadata on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure Contains fields: path, format, duration, length, sample_rate, channels, bitrate """ try: if aid not in self._aid_to_filepath: return (AP_DS_ERR_INVALID_AID, f"Invalid AID: {aid}", "Check that the AID is valid") file_path = self._aid_to_filepath[aid] # Opus metadata: use opusplayer (delegate to its class method) if _is_opus_file(file_path): opus = self._get_opus_player() if opus is None: return (AP_DS_ERR_METADATA_PARSE_FAILED, "Opus player not available", "Install opusplayer module") return opus.get_audio_metadata_by_path(file_path) if AUDIO_PARSER_AVAILABLE: from .audio_parser import get_audio_metadata result = get_audio_metadata(file_path) if result is None: return (AP_DS_ERR_METADATA_PARSE_FAILED, f"Failed to parse metadata for: {file_path}", "File may be corrupted or unsupported") return result else: return (AP_DS_ERR_METADATA_PARSE_FAILED, "audio_parser not available", "Install audio_parser module") except Exception as e: return (AP_DS_ERR_UNKNOWN, f"Error getting audio metadata by AID: {str(e)}", "Check file integrity and try again") def get_audio_metadata_by_path(self, file_path: str) -> Union[Dict, Tuple[int, str, str]]: """ Directly obtain the complete metadata of an audio file based on its file path Args: file_path: Audio file path Returns: dict: Dictionary containing complete audio metadata on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure Fields included: path, format, duration, length, sample_rate, channels, bitrate """ try: if not os.path.exists(file_path): return (AP_DS_ERR_FILE_NOT_FOUND, f"File not found: {file_path}", "Verify the file path exists") # Opus metadata: use opusplayer if _is_opus_file(file_path): opus = self._get_opus_player() if opus is not None: return opus.get_audio_metadata_by_path(file_path) if AUDIO_PARSER_AVAILABLE: from .audio_parser import get_audio_metadata result = get_audio_metadata(file_path) if result is None: return (AP_DS_ERR_METADATA_PARSE_FAILED, f"Failed to parse metadata for: {file_path}", "File may be corrupted or unsupported") return result else: return (AP_DS_ERR_METADATA_PARSE_FAILED, "audio_parser not available", "Install audio_parser module") except Exception as e: return (AP_DS_ERR_UNKNOWN, f"Error getting audio metadata by path: {str(e)}", "Check file integrity and try again") def get_audio_metadata(self, source: Union[str, int], is_file: bool = False) -> Union[Dict, Tuple[int, str, str]]: """ Retrieve audio metadata based on AID (Audio Identifier) or audio file path. Args: source: AID or audio file path. is_file: If True, treats source as file path; if False, as AID Returns: Dict: Dictionary containing complete audio metadata on success Tuple[int, str, str]: (error_code, error_msg, suggestion) on failure Contains fields: path, format, duration, length, sample_rate, channels, bitrate """ if is_file or isinstance(source, str): return self.get_audio_metadata_by_path(str(source)) elif isinstance(source, int): return self.get_audio_metadata_by_aid(source) else: return (AP_DS_ERR_INVALID_SOURCE, f"Invalid source type: {type(source)}. Expected str or int.", "Use file path (str) or AID (int)") def _get_sample_rate(self, source: Union[str, int]) -> int: """Get the actual sample rate from audio metadata. Args: source: Audio source (file path string or AID integer) Returns: int: Sample rate in Hz. Returns 44100 as fallback if metadata not available. """ try: metadata = self.get_audio_metadata(source, is_file=isinstance(source, str)) if isinstance(metadata, dict) and 'sample_rate' in metadata: return metadata['sample_rate'] except Exception: pass return 44100 # Fallback default value def _get_channels(self, source: Union[str, int]) -> int: """Get the actual channel count from audio metadata. Args: source: Audio source (file path string or AID integer) Returns: int: Number of audio channels. Returns 2 as fallback if metadata not available. """ try: metadata = self.get_audio_metadata(source, is_file=isinstance(source, str)) if isinstance(metadata, dict) and 'channels' in metadata: return metadata['channels'] except Exception: pass return 2 # Fallback default value def _get_aid_for_audio(self, file_path: str) -> Union[int, Tuple[int, str, str]]: """Find corresponding AID by file path""" for channel, info in self._channel_info.items(): if info.get('file_path') == file_path and not info.get('is_music', True): return info.get('aid') return (AP_DS_ERR_INVALID_AID, f"No AID found for file: {file_path}", "File may not be loaded or is a music file") def _get_aid_for_music(self, file_path: str) -> Union[int, Tuple[int, str, str]]: """Find corresponding AID by music file path""" for channel, info in self._channel_info.items(): if info.get('file_path') == file_path and info.get('is_music', False): return info.get('aid') return (AP_DS_ERR_INVALID_AID, f"No AID found for music file: {file_path}", "File may not be loaded or is not a music file") def _get_playing_duration(self, aid: int) -> float: """Get total duration of playing audio""" file_path = self._get_file_path_by_aid(aid) return self._get_file_duration(file_path) if file_path else 0.0 def _get_file_duration(self, file_path: str) -> float: result = self._get_duration_by_filepath(file_path) if isinstance(result, tuple): return 0.0 return float(result) # Resource management ====================================================== def clear_memory_cache(self) -> None: """Clear memory cache""" for audio in self._audio_cache.values(): if audio: Mix_FreeChunk(audio) for music in self._music_cache.values(): if music: Mix_FreeMusic(music) self._audio_cache.clear() self._music_cache.clear() def cleanup_function(self): """Clean up resources""" self.clear_memory_cache() Mix_CloseAudio() SDL_Quit()