LFV
v4.0.0
player.py
46.2 KB
# 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")
# ============================================================
# 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
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):
if not AUDIO_PARSER_AVAILABLE:
raise RuntimeError("audio_parser not available")
return batch_get_metadata(file_paths, max_workers, show_progress)
def batch_get_duration(self, file_paths, max_workers=None):
if not AUDIO_PARSER_AVAILABLE:
raise RuntimeError("audio_parser not available")
return batch_get_duration(file_paths, max_workers)
def batch_get_metadata_by_type(self, file_paths, file_type, max_workers=None):
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
"""
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
# 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
# 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
# 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
"""
# 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()")
# 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
"""
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
"""
return Mix_PlayingMusic() == 1
def is_music_paused(self) -> bool:
"""
Check if music is paused
Returns:
bool: True if paused, False otherwise
"""
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
"""
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
"""
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
"""
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
"""
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
"""
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
"""
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")
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]
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")
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()