103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import struct
|
|
from array import array
|
|
|
|
try:
|
|
import audioop as _audioop
|
|
except ModuleNotFoundError:
|
|
try:
|
|
import audioop_lts as _audioop # type: ignore[import-not-found]
|
|
except ModuleNotFoundError:
|
|
_audioop = None
|
|
|
|
|
|
def _native_is_little_endian() -> bool:
|
|
return struct.pack("=h", 1) == struct.pack("<h", 1)
|
|
|
|
|
|
def _read_int16_samples(fragment: bytes, width: int) -> array:
|
|
if width != 2:
|
|
raise NotImplementedError("audioop_compat fallback currently supports only 16-bit PCM")
|
|
samples = array("h")
|
|
samples.frombytes(fragment)
|
|
if not _native_is_little_endian():
|
|
samples.byteswap()
|
|
return samples
|
|
|
|
|
|
def _write_int16_samples(samples: array) -> bytes:
|
|
output = array("h", samples)
|
|
if not _native_is_little_endian():
|
|
output.byteswap()
|
|
return output.tobytes()
|
|
|
|
|
|
def _clip_int16(value: float) -> int:
|
|
return max(-32768, min(32767, int(round(value))))
|
|
|
|
|
|
class _AudioopFallback:
|
|
@staticmethod
|
|
def rms(fragment: bytes, width: int) -> int:
|
|
samples = _read_int16_samples(fragment, width)
|
|
if not samples:
|
|
return 0
|
|
mean_square = sum(sample * sample for sample in samples) / len(samples)
|
|
return int(math.sqrt(mean_square))
|
|
|
|
@staticmethod
|
|
def tomono(fragment: bytes, width: int, lfactor: float, rfactor: float) -> bytes:
|
|
samples = _read_int16_samples(fragment, width)
|
|
if len(samples) % 2 != 0:
|
|
raise ValueError("Stereo PCM must contain an even number of samples")
|
|
mono = array("h")
|
|
for index in range(0, len(samples), 2):
|
|
left = samples[index]
|
|
right = samples[index + 1]
|
|
mono.append(_clip_int16((left * lfactor) + (right * rfactor)))
|
|
return _write_int16_samples(mono)
|
|
|
|
@staticmethod
|
|
def ratecv(
|
|
fragment: bytes,
|
|
width: int,
|
|
nchannels: int,
|
|
inrate: int,
|
|
outrate: int,
|
|
state,
|
|
weightA: int = 1,
|
|
weightB: int = 0,
|
|
) -> tuple[bytes, None]:
|
|
if nchannels <= 0:
|
|
raise ValueError("nchannels must be positive")
|
|
if inrate <= 0 or outrate <= 0:
|
|
raise ValueError("Sample rates must be positive")
|
|
samples = _read_int16_samples(fragment, width)
|
|
if not samples or inrate == outrate:
|
|
return fragment, None
|
|
if len(samples) % nchannels != 0:
|
|
raise ValueError("PCM fragment size does not match channel count")
|
|
|
|
frame_count = len(samples) // nchannels
|
|
output_frame_count = max(1, int(round(frame_count * outrate / inrate)))
|
|
output = array("h")
|
|
|
|
for out_index in range(output_frame_count):
|
|
position = out_index * inrate / outrate
|
|
left_index = min(int(position), frame_count - 1)
|
|
right_index = min(left_index + 1, frame_count - 1)
|
|
fraction = max(0.0, min(1.0, position - left_index))
|
|
for channel_index in range(nchannels):
|
|
left_sample = samples[(left_index * nchannels) + channel_index]
|
|
right_sample = samples[(right_index * nchannels) + channel_index]
|
|
interpolated = left_sample + ((right_sample - left_sample) * fraction)
|
|
output.append(_clip_int16(interpolated))
|
|
return _write_int16_samples(output), None
|
|
|
|
|
|
audioop = _audioop or _AudioopFallback()
|
|
|
|
__all__ = ["audioop"]
|