diff --git a/python/_lc3.pyi b/python/_lc3.pyi new file mode 100644 index 0000000..1477d87 --- /dev/null +++ b/python/_lc3.pyi @@ -0,0 +1,87 @@ +# +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +import collections.abc + +class EncoderContext: + def __init__( + self, + hrmode: bool, + dt_us: int, + sr_hz: int, + sr_pcm_hz: int, + nchannels: int, + ) -> None: ... + def encode( + self, + pcm: collections.abc.Buffer, + num_bytes: int, + pcm_format: int, + ) -> bytes: ... + def disable_ltpf(self) -> None: ... + +class DecoderContext: + def __init__( + self, + hrmode: bool, + dt_us: int, + sr_hz: int, + sr_pcm_hz: int, + nchannels: int, + ) -> None: ... + def decode( + self, + data: collections.abc.Buffer | None, + pcm_format: int, + ) -> bytes: ... + +def hr_frame_samples( + hrmode: bool, + dt_us: int, + sr_hz: int, +) -> int: ... + +def hr_frame_block_bytes( + hrmode: bool, + dt_us: int, + sr_hz: int, + nchannels: int, + bitrate: int, +) -> int: ... + +def hr_resolve_bitrate( + hrmode: bool, + dt_us: int, + sr_hz: int, + nbytes: int, +) -> int: ... + +def hr_delay_samples( + hrmode: bool, + dt_us: int, + sr_hz: int, +) -> int: ... + +def hr_encoder_size( + hrmode: bool, + dt_us: int, + sr_hz: int, +) -> int: ... + +def hr_decoder_size( + hrmode: bool, + dt_us: int, + sr_hz: int, +) -> int: ... diff --git a/python/lc3.py b/python/lc3.py index eda76fa..3935824 100644 --- a/python/lc3.py +++ b/python/lc3.py @@ -16,16 +16,34 @@ from __future__ import annotations import array -import ctypes import enum -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 +try: + from _lc3 import ( + DecoderContext as _DecoderContext, + ) + from _lc3 import ( + EncoderContext as _EncoderContext, + ) + from _lc3 import ( + hr_delay_samples as _hr_delay_samples, + ) + from _lc3 import ( + hr_frame_block_bytes as _hr_frame_block_bytes, + ) + from _lc3 import ( + hr_frame_samples as _hr_frame_samples, + ) + from _lc3 import ( + hr_resolve_bitrate as _hr_resolve_bitrate, + ) +except ImportError as err: + raise RuntimeError( + "Failed to import native LC3 extension (_lc3). Ensure the package is built with meson." + ) from err + class BaseError(Exception): """Base error raised by liblc3.""" @@ -47,7 +65,6 @@ class _PcmFormat(enum.IntEnum): class _Base: - def __init__( self, frame_duration_us: int, @@ -76,79 +93,16 @@ 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] - def get_frame_samples(self) -> int: """ Returns the number of PCM samples in an LC3 frame. """ - ret = self.lib.lc3_hr_frame_samples( - self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz - ) - if ret < 0: - raise InvalidArgumentError("Bad parameters") - return ret + try: + return _hr_frame_samples( + self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz + ) + except ValueError as e: + raise InvalidArgumentError("Bad parameters") from e def get_frame_bytes(self, bitrate: int) -> int: """ @@ -156,51 +110,48 @@ def get_frame_bytes(self, bitrate: int) -> int: A target `bitrate` equals 0 or `INT32_MAX` returns respectively the minimum and maximum allowed size. """ - ret = self.lib.lc3_hr_frame_block_bytes( - self.hrmode, - self.frame_duration_us, - self.sample_rate_hz, - self.num_channels, - bitrate, - ) - if ret < 0: - raise InvalidArgumentError("Bad parameters") - return ret + try: + return _hr_frame_block_bytes( + self.hrmode, + self.frame_duration_us, + self.sample_rate_hz, + self.num_channels, + bitrate, + ) + except ValueError as e: + raise InvalidArgumentError("Bad parameters") from e def resolve_bitrate(self, num_bytes: int) -> int: """ Returns the bitrate in bits per seconds, from the size of LC3 frames. """ - ret = self.lib.lc3_hr_resolve_bitrate( - self.hrmode, self.frame_duration_us, self.sample_rate_hz, num_bytes - ) - if ret < 0: - raise InvalidArgumentError("Bad parameters") - return ret + try: + return _hr_resolve_bitrate( + self.hrmode, self.frame_duration_us, self.sample_rate_hz, num_bytes + ) + except ValueError as e: + raise InvalidArgumentError("Bad parameters") from e def get_delay_samples(self) -> int: """ Returns the algorithmic delay, as a number of samples. """ - ret = self.lib.lc3_hr_delay_samples( - self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz - ) - if ret < 0: - raise InvalidArgumentError("Bad parameters") - return ret + try: + return _hr_delay_samples( + self.hrmode, self.frame_duration_us, self.pcm_sample_rate_hz + ) + except ValueError as e: + raise InvalidArgumentError("Bad parameters") from e @classmethod - 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], - ]: + def _resolve_pcm_format(cls, bit_depth: int | None) -> tuple[_PcmFormat, int]: match bit_depth: case 16: - return (_PcmFormat.S16, ctypes.c_int16) + return (_PcmFormat.S16, 2) case 24: - return (_PcmFormat.S24_3LE, 3 * ctypes.c_byte) + return (_PcmFormat.S24_3LE, 3) case None: - return (_PcmFormat.FLOAT, ctypes.c_float) + return (_PcmFormat.FLOAT, 4) case _: raise InvalidArgumentError("Could not interpret PCM bit_depth") @@ -221,12 +172,9 @@ class Encoder(_Base): Optional arguments: hrmode : Enable High-Resolution mode, default is `False`. input_sample_rate_hz : Input PCM samplerate, enable downsampling of input. - libpath : LC3 library path and name + libpath : LC3 library path and name (compatibility option) """ - class c_encoder_t(c_void_p): - pass - def __init__( self, frame_duration_us: int, @@ -246,63 +194,16 @@ def __init__( libpath, ) - lib = self.lib - - 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( + try: + self._ctx = _EncoderContext( 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 - ) - ), + self.num_channels, ) - - self.__encoders = [new_encoder() for _ in range(num_channels)] - - def __del__(self) -> None: - - try: - (self.free(encoder) for encoder in self.__encoders) - finally: - return + except Exception as e: + raise InitializationError(f"Failed to initialize LC3 encoder: {e}") from e @typing.overload def encode( @@ -332,43 +233,27 @@ def encode(self, pcm, num_bytes: int, bit_depth: int | None = None) -> bytes: Channels concatenation of encoded LC3 frames, of `nbytes`, is returned. """ - nchannels = self.num_channels + pcm_fmt, _ = self._resolve_pcm_format(bit_depth) frame_samples = self.get_frame_samples() - (pcm_fmt, pcm_t) = self._resolve_pcm_format(bit_depth) - pcm_len = nchannels * frame_samples - if bit_depth is None: - pcm_buffer = array.array("f", pcm) + if isinstance(pcm, array.array) and pcm.typecode == "f": + pcm_buffer = pcm + else: + pcm_buffer = array.array("f", pcm) # Invert test to catch NaN if not abs(sum(pcm_buffer)) / frame_samples < 2: raise InvalidArgumentError("Out of range PCM input") - padding = max(pcm_len - frame_samples, 0) - pcm_buffer.extend(array.array("f", [0] * padding)) - + pcm_bytes = pcm_buffer.tobytes() else: - padding = max(pcm_len * ctypes.sizeof(pcm_t) - len(pcm), 0) - pcm_buffer = bytearray(pcm) + bytearray(padding) # type: ignore - - 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) - - 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 + pcm_bytes = pcm - ret = self.lib.lc3_encode(encoder, pcm_fmt, pcm, nchannels, len(data), data) - if ret < 0: - raise InvalidArgumentError("Bad parameters") - - return bytes(data_buffer) + try: + return self._ctx.encode(pcm_bytes, num_bytes, int(pcm_fmt)) + except (ValueError, RuntimeError) as e: + raise InvalidArgumentError("Bad parameters") from e class Decoder(_Base): @@ -380,19 +265,16 @@ 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. Optional arguments: hrmode : Enable High-Resolution mode, default is `False`. output_sample_rate_hz : Output PCM sample_rate_hz, enable upsampling of output. - libpath : LC3 library path and name + libpath : LC3 library path and name (compatibility option) """ - class c_decoder_t(c_void_p): - pass - def __init__( self, frame_duration_us: int, @@ -412,63 +294,16 @@ def __init__( libpath, ) - lib = self.lib - - 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( + try: + self._ctx = _DecoderContext( 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 - ) - ), + self.num_channels, ) - - 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 + except Exception as e: + raise InitializationError(f"Failed to initialize LC3 decoder: {e}") from e @typing.overload def decode( @@ -476,7 +311,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 @@ -496,35 +333,15 @@ def decode( width, respectively. """ - num_channels = self.num_channels - - (pcm_fmt, pcm_t) = self._resolve_pcm_format(bit_depth) - pcm_len = num_channels * self.get_frame_samples() - pcm_buffer = (pcm_t * pcm_len)() + pcm_fmt, _ = self._resolve_pcm_format(bit_depth) - if data is not None: - data_buffer = bytearray(data) - data_offset = 0 - - 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) + try: + raw_bytes = self._ctx.decode(data, int(pcm_fmt)) + except (ValueError, RuntimeError) as e: + raise InvalidArgumentError("Bad parameters") from e - if data is None: - ret = self.lib.lc3_decode( - decoder, None, 0, pcm_fmt, pcm, self.num_channels - ) - else: - data_size = len(data_buffer) // num_channels + int( - ich < len(data_buffer) % num_channels - ) - 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 - ) - - if ret < 0: - raise InvalidArgumentError("Bad parameters") - - return array.array("f", pcm_buffer) if bit_depth is None else bytes(pcm_buffer) + if bit_depth is None: + res = array.array("f") + res.frombytes(raw_bytes) + return res + return raw_bytes diff --git a/python/lc3_native.c b/python/lc3_native.c new file mode 100644 index 0000000..6436578 --- /dev/null +++ b/python/lc3_native.c @@ -0,0 +1,610 @@ +/****************************************************************************** + * + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +#define PY_SSIZE_T_CLEAN +#include +#include +#include +#include +#include + +#include "lc3.h" + +/* ------------------------------------------------------------------------- + * Helper functions + * ------------------------------------------------------------------------- */ + +static PyObject *py_hr_frame_samples(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "pii", kwlist, &hrmode, &dt_us, &sr_hz)) + return NULL; + + int ret = lc3_hr_frame_samples(hrmode != 0, dt_us, sr_hz); + if (ret < 0) { + PyErr_SetString(PyExc_ValueError, "Bad parameters"); + return NULL; + } + + return PyLong_FromLong(ret); +} + +static PyObject *py_hr_frame_block_bytes(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", "nchannels", "bitrate", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0, nchannels = 0, bitrate = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "piiii", kwlist, + &hrmode, &dt_us, &sr_hz, &nchannels, &bitrate)) + return NULL; + + int ret = lc3_hr_frame_block_bytes(hrmode != 0, dt_us, sr_hz, nchannels, bitrate); + if (ret < 0) { + PyErr_SetString(PyExc_ValueError, "Bad parameters"); + return NULL; + } + + return PyLong_FromLong(ret); +} + +static PyObject *py_hr_resolve_bitrate(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", "nbytes", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0, nbytes = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "piii", kwlist, &hrmode, &dt_us, &sr_hz, &nbytes)) + return NULL; + + int ret = lc3_hr_resolve_bitrate(hrmode != 0, dt_us, sr_hz, nbytes); + if (ret < 0) { + PyErr_SetString(PyExc_ValueError, "Bad parameters"); + return NULL; + } + + return PyLong_FromLong(ret); +} + +static PyObject *py_hr_delay_samples(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "pii", kwlist, &hrmode, &dt_us, &sr_hz)) + return NULL; + + int ret = lc3_hr_delay_samples(hrmode != 0, dt_us, sr_hz); + if (ret < 0) { + PyErr_SetString(PyExc_ValueError, "Bad parameters"); + return NULL; + } + + return PyLong_FromLong(ret); +} + +static PyObject *py_hr_encoder_size(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "pii", kwlist, &hrmode, &dt_us, &sr_hz)) + return NULL; + + unsigned ret = lc3_hr_encoder_size(hrmode != 0, dt_us, sr_hz); + return PyLong_FromUnsignedLong(ret); +} + +static PyObject *py_hr_decoder_size(PyObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", NULL}; + int hrmode = 0, dt_us = 0, sr_hz = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "pii", kwlist, &hrmode, &dt_us, &sr_hz)) + return NULL; + + unsigned ret = lc3_hr_decoder_size(hrmode != 0, dt_us, sr_hz); + return PyLong_FromUnsignedLong(ret); +} + + +/* ------------------------------------------------------------------------- + * EncoderContext Object + * ------------------------------------------------------------------------- */ + +typedef struct { + PyObject_HEAD + bool hrmode; + int dt_us; + int sr_hz; + int sr_pcm_hz; + int nchannels; + int frame_samples; + size_t enc_size; + lc3_encoder_t *handles; + void **mem_blocks; +} EncoderContextObject; + +static void EncoderContext_dealloc(EncoderContextObject *self) +{ + if (self->mem_blocks) { + for (int ich = 0; ich < self->nchannels; ich++) { + if (self->mem_blocks[ich]) { + PyMem_Free(self->mem_blocks[ich]); + } + } + PyMem_Free(self->mem_blocks); + self->mem_blocks = NULL; + } + if (self->handles) { + PyMem_Free(self->handles); + self->handles = NULL; + } + Py_TYPE(self)->tp_free((PyObject *)self); +} + +static int EncoderContext_init(EncoderContextObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", "sr_pcm_hz", "nchannels", NULL}; + int hrmode_int = 0; + int dt_us = 0, sr_hz = 0, sr_pcm_hz = 0, nchannels = 1; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "piiii", kwlist, + &hrmode_int, &dt_us, &sr_hz, &sr_pcm_hz, &nchannels)) + return -1; + + if (nchannels <= 0 || nchannels > 16) { + PyErr_SetString(PyExc_ValueError, "Invalid number of channels"); + return -1; + } + + bool hrmode = (hrmode_int != 0); + unsigned enc_size = lc3_hr_encoder_size(hrmode, dt_us, sr_pcm_hz); + if (enc_size == 0) { + PyErr_SetString(PyExc_ValueError, "Invalid encoder parameters"); + return -1; + } + + int frame_samples = lc3_hr_frame_samples(hrmode, dt_us, sr_pcm_hz); + if (frame_samples <= 0) { + PyErr_SetString(PyExc_ValueError, "Invalid encoder frame samples"); + return -1; + } + + self->hrmode = hrmode; + self->dt_us = dt_us; + self->sr_hz = sr_hz; + self->sr_pcm_hz = sr_pcm_hz; + self->nchannels = nchannels; + self->frame_samples = frame_samples; + self->enc_size = enc_size; + + self->handles = (lc3_encoder_t *)PyMem_Calloc(nchannels, sizeof(lc3_encoder_t)); + self->mem_blocks = (void **)PyMem_Calloc(nchannels, sizeof(void *)); + + if (!self->handles || !self->mem_blocks) { + PyErr_NoMemory(); + return -1; + } + + for (int ich = 0; ich < nchannels; ich++) { + self->mem_blocks[ich] = PyMem_Malloc(enc_size); + if (!self->mem_blocks[ich]) { + PyErr_NoMemory(); + return -1; + } + self->handles[ich] = lc3_hr_setup_encoder( + hrmode, dt_us, sr_hz, sr_pcm_hz, self->mem_blocks[ich] + ); + if (!self->handles[ich]) { + PyErr_SetString(PyExc_RuntimeError, "Failed to initialize LC3 encoder"); + return -1; + } + } + + return 0; +} + +static PyObject *EncoderContext_encode(EncoderContextObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"pcm", "num_bytes", "pcm_format", NULL}; + Py_buffer pcm_buf; + Py_ssize_t num_bytes = 0; + int pcm_fmt = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "y*ni", kwlist, + &pcm_buf, &num_bytes, &pcm_fmt)) + return NULL; + + if (num_bytes <= 0) { + PyBuffer_Release(&pcm_buf); + PyErr_SetString(PyExc_ValueError, "Invalid num_bytes"); + return NULL; + } + + int sample_size = 2; + if (pcm_fmt == LC3_PCM_FORMAT_S16) sample_size = 2; + else if (pcm_fmt == LC3_PCM_FORMAT_S24_3LE) sample_size = 3; + else if (pcm_fmt == LC3_PCM_FORMAT_FLOAT) sample_size = 4; + else if (pcm_fmt == LC3_PCM_FORMAT_S24) sample_size = 4; + else { + PyBuffer_Release(&pcm_buf); + PyErr_SetString(PyExc_ValueError, "Invalid PCM format"); + return NULL; + } + + int nchannels = self->nchannels; + int frame_samples = self->frame_samples; + Py_ssize_t required_pcm_bytes = (Py_ssize_t)nchannels * frame_samples * sample_size; + + const uint8_t *pcm_ptr = (const uint8_t *)pcm_buf.buf; + uint8_t *padded_pcm = NULL; + + if (pcm_buf.len < required_pcm_bytes) { + padded_pcm = (uint8_t *)PyMem_Calloc(1, required_pcm_bytes); + if (!padded_pcm) { + PyBuffer_Release(&pcm_buf); + return PyErr_NoMemory(); + } + memcpy(padded_pcm, pcm_buf.buf, pcm_buf.len); + pcm_ptr = padded_pcm; + } + + PyObject *out_bytes = PyBytes_FromStringAndSize(NULL, num_bytes); + if (!out_bytes) { + if (padded_pcm) PyMem_Free(padded_pcm); + PyBuffer_Release(&pcm_buf); + return NULL; + } + + uint8_t *out_ptr = (uint8_t *)PyBytes_AsString(out_bytes); + int ret = 0; + + Py_BEGIN_ALLOW_THREADS + + int data_offset = 0; + for (int ich = 0; ich < nchannels; ich++) { + int pcm_offset = ich * sample_size; + const void *ch_pcm = (const void *)(pcm_ptr + pcm_offset); + int data_size = (int)(num_bytes / nchannels + (ich < (num_bytes % nchannels))); + void *ch_out = (void *)(out_ptr + data_offset); + data_offset += data_size; + + int ch_ret = lc3_encode( + self->handles[ich], + (enum lc3_pcm_format)pcm_fmt, + ch_pcm, + nchannels, + data_size, + ch_out + ); + if (ch_ret < 0) { + ret = -1; + break; + } + } + + Py_END_ALLOW_THREADS + + if (padded_pcm) PyMem_Free(padded_pcm); + PyBuffer_Release(&pcm_buf); + + if (ret < 0) { + Py_DECREF(out_bytes); + PyErr_SetString(PyExc_ValueError, "Bad parameters in lc3_encode"); + return NULL; + } + + return out_bytes; +} + +static PyObject *EncoderContext_disable_ltpf(EncoderContextObject *self, PyObject *Py_UNUSED(ignored)) +{ + for (int ich = 0; ich < self->nchannels; ich++) { + lc3_encoder_disable_ltpf(self->handles[ich]); + } + Py_RETURN_NONE; +} + +static PyMethodDef EncoderContext_methods[] = { + {"encode", (PyCFunction)EncoderContext_encode, METH_VARARGS | METH_KEYWORDS, "Encode a PCM frame"}, + {"disable_ltpf", (PyCFunction)EncoderContext_disable_ltpf, METH_NOARGS, "Disable LTPF analysis"}, + {NULL} +}; + +static PyTypeObject EncoderContextType = { + PyVarObject_HEAD_INIT(NULL, 0) + .tp_name = "_lc3.EncoderContext", + .tp_doc = "LC3 Native Encoder Context", + .tp_basicsize = sizeof(EncoderContextObject), + .tp_itemsize = 0, + .tp_flags = Py_TPFLAGS_DEFAULT, + .tp_new = PyType_GenericNew, + .tp_init = (initproc)EncoderContext_init, + .tp_dealloc = (destructor)EncoderContext_dealloc, + .tp_methods = EncoderContext_methods, +}; + + +/* ------------------------------------------------------------------------- + * DecoderContext Object + * ------------------------------------------------------------------------- */ + +typedef struct { + PyObject_HEAD + bool hrmode; + int dt_us; + int sr_hz; + int sr_pcm_hz; + int nchannels; + int frame_samples; + size_t dec_size; + lc3_decoder_t *handles; + void **mem_blocks; +} DecoderContextObject; + +static void DecoderContext_dealloc(DecoderContextObject *self) +{ + if (self->mem_blocks) { + for (int ich = 0; ich < self->nchannels; ich++) { + if (self->mem_blocks[ich]) { + PyMem_Free(self->mem_blocks[ich]); + } + } + PyMem_Free(self->mem_blocks); + self->mem_blocks = NULL; + } + if (self->handles) { + PyMem_Free(self->handles); + self->handles = NULL; + } + Py_TYPE(self)->tp_free((PyObject *)self); +} + +static int DecoderContext_init(DecoderContextObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"hrmode", "dt_us", "sr_hz", "sr_pcm_hz", "nchannels", NULL}; + int hrmode_int = 0; + int dt_us = 0, sr_hz = 0, sr_pcm_hz = 0, nchannels = 1; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "piiii", kwlist, + &hrmode_int, &dt_us, &sr_hz, &sr_pcm_hz, &nchannels)) + return -1; + + if (nchannels <= 0 || nchannels > 16) { + PyErr_SetString(PyExc_ValueError, "Invalid number of channels"); + return -1; + } + + bool hrmode = (hrmode_int != 0); + unsigned dec_size = lc3_hr_decoder_size(hrmode, dt_us, sr_pcm_hz); + if (dec_size == 0) { + PyErr_SetString(PyExc_ValueError, "Invalid decoder parameters"); + return -1; + } + + int frame_samples = lc3_hr_frame_samples(hrmode, dt_us, sr_pcm_hz); + if (frame_samples <= 0) { + PyErr_SetString(PyExc_ValueError, "Invalid decoder frame samples"); + return -1; + } + + self->hrmode = hrmode; + self->dt_us = dt_us; + self->sr_hz = sr_hz; + self->sr_pcm_hz = sr_pcm_hz; + self->nchannels = nchannels; + self->frame_samples = frame_samples; + self->dec_size = dec_size; + + self->handles = (lc3_decoder_t *)PyMem_Calloc(nchannels, sizeof(lc3_decoder_t)); + self->mem_blocks = (void **)PyMem_Calloc(nchannels, sizeof(void *)); + + if (!self->handles || !self->mem_blocks) { + PyErr_NoMemory(); + return -1; + } + + for (int ich = 0; ich < nchannels; ich++) { + self->mem_blocks[ich] = PyMem_Malloc(dec_size); + if (!self->mem_blocks[ich]) { + PyErr_NoMemory(); + return -1; + } + self->handles[ich] = lc3_hr_setup_decoder( + hrmode, dt_us, sr_hz, sr_pcm_hz, self->mem_blocks[ich] + ); + if (!self->handles[ich]) { + PyErr_SetString(PyExc_RuntimeError, "Failed to initialize LC3 decoder"); + return -1; + } + } + + return 0; +} + +static PyObject *DecoderContext_decode(DecoderContextObject *self, PyObject *args, PyObject *kwds) +{ + static char *kwlist[] = {"data", "pcm_format", NULL}; + PyObject *data_obj = NULL; + int pcm_fmt = 0; + + if (!PyArg_ParseTupleAndKeywords(args, kwds, "Oi", kwlist, &data_obj, &pcm_fmt)) + return NULL; + + Py_buffer in_buf; + bool has_in_buf = false; + const uint8_t *in_ptr = NULL; + Py_ssize_t in_len = 0; + + if (data_obj != Py_None) { + if (PyObject_GetBuffer(data_obj, &in_buf, PyBUF_SIMPLE) != 0) { + return NULL; + } + has_in_buf = true; + in_ptr = (const uint8_t *)in_buf.buf; + in_len = in_buf.len; + } + + int sample_size = 2; + if (pcm_fmt == LC3_PCM_FORMAT_S16) sample_size = 2; + else if (pcm_fmt == LC3_PCM_FORMAT_S24_3LE) sample_size = 3; + else if (pcm_fmt == LC3_PCM_FORMAT_FLOAT) sample_size = 4; + else if (pcm_fmt == LC3_PCM_FORMAT_S24) sample_size = 4; + else { + if (has_in_buf) PyBuffer_Release(&in_buf); + PyErr_SetString(PyExc_ValueError, "Invalid PCM format"); + return NULL; + } + + int nchannels = self->nchannels; + int frame_samples = self->frame_samples; + Py_ssize_t out_len = (Py_ssize_t)nchannels * frame_samples * sample_size; + + PyObject *out_bytes = PyBytes_FromStringAndSize(NULL, out_len); + if (!out_bytes) { + if (has_in_buf) PyBuffer_Release(&in_buf); + return NULL; + } + + uint8_t *pcm_out = (uint8_t *)PyBytes_AsString(out_bytes); + memset(pcm_out, 0, out_len); + + int ret = 0; + + Py_BEGIN_ALLOW_THREADS + + int data_offset = 0; + for (int ich = 0; ich < nchannels; ich++) { + int pcm_offset = ich * sample_size; + void *ch_pcm = (void *)(pcm_out + pcm_offset); + + if (!has_in_buf) { + int ch_ret = lc3_decode( + self->handles[ich], + NULL, + 0, + (enum lc3_pcm_format)pcm_fmt, + ch_pcm, + nchannels + ); + if (ch_ret < 0) { + ret = -1; + break; + } + } else { + int data_size = (int)(in_len / nchannels + (ich < (in_len % nchannels))); + const void *ch_in = (const void *)(in_ptr + data_offset); + data_offset += data_size; + + int ch_ret = lc3_decode( + self->handles[ich], + ch_in, + data_size, + (enum lc3_pcm_format)pcm_fmt, + ch_pcm, + nchannels + ); + if (ch_ret < 0) { + ret = -1; + break; + } + } + } + + Py_END_ALLOW_THREADS + + if (has_in_buf) PyBuffer_Release(&in_buf); + + if (ret < 0) { + Py_DECREF(out_bytes); + PyErr_SetString(PyExc_ValueError, "Bad parameters in lc3_decode"); + return NULL; + } + + return out_bytes; +} + +static PyMethodDef DecoderContext_methods[] = { + {"decode", (PyCFunction)DecoderContext_decode, METH_VARARGS | METH_KEYWORDS, "Decode an LC3 frame"}, + {NULL} +}; + +static PyTypeObject DecoderContextType = { + PyVarObject_HEAD_INIT(NULL, 0) + .tp_name = "_lc3.DecoderContext", + .tp_doc = "LC3 Native Decoder Context", + .tp_basicsize = sizeof(DecoderContextObject), + .tp_itemsize = 0, + .tp_flags = Py_TPFLAGS_DEFAULT, + .tp_new = PyType_GenericNew, + .tp_init = (initproc)DecoderContext_init, + .tp_dealloc = (destructor)DecoderContext_dealloc, + .tp_methods = DecoderContext_methods, +}; + + +/* ------------------------------------------------------------------------- + * Module Definition + * ------------------------------------------------------------------------- */ + +static PyMethodDef module_methods[] = { + {"hr_frame_samples", (PyCFunction)py_hr_frame_samples, METH_VARARGS | METH_KEYWORDS, "Return PCM samples in frame"}, + {"hr_frame_block_bytes", (PyCFunction)py_hr_frame_block_bytes, METH_VARARGS | METH_KEYWORDS, "Return frame block size"}, + {"hr_resolve_bitrate", (PyCFunction)py_hr_resolve_bitrate, METH_VARARGS | METH_KEYWORDS, "Resolve bitrate from size"}, + {"hr_delay_samples", (PyCFunction)py_hr_delay_samples, METH_VARARGS | METH_KEYWORDS, "Return algorithmic delay samples"}, + {"hr_encoder_size", (PyCFunction)py_hr_encoder_size, METH_VARARGS | METH_KEYWORDS, "Return encoder size"}, + {"hr_decoder_size", (PyCFunction)py_hr_decoder_size, METH_VARARGS | METH_KEYWORDS, "Return decoder size"}, + {NULL} +}; + +static struct PyModuleDef module_def = { + PyModuleDef_HEAD_INIT, + .m_name = "_lc3", + .m_doc = "Native C extension for liblc3", + .m_size = -1, + .m_methods = module_methods, +}; + +PyMODINIT_FUNC PyInit__lc3(void) +{ + if (PyType_Ready(&EncoderContextType) < 0) + return NULL; + if (PyType_Ready(&DecoderContextType) < 0) + return NULL; + + PyObject *m = PyModule_Create(&module_def); + if (!m) + return NULL; + + Py_INCREF(&EncoderContextType); + if (PyModule_AddObject(m, "EncoderContext", (PyObject *)&EncoderContextType) < 0) { + Py_DECREF(&EncoderContextType); + Py_DECREF(m); + return NULL; + } + + Py_INCREF(&DecoderContextType); + if (PyModule_AddObject(m, "DecoderContext", (PyObject *)&DecoderContextType) < 0) { + Py_DECREF(&DecoderContextType); + Py_DECREF(m); + return NULL; + } + + return m; +} diff --git a/python/meson.build b/python/meson.build index e63c768..d091607 100644 --- a/python/meson.build +++ b/python/meson.build @@ -1,3 +1,9 @@ py = import('python').find_installation() -py.install_sources('lc3.py') +py.extension_module('_lc3', + 'lc3_native.c', + dependencies: [liblc3_dep], + install: true +) + +py.install_sources('lc3.py', 'py.typed', '_lc3.pyi') diff --git a/python/py.typed b/python/py.typed new file mode 100644 index 0000000..7632ecf --- /dev/null +++ b/python/py.typed @@ -0,0 +1 @@ +# Marker file for PEP 561