From e5bba32c7aa3894a06948733311d556801535534 Mon Sep 17 00:00:00 2001 From: Josh Wu Date: Wed, 26 Aug 2026 23:39:22 +0800 Subject: [PATCH 1/2] python: Refactor ctypes library loading and memory management - Cache library loading and configure ctypes function signatures once - Support LIBLC3_PATH and multi-platform library search paths (.so, .dylib, .dll) - Use Python GC-managed buffers instead of manual libc malloc/free - Fix docstring typos --- python/lc3.py | 381 ++++++++++++++++++++++++++------------------------ 1 file changed, 199 insertions(+), 182 deletions(-) diff --git a/python/lc3.py b/python/lc3.py index eda76fa..ca92965 100644 --- a/python/lc3.py +++ b/python/lc3.py @@ -21,10 +21,9 @@ import glob import os import typing - -from ctypes import c_bool, c_byte, c_int, c_uint, c_size_t, c_void_p -from ctypes.util import find_library from collections.abc import Iterable +from ctypes import c_bool, c_byte, c_int, c_uint, c_void_p +from ctypes.util import find_library class BaseError(Exception): @@ -46,8 +45,158 @@ class _PcmFormat(enum.IntEnum): FLOAT = 3 -class _Base: +_LIB_CACHE: dict[str, ctypes.CDLL] = {} + + +def _find_library(libpath: str | None = None) -> str: + """Finds the liblc3 shared library via explicit path, bundled wheel, or dynamic linker.""" + if libpath: + if os.path.exists(libpath): + return libpath + raise InitializationError( + f"Specified LC3 library path does not exist: {libpath}" + ) + + if (env_path := os.environ.get("LIBLC3_PATH")) and os.path.exists(env_path): + return env_path + + # Search package directory and wheel directory (.lc3py.mesonpy.libs) + pkg_dir = os.path.dirname(os.path.abspath(__file__)) + search_dirs = [ + pkg_dir, + os.path.join(pkg_dir, ".lc3py.mesonpy.libs"), + ] + for directory in search_dirs: + for ext in ("so*", "dylib", "dll"): + for match in glob.glob(os.path.join(directory, f"*lc3*.{ext}")): + if os.path.isfile(match) and "cpython" not in match: + return match + + # Search standard system library and dynamic linker paths + if sys_lib := find_library("lc3"): + return sys_lib + + for soname in ("liblc3.so.1", "liblc3.so", "liblc3.dylib", "lc3.dll"): + try: + ctypes.cdll.LoadLibrary(soname) + return soname + except OSError: + pass + + raise InitializationError( + "LC3 library not found. Please ensure liblc3 is installed or set the LIBLC3_PATH environment variable." + ) + + +def _load_lc3_library(libpath: str | None = None) -> ctypes.CDLL: + """Loads and configures the liblc3 ctypes library once (cached singleton).""" + resolved_path = _find_library(libpath) + if resolved_path in _LIB_CACHE: + return _LIB_CACHE[resolved_path] + + try: + lib = ctypes.cdll.LoadLibrary(resolved_path) + except Exception as e: + raise InitializationError( + f"Failed to load LC3 library from {resolved_path}: {e}" + ) from e + + if not all( + hasattr(lib, func) + for func in ( + "lc3_hr_frame_samples", + "lc3_hr_frame_block_bytes", + "lc3_hr_resolve_bitrate", + "lc3_hr_delay_samples", + ) + ): + lc3_hr_frame_samples = lambda hrmode, dt_us, sr_hz: lib.lc3_frame_samples( + dt_us, sr_hz + ) + lc3_hr_frame_block_bytes = lambda hrmode, dt_us, sr_hz, num_channels, bitrate: ( + num_channels * lib.lc3_frame_bytes(dt_us, bitrate // 2) + ) + lc3_hr_resolve_bitrate = lambda hrmode, dt_us, sr_hz, nbytes: ( + lib.lc3_resolve_bitrate(dt_us, nbytes) + ) + lc3_hr_delay_samples = lambda hrmode, dt_us, sr_hz: lib.lc3_delay_samples( + dt_us, sr_hz + ) + lib.lc3_hr_frame_samples = lc3_hr_frame_samples + lib.lc3_hr_frame_block_bytes = lc3_hr_frame_block_bytes + lib.lc3_hr_resolve_bitrate = lc3_hr_resolve_bitrate + lib.lc3_hr_delay_samples = lc3_hr_delay_samples + lib._has_hr = False + else: + lib.lc3_hr_frame_samples.argtypes = [c_bool, c_int, c_int] + lib.lc3_hr_frame_samples.restype = c_int + lib.lc3_hr_frame_block_bytes.argtypes = [c_bool, c_int, c_int, c_int, c_int] + lib.lc3_hr_frame_block_bytes.restype = c_int + lib.lc3_hr_resolve_bitrate.argtypes = [c_bool, c_int, c_int, c_int] + lib.lc3_hr_resolve_bitrate.restype = c_int + lib.lc3_hr_delay_samples.argtypes = [c_bool, c_int, c_int] + lib.lc3_hr_delay_samples.restype = c_int + lib._has_hr = True + + if not all( + hasattr(lib, func) for func in ("lc3_hr_encoder_size", "lc3_hr_setup_encoder") + ): + lc3_hr_encoder_size = lambda hrmode, dt_us, sr_hz: lib.lc3_encoder_size( + dt_us, sr_hz + ) + lc3_hr_setup_encoder = lambda hrmode, dt_us, sr_hz, sr_pcm_hz, mem: ( + lib.lc3_setup_encoder(dt_us, sr_hz, sr_pcm_hz, mem) + ) + lib.lc3_hr_encoder_size = lc3_hr_encoder_size + lib.lc3_hr_setup_encoder = lc3_hr_setup_encoder + else: + lib.lc3_hr_encoder_size.argtypes = [c_bool, c_int, c_int] + lib.lc3_hr_encoder_size.restype = c_uint + lib.lc3_hr_setup_encoder.argtypes = [c_bool, c_int, c_int, c_int, c_void_p] + lib.lc3_hr_setup_encoder.restype = c_void_p + if not all( + hasattr(lib, func) for func in ("lc3_hr_decoder_size", "lc3_hr_setup_decoder") + ): + lc3_hr_decoder_size = lambda hrmode, dt_us, sr_hz: lib.lc3_decoder_size( + dt_us, sr_hz + ) + lc3_hr_setup_decoder = lambda hrmode, dt_us, sr_hz, sr_pcm_hz, mem: ( + lib.lc3_setup_decoder(dt_us, sr_hz, sr_pcm_hz, mem) + ) + lib.lc3_hr_decoder_size = lc3_hr_decoder_size + lib.lc3_hr_setup_decoder = lc3_hr_setup_decoder + else: + lib.lc3_hr_decoder_size.argtypes = [c_bool, c_int, c_int] + lib.lc3_hr_decoder_size.restype = c_uint + lib.lc3_hr_setup_decoder.argtypes = [c_bool, c_int, c_int, c_int, c_void_p] + lib.lc3_hr_setup_decoder.restype = c_void_p + + lib.lc3_encode.argtypes = [ + c_void_p, + c_int, + c_void_p, + c_int, + c_int, + c_void_p, + ] + lib.lc3_encode.restype = c_int + + lib.lc3_decode.argtypes = [ + c_void_p, + c_void_p, + c_int, + c_int, + c_void_p, + c_int, + ] + lib.lc3_decode.restype = c_int + + _LIB_CACHE[resolved_path] = lib + return lib + + +class _Base: def __init__( self, frame_duration_us: int, @@ -76,68 +225,9 @@ def __init__( if self.sample_rate_hz not in allowed_samplerate: raise InvalidArgumentError(f"Invalid sample rate: {sample_rate_hz} Hz") - if libpath is None: - mesonpy_lib = glob.glob( - os.path.join(os.path.dirname(__file__), ".lc3py.mesonpy.libs", "*lc3*") - ) - - if mesonpy_lib: - libpath = mesonpy_lib[0] - else: - libpath = find_library("lc3") - if not libpath: - raise InitializationError("LC3 library not found") - - lib = ctypes.cdll.LoadLibrary(libpath) - - if not all( - hasattr(lib, func) - for func in ( - "lc3_hr_frame_samples", - "lc3_hr_frame_block_bytes", - "lc3_hr_resolve_bitrate", - "lc3_hr_delay_samples", - ) - ): - if self.hrmode: - raise InitializationError("High-Resolution interface not available") - - lc3_hr_frame_samples = lambda hrmode, dt_us, sr_hz: lib.lc3_frame_samples( - dt_us, sr_hz - ) - lc3_hr_frame_block_bytes = ( - lambda hrmode, dt_us, sr_hz, num_channels, bitrate: num_channels - * lib.lc3_frame_bytes(dt_us, bitrate // 2) - ) - lc3_hr_resolve_bitrate = ( - lambda hrmode, dt_us, sr_hz, nbytes: lib.lc3_resolve_bitrate( - dt_us, nbytes - ) - ) - lc3_hr_delay_samples = lambda hrmode, dt_us, sr_hz: lib.lc3_delay_samples( - dt_us, sr_hz - ) - setattr(lib, "lc3_hr_frame_samples", lc3_hr_frame_samples) - setattr(lib, "lc3_hr_frame_block_bytes", lc3_hr_frame_block_bytes) - setattr(lib, "lc3_hr_resolve_bitrate", lc3_hr_resolve_bitrate) - setattr(lib, "lc3_hr_delay_samples", lc3_hr_delay_samples) - - lib.lc3_hr_frame_samples.argtypes = [c_bool, c_int, c_int] - lib.lc3_hr_frame_block_bytes.argtypes = [c_bool, c_int, c_int, c_int, c_int] - lib.lc3_hr_resolve_bitrate.argtypes = [c_bool, c_int, c_int, c_int] - lib.lc3_hr_delay_samples.argtypes = [c_bool, c_int, c_int] - self.lib = lib - - if not (libc_path := find_library("c")): - raise InitializationError("Unable to find libc") - libc = ctypes.cdll.LoadLibrary(libc_path) - - self.malloc = libc.malloc - self.malloc.argtypes = [c_size_t] - self.malloc.restype = c_void_p - - self.free = libc.free - self.free.argtypes = [c_void_p] + self.lib = _load_lc3_library(libpath) + if self.hrmode and not getattr(self.lib, "_has_hr", True): + raise InitializationError("High-Resolution interface not available") def get_frame_samples(self) -> int: """ @@ -190,9 +280,11 @@ def get_delay_samples(self) -> int: return ret @classmethod - def _resolve_pcm_format(cls, bit_depth: int | None) -> tuple[ + def _resolve_pcm_format( + cls, bit_depth: int | None + ) -> tuple[ _PcmFormat, - type[ctypes.c_int16] | type[ctypes.Array[ctypes.c_byte]] | type[ctypes.c_float], + type[ctypes.c_int16 | ctypes.Array[ctypes.c_byte] | ctypes.c_float], ]: match bit_depth: case 16: @@ -224,9 +316,6 @@ class Encoder(_Base): libpath : LC3 library path and name """ - class c_encoder_t(c_void_p): - pass - def __init__( self, frame_duration_us: int, @@ -247,62 +336,26 @@ def __init__( ) lib = self.lib + enc_size = lib.lc3_hr_encoder_size( + self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz + ) + if enc_size == 0: + raise InitializationError("Failed to determine LC3 encoder size") - if not all( - hasattr(lib, func) - for func in ("lc3_hr_encoder_size", "lc3_hr_setup_encoder") - ): - if self.hrmode: - raise InitializationError("High-Resolution interface not available") - - lc3_hr_encoder_size = lambda hrmode, dt_us, sr_hz: lib.lc3_encoder_size( - dt_us, sr_hz - ) - - lc3_hr_setup_encoder = ( - lambda hrmode, dt_us, sr_hz, sr_pcm_hz, mem: lib.lc3_setup_encoder( - dt_us, sr_hz, sr_pcm_hz, mem - ) - ) - setattr(lib, "lc3_hr_encoder_size", lc3_hr_encoder_size) - setattr(lib, "lc3_hr_setup_encoder", lc3_hr_setup_encoder) - - lib.lc3_hr_encoder_size.argtypes = [c_bool, c_int, c_int] - lib.lc3_hr_encoder_size.restype = c_uint - - lib.lc3_hr_setup_encoder.argtypes = [c_bool, c_int, c_int, c_int, c_void_p] - lib.lc3_hr_setup_encoder.restype = self.c_encoder_t - - lib.lc3_encode.argtypes = [ - self.c_encoder_t, - c_int, - c_void_p, - c_int, - c_int, - c_void_p, - ] - - def new_encoder(): - return lib.lc3_hr_setup_encoder( + # Allocate memory buffers managed by Python GC - no libc.malloc/free needed + self._mem_buffers = [(c_byte * enc_size)() for _ in range(num_channels)] + self.__encoders = [ + lib.lc3_hr_setup_encoder( self.hrmode, self.frame_duration_us, self.sample_rate_hz, self.pcm_sample_rate_hz, - self.malloc( - lib.lc3_hr_encoder_size( - self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz - ) - ), + ctypes.byref(buf), ) - - self.__encoders = [new_encoder() for _ in range(num_channels)] - - def __del__(self) -> None: - - try: - (self.free(encoder) for encoder in self.__encoders) - finally: - return + for buf in self._mem_buffers + ] + if any(not enc for enc in self.__encoders): + raise InitializationError("Failed to initialize LC3 encoder") @typing.overload def encode( @@ -356,15 +409,16 @@ def encode(self, pcm, num_bytes: int, bit_depth: int | None = None) -> bytes: data_offset = 0 for ich, encoder in enumerate(self.__encoders): - pcm_offset = ich * ctypes.sizeof(pcm_t) - pcm = (pcm_t * (pcm_len - ich)).from_buffer(pcm_buffer, pcm_offset) + pcm_slice = (pcm_t * (pcm_len - ich)).from_buffer(pcm_buffer, pcm_offset) data_size = num_bytes // nchannels + int(ich < num_bytes % nchannels) data = (c_byte * data_size).from_buffer(data_buffer, data_offset) data_offset += data_size - ret = self.lib.lc3_encode(encoder, pcm_fmt, pcm, nchannels, len(data), data) + ret = self.lib.lc3_encode( + encoder, pcm_fmt, pcm_slice, nchannels, len(data), data + ) if ret < 0: raise InvalidArgumentError("Bad parameters") @@ -380,7 +434,7 @@ class Decoder(_Base): or 48000, unless High-Resolution mode is enabled. In High-Resolution mode, the `sample_rate_hz` is 48000 or 96000. - By default, one channel is processed. When `num_chanels` is greater than one, + By default, one channel is processed. When `num_channels` is greater than one, the PCM input stream is read interleaved and consecutives LC3 frames are output, for each channel. @@ -390,9 +444,6 @@ class Decoder(_Base): libpath : LC3 library path and name """ - class c_decoder_t(c_void_p): - pass - def __init__( self, frame_duration_us: int, @@ -413,62 +464,26 @@ def __init__( ) lib = self.lib + dec_size = lib.lc3_hr_decoder_size( + self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz + ) + if dec_size == 0: + raise InitializationError("Failed to determine LC3 decoder size") - if not all( - hasattr(lib, func) - for func in ("lc3_hr_decoder_size", "lc3_hr_setup_decoder") - ): - if self.hrmode: - raise InitializationError("High-Resolution interface not available") - - lc3_hr_decoder_size = lambda hrmode, dt_us, sr_hz: lib.lc3_decoder_size( - dt_us, sr_hz - ) - - lc3_hr_setup_decoder = ( - lambda hrmode, dt_us, sr_hz, sr_pcm_hz, mem: lib.lc3_setup_decoder( - dt_us, sr_hz, sr_pcm_hz, mem - ) - ) - setattr(lib, "lc3_hr_decoder_size", lc3_hr_decoder_size) - setattr(lib, "lc3_hr_setup_decoder", lc3_hr_setup_decoder) - - lib.lc3_hr_decoder_size.argtypes = [c_bool, c_int, c_int] - lib.lc3_hr_decoder_size.restype = c_uint - - lib.lc3_hr_setup_decoder.argtypes = [c_bool, c_int, c_int, c_int, c_void_p] - lib.lc3_hr_setup_decoder.restype = self.c_decoder_t - - lib.lc3_decode.argtypes = [ - self.c_decoder_t, - c_void_p, - c_int, - c_int, - c_void_p, - c_int, - ] - - def new_decoder(): - return lib.lc3_hr_setup_decoder( + # Allocate memory buffers managed by Python GC - no libc.malloc/free needed + self._mem_buffers = [(c_byte * dec_size)() for _ in range(num_channels)] + self.__decoders = [ + lib.lc3_hr_setup_decoder( self.hrmode, self.frame_duration_us, self.sample_rate_hz, self.pcm_sample_rate_hz, - self.malloc( - lib.lc3_hr_decoder_size( - self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz - ) - ), + ctypes.byref(buf), ) - - self.__decoders = [new_decoder() for i in range(num_channels)] - - def __del__(self) -> None: - - try: - (self.free(decoder) for decoder in self.__decoders) - finally: - return + for buf in self._mem_buffers + ] + if any(not dec for dec in self.__decoders): + raise InitializationError("Failed to initialize LC3 decoder") @typing.overload def decode( @@ -476,7 +491,9 @@ def decode( ) -> array.array[float]: ... @typing.overload - def decode(self, data: bytes | bytearray | memoryview | None, bit_depth: int) -> bytes: ... + def decode( + self, data: bytes | bytearray | memoryview | None, bit_depth: int + ) -> bytes: ... def decode( self, data: bytes | bytearray | memoryview | None, bit_depth: int | None = None @@ -508,11 +525,11 @@ def decode( for ich, decoder in enumerate(self.__decoders): pcm_offset = ich * ctypes.sizeof(pcm_t) - pcm = (pcm_t * (pcm_len - ich)).from_buffer(pcm_buffer, pcm_offset) + pcm_slice = (pcm_t * (pcm_len - ich)).from_buffer(pcm_buffer, pcm_offset) if data is None: ret = self.lib.lc3_decode( - decoder, None, 0, pcm_fmt, pcm, self.num_channels + decoder, None, 0, pcm_fmt, pcm_slice, self.num_channels ) else: data_size = len(data_buffer) // num_channels + int( @@ -521,7 +538,7 @@ def decode( buf = (c_byte * data_size).from_buffer(data_buffer, data_offset) data_offset += data_size ret = self.lib.lc3_decode( - decoder, buf, len(buf), pcm_fmt, pcm, self.num_channels + decoder, buf, len(buf), pcm_fmt, pcm_slice, self.num_channels ) if ret < 0: From 6a0ce130b78d08fefe12f2bfc78a69985ad46a12 Mon Sep 17 00:00:00 2001 From: Josh Wu Date: Thu, 27 Aug 2026 20:03:59 +0800 Subject: [PATCH 2/2] ci: Add cross-platform Python tests and adopt dependency-groups - Test python wrapper and basic_test across Linux, macOS, and Windows matrix - Migrate optional-dependencies to dependency-groups in pyproject.toml --- .github/workflows/ci.yaml | 15 ++++++++++++--- pyproject.toml | 5 ++++- python/lc3.py | 19 ++++++++++++++++--- 3 files changed, 32 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 7938dbf..8c1c0f3 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -82,12 +82,21 @@ jobs: - run: gcc -v - run: make test - install-python-linux: - runs-on: ubuntu-latest + test-python: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: '3.10' cache: 'pip' - - run: pip install . + - uses: TheMrMilchmann/setup-msvc-dev@v3 + if: runner.os == 'Windows' + with: + arch: x64 + - run: pip install . pytest + - run: pytest python/tests diff --git a/pyproject.toml b/pyproject.toml index 3de5361..d90962e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,11 +12,14 @@ authors = [ description = "LC3 Codec library wrapper" requires-python = ">=3.10" -[project.optional-dependencies] +[dependency-groups] dev = ["pytest"] [project.urls] Homepage = "https://github.com/google/liblc3" +[tool.meson-python] +allow-windows-internal-shared-libs = true + [tool.meson-python.args] setup = ['-Dpython=true'] diff --git a/python/lc3.py b/python/lc3.py index ca92965..a5a938b 100644 --- a/python/lc3.py +++ b/python/lc3.py @@ -20,6 +20,7 @@ import enum import glob import os +import sys import typing from collections.abc import Iterable from ctypes import c_bool, c_byte, c_int, c_uint, c_void_p @@ -60,6 +61,16 @@ def _find_library(libpath: str | None = None) -> str: if (env_path := os.environ.get("LIBLC3_PATH")) and os.path.exists(env_path): return env_path + if sys.platform == "win32": + exts = ("dll",) + sonames = ("lc3-1.dll", "lc3.dll", "liblc3.dll") + elif sys.platform == "darwin": + exts = ("dylib",) + sonames = ("liblc3.1.dylib", "liblc3.dylib") + else: + exts = ("so*",) + sonames = ("liblc3.so.1", "liblc3.so") + # Search package directory and wheel directory (.lc3py.mesonpy.libs) pkg_dir = os.path.dirname(os.path.abspath(__file__)) search_dirs = [ @@ -67,16 +78,18 @@ def _find_library(libpath: str | None = None) -> str: os.path.join(pkg_dir, ".lc3py.mesonpy.libs"), ] for directory in search_dirs: - for ext in ("so*", "dylib", "dll"): + for ext in exts: for match in glob.glob(os.path.join(directory, f"*lc3*.{ext}")): if os.path.isfile(match) and "cpython" not in match: + if sys.platform == "win32": + os.add_dll_directory(os.path.dirname(match)) return match # Search standard system library and dynamic linker paths if sys_lib := find_library("lc3"): return sys_lib - for soname in ("liblc3.so.1", "liblc3.so", "liblc3.dylib", "lc3.dll"): + for soname in sonames: try: ctypes.cdll.LoadLibrary(soname) return soname @@ -403,7 +416,7 @@ def encode(self, pcm, num_bytes: int, bit_depth: int | None = None) -> bytes: else: padding = max(pcm_len * ctypes.sizeof(pcm_t) - len(pcm), 0) - pcm_buffer = bytearray(pcm) + bytearray(padding) # type: ignore + pcm_buffer = bytearray(pcm) + bytearray(padding) data_buffer = (c_byte * num_bytes)() data_offset = 0