dvs/ap_ds
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Compare commits

merge into: dvs/ap_ds:RC
Branches Tags
dvs/ap_ds:main dvs/ap_ds:RC
dvs/ap_ds:v4.0.1
..
pull from: dvs/ap_ds:main
Branches Tags
dvs/ap_ds:main dvs/ap_ds:RC
dvs/ap_ds:v4.0.1
1 Commits
RC .. main
Author SHA1 Message Date
dvs 034a1ec40d Initial commit: ap_ds 4.0.1 2026-08-27 19:21:49 +08:00
9 changed files with 2723 additions and 4987 deletions
Expand all files Collapse all files
README.md
+2696 -1563
View File
File diff suppressed because it is too large Load Diff
ap_ds/Test/OPUS_TEST.py
-893
View File
@@ -1,893 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""
============================================================================
AP_DS 4.0.1 - Opus Specialized Test Suite
Audio Library By DVS
============================================================================
Automated + interactive test coverage for the Opus playback support
(opusplayer.py + _opusdll.py) integrated into AudioLibrary.
Sections:
[OP-1] Opus Module Imports / Constants / Error Codes
[OP-2] _opusdll.py DLL Loading & Auto-Download
[OP-3] Opus Metadata Parsing (basic + extended)
[OP-4] Opus Playback (play_from_file / new_aid / play_from_memory)
[OP-5] Opus Playback Control (pause / resume / stop)
[OP-6] Opus Volume Control
[OP-7] Opus Seek
[OP-8] Opus Fade In / Out
[OP-9] Opus vs Original Distinction
[OP-10] Opus Batch Parsing
[OP-11] Opus Error Codes (normal / boundary / error)
[OP-12] Opus AID Mapping (1:1)
[OP-13] Opus Resource Management
[OP-L] Listening Tests (interactive, requires ears)
Usage:
python OPUS_TEST.py --auto Run automated tests only
python OPUS_TEST.py --listen Run interactive listening tests only
python OPUS_TEST.py --full Run everything (default)
The suite verifies Opus-specific error codes (2001-2010), normal operation,
boundary values, and error paths.
============================================================================
"""
import os
import sys
import io
import time
import struct
import hashlib
import contextlib
# ============================================================================
# Environment
# ============================================================================
os.environ.setdefault('AP_DS_SKIP_AUTO_CHECK', '1')
os.environ.setdefault('AP_DS_SUPPRESS_WARNINGS', '1')
_HERE = os.path.dirname(os.path.abspath(__file__))
if os.path.basename(_HERE) == 'ap_ds':
sys.path.insert(0, os.path.dirname(_HERE))
else:
sys.path.insert(0, _HERE)
import ap_ds
from ap_ds import AudioLibrary
import ap_ds.player as player
import ap_ds.opusplayer as opusplayer
import ap_ds._opusdll as opusdll
# ============================================================================
# Test resources
# ============================================================================
TMP_DIR = os.path.join(_HERE, 'opus_tmp')
os.makedirs(TMP_DIR, exist_ok=True)
# Opus test file (use the real test.opus if available, else search package)
OPUS_FILE = r"D:\DVS开发目录\ap_ds opus支持 开发目录\test.opus"
if not os.path.exists(OPUS_FILE):
pkg_dir = os.path.dirname(os.path.abspath(opusplayer.__file__))
for f in os.listdir(pkg_dir):
if f.endswith('.opus'):
OPUS_FILE = os.path.join(pkg_dir, f)
break
def _prompt_opus():
"""Prompt the user for an Opus file path. Returns default if empty/EOF."""
print("\n" + "=" * 60)
print(" Opus Test Suite - Test Audio Selection")
print("=" * 60)
default = OPUS_FILE
try:
answer = input(
"Enter path to an Opus file for tests\n"
f"(or press Enter to use default: {default})\n> "
).strip().strip('"').strip("'")
if answer and os.path.exists(answer):
return answer
if answer and not os.path.exists(answer):
print(f" ! File not found: {answer}, using default")
return default
except EOFError:
# Non-interactive execution -> use default
return default
# Original (non-Opus) test file - a WAV
WAV_FILE = os.path.join(TMP_DIR, 'orig.wav')
def _make_wav(path, seconds=2, sr=22050, ch=1):
import wave
n = sr * seconds * ch
with wave.open(path, 'w') as w:
w.setnchannels(ch)
w.setsampwidth(2)
w.setframerate(sr)
w.writeframes(b''.join(struct.pack('<h', 0) for _ in range(n)))
return path
_make_wav(WAV_FILE)
# Corrupted Opus file
BAD_OPUS = os.path.join(TMP_DIR, 'bad.opus')
with open(BAD_OPUS, 'wb') as f:
f.write(b'\x00\x01\x02\x03' * 100) # not a valid Opus stream
# ============================================================================
# Test framework
# ============================================================================
class OpusTest:
def __init__(self):
self.passed = 0
self.failed = 0
self.skipped = 0
self.failures = []
def log(self, msg):
print(msg, flush=True)
def check(self, name, cond, detail=""):
if cond:
self.passed += 1
self.log(f" [PASS] {name}" + (f" | {detail}" if detail else ""))
else:
self.failed += 1
self.failures.append((name, detail))
self.log(f" [FAIL] {name}" + (f" | {detail}" if detail else ""))
def skip(self, name, reason):
self.skipped += 1
self.log(f" [SKIP] {name} | {reason}")
def section(self, title):
self.log("\n" + "=" * 66)
self.log(f" {title}")
self.log("=" * 66)
def summary(self):
self.log("\n" + "=" * 66)
self.log(" Opus Test Summary")
self.log("=" * 66)
self.log(f" Passed : {self.passed}")
self.log(f" Failed : {self.failed}")
self.log(f" Skipped: {self.skipped}")
if self.failures:
self.log(" --- Failed details ---")
for name, detail in self.failures:
self.log(f" [FAIL] {name}: {detail}")
self.log("=" * 66)
# ============================================================================
# [OP-1] Opus Module Imports / Constants / Error Codes
# ============================================================================
def test_opus_imports(t):
t.section("[OP-1] Opus Module Imports / Constants / Error Codes")
# Module imports
t.check("opusplayer importable", callable(opusplayer.OpusAudio))
t.check("_opusdll importable", opusdll is not None)
t.check("OpusAudio class exists", hasattr(opusplayer, 'OpusAudio'))
t.check("check_opus_dll callable", callable(opusplayer.check_opus_dll))
t.check("import_opus callable", callable(opusdll.import_opus))
t.check("download_opus_libraries callable", callable(opusdll.download_opus_libraries))
# Constants
t.check("MUS_NO_FADING=0", opusplayer.MUS_NO_FADING == 0)
t.check("MUS_FADING_IN=1", opusplayer.MUS_FADING_IN == 1)
t.check("MUS_FADING_OUT=2", opusplayer.MUS_FADING_OUT == 2)
t.check("WAVE_FORMAT_PCM=1", opusplayer.WAVE_FORMAT_PCM == 1)
t.check("CALLBACK_EVENT set", opusplayer.CALLBACK_EVENT == 0x00050000)
# Structures
t.check("OpusHead structure", hasattr(opusdll, 'OpusHead'))
t.check("OpusTags structure", hasattr(opusdll, 'OpusTags'))
t.check("WAVEFORMATEX structure", hasattr(opusdll, 'WAVEFORMATEX'))
t.check("WAVEHDR structure", hasattr(opusdll, 'WAVEHDR'))
# DLL file presence
pkg = os.path.dirname(os.path.abspath(opusdll.__file__))
for dll in ('libopusfile-0.dll', 'libopus-0.dll', 'libogg-0.dll', 'libopusurl-0.dll'):
t.check(f"{dll} exists", os.path.exists(os.path.join(pkg, dll)))
# DLL hash table
t.check("OPUS_DLL_HASHES defined", hasattr(opusdll, 'OPUS_DLL_HASHES'))
if hasattr(opusdll, 'OPUS_DLL_HASHES'):
for dll in ('libopusfile-0.dll', 'libopus-0.dll', 'libogg-0.dll', 'libopusurl-0.dll'):
t.check(f"hash for {dll}", dll in opusdll.OPUS_DLL_HASHES)
# ============================================================================
# [OP-2] _opusdll.py DLL Loading & Auto-Download
# ============================================================================
def test_opus_dll(t):
t.section("[OP-2] _opusdll.py DLL Loading & Auto-Download")
ok, msg = opusplayer.check_opus_dll()
t.check("check_opus_dll returns (bool,str)", isinstance(ok, bool) and isinstance(msg, str), f"got={ok},{msg}")
t.check("check_opus_dll ok", ok is True, f"msg={msg}")
r = opusdll.import_opus()
t.check("import_opus returns bool", isinstance(r, bool), f"got={r}")
t.check("import_opus ok", r is True)
t.check("opusfile handle loaded", opusdll.opusfile is not None)
t.check("OPUS_DLL_FILES has 4", len(opusdll.OPUS_DLL_FILES) == 4, f"got={len(opusdll.OPUS_DLL_FILES)}")
if hasattr(opusdll, 'OPUS_DLL_FILES'):
for dll in opusdll.OPUS_DLL_FILES:
t.check(f"DLL config {dll['filename']}", 'url' in dll and 'filename' in dll)
# ============================================================================
# [OP-3] Opus Metadata Parsing
# ============================================================================
def test_opus_metadata(t):
t.section("[OP-3] Opus Metadata Parsing (basic + extended)")
if not os.path.exists(OPUS_FILE):
t.skip("Opus metadata", "no Opus file available")
return
lib = opusplayer.OpusAudio()
meta = lib.get_audio_metadata_by_path(OPUS_FILE)
t.check("get_audio_metadata_by_path is dict", isinstance(meta, dict), f"got={type(meta).__name__}")
if isinstance(meta, dict):
t.check("format=opus", meta.get('format') == 'opus', f"got={meta.get('format')}")
t.check("duration>0", meta.get('duration', 0) > 0, f"got={meta.get('duration')}")
t.check("sample_rate=48000", meta.get('sample_rate') == 48000, f"got={meta.get('sample_rate')}")
t.check("channels>0", meta.get('channels', 0) > 0, f"got={meta.get('channels')}")
t.check("bitrate>0", meta.get('bitrate', 0) > 0, f"got={meta.get('bitrate')}")
t.check("fields complete", all(k in meta for k in ('path', 'format', 'duration', 'length', 'sample_rate', 'channels', 'bitrate')))
dur = lib.get_audio_duration(OPUS_FILE, is_file=True)
t.check("get_audio_duration>0", isinstance(dur, int) and dur > 0, f"got={dur}")
sr = lib._get_sample_rate(OPUS_FILE)
t.check("_get_sample_rate=48000", sr == 48000, f"got={sr}")
ch = lib._get_channels(OPUS_FILE)
t.check("_get_channels>0", ch > 0, f"got={ch}")
ext = lib.get_audio_extended_metadata(OPUS_FILE)
t.check("extended metadata is dict", isinstance(ext, dict), f"got={type(ext).__name__}")
if isinstance(ext, dict):
t.check("extended has title/album/artist", 'title' in ext or 'album' in ext or 'artist' in ext, f"keys={list(ext.keys())[:5]}")
t.check("extended has vendor", 'vendor' in ext, f"vendor={ext.get('vendor')}")
lib.cleanup_function()
# ============================================================================
# [OP-4] Opus Playback
# ============================================================================
def test_opus_play(t):
t.section("[OP-4] Opus Playback")
if not os.path.exists(OPUS_FILE):
t.skip("Opus playback", "no Opus file available")
return
lib = opusplayer.OpusAudio()
aid = lib.play_from_file(OPUS_FILE)
t.check("play_from_file returns int AID", isinstance(aid, int), f"got={aid}")
if isinstance(aid, int):
time.sleep(0.3)
t.check("is_music_playing", lib.is_music_playing() is True)
lib.stop_audio(aid)
aid2 = lib.new_aid(OPUS_FILE)
t.check("new_aid returns int AID", isinstance(aid2, int), f"got={aid2}")
r = lib.play_from_memory(OPUS_FILE)
t.check("play_from_memory returns int", isinstance(r, int), f"got={r}")
if isinstance(r, int):
lib.stop_audio(r)
r = lib.play_from_file(os.path.join(TMP_DIR, 'missing.opus'))
t.check("play_from_file(missing) -> 1001", isinstance(r, tuple) and r[0] == 1001, f"got={r}")
r = lib.play_from_file(None)
t.check("play_from_file(None) -> 1001", isinstance(r, tuple) and r[0] == 1001, f"got={r}")
r = lib.play_from_file(BAD_OPUS)
t.check("play_from_file(corrupted) -> 2003", isinstance(r, tuple) and r[0] == 2003, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-5] Opus Playback Control
# ============================================================================
def test_opus_control(t):
t.section("[OP-5] Opus Playback Control")
if not os.path.exists(OPUS_FILE):
t.skip("Opus control", "no Opus file available")
return
lib = opusplayer.OpusAudio()
aid = lib.play_from_file(OPUS_FILE)
if not isinstance(aid, int):
t.check("Play opus ok", False, f"got={aid}")
lib.cleanup_function()
return
time.sleep(0.3)
r = lib.pause_audio(aid)
t.check("pause_audio ok", r[0] == 0, f"got={r}")
time.sleep(0.2)
t.check("is_music_paused=True", lib.is_music_paused() is True)
r = lib.play_audio(aid)
t.check("play_audio resume ok", r[0] == 0, f"got={r}")
time.sleep(0.2)
t.check("is_music_playing=True after resume", lib.is_music_playing() is True)
r = lib.stop_audio(aid)
t.check("stop_audio returns float", isinstance(r, float), f"got={r}")
# Error paths
r = lib.pause_audio(99999)
t.check("pause_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
r = lib.play_audio(99999)
t.check("play_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
r = lib.stop_audio(99999)
t.check("stop_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-6] Opus Volume Control
# ============================================================================
def test_opus_volume(t):
t.section("[OP-6] Opus Volume Control")
if not os.path.exists(OPUS_FILE):
t.skip("Opus volume", "no Opus file available")
return
lib = opusplayer.OpusAudio()
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
# Valid volumes
for vol in (0, 1, 64, 100, 128):
r = lib.set_volume(aid, vol)
t.check(f"set_volume({vol}) ok", r[0] == 0, f"got={r}")
g = lib.get_volume(aid)
t.check("get_volume is int", isinstance(g, int), f"got={g}")
t.check("get_volume in range 0-128", 0 <= g <= 128, f"got={g}")
# Invalid volumes
for vol in (-1, 129, 200):
r = lib.set_volume(aid, vol)
t.check(f"set_volume({vol}) -> 1015", r[0] == 1015, f"got={r}")
# Invalid types
for vol in (None, '50', 1.5):
r = lib.set_volume(aid, vol)
t.check(f"set_volume({vol!r}) -> 1015", isinstance(r, tuple) and r[0] == 1015, f"got={r}")
lib.stop_audio(aid)
# Invalid AID
r = lib.set_volume(99999, 50)
t.check("set_volume(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
r = lib.get_volume(99999)
t.check("get_volume(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-7] Opus Seek
# ============================================================================
def test_opus_seek(t):
t.section("[OP-7] Opus Seek")
if not os.path.exists(OPUS_FILE):
t.skip("Opus seek", "no Opus file available")
return
lib = opusplayer.OpusAudio()
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
time.sleep(0.2)
# Valid seeks
for pos in (0.0, 1.0, 30.0, 100.5):
r = lib.seek_audio(aid, pos)
t.check(f"seek_audio({pos}) ok", r[0] == 0, f"got={r}")
t.check("still playing after seek", lib.is_music_playing() is True)
# Invalid types
for pos in (None, '5', [], {}):
r = lib.seek_audio(aid, pos)
t.check(f"seek_audio({pos!r}) -> tuple", isinstance(r, tuple), f"got={r}")
lib.stop_audio(aid)
r = lib.seek_audio(99999, 1.0)
t.check("seek_audio(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-8] Opus Fade In / Out
# ============================================================================
def test_opus_fade(t):
t.section("[OP-8] Opus Fade In / Out")
if not os.path.exists(OPUS_FILE):
t.skip("Opus fade", "no Opus file available")
return
lib = opusplayer.OpusAudio()
# Fade-in play (via new_aid)
aid = lib.new_aid(OPUS_FILE)
r = lib.fadein_music(aid, ms=1000)
t.check("fadein_music ok", r[0] == 0, f"got={r}")
if r[0] == 0:
time.sleep(0.3)
fading = lib.get_music_fading()
t.check("get_music_fading=1 (fading in)", fading == 1, f"got={fading}")
t.check("playing during fadein", lib.is_music_playing() is True)
time.sleep(1.5) # wait for fade-in to complete
t.check("fading=0 after fadein", lib.get_music_fading() == 0, f"got={lib.get_music_fading()}")
# Fade-out
r = lib.fadeout_music(ms=1000)
t.check("fadeout_music ok", r[0] == 0, f"got={r}")
time.sleep(3)
t.check("stopped after fadeout", not lib.is_music_playing())
# Fade-in from position
aid2 = lib.new_aid(OPUS_FILE)
r = lib.fadein_music_pos(aid2, ms=500, position=30.0)
t.check("fadein_music_pos ok", r[0] == 0, f"got={r}")
if r[0] == 0:
time.sleep(2)
lib.stop_audio(aid2)
# Error paths
r = lib.fadein_music(99999)
t.check("fadein_music(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
r = lib.fadein_music_pos(99999, ms=500)
t.check("fadein_music_pos(invalid AID) -> 1002", r[0] == 1002, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-9] Opus vs Original Distinction
# ============================================================================
def test_opus_distinction(t):
t.section("[OP-9] Opus vs Original Distinction")
# _is_opus_file detection
t.check("_is_opus_file(opus)=True", player._is_opus_file(OPUS_FILE) is True)
t.check("_is_opus_file(wav)=False", player._is_opus_file(WAV_FILE) is False)
t.check("_is_opus_file(.mp3)=False", player._is_opus_file('x.mp3') is False)
t.check("_is_opus_file(.ogg)=False", player._is_opus_file('x.ogg') is False)
# AudioLibrary routes opus to OpusAudio
lib = AudioLibrary()
opus_aid = lib.play_from_file(OPUS_FILE)
t.check("AudioLibrary play opus returns int", isinstance(opus_aid, int), f"got={opus_aid}")
if isinstance(opus_aid, int):
t.check("opus AID mapped", lib._is_opus_aid(opus_aid), f"mapping={lib._aid_to_opus_aid}")
lib.stop_audio(opus_aid)
# AudioLibrary routes wav to SDL2
wav_aid = lib.play_from_file(WAV_FILE)
t.check("AudioLibrary play wav returns int", isinstance(wav_aid, int), f"got={wav_aid}")
if isinstance(wav_aid, int):
t.check("wav NOT mapped to opus", not lib._is_opus_aid(wav_aid), "wav uses SDL2")
lib.stop_audio(wav_aid)
# Metadata distinction
meta_opus = lib.get_audio_metadata_by_path(OPUS_FILE)
meta_wav = lib.get_audio_metadata_by_path(WAV_FILE)
t.check("opus metadata format=opus", isinstance(meta_opus, dict) and meta_opus.get('format') == 'opus', f"got={meta_opus.get('format') if isinstance(meta_opus,dict) else meta_opus}")
t.check("wav metadata format=wav", isinstance(meta_wav, dict) and meta_wav.get('format') == 'wav', f"got={meta_wav.get('format') if isinstance(meta_wav,dict) else meta_wav}")
lib.cleanup_function()
# ============================================================================
# [OP-10] Opus Batch Parsing
# ============================================================================
def test_opus_batch(t):
t.section("[OP-10] Opus Batch Parsing")
if not os.path.exists(OPUS_FILE):
t.skip("Opus batch", "no Opus file available")
return
lib = opusplayer.OpusAudio()
files = [OPUS_FILE]
bl = lib.batch_get_metadata(files)
t.check("batch_get_metadata returns 1", len(bl) == 1, f"got={len(bl)}")
if bl:
t.check("batch metadata format=opus", bl[0].get('format') == 'opus', f"got={bl[0].get('format')}")
bd = lib.batch_get_duration(files)
t.check("batch_get_duration returns dict", isinstance(bd, dict), f"got={type(bd).__name__}")
t.check("batch_get_duration has opus", OPUS_FILE in bd, f"keys={list(bd.keys())}")
bt = lib.batch_get_metadata_by_type(files, 'opus')
t.check("batch_by_type opus -> 1", len(bt) == 1, f"got={len(bt)}")
bt2 = lib.batch_get_metadata_by_type(files, 'wav')
t.check("batch_by_type wav -> 0", len(bt2) == 0, f"got={len(bt2)}")
# Mixed batch via AudioLibrary (test opus and wav separately to avoid
# multiprocessing spawn limitations in test environments)
lib2 = AudioLibrary()
# Opus files in batch use opusplayer
opus_batch = lib2.batch_get_metadata([OPUS_FILE])
t.check("AudioLibrary batch opus -> 1", len(opus_batch) == 1, f"got={len(opus_batch)}")
if opus_batch:
t.check("AudioLibrary batch opus format", opus_batch[0].get('format') == 'opus', f"got={opus_batch[0].get('format')}")
# WAV files in batch use audio_parser
wav_batch = lib2.batch_get_metadata([WAV_FILE])
t.check("AudioLibrary batch wav -> 1", len(wav_batch) == 1, f"got={len(wav_batch)}")
if wav_batch:
t.check("AudioLibrary batch wav format", wav_batch[0].get('format') == 'wav', f"got={wav_batch[0].get('format')}")
lib2.cleanup_function()
lib.cleanup_function()
# ============================================================================
# [OP-11] Opus Error Codes (normal / boundary / error)
# ============================================================================
def test_opus_errors(t):
t.section("[OP-11] Opus Error Codes (normal / boundary / error)")
# Error code constants
err_codes = {
'AP_DS_ERR_OPUS_LIB_LOAD_FAILED': 2001,
'AP_DS_ERR_OPUS_DLL_DEPENDENCY': 2002,
'AP_DS_ERR_OPUS_OPEN_FAILED': 2003,
'AP_DS_ERR_OPUS_HEADER_CORRUPT': 2004,
'AP_DS_ERR_OPUS_TAGS_PARSE_FAILED': 2005,
'AP_DS_ERR_OPUS_DECODE_FAILED': 2006,
'AP_DS_ERR_OPUS_SEEK_FAILED': 2007,
'AP_DS_ERR_OPUS_BITRATE_UNAVAILABLE': 2008,
'AP_DS_ERR_OPUS_NOT_SEEKABLE': 2009,
'AP_DS_ERR_OPUS_CHANNEL_INVALID': 2010,
}
for name, code in err_codes.items():
t.check(f"{name}={code}", getattr(opusplayer, name, None) == code, f"got={getattr(opusplayer, name, None)}")
# OPUS_ERR_INFO table
t.check("OPUS_ERR_INFO defined", hasattr(opusplayer, 'OPUS_ERR_INFO'))
if hasattr(opusplayer, 'OPUS_ERR_INFO'):
for code in err_codes.values():
t.check(f"OPUS_ERR_INFO[{code}]", code in opusplayer.OPUS_ERR_INFO)
# Normal error paths (corrupted file -> 2003)
lib = opusplayer.OpusAudio()
r = lib.get_audio_metadata_by_path(BAD_OPUS)
t.check("corrupted metadata -> 2003", isinstance(r, tuple) and r[0] == 2003, f"got={r}")
# Invalid source type
r = lib.get_audio_metadata(1.5)
t.check("invalid source type -> 1014", isinstance(r, tuple) and r[0] == 1014, f"got={r}")
# Missing file
r = lib.get_audio_metadata_by_path(os.path.join(TMP_DIR, 'missing.opus'))
t.check("missing file -> 1001", isinstance(r, tuple) and r[0] == 1001, f"got={r}")
lib.cleanup_function()
# ============================================================================
# [OP-12] Opus AID Mapping (1:1)
# ============================================================================
def test_opus_aid_mapping(t):
t.section("[OP-12] Opus AID Mapping (1:1)")
if not os.path.exists(OPUS_FILE):
t.skip("Opus AID mapping", "no Opus file available")
return
lib = AudioLibrary()
main_aid = lib.play_from_file(OPUS_FILE)
t.check("main AID is int", isinstance(main_aid, int), f"got={main_aid}")
if isinstance(main_aid, int):
t.check("AID mapped to opus", lib._is_opus_aid(main_aid), f"mapping={lib._aid_to_opus_aid}")
opus_aid = lib._get_opus_aid(main_aid)
t.check("opus AID is int", isinstance(opus_aid, int), f"got={opus_aid}")
t.check("mapping is 1:1", isinstance(main_aid, int) and isinstance(opus_aid, int))
r = lib.pause_audio(main_aid)
t.check("pause via main AID -> opus", r[0] == 0, f"got={r}")
r = lib.play_audio(main_aid)
t.check("resume via main AID -> opus", r[0] == 0, f"got={r}")
r = lib.set_volume(main_aid, 80)
t.check("set_volume via main AID -> opus", r[0] == 0, f"got={r}")
g = lib.get_volume(main_aid)
t.check("get_volume via main AID -> opus", isinstance(g, int), f"got={g}")
r = lib.seek_audio(main_aid, 30.0)
t.check("seek via main AID -> opus", r[0] == 0, f"got={r}")
r = lib.stop_audio(main_aid)
t.check("stop via main AID -> opus", isinstance(r, float), f"got={r}")
t.check("mapping removed after stop", not lib._is_opus_aid(main_aid), f"mapping={lib._aid_to_opus_aid}")
lib.cleanup_function()
# ============================================================================
# [OP-13] Opus Resource Management
# ============================================================================
def test_opus_resources(t):
t.section("[OP-13] Opus Resource Management")
if not os.path.exists(OPUS_FILE):
t.skip("Opus resources", "no Opus file available")
return
lib = opusplayer.OpusAudio()
lib.play_from_file(OPUS_FILE)
t.check("playing before cleanup", lib.is_music_playing() is True)
lib.cleanup_function()
t.check("not playing after cleanup", not lib.is_music_playing())
aid = lib.play_from_file(OPUS_FILE)
t.check("reusable after cleanup", isinstance(aid, int), f"got={aid}")
if isinstance(aid, int):
lib.stop_audio(aid)
lib.cleanup_function()
# ============================================================================
# [OP-14] Opus Boundary & Error Tests (reference CI/CD test_edge_cases)
# ============================================================================
def test_opus_boundary(t):
t.section("[OP-14] Opus Boundary & Error Tests")
lib = opusplayer.OpusAudio()
# --- B1: invalid argument types -> error tuples, never crash ---
t.log(" --- B1 invalid argument types ---")
for bad in (None, 1.5, [], {}, ('a',)):
r = lib.play_from_file(bad)
ok = isinstance(r, tuple) and len(r) == 3 and r[0] == 1001
t.check(f"play_from_file({type(bad).__name__}) -> tuple(1001)", ok, f"got={r!r}")
for bad in (None, 1.5, []):
r = lib.new_aid(bad)
ok = isinstance(r, tuple) and len(r) == 3 and r[0] == 1001
t.check(f"new_aid({type(bad).__name__}) -> tuple(1001)", ok, f"got={r!r}")
# --- B2: seek / volume / fade boundary values ---
t.log(" --- B2 boundary values ---")
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
# Seek boundary
for pos in (None, '5', [], {}, -1.0, 0.0, 999999.0):
r = lib.seek_audio(aid, pos)
ok = isinstance(r, tuple)
t.check(f"seek_audio({pos!r}) -> tuple", ok, f"got={r!r}")
r = lib.seek_audio(aid, 0.0)
t.check("seek_audio(0.0) -> success", isinstance(r, tuple) and r[0] == 0, f"got={r!r}")
# Volume boundary
for v in (None, '50', 1.5, [], {}):
r = lib.set_volume(aid, v)
t.check(f"set_volume({v!r}) -> 1015", isinstance(r, tuple) and r[0] == 1015, f"got={r!r}")
for v in (0, 1, 64, 100, 128):
r = lib.set_volume(aid, v)
t.check(f"set_volume({v}) boundary -> success", r[0] == 0, f"got={r}")
for v in (-1, 129, 200, 1000):
r = lib.set_volume(aid, v)
t.check(f"set_volume({v}) out of range -> 1015", r[0] == 1015, f"got={r}")
# Fade boundary
r = lib.fadein_music(aid, ms=None)
t.check("fadein_music(ms=None) -> tuple", isinstance(r, tuple), f"got={r!r}")
r = lib.fadein_music_pos(aid, ms=100, position=None)
t.check("fadein_music_pos(position=None) -> tuple", isinstance(r, tuple), f"got={r!r}")
lib.stop_audio(aid)
# --- B3: exact error codes ---
t.log(" --- B3 exact error codes ---")
cases = {
'1001 missing file': (lambda: lib.play_from_file(os.path.join(TMP_DIR, 'no_such.opus')), 1001),
'1002 invalid AID': (lambda: lib.pause_audio(99999), 1002),
'1014 invalid source type': (lambda: lib.get_audio_metadata(1.5), 1014),
'2003 corrupted opus': (lambda: lib.get_audio_metadata_by_path(BAD_OPUS), 2003),
}
for name, (fn, expect_code) in cases.items():
r = fn()
ok = isinstance(r, tuple) and r[0] == expect_code
t.check(f"{name} -> {expect_code}", ok, f"got={r!r}")
lib.cleanup_function()
# ============================================================================
# [OP-15] Opus DLL Specialized Tests (reference CI/CD test_sdl2_bindings)
# ============================================================================
def test_opus_dll_deep(t):
t.section("[OP-15] Opus DLL Specialized Tests")
# DLL file presence and sizes
pkg = os.path.dirname(os.path.abspath(opusdll.__file__))
dll_info = {
'libopusfile-0.dll': 55884,
'libopus-0.dll': 500112,
'libogg-0.dll': 40580,
'libopusurl-0.dll': 76772,
}
for dll, size in dll_info.items():
path = os.path.join(pkg, dll)
t.check(f"{dll} exists", os.path.exists(path))
if os.path.exists(path):
actual = os.path.getsize(path)
t.check(f"{dll} size={size}", actual == size, f"got={actual}")
# DLL hash verification
for dll in dll_info:
path = os.path.join(pkg, dll)
if os.path.exists(path):
with open(path, 'rb') as f:
content = f.read()
h = hashlib.sha256(content).hexdigest()
expected = opusdll.OPUS_DLL_HASHES.get(dll)
t.check(f"{dll} hash matches", h == expected, f"got={h[:16]}...")
# import_opus idempotent
r1 = opusdll.import_opus()
r2 = opusdll.import_opus()
t.check("import_opus idempotent", r1 == r2 == True, f"r1={r1} r2={r2}")
# opusfile handle
t.check("opusfile handle valid", opusdll.opusfile is not None)
# OPUS_DLL_FILES structure
t.check("OPUS_DLL_FILES has 4", len(opusdll.OPUS_DLL_FILES) == 4)
for dll in opusdll.OPUS_DLL_FILES:
t.check(f"DLL {dll['filename']} has url", 'url' in dll and dll['url'].startswith('http'))
t.check(f"DLL {dll['filename']} has size", 'size' in dll and dll['size'] > 0)
# check_opus_dll
ok, msg = opusplayer.check_opus_dll()
t.check("check_opus_dll ok", ok is True, f"msg={msg}")
# ============================================================================
# [OP-16] Opus Error Code Trigger Tests (all 2001-2010)
# ============================================================================
def test_opus_error_triggers(t):
t.section("[OP-16] Opus Error Code Trigger Tests")
lib = opusplayer.OpusAudio()
# 2003: corrupted file open failure
r = lib.play_from_file(BAD_OPUS)
t.check("2003 OPEN_FAILED trigger", isinstance(r, tuple) and r[0] == 2003, f"got={r}")
# 2004: HEADER_CORRUPT (monkey-patch op_head to return empty)
import ctypes
orig_head = opusdll.opusfile.op_head
opusdll.opusfile.op_head = lambda of, li: ctypes.POINTER(opusdll.OpusHead)()
r = opusplayer._get_opus_metadata(OPUS_FILE)
t.check("2004 HEADER_CORRUPT trigger", isinstance(r, tuple) and r[0] == 2004, f"got={r}")
opusdll.opusfile.op_head = orig_head
# 2008: BITRATE_UNAVAILABLE (monkey-patch op_bitrate to return 0)
orig_bitrate = opusdll.opusfile.op_bitrate
opusdll.opusfile.op_bitrate = lambda of, li: 0
r = opusplayer._get_opus_metadata(OPUS_FILE)
t.check("2008 BITRATE_UNAVAILABLE trigger", isinstance(r, tuple) and r[0] == 2008, f"got={r}")
opusdll.opusfile.op_bitrate = orig_bitrate
# 2010: CHANNEL_INVALID (monkey-patch op_channel_count to return 0)
orig_channels = opusdll.opusfile.op_channel_count
opusdll.opusfile.op_channel_count = lambda of, li: 0
r = opusplayer._get_opus_metadata(OPUS_FILE)
t.check("2010 CHANNEL_INVALID trigger", isinstance(r, tuple) and r[0] == 2010, f"got={r}")
opusdll.opusfile.op_channel_count = orig_channels
# 2009: NOT_SEEKABLE (monkey-patch op_seekable to return 0)
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
orig_seekable = opusdll.opusfile.op_seekable
opusdll.opusfile.op_seekable = lambda of: 0
r = lib.seek_audio(aid, 10.0)
t.check("2009 NOT_SEEKABLE trigger", isinstance(r, tuple) and r[0] == 2009, f"got={r}")
opusdll.opusfile.op_seekable = orig_seekable
lib.stop_audio(aid)
# 2007: SEEK_FAILED (monkey-patch op_pcm_seek to return -1)
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
orig_seek = opusdll.opusfile.op_pcm_seek
opusdll.opusfile.op_pcm_seek = lambda of, pos: -1
r = lib.seek_audio(aid, 10.0)
t.check("2007 SEEK_FAILED trigger", isinstance(r, tuple) and r[0] == 2007, f"got={r}")
opusdll.opusfile.op_pcm_seek = orig_seek
lib.stop_audio(aid)
# 2005: TAGS_PARSE_FAILED (monkey-patch op_tags to raise)
class BadTags:
@property
def contents(self):
raise ValueError("corrupt tags")
orig_tags = opusdll.opusfile.op_tags
opusdll.opusfile.op_tags = lambda of, li: BadTags()
r = opusplayer._get_opus_extended_metadata(OPUS_FILE)
t.check("2005 TAGS_PARSE_FAILED trigger", isinstance(r, tuple) and r[0] == 2005, f"got={r}")
opusdll.opusfile.op_tags = orig_tags
# 2006: DECODE_FAILED (monkey-patch op_read_stereo to return -1)
orig_read = opusdll.opusfile.op_read_stereo
opusdll.opusfile.op_read_stereo = lambda of, pcm, size: -1
aid = lib.play_from_file(OPUS_FILE)
time.sleep(0.5)
r = lib._decode_error
t.check("2006 DECODE_FAILED trigger", r is not None and r[0] == 2006, f"got={r}")
opusdll.opusfile.op_read_stereo = orig_read
lib.stop_audio(aid)
lib.cleanup_function()
# ============================================================================
# [OP-L] Listening Tests (interactive, requires ears)
# ============================================================================
def test_opus_listen(t):
t.section("[OP-L] Opus Listening Tests (interactive)")
if not os.path.exists(OPUS_FILE):
t.skip("Opus listening", "no Opus file available")
return
try:
ans = input("\nRun Opus listening tests? (y/n): ").strip().lower()
if ans != 'y':
t.skip("Opus listening", "user skipped")
return
except EOFError:
t.skip("Opus listening", "non-interactive")
return
lib = opusplayer.OpusAudio()
t.log("\n--- Listening 1: Normal playback ---")
t.log(" Playing Opus at normal volume...")
aid = lib.play_from_file(OPUS_FILE)
if isinstance(aid, int):
time.sleep(3)
try:
ans = input(" Did you hear audio? (y/n): ").strip().lower()
t.check("normal playback audible", ans == 'y')
except EOFError:
t.skip("normal playback audible", "no input")
lib.stop_audio(aid)
t.log("\n--- Listening 2: Fade-in ---")
t.log(" Fading in over 3 seconds...")
aid = lib.new_aid(OPUS_FILE)
r = lib.fadein_music(aid, ms=3000)
if r[0] == 0:
time.sleep(3)
try:
ans = input(" Did volume gradually increase? (y/n): ").strip().lower()
t.check("fade-in audible", ans == 'y')
except EOFError:
t.skip("fade-in audible", "no input")
time.sleep(1)
lib.stop_audio(aid)
t.log("\n--- Listening 3: Fade-out ---")
t.log(" Playing at normal volume for 3 seconds, then fading out...")
aid = lib.play_from_file(OPUS_FILE)
time.sleep(3) # Play at normal volume first
r = lib.fadeout_music(ms=3000)
if r[0] == 0:
time.sleep(3)
try:
ans = input(" Did volume gradually decrease then stop? (y/n): ").strip().lower()
t.check("fade-out audible", ans == 'y')
except EOFError:
t.skip("fade-out audible", "no input")
time.sleep(1)
t.log("\n--- Listening 4: Volume change ---")
t.log(" Playing at normal volume for 2s, then volume 128 -> 0 -> 128...")
aid = lib.play_from_file(OPUS_FILE)
time.sleep(2) # Play at normal volume first
lib.set_volume(aid, 128)
time.sleep(1)
lib.set_volume(aid, 0)
time.sleep(1)
try:
ans = input(" Did sound become silent at volume 0? (y/n): ").strip().lower()
t.check("volume 0 silent", ans == 'y')
except EOFError:
t.skip("volume 0 silent", "no input")
lib.set_volume(aid, 100)
time.sleep(1)
lib.stop_audio(aid)
t.log("\n--- Listening 5: Seek position ---")
t.log(" Playing for 3s, then seeking to 30s, 60s, 120s...")
aid = lib.play_from_file(OPUS_FILE)
time.sleep(3) # Play at normal position first
for pos in (30, 60, 120):
lib.seek_audio(aid, pos)
time.sleep(1)
try:
ans = input(f" After seek to {pos}s, did position change? (y/n): ").strip().lower()
t.check(f"seek to {pos}s audible", ans == 'y')
except EOFError:
t.skip(f"seek to {pos}s audible", "no input")
lib.stop_audio(aid)
lib.cleanup_function()
# ============================================================================
# main()
# ============================================================================
def main():
global OPUS_FILE
mode = sys.argv[1].lower().lstrip('-') if len(sys.argv) > 1 else 'full'
if mode not in ('auto', 'listen', 'full'):
print(f"Unknown mode: {mode}, using full")
mode = 'full'
# Prompt for Opus file path (use default if empty)
OPUS_FILE = _prompt_opus()
print(f"\n Python: {sys.version.split()[0]} | Platform: {sys.platform} | Mode: {mode}")
print(f" ap_ds version: {ap_ds.__version__} | Path: {os.path.dirname(ap_ds.__file__)}")
print(f" Opus test file: {OPUS_FILE} (exists={os.path.exists(OPUS_FILE)})")
t = OpusTest()
if mode in ('auto', 'full'):
test_opus_imports(t)
test_opus_dll(t)
test_opus_metadata(t)
test_opus_play(t)
test_opus_control(t)
test_opus_volume(t)
test_opus_seek(t)
test_opus_fade(t)
test_opus_distinction(t)
test_opus_batch(t)
test_opus_errors(t)
test_opus_aid_mapping(t)
test_opus_resources(t)
test_opus_boundary(t)
test_opus_dll_deep(t)
test_opus_error_triggers(t)
if mode in ('listen', 'full'):
test_opus_listen(t)
t.summary()
return 0 if t.failed == 0 else 1
if __name__ == '__main__':
sys.exit(main())
ap_ds/__init__.py
+3 -89
View File
@@ -93,7 +93,7 @@ def show_tech_manual() -> None:
manual = r"""
╔═══════════════════════════════════════════════════════════════════════════════╗
║ ║
║ AP_DS 4.1.0 TECHNICAL MANUAL ║
║ AP_DS 4.0.1 TECHNICAL MANUAL ║
║ Audio Library By DVS - https://apds.top ║
║ ║
╚═══════════════════════════════════════════════════════════════════════════════╝
@@ -125,53 +125,8 @@ audio library for Python applications. Built on SDL2 and SDL2_mixer, it provides
│ FLAC │ .flac │ STREAMINFO block, 100% accuracy │
│ OGG Vorbis │ .ogg │ Granule position, 99.99% accuracy │
│ AAC (ADTS) │ .aac │ ADTS frame parsing, >99% accuracy │
│ OPUS │ .opus │ libopusfile decode, waveOut playback │
└──────────────┴─────────────┴─────────────────────────────────────────┘
┌───────────────────────────────────────────────────────────────────────────────┐
│ 2.1 OPUS SUPPORT (NEW in 4.1.0) │
└───────────────────────────────────────────────────────────────────────────────┘
AP_DS 4.1.0 adds native Opus playback support via libopusfile + winmm waveOut.
This is a separate playback path from SDL2, automatically selected when the
audio file is an Opus (.opus) file.
Key Features:
• Automatic detection: Opus files are routed to OpusAudio sub-player
• AID mapping: Main library AID <-> Opus sub-library AID (1:1)
• Auto-download: Opus DLLs downloaded from https://dvsyun.top
• Hash verification: SHA256 verified after download
• Full control: play / pause / resume / stop / seek / volume / fade
Modules:
• opusplayer.py - OpusAudio class (Opus playback engine)
• _opusdll.py - DLL loader + auto-download + hash verification
OpusAudio class:
class OpusAudio(frequency=48000, channels=2, volume_pct=80)
Same API as AudioLibrary for Opus files:
play_from_file / new_aid / play_from_memory
pause_audio / play_audio / stop_audio / seek_audio
set_volume / get_volume
fadein_music / fadein_music_pos / fadeout_music
get_audio_metadata / get_audio_duration / batch_*
Usage (automatic routing through AudioLibrary):
from ap_ds import AudioLibrary
lib = AudioLibrary()
aid = lib.play_from_file("song.opus") # auto-routed to OpusAudio
Direct usage:
from ap_ds import OpusAudio
opus = OpusAudio()
aid = opus.play_from_file("song.opus")
Opus DLLs (auto-downloaded from https://dvsyun.top/ap_ds/download/):
• libopusfile-0.dll (Opus file decoding)
• libopus-0.dll (Opus codec core)
• libogg-0.dll (Ogg container)
• libopusurl-0.dll (Opus URL streaming)
┌───────────────────────────────────────────────────────────────────────────────┐
│ 3. CORE COMPONENTS │
└───────────────────────────────────────────────────────────────────────────────┘
@@ -474,41 +429,12 @@ Configure via: AP_DS_WAV_THRESHOLD environment variable
• Supports seeking and fading
• Best for long tracks (>=6s)
┌───────────────────────────────────────────────────────────────────────────────┐
│ 10.4 Opus Error Codes (NEW in 4.1.0) │
└───────────────────────────────────────────────────────────────────────────────┘
Opus-specific error codes (2000+):
┌──────────┬────────────────────────────────────────────┬──────────────────────────────────┐
│ Code │ Name │ Description │
├──────────┼────────────────────────────────────────────┼──────────────────────────────────┤
│ 2001 │ AP_DS_ERR_OPUS_LIB_LOAD_FAILED │ libopusfile-0.dll load failed │
│ 2002 │ AP_DS_ERR_OPUS_DLL_DEPENDENCY │ DLL dependency missing │
│ 2003 │ AP_DS_ERR_OPUS_OPEN_FAILED │ Opus file open failed │
│ 2004 │ AP_DS_ERR_OPUS_HEADER_CORRUPT │ OpusHead header corrupt │
│ 2005 │ AP_DS_ERR_OPUS_TAGS_PARSE_FAILED │ OpusTags tag parse failed │
│ 2006 │ AP_DS_ERR_OPUS_DECODE_FAILED │ Opus decode failed │
│ 2007 │ AP_DS_ERR_OPUS_SEEK_FAILED │ Opus seek failed │
│ 2008 │ AP_DS_ERR_OPUS_BITRATE_UNAVAILABLE │ Bitrate unavailable │
│ 2009 │ AP_DS_ERR_OPUS_NOT_SEEKABLE │ Stream not seekable │
│ 2010 │ AP_DS_ERR_OPUS_CHANNEL_INVALID │ Invalid channel count │
└──────────┴────────────────────────────────────────────┴──────────────────────────────────┘
┌───────────────────────────────────────────────────────────────────────────────┐
│ 11. VERSION HISTORY │
└───────────────────────────────────────────────────────────────────────────────┘
Version 4.1.0 (Current)
Version 4.0.1 (Current)
────────────────────────
• Opus playback support (libopusfile + waveOut)
• OpusAudio class with automatic routing
• Opus DLL auto-download + SHA256 verification
• Opus-specific error codes (2001-2010)
• Opus batch parsing
• Opus vs original format distinction
Version 4.0.1
─────────────
• Python 3.15t free-threading support
• Lazy imports for Python 3.15+
• DAP O(1) deduplication
@@ -539,7 +465,7 @@ Configure via: AP_DS_WAV_THRESHOLD environment variable
╔═══════════════════════════════════════════════════════════════════════════════╗
║ END OF MANUAL ║
║ AP_DS 4.1.0 - August 2026 ║
║ AP_DS 4.0.1 - December 2025 ║
║ ║
║ 📖 For detailed Markdown documentation, visit: ║
║ https://apds.top ║
@@ -724,17 +650,6 @@ try:
except ImportError:
from player import *
# ============================================================
# Opus Support (opusplayer.py)
# ============================================================
try:
from .opusplayer import OpusAudio
except ImportError:
try:
from opusplayer import OpusAudio
except ImportError:
OpusAudio = None
# ============================================================
# Export Self-Check Functions (users can call manually)
# ============================================================
@@ -760,7 +675,6 @@ except Exception:
__all__ = [
"__version__",
"AudioLibrary",
"OpusAudio",
"get_audio_duration",
"get_audio_metadata",
"batch_get_metadata",
ap_ds/_opusdll.py
-946
View File
@@ -1,946 +0,0 @@
# _opusdll.py - Opus DLL constants, structures, bindings and auto-download loader
# Reference: ap_ds/_sdl2.py structure
import os
import sys
import ssl
import hashlib
import tempfile
import shutil
import urllib.request
import ctypes
import ctypes.wintypes as wt
from ctypes import *
# ============================================================
# Opus DLL Library Loader
# ============================================================
# Global library handles
opusfile = None # libopusfile-0.dll
winmm = None # winmm.dll (system)
kernel32 = None # kernel32.dll (system)
_opus_dll_error = None # Records DLL load failure reason
# DLL file list (from the download API)
OPUS_DLL_FILES = [
{
"filename": "libopusurl-0.dll",
"url": "https://dvsyun.top/ap_ds/download/libopusurl-0.dll",
"size": 76772,
},
{
"filename": "libopus-0.dll",
"url": "https://dvsyun.top/ap_ds/download/libopus-0.dll",
"size": 500112,
},
{
"filename": "libogg-0.dll",
"url": "https://dvsyun.top/ap_ds/download/libogg-0.dll",
"size": 40580,
},
{
"filename": "libopusfile-0.dll",
"url": "https://dvsyun.top/ap_ds/download/libopusfile-0.dll",
"size": 55884,
},
]
# DLL SHA256 hashes for verification (after download)
OPUS_DLL_HASHES = {
"libopusfile-0.dll": "fc8ff75c5e0180e73b0528dc78c51ed0fb493741375cdc227f50c2a33cabf727",
"libopus-0.dll": "90aa25a0a6525d7da48a7ae8dd3306e45b0c28ce09a73d2a02b56cd95418d5be",
"libogg-0.dll": "3038ce8d161324a6349bf7c83b78493857ff6a3501e3adb3d541c6a07bd94a57",
"libopusurl-0.dll": "a6cde968a23f2d0067332a13718c52e265653a2c35d65862e8dff4cf2a0346d9",
}
def _get_package_dir():
"""Get the directory containing this module."""
return os.path.dirname(os.path.abspath(__file__))
def _check_opus_libraries_exist(directory):
"""Check if all Opus DLLs exist in the given directory."""
required = ["libopusfile-0.dll", "libopus-0.dll", "libogg-0.dll"]
for dll in required:
if not os.path.exists(os.path.join(directory, dll)):
return False
return True
def _load_from_directory(directory):
"""Load Opus libraries from the specified directory.
Supports both Windows (.dll) and Linux (.so) library names.
Returns:
bool: True if the Opus library loaded successfully
"""
global opusfile
platform = sys.platform
# Determine library filename based on platform
if platform == "win32":
libopusfile_path = os.path.join(directory, "libopusfile-0.dll")
elif platform.startswith("linux"):
# Try multiple Linux .so naming conventions
names_to_try = [
"libopusfile.so",
"libopusfile.so.0",
"libopusfile-0.so",
]
libopusfile_path = None
for name in names_to_try:
candidate = os.path.join(directory, name)
if os.path.exists(candidate):
libopusfile_path = candidate
break
if libopusfile_path is None:
return False
else:
# macOS (not added yet) or other
libopusfile_path = os.path.join(directory, "libopusfile-0.dll")
if not os.path.exists(libopusfile_path):
return False
if not os.path.exists(libopusfile_path):
return False
try:
# Add directory to library search path
if platform == "win32":
if hasattr(os, 'add_dll_directory'):
os.add_dll_directory(directory)
os.environ['PATH'] = directory + os.pathsep + os.environ.get('PATH', '')
elif platform.startswith("linux"):
if 'LD_LIBRARY_PATH' not in os.environ:
os.environ['LD_LIBRARY_PATH'] = directory
else:
os.environ['LD_LIBRARY_PATH'] = directory + ':' + os.environ['LD_LIBRARY_PATH']
opusfile = ctypes.CDLL(libopusfile_path)
return True
except Exception as e:
_opus_dll_error = f"Opus library load error: {e}"
return False
def _load_from_system():
"""Try loading libopusfile from system paths."""
global opusfile
try:
import ctypes.util
found = ctypes.util.find_library("opusfile")
if found:
opusfile = ctypes.CDLL(found)
return True
except Exception:
pass
return False
def _load_user_config():
"""Load user-saved Opus library paths from config file."""
global opusfile
try:
config_file = os.path.expanduser('~/.config/ap_ds/opus_paths.conf')
if os.path.exists(config_file):
with open(config_file, 'r') as f:
opusfile_path = None
for line in f:
if line.startswith('OPUSFILE_PATH='):
opusfile_path = line.strip().split('=', 1)[1]
if opusfile_path and os.path.exists(opusfile_path):
opusfile = ctypes.CDLL(opusfile_path)
return True
except Exception:
pass
return False
def _run_sudo_command(cmd, packages):
"""Run a sudo command with interactive password input.
First tries without password (if already root or passwordless sudo).
If that fails, prompts for the sudo password interactively.
Args:
cmd: Base command list (e.g. ['apt-get', 'install', '-y'])
packages: Package names to install
Returns:
bool: True if command succeeded
"""
import subprocess
import getpass
full_cmd = cmd + packages
# Try without password first (if already root / passwordless sudo)
try:
result = subprocess.run(
['sudo', '-n'] + full_cmd,
capture_output=True, text=True, timeout=120
)
if result.returncode == 0:
return True
except Exception:
pass
# Prompt for sudo password interactively
print("🔑 sudo password required for package installation")
try:
password = getpass.getpass("Enter sudo password: ")
except (EOFError, KeyboardInterrupt):
print("\n❌ Password input cancelled")
return False
# Use sudo -S to read password from stdin
try:
result = subprocess.run(
['sudo', '-S'] + full_cmd,
input=password + '\n',
capture_output=True, text=True, timeout=180
)
if result.returncode == 0:
print("✅ Packages installed successfully")
return True
else:
print(f"❌ Installation failed: {result.stderr.strip()}")
return False
except Exception as e:
print(f"❌ Installation error: {e}")
return False
def _linux_auto_install():
"""Try automatic package manager installation on Linux.
Uses interactive sudo password input to avoid hanging.
"""
try:
import subprocess
import shutil
if shutil.which('apt-get'):
print("📦 Detected apt-based system (Ubuntu/Debian)")
if _run_sudo_command(['apt-get', 'install', '-y'],
['libopusfile-dev', 'libopus-dev', 'libogg-dev']):
if _load_from_system():
print("✅ Opus libraries installed and loaded")
return True
elif shutil.which('dnf'):
print("📦 Detected dnf-based system (Fedora)")
if _run_sudo_command(['dnf', 'install', '-y'],
['opusfile-devel', 'opus-devel', 'libogg-devel']):
if _load_from_system():
print("✅ Opus libraries installed and loaded")
return True
elif shutil.which('pacman'):
print("📦 Detected pacman-based system (Arch)")
if _run_sudo_command(['pacman', '-S', '--noconfirm'],
['opusfile', 'opus', 'libogg']):
if _load_from_system():
print("✅ Opus libraries installed and loaded")
return True
except Exception as e:
print(f"⚠️ Automatic installation failed: {e}")
return False
def _linux_interactive_setup():
"""Linux interactive setup for Opus libraries."""
global opusfile
print("\n" + "=" * 70)
print("Linux Opus Library Loading")
print("=" * 70)
print("Options:")
print("1. Use system-installed libraries (re-check)")
print("2. Specify path to your compiled .so files")
print("3. Show installation instructions")
print("=" * 70)
while True:
choice = input("\nChoose option (1/2/3): ").strip()
if choice == "1":
if _load_from_system():
print("✅ Opus libraries loaded from system")
return True
print("❌ System libraries not found")
continue
elif choice == "2":
opusfile_path = input("Enter full path to libopusfile.so: ").strip()
if os.path.exists(opusfile_path):
try:
opusfile = ctypes.CDLL(opusfile_path)
# Save for future
try:
config_dir = os.path.expanduser('~/.config/ap_ds')
os.makedirs(config_dir, exist_ok=True)
with open(os.path.join(config_dir, 'opus_paths.conf'), 'w') as f:
f.write(f"OPUSFILE_PATH={opusfile_path}\n")
except Exception:
pass
print("✅ Libraries loaded from user-specified paths")
return True
except Exception as e:
print(f"❌ Failed to load: {e}")
else:
print("❌ File not found")
continue
elif choice == "3":
print("\n📚 Installation instructions:")
print("=" * 50)
print("For Ubuntu/Debian:")
print(" sudo apt-get install libopusfile-dev libopus-dev libogg-dev")
print("\nFor Fedora:")
print(" sudo dnf install opusfile-devel opus-devel libogg-devel")
print("\nFor Arch:")
print(" sudo pacman -S opusfile opus libogg")
print("=" * 50)
continue
else:
print("❌ Invalid choice. Please enter 1, 2, or 3.")
def _macos_apology():
"""Print a humble apology about missing Opus precompiled packages on macOS."""
print("\n" + "=" * 70)
print("⚠️ macOS Opus Support Notice")
print("=" * 70)
print("We sincerely apologize.")
print("On macOS, SDL2 libraries can be downloaded automatically, but we")
print("could NOT find any precompiled Opus framework packages for macOS.")
print("This is a limitation of the Opus ecosystem, not of ap_ds.")
print("")
print("We recommend using a package manager to install the Opus libraries.")
print("ap_ds will first try to auto-install them for you.")
print("If that fails, we will guide you through manual installation.")
print("=" * 70 + "\n")
def _macos_detect_system():
"""Detect if Opus libraries are already present on macOS system paths."""
global opusfile
try:
import ctypes.util
found = ctypes.util.find_library("opusfile")
if found:
opusfile = ctypes.CDLL(found)
return True
except Exception:
pass
# Check common macOS library paths
for path in [
"/opt/homebrew/lib/libopusfile.dylib", # Apple Silicon Homebrew
"/usr/local/lib/libopusfile.dylib", # Intel Homebrew
"/opt/local/lib/libopusfile.dylib", # MacPorts
]:
if os.path.exists(path):
try:
opusfile = ctypes.CDLL(path)
return True
except Exception:
pass
return False
def _macos_install_macports():
"""Try installing Opus libraries via MacPorts."""
import subprocess
import shutil
if shutil.which('port'):
print("📦 Detected MacPorts")
try:
result = subprocess.run(
['sudo', '-n', 'port', 'install', 'opus', 'opusfile', 'libogg'],
capture_output=True, text=True, timeout=180
)
if result.returncode == 0:
if _macos_detect_system():
print("✅ Opus libraries installed via MacPorts")
return True
except Exception:
pass
# If sudo -n failed (needs password), prompt for password
import getpass
print("🔑 sudo password required for MacPorts installation")
try:
password = getpass.getpass("Enter sudo password: ")
result = subprocess.run(
['sudo', '-S', 'port', 'install', 'opus', 'opusfile', 'libogg'],
input=password + '\n', capture_output=True, text=True, timeout=300
)
if result.returncode == 0 and _macos_detect_system():
print("✅ Opus libraries installed via MacPorts")
return True
except Exception as e:
print(f"❌ MacPorts installation failed: {e}")
return False
def _macos_install_homebrew():
"""Try installing Opus libraries via Homebrew."""
import subprocess
import shutil
if shutil.which('brew'):
print("📦 Detected Homebrew")
try:
result = subprocess.run(
['brew', 'install', 'opus', 'opusfile', 'libogg'],
capture_output=True, text=True, timeout=300
)
if result.returncode == 0:
if _macos_detect_system():
print("✅ Opus libraries installed via Homebrew")
return True
except Exception as e:
print(f"❌ Homebrew installation failed: {e}")
return False
def _macos_guide_manual_install():
"""Guide the user through manual installation of Opus libraries."""
print("\n" + "=" * 70)
print("📚 Manual Installation Guide (macOS)")
print("=" * 70)
print("We could not auto-install the Opus libraries. Please install them")
print("using one of the following methods:")
print("")
print("Method 1: Install Homebrew (if not installed)")
print(" /bin/bash -c \"$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)\"")
print(" Then: brew install opus opusfile libogg")
print("")
print("Method 2: Install MacPorts")
print(" https://www.macports.org/install.php")
print(" Then: sudo port install opus opusfile libogg")
print("")
print("Method 3: Compile from source")
print(" Download from https://opus-codec.org/downloads/")
print(" opus-1.6.1.tar.gz, opusfile-0.12.tar.gz, libogg-1.3.6.tar.gz")
print(" Compile each with: ./configure && make && sudo make install")
print("=" * 70 + "\n")
def _macos_auto_install():
"""Try to auto-install Opus libraries on macOS.
Order: detect system -> MacPorts -> Homebrew -> manual guide.
"""
_macos_apology()
# Step 1: Detect if already present
if _macos_detect_system():
print("✅ Opus libraries already present on system")
return True
# Step 2: Try MacPorts
print("\n📦 Attempting MacPorts installation...")
if _macos_install_macports():
return True
# Step 3: Try Homebrew
print("\n📦 Attempting Homebrew installation...")
if _macos_install_homebrew():
return True
# Step 4: Guide manual installation
_macos_guide_manual_install()
return False
def _check_opus_libraries_exist_linux(directory):
"""Check if Opus .so libraries exist in directory (Linux)."""
return os.path.exists(os.path.join(directory, "libopusfile.so"))
def verify_file_hash(file_path, expected_hash):
"""Verify the SHA256 hash of a file.
Args:
file_path: Path to the file to verify
expected_hash: Expected SHA256 hash (hex string)
Returns:
bool: True if hash matches (or no hash configured)
"""
if not expected_hash:
print(f" ⚠️ No hash configured for {os.path.basename(file_path)}, skipping verification")
return True
try:
with open(file_path, 'rb') as f:
content = f.read()
file_hash = hashlib.sha256(content).hexdigest()
print(f" Existing file SHA256: {file_hash}")
if file_hash.lower() == expected_hash.lower():
print(f" ✅ Hash verification passed")
return True
else:
print(f" ❌ Hash verification failed! Expected: {expected_hash}, Got: {file_hash}")
return False
except Exception as e:
print(f" ❌ Error verifying hash: {e}")
return False
def download_opus_libraries():
"""Download all Opus DLLs to the package directory with auto-download and hash verification.
Returns:
bool: True if all DLLs downloaded successfully
"""
current_dir = _get_package_dir()
print(f"Package directory: {current_dir}")
def download_file(url, filename, expected_hash=None):
"""Download a single file with SSL fallback and hash verification."""
temp_file = tempfile.NamedTemporaryFile(delete=False)
temp_file.close()
try:
print(f" Downloading {filename} from {url}...")
try:
# Try with SSL verification first
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=30) as response:
content = response.read()
print(f" ✅ Download successful with SSL verification")
except (urllib.error.URLError, ssl.SSLError) as e:
print(f" SSL verification failed: {e}")
print(f" Retrying without SSL verification...")
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=30, context=context) as response:
content = response.read()
print(f" ✅ Download successful without SSL verification")
with open(temp_file.name, 'wb') as f:
f.write(content)
print(f" Downloaded {len(content)} bytes")
# Verify hash before moving
if expected_hash:
file_hash = hashlib.sha256(content).hexdigest()
print(f" Downloaded file SHA256: {file_hash}")
if file_hash.lower() != expected_hash.lower():
print(f" ❌ Hash verification failed! Expected: {expected_hash}, Got: {file_hash}")
try:
os.unlink(temp_file.name)
except Exception:
pass
return False
print(f" ✅ Hash verification passed")
# Move to package directory
dest_path = os.path.join(current_dir, filename)
shutil.move(temp_file.name, dest_path)
print(f" ✅ Saved to {dest_path}")
return True
except Exception as e:
print(f" ❌ Download failed for {filename}: {e}")
try:
os.unlink(temp_file.name)
except Exception:
pass
return False
# Download each DLL with hash verification
success_count = 0
for dll_info in OPUS_DLL_FILES:
filename = dll_info["filename"]
expected_hash = OPUS_DLL_HASHES.get(filename)
file_path = os.path.join(current_dir, filename)
if os.path.exists(file_path):
print(f"\n📁 {filename} already exists, verifying hash...")
if verify_file_hash(file_path, expected_hash):
print(f"✅ {filename} is valid, skipping download")
success_count += 1
continue
else:
print(f"⚠️ {filename} hash mismatch, re-downloading...")
try:
os.remove(file_path)
except Exception:
pass
print(f"\n📥 Downloading {filename}...")
if download_file(dll_info["url"], filename, expected_hash):
success_count += 1
print(f"\n✅ Downloaded {success_count}/{len(OPUS_DLL_FILES)} DLLs")
return success_count == len(OPUS_DLL_FILES)
def import_opus():
"""Main function: Import Opus libraries with cross-platform support.
Returns:
bool: True if Opus libraries loaded successfully
"""
global opusfile
# Already loaded?
if opusfile is not None:
return True
current_dir = _get_package_dir()
platform = sys.platform
# Windows: use DLL + auto-download
if platform == "win32":
# Layer 1: Load from current directory
if _load_from_directory(current_dir):
return True
# Layer 2: Load from system
if _load_from_system():
return True
# Layer 3: Auto-download DLLs
print("Opus DLLs not found, downloading...")
if download_opus_libraries():
if _load_from_directory(current_dir):
print("✅ Opus DLLs loaded after download")
return True
# Linux: use system .so libraries (reference _sdl2.py)
# NOTE: The package directory contains Windows .dll files only.
# Linux Opus libraries (.so) are installed via system package manager.
elif platform.startswith("linux"):
# Layer 1: User config (custom .so paths)
if _load_user_config():
print("✅ Opus loaded from user config")
return True
# Layer 2: System libraries
if _load_from_system():
print("✅ Opus loaded from system")
return True
# Layer 3: Auto install via package manager
if _linux_auto_install():
return True
# Layer 4: Interactive setup
if _linux_interactive_setup():
return True
# macOS: no precompiled Opus framework, use package manager
# Order: detect system -> MacPorts -> Homebrew -> manual guide
elif platform == "darwin":
if _macos_auto_install():
return True
global _opus_dll_error
if not _opus_dll_error:
_opus_dll_error = "Failed to load Opus libraries"
return False
def check_opus_dll():
"""Check whether the Opus DLL can be loaded normally.
Returns:
(bool, str): (whether normal, error message)
"""
if not import_opus():
return (False, _opus_dll_error or "Opus DLL load failed")
try:
if not hasattr(opusfile, 'op_open_file'):
return (False, "libopusfile-0.dll not loaded correctly (missing op_open_file)")
return (True, "Opus DLL OK")
except Exception as e:
return (False, f"Opus DLL check failed: {e}")
# ============================================================
# Opus structures
# ============================================================
class OpusHead(Structure):
_fields_ = [
("version", c_int),
("channel_count", c_int),
("pre_skip", c_uint),
("input_sample_rate", c_uint),
("output_gain", c_int),
("mapping_family", c_int),
("stream_count", c_int),
("coupled_count", c_int),
("mapping", c_ubyte * 255),
]
class OpusTags(Structure):
_fields_ = [
("user_comments", POINTER(c_char_p)),
("comment_lengths", POINTER(c_int)),
("comments", c_int),
("vendor", c_char_p),
]
# ============================================================
# Windows Wave API structures and constants
# ============================================================
WAVE_FORMAT_PCM = 1
WAVE_MAPPER = 0xFFFFFFFF
CALLBACK_EVENT = 0x00050000
WHDR_DONE = 0x1
MMSYSERR_NOERROR = 0
WAIT_OBJECT_0 = 0
class WAVEFORMATEX(Structure):
_fields_ = [
("wFormatTag", wt.WORD),
("nChannels", wt.WORD),
("nSamplesPerSec", wt.DWORD),
("nAvgBytesPerSec", wt.DWORD),
("nBlockAlign", wt.WORD),
("wBitsPerSample", wt.WORD),
("cbSize", wt.WORD),
]
class WAVEHDR(Structure):
pass
WAVEHDR._fields_ = [
("lpData", wt.LPSTR),
("dwBufferLength", wt.DWORD),
("dwBytesRecorded", wt.DWORD),
("dwUser", c_void_p), # DWORD_PTR (8 bytes)
("dwFlags", wt.DWORD),
("dwLoops", wt.DWORD),
("lpNext", POINTER(WAVEHDR)),
("reserved", c_void_p), # DWORD_PTR (8 bytes)
]
# ============================================================
# Opus file function bindings (wrapper functions)
# ============================================================
def op_open_file(path, error):
"""Open an Opus file. Returns handle or None."""
return opusfile.op_open_file(path, error)
def op_free(of):
"""Free an Opus file handle."""
opusfile.op_free(of)
def op_head(of, li):
"""Get OpusHead for a link."""
return opusfile.op_head(of, li)
def op_tags(of, li):
"""Get OpusTags for a link."""
return opusfile.op_tags(of, li)
def op_channel_count(of, li):
"""Get channel count."""
return opusfile.op_channel_count(of, li)
def op_pcm_total(of, li):
"""Get total PCM samples."""
return opusfile.op_pcm_total(of, li)
def op_bitrate(of, li):
"""Get average bitrate."""
return opusfile.op_bitrate(of, li)
def op_seekable(of):
"""Check if stream is seekable."""
return opusfile.op_seekable(of)
def op_link_count(of):
"""Get number of links."""
return opusfile.op_link_count(of)
def op_read_stereo(of, pcm, buf_size):
"""Read decoded stereo PCM."""
return opusfile.op_read_stereo(of, pcm, buf_size)
def op_pcm_seek(of, pos):
"""Seek to PCM sample position."""
return opusfile.op_pcm_seek(of, pos)
def op_pcm_tell(of):
"""Get current PCM position."""
return opusfile.op_pcm_tell(of)
# ============================================================
# Windows Wave API function bindings
# ============================================================
def waveOutOpen(phwo, device_id, fmt, callback, instance, flags):
return winmm.waveOutOpen(phwo, device_id, fmt, callback, instance, flags)
def waveOutPrepareHeader(hwo, pwh, cbwh):
return winmm.waveOutPrepareHeader(hwo, pwh, cbwh)
def waveOutWrite(hwo, pwh, cbwh):
return winmm.waveOutWrite(hwo, pwh, cbwh)
def waveOutUnprepareHeader(hwo, pwh, cbwh):
return winmm.waveOutUnprepareHeader(hwo, pwh, cbwh)
def waveOutClose(hwo):
return winmm.waveOutClose(hwo)
def waveOutSetVolume(hwo, volume):
return winmm.waveOutSetVolume(hwo, volume)
def waveOutGetVolume(hwo, volume):
return winmm.waveOutGetVolume(hwo, volume)
def waveOutPause(hwo):
return winmm.waveOutPause(hwo)
def waveOutRestart(hwo):
return winmm.waveOutRestart(hwo)
def waveOutReset(hwo):
return winmm.waveOutReset(hwo)
def waveOutGetErrorTextW(code, buf, size):
return winmm.waveOutGetErrorTextW(code, buf, size)
# ============================================================
# Kernel32 function bindings
# ============================================================
def CreateEventW(lp_attrs, b_manual, b_initial, name):
return kernel32.CreateEventW(lp_attrs, b_manual, b_initial, name)
def WaitForSingleObject(handle, ms):
return kernel32.WaitForSingleObject(handle, ms)
def ResetEvent(handle):
return kernel32.ResetEvent(handle)
def CloseHandle(handle):
return kernel32.CloseHandle(handle)
# ============================================================
# Function prototypes (set after libraries are loaded)
# ============================================================
def _setup_prototypes():
"""Set up function prototypes for opusfile, winmm, and kernel32."""
# --- opusfile prototypes ---
opusfile.op_open_file.restype = c_void_p
opusfile.op_open_file.argtypes = [c_char_p, POINTER(c_int)]
opusfile.op_free.restype = None
opusfile.op_free.argtypes = [c_void_p]
opusfile.op_head.restype = POINTER(OpusHead)
opusfile.op_head.argtypes = [c_void_p, c_int]
opusfile.op_tags.restype = POINTER(OpusTags)
opusfile.op_tags.argtypes = [c_void_p, c_int]
opusfile.op_channel_count.restype = c_int
opusfile.op_channel_count.argtypes = [c_void_p, c_int]
opusfile.op_pcm_total.restype = c_longlong
opusfile.op_pcm_total.argtypes = [c_void_p, c_int]
opusfile.op_bitrate.restype = c_int
opusfile.op_bitrate.argtypes = [c_void_p, c_int]
opusfile.op_seekable.restype = c_int
opusfile.op_seekable.argtypes = [c_void_p]
opusfile.op_link_count.restype = c_int
opusfile.op_link_count.argtypes = [c_void_p]
opusfile.op_read_stereo.restype = c_int
opusfile.op_read_stereo.argtypes = [c_void_p, POINTER(c_int16), c_int]
opusfile.op_pcm_seek.restype = c_int
opusfile.op_pcm_seek.argtypes = [c_void_p, c_longlong]
opusfile.op_pcm_tell.restype = c_longlong
opusfile.op_pcm_tell.argtypes = [c_void_p]
# --- winmm prototypes (Windows only) ---
if winmm is None:
return
winmm.waveOutOpen.restype = wt.DWORD
winmm.waveOutOpen.argtypes = [
POINTER(wt.HANDLE), wt.UINT, POINTER(WAVEFORMATEX),
wt.DWORD, wt.DWORD, wt.DWORD]
winmm.waveOutPrepareHeader.restype = wt.DWORD
winmm.waveOutPrepareHeader.argtypes = [wt.HANDLE, POINTER(WAVEHDR), wt.UINT]
winmm.waveOutWrite.restype = wt.DWORD
winmm.waveOutWrite.argtypes = [wt.HANDLE, POINTER(WAVEHDR), wt.UINT]
winmm.waveOutUnprepareHeader.restype = wt.DWORD
winmm.waveOutUnprepareHeader.argtypes = [wt.HANDLE, POINTER(WAVEHDR), wt.UINT]
winmm.waveOutClose.restype = wt.DWORD
winmm.waveOutClose.argtypes = [wt.HANDLE]
winmm.waveOutSetVolume.restype = wt.DWORD
winmm.waveOutSetVolume.argtypes = [wt.HANDLE, wt.DWORD]
winmm.waveOutGetVolume.restype = wt.DWORD
winmm.waveOutGetVolume.argtypes = [wt.HANDLE, POINTER(wt.DWORD)]
winmm.waveOutPause.restype = wt.DWORD
winmm.waveOutPause.argtypes = [wt.HANDLE]
winmm.waveOutRestart.restype = wt.DWORD
winmm.waveOutRestart.argtypes = [wt.HANDLE]
winmm.waveOutReset.restype = wt.DWORD
winmm.waveOutReset.argtypes = [wt.HANDLE]
winmm.waveOutGetErrorTextW.restype = wt.DWORD
winmm.waveOutGetErrorTextW.argtypes = [wt.DWORD, wt.LPWSTR, wt.UINT]
# --- kernel32 prototypes ---
kernel32.CreateEventW.restype = wt.HANDLE
kernel32.CreateEventW.argtypes = [c_void_p, wt.BOOL, wt.BOOL, wt.LPCWSTR]
kernel32.WaitForSingleObject.restype = wt.DWORD
kernel32.WaitForSingleObject.argtypes = [wt.HANDLE, wt.DWORD]
kernel32.ResetEvent.restype = wt.BOOL
kernel32.ResetEvent.argtypes = [wt.HANDLE]
kernel32.CloseHandle.restype = wt.BOOL
kernel32.CloseHandle.argtypes = [wt.HANDLE]
# ============================================================
# Initialize: Load system DLLs and Opus DLLs
# ============================================================
# Windows-specific system DLLs (winmm/kernel32) are only available on Windows.
# On Linux/macOS, these are set to None; Opus decoding still works via opusfile,
# but waveOut playback is Windows-only.
if sys.platform == "win32":
winmm = ctypes.WinDLL("winmm")
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
else:
winmm = None
kernel32 = None
# Load Opus libraries (with auto-download on Windows, system .so on Linux)
if import_opus():
_setup_prototypes()
ap_ds/_sdl2.py
+16 -61
View File
@@ -112,61 +112,6 @@ def _load_user_config():
return False
def _run_sudo_command(cmd, packages):
"""Run a sudo command with interactive password input.
First tries without password (if already root or passwordless sudo).
If that fails, prompts for the sudo password interactively.
Args:
cmd: Base command list (e.g. ['apt-get', 'install', '-y'])
packages: Package names to install
Returns:
bool: True if command succeeded
"""
import subprocess
import getpass
full_cmd = cmd + packages
# Try without password first (if already root / passwordless sudo)
try:
result = subprocess.run(
['sudo', '-n'] + full_cmd,
capture_output=True, text=True, timeout=120
)
if result.returncode == 0:
return True
except Exception:
pass
# Prompt for sudo password interactively
print("🔑 sudo password required for package installation")
try:
password = getpass.getpass("Enter sudo password: ")
except (EOFError, KeyboardInterrupt):
print("\n❌ Password input cancelled")
return False
# Use sudo -S to read password from stdin
try:
result = subprocess.run(
['sudo', '-S'] + full_cmd,
input=password + '\n',
capture_output=True, text=True, timeout=180
)
if result.returncode == 0:
print("✅ Packages installed successfully")
return True
else:
print(f"❌ Installation failed: {result.stderr.strip()}")
return False
except Exception as e:
print(f"❌ Installation error: {e}")
return False
def _linux_auto_install():
"""Try automatic package manager installation on Linux"""
try:
@@ -175,24 +120,33 @@ def _linux_auto_install():
if shutil.which('apt-get'):
print("📦 Detected apt-based system (Ubuntu/Debian)")
if _run_sudo_command(['apt-get', 'install', '-y'],
['libsdl2-dev', 'libsdl2-mixer-dev']):
result = subprocess.run(
['sudo', 'apt-get', 'install', '-y', 'libsdl2-dev', 'libsdl2-mixer-dev'],
capture_output=True, text=True
)
if result.returncode == 0:
if _load_from_system():
print("✅ SDL2 libraries installed and loaded")
return True
elif shutil.which('dnf'):
print("📦 Detected dnf-based system (Fedora)")
if _run_sudo_command(['dnf', 'install', '-y'],
['SDL2-devel', 'SDL2_mixer-devel']):
result = subprocess.run(
['sudo', 'dnf', 'install', '-y', 'SDL2-devel', 'SDL2_mixer-devel'],
capture_output=True, text=True
)
if result.returncode == 0:
if _load_from_system():
print("✅ SDL2 libraries installed and loaded")
return True
elif shutil.which('pacman'):
print("📦 Detected pacman-based system (Arch)")
if _run_sudo_command(['pacman', '-S', '--noconfirm'],
['sdl2', 'sdl2_mixer']):
result = subprocess.run(
['sudo', 'pacman', '-S', '--noconfirm', 'sdl2', 'sdl2_mixer'],
capture_output=True, text=True
)
if result.returncode == 0:
if _load_from_system():
print("✅ SDL2 libraries installed and loaded")
return True
@@ -201,6 +155,7 @@ def _linux_auto_install():
print(f"⚠️ Automatic installation failed: {e}")
return False
def _linux_interactive_setup():
"""Linux interactive setup for SDL2"""
global _sdl_lib, _mix_lib
ap_ds/_version.py
+1 -1
View File
@@ -1 +1 @@
__version__ = "4.1.0rc1"
__version__ = "4.0.1"
ap_ds/opusplayer.py
-1221
View File
File diff suppressed because it is too large Load Diff
ap_ds/player.py
+6 -212
View File
@@ -140,30 +140,6 @@ 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
# ============================================================
@@ -197,31 +173,6 @@ class AudioLibrary:
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)
@@ -230,46 +181,16 @@ class AudioLibrary:
# ============================================================
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
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):
# 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
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 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)
@@ -283,13 +204,6 @@ class AudioLibrary:
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")
@@ -339,20 +253,6 @@ class AudioLibrary:
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):
@@ -426,18 +326,6 @@ class AudioLibrary:
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:
@@ -491,16 +379,6 @@ class AudioLibrary:
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:
@@ -533,13 +411,6 @@ class AudioLibrary:
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']:
@@ -594,20 +465,6 @@ class AudioLibrary:
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']:
@@ -655,10 +512,6 @@ class AudioLibrary:
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, "", "")
@@ -672,9 +525,6 @@ class AudioLibrary:
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:
@@ -684,9 +534,6 @@ class AudioLibrary:
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:
@@ -699,9 +546,6 @@ class AudioLibrary:
- 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()
# ============================================================
@@ -869,13 +713,6 @@ class AudioLibrary:
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")
@@ -899,16 +736,6 @@ class AudioLibrary:
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")
@@ -932,13 +759,6 @@ class AudioLibrary:
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")
@@ -1009,13 +829,6 @@ class AudioLibrary:
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")
@@ -1051,13 +864,6 @@ class AudioLibrary:
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")
@@ -1130,12 +936,6 @@ class AudioLibrary:
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
@@ -1237,12 +1037,6 @@ class AudioLibrary:
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)
setup.py
+1 -1
View File
@@ -10,7 +10,7 @@ def read_file(filename):
return "Audio Player By DVS - Advanced audio processing and playback library"
# 定义版本常量
VERSION = "4.1.0rc1"
VERSION = "4.0.1"
# 自动生成或更新 _version.py
def write_version_file():