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 eda76fa..a5a938b 100644 --- a/python/lc3.py +++ b/python/lc3.py @@ -20,11 +20,11 @@ import enum import glob import os +import sys 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 +46,170 @@ 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 + + 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 = [ + pkg_dir, + os.path.join(pkg_dir, ".lc3py.mesonpy.libs"), + ] + for directory in search_dirs: + 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 sonames: + 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 +238,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 +293,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 +329,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 +349,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( @@ -350,21 +416,22 @@ 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 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 +447,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 +457,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 +477,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 +504,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 +538,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 +551,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: diff --git a/python/tests/basic_test.py b/python/tests/basic_test.py index de61088..82336e9 100644 --- a/python/tests/basic_test.py +++ b/python/tests/basic_test.py @@ -1,4 +1,9 @@ +from __future__ import annotations + import array +import math +import struct + import lc3 import pytest @@ -45,3 +50,281 @@ def test_encode_with_bad_bit_depth() -> None: encoder = lc3.Encoder(frame_duration_us=10000, sample_rate_hz=48000) with pytest.raises(lc3.InvalidArgumentError): encoder.encode(bytes(1920), num_bytes=120, bit_depth=128) + + +def _generate_sine( + num_samples: int, sample_rate_hz: int, freq_hz: float = 440.0, amp: float = 0.5 +) -> list[float]: + """Generates a sinusoidal test signal with floating-point amplitudes.""" + return [ + amp * math.sin(2.0 * math.pi * freq_hz * i / sample_rate_hz) + for i in range(num_samples) + ] + + +def _floats_to_pcm(floats: list[float], bit_depth: int | None) -> bytes | list[float]: + """Converts a float sample sequence to the appropriate PCM representation.""" + if bit_depth is None: + return list(floats) + if bit_depth == 16: + raw = bytearray() + for x in floats: + val = max(-32768, min(32767, int(x * 32767))) + raw.extend(struct.pack(" list[float]: + """Converts decoded PCM output back to normalized float samples.""" + if bit_depth is None: + return list(pcm) + if bit_depth == 16: + ints = struct.unpack(f"<{len(pcm) // 2}h", pcm) + return [x / 32768.0 for x in ints] + if bit_depth == 24: + res = [] + for i in range(0, len(pcm), 3): + val = int.from_bytes(pcm[i : i + 3], byteorder="little", signed=True) + res.append(val / 8388608.0) + return res + raise ValueError(f"Unsupported bit_depth: {bit_depth}") + + +def _deinterleave(interleaved: list[float], num_channels: int) -> list[list[float]]: + """Separates an interleaved multi-channel sample stream into per-channel lists.""" + return [interleaved[ch::num_channels] for ch in range(num_channels)] + + +def _pearson_correlation(x: list[float], y: list[float]) -> float: + """Computes the Pearson correlation coefficient between two signals.""" + n = min(len(x), len(y)) + if n == 0: + return 0.0 + x_s = x[:n] + y_s = y[:n] + mx = sum(x_s) / n + my = sum(y_s) / n + num = sum((xi - mx) * (yi - my) for xi, yi in zip(x_s, y_s)) + den_x = sum((xi - mx) ** 2 for xi in x_s) + den_y = sum((yi - my) ** 2 for yi in y_s) + den = math.sqrt(den_x * den_y) + return num / den if den > 0 else 0.0 + + +def _calculate_rms(samples: list[float]) -> float: + """Calculates the root mean square (RMS) amplitude of samples.""" + if not samples: + return 0.0 + return math.sqrt(sum(x * x for x in samples) / len(samples)) + + +@pytest.mark.parametrize("bit_depth", [None, 16, 24]) +def test_single_frame_roundtrip(bit_depth: int | None) -> None: + """Validates single-frame encoding and decoding roundtrip.""" + encoder = lc3.Encoder(frame_duration_us=10000, sample_rate_hz=48000) + decoder = lc3.Decoder(frame_duration_us=10000, sample_rate_hz=48000) + samples_per_frame = encoder.get_frame_samples() + num_bytes = encoder.get_frame_bytes(64000) + + sine_frame = _generate_sine(samples_per_frame, 48000, freq_hz=440.0) + pcm_in = _floats_to_pcm(sine_frame, bit_depth) + + encoded = encoder.encode(pcm_in, num_bytes=num_bytes, bit_depth=bit_depth) + assert isinstance(encoded, bytes) + assert len(encoded) == num_bytes + + decoded = decoder.decode(encoded, bit_depth=bit_depth) + if bit_depth is None: + assert isinstance(decoded, array.array) + assert len(decoded) == samples_per_frame + else: + assert isinstance(decoded, bytes) + bytes_per_sample = 2 if bit_depth == 16 else 3 + assert len(decoded) == samples_per_frame * bytes_per_sample + + decoded_floats = _pcm_to_floats(decoded, bit_depth) + assert any(abs(x) > 0.0 for x in decoded_floats) + assert not any(math.isnan(x) or math.isinf(x) for x in decoded_floats) + + +@pytest.mark.parametrize("frame_duration_us", [10000, 7500]) +@pytest.mark.parametrize("sample_rate_hz", [48000, 16000]) +@pytest.mark.parametrize("num_channels", [1, 2]) +@pytest.mark.parametrize("bit_depth", [None, 16, 24]) +def test_stream_roundtrip( + frame_duration_us: int, + sample_rate_hz: int, + num_channels: int, + bit_depth: int | None, +) -> None: + """Validates multi-frame audio streaming roundtrip with delay compensation.""" + encoder = lc3.Encoder( + frame_duration_us=frame_duration_us, + sample_rate_hz=sample_rate_hz, + num_channels=num_channels, + ) + decoder = lc3.Decoder( + frame_duration_us=frame_duration_us, + sample_rate_hz=sample_rate_hz, + num_channels=num_channels, + ) + + spf = encoder.get_frame_samples() + delay = encoder.get_delay_samples() + bitrate = 64000 * num_channels + num_bytes = encoder.get_frame_bytes(bitrate) + num_frames = 10 + + # Generate test tones: 440 Hz for ch0, 880 Hz for ch1 (if stereo) + channel_inputs = [ + _generate_sine(num_frames * spf, sample_rate_hz, freq_hz=440.0 * (c + 1)) + for c in range(num_channels) + ] + + decoded_stream: list[float] = [] + for f in range(num_frames): + frame_floats: list[float] = [] + for s in range(spf): + for c in range(num_channels): + frame_floats.append(channel_inputs[c][f * spf + s]) + + pcm_in = _floats_to_pcm(frame_floats, bit_depth) + encoded = encoder.encode(pcm_in, num_bytes=num_bytes, bit_depth=bit_depth) + decoded = decoder.decode(encoded, bit_depth=bit_depth) + decoded_stream.extend(_pcm_to_floats(decoded, bit_depth)) + + decoded_channels = _deinterleave(decoded_stream, num_channels) + for c in range(num_channels): + in_ch = channel_inputs[c] + out_ch = decoded_channels[c] + + # Delay compensation: align input with output + comp_in = in_ch[: len(out_ch) - delay] + comp_out = out_ch[delay : delay + len(comp_in)] + + # Discard the first frame to allow encoder/decoder filter warm-up + eval_in = comp_in[spf:] + eval_out = comp_out[spf:] + + corr = _pearson_correlation(eval_in, eval_out) + rms = _calculate_rms(eval_out) + + assert corr > 0.95, f"Correlation {corr:.4f} below threshold for channel {c}" + assert 0.20 <= rms <= 0.60, f"RMS {rms:.4f} outside expected range" + assert not any(math.isnan(x) or math.isinf(x) for x in eval_out) + + +def test_stereo_channel_separation() -> None: + """Verifies that stereo encoding and decoding maintains channel separation.""" + sample_rate_hz = 48000 + frame_duration_us = 10000 + encoder = lc3.Encoder( + frame_duration_us=frame_duration_us, + sample_rate_hz=sample_rate_hz, + num_channels=2, + ) + decoder = lc3.Decoder( + frame_duration_us=frame_duration_us, + sample_rate_hz=sample_rate_hz, + num_channels=2, + ) + + spf = encoder.get_frame_samples() + delay = encoder.get_delay_samples() + num_bytes = encoder.get_frame_bytes(128000) + num_frames = 10 + + # Ch0: 440 Hz, Ch1: 880 Hz + ch0 = _generate_sine(num_frames * spf, sample_rate_hz, freq_hz=440.0) + ch1 = _generate_sine(num_frames * spf, sample_rate_hz, freq_hz=880.0) + + decoded_stream: list[float] = [] + for f in range(num_frames): + frame_floats: list[float] = [] + for s in range(spf): + frame_floats.append(ch0[f * spf + s]) + frame_floats.append(ch1[f * spf + s]) + + pcm_in = _floats_to_pcm(frame_floats, 16) + encoded = encoder.encode(pcm_in, num_bytes=num_bytes, bit_depth=16) + decoded = decoder.decode(encoded, bit_depth=16) + decoded_stream.extend(_pcm_to_floats(decoded, 16)) + + decoded_channels = _deinterleave(decoded_stream, 2) + comp_ch0_in = ch0[spf : len(decoded_channels[0]) - delay] + comp_ch0_out = decoded_channels[0][spf + delay : spf + delay + len(comp_ch0_in)] + comp_ch1_in = ch1[spf : len(decoded_channels[1]) - delay] + comp_ch1_out = decoded_channels[1][spf + delay : spf + delay + len(comp_ch1_in)] + + corr_ch0_match = _pearson_correlation(comp_ch0_in, comp_ch0_out) + corr_ch0_cross = _pearson_correlation(comp_ch0_in, comp_ch1_out) + corr_ch1_match = _pearson_correlation(comp_ch1_in, comp_ch1_out) + corr_ch1_cross = _pearson_correlation(comp_ch1_in, comp_ch0_out) + + assert corr_ch0_match > 0.95 + assert abs(corr_ch0_cross) < 0.15 + assert corr_ch1_match > 0.95 + assert abs(corr_ch1_cross) < 0.15 + + +@pytest.mark.parametrize("bit_depth", [None, 16, 24]) +def test_silence_roundtrip(bit_depth: int | None) -> None: + """Verifies that encoding silence produces silent output without noise.""" + encoder = lc3.Encoder(frame_duration_us=10000, sample_rate_hz=48000) + decoder = lc3.Decoder(frame_duration_us=10000, sample_rate_hz=48000) + spf = encoder.get_frame_samples() + num_bytes = encoder.get_frame_bytes(64000) + + silent_frame = [0.0] * spf + pcm_in = _floats_to_pcm(silent_frame, bit_depth) + + for _ in range(5): + encoded = encoder.encode(pcm_in, num_bytes=num_bytes, bit_depth=bit_depth) + decoded = decoder.decode(encoded, bit_depth=bit_depth) + decoded_floats = _pcm_to_floats(decoded, bit_depth) + rms = _calculate_rms(decoded_floats) + assert rms < 1e-4, f"Expected silence, but got RMS {rms:.6f}" + + +@pytest.mark.parametrize("bit_depth", [None, 16, 24]) +def test_packet_loss_concealment(bit_depth: int | None) -> None: + """Verifies PLC generates smooth concealed frames when packet loss occurs.""" + encoder = lc3.Encoder(frame_duration_us=10000, sample_rate_hz=48000) + decoder = lc3.Decoder(frame_duration_us=10000, sample_rate_hz=48000) + spf = encoder.get_frame_samples() + num_bytes = encoder.get_frame_bytes(64000) + + audio = _generate_sine(3 * spf, 48000, freq_hz=440.0) + + # Feed 3 good frames + for f in range(3): + pcm_in = _floats_to_pcm(audio[f * spf : (f + 1) * spf], bit_depth) + encoded = encoder.encode(pcm_in, num_bytes=num_bytes, bit_depth=bit_depth) + decoder.decode(encoded, bit_depth=bit_depth) + + # Feed 2 lost frames (PLC) + plc1 = decoder.decode(None, bit_depth=bit_depth) + plc2 = decoder.decode(None, bit_depth=bit_depth) + + plc1_floats = _pcm_to_floats(plc1, bit_depth) + plc2_floats = _pcm_to_floats(plc2, bit_depth) + + assert len(plc1_floats) == spf + assert len(plc2_floats) == spf + + rms1 = _calculate_rms(plc1_floats) + rms2 = _calculate_rms(plc2_floats) + + assert rms1 > 0.05, f"PLC frame 1 should have energy, got {rms1:.4f}" + assert rms2 <= rms1 + 1e-3, ( + f"PLC frame 2 should decay or stay bounded, got {rms2:.4f} vs {rms1:.4f}" + )