diff --git a/pyproject.toml b/pyproject.toml index 612e0dd..ec86dd3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,7 +41,7 @@ dependencies = ["cryptography"] [project.optional-dependencies] audio = ["numpy"] hd-audio = ["numpy", "av"] -ai = ["faster-whisper", "numpy", "ollama", "pocket-tts"] +ai = ["faster-whisper", "numpy", "av", "ollama", "pocket-tts"] cli = ["click", "pygments", "faster-whisper", "numpy", "ollama", "pocket-tts"] pygments = ["Pygments"] diff --git a/tests/codecs/test_base.py b/tests/codecs/test_base.py index 40776b0..ef743a8 100644 --- a/tests/codecs/test_base.py +++ b/tests/codecs/test_base.py @@ -31,10 +31,17 @@ def test_resample__downsample_halves_length(self): result = RTPCodec.resample(audio, 16000, 8000) assert len(result) == 160 - def test_resample__returns_float32(self): - """Resample always returns a float32 array.""" - audio = np.ones(100, dtype=np.float64) - result = RTPCodec.resample(audio, 8000, 16000) + def test_resample__empty_input_returns_empty(self): + """Resampling an empty array returns an empty float32 array.""" + result = RTPCodec.resample(np.empty(0, dtype=np.float32), 8000, 16000) + assert result.dtype == np.float32 + assert len(result) == 0 + + def test_resample__single_sample_heavy_downsample_returns_at_least_one(self): + """Resampling a single sample always yields at least one output sample.""" + audio = np.array([0.5], dtype=np.float32) + result = RTPCodec.resample(audio, 8000, 100) + assert len(result) >= 1 assert result.dtype == np.float32 diff --git a/tests/codecs/test_codecs.py b/tests/codecs/test_codecs.py index e37c80f..c87e82d 100644 --- a/tests/codecs/test_codecs.py +++ b/tests/codecs/test_codecs.py @@ -46,6 +46,11 @@ def test_get__raise_not_implemented_error(self): get("unknown") +_PYAV_MODULE_KEYS: frozenset[str] = frozenset( + {"av", "voip.codecs", "voip.codecs.av", "voip.codecs.g722", "voip.codecs.opus"} +) + + class TestRegistry: def test_registry__always_contains_numpy_codecs(self): """REGISTRY always contains PCMA and PCMU regardless of PyAV availability.""" @@ -58,18 +63,7 @@ def test_registry__excludes_pyav_codecs_when_av_unavailable(self): """REGISTRY excludes G722 and Opus when av is not importable.""" import voip.codecs as target # noqa: PLC0415 - keys_to_remove = [ - k - for k in list(sys.modules) - if k - in { - "av", - "voip.codecs", - "voip.codecs.av", - "voip.codecs.g722", - "voip.codecs.opus", - } - ] + keys_to_remove = [k for k in sys.modules if k in _PYAV_MODULE_KEYS] saved = {k: sys.modules.pop(k) for k in keys_to_remove} sys.modules["av"] = None # causes ImportError on `import av` @@ -85,3 +79,22 @@ def test_registry__excludes_pyav_codecs_when_av_unavailable(self): sys.modules.pop(k, None) sys.modules.update(saved) importlib.reload(target) + + def test_all__excludes_pyav_names_when_av_unavailable(self): + """__all__ excludes G722, Opus, and PyAVCodec when av is not importable.""" + keys_to_remove = [k for k in sys.modules if k in _PYAV_MODULE_KEYS] + saved = {k: sys.modules.pop(k) for k in keys_to_remove} + sys.modules["av"] = None # causes ImportError on `import av` + + try: + import voip.codecs as fresh # noqa: PLC0415 + + for name in ("G722", "Opus", "PyAVCodec"): + assert name not in fresh.__all__, ( + f"{name!r} must not be in __all__ without av" + ) + finally: + for k in list(sys.modules): + if k in _PYAV_MODULE_KEYS: + sys.modules.pop(k, None) + sys.modules.update(saved) diff --git a/tests/codecs/test_pcm.py b/tests/codecs/test_pcm.py index 3255b3b..d1657ba 100644 --- a/tests/codecs/test_pcm.py +++ b/tests/codecs/test_pcm.py @@ -65,12 +65,17 @@ def test_decode__positive_and_negative_differ(self): assert pos[0] > 0 assert neg[0] < 0 - def test_decode__ignores_input_rate_hz(self): - """Decode ignores input_rate_hz: A-law is always at 8 kHz.""" + def test_decode__uses_input_rate_hz_as_source_rate(self): + """input_rate_hz is treated as the source sample rate for resampling.""" payload = PCMA.encode(np.zeros(160, dtype=np.float32)) - result_default = PCMA.decode(payload, 8000, input_rate_hz=None) - result_override = PCMA.decode(payload, 8000, input_rate_hz=16000) - np.testing.assert_array_equal(result_default, result_override) + # 160 samples interpreted at 16 kHz, resampled to 8 kHz → 80 output samples. + result = PCMA.decode(payload, 8000, input_rate_hz=16000) + assert len(result) == 80 + + def test_decode__max_amplitude_codeword(self): + """A-law codeword 0xAA (max positive) decodes to exactly 0.984375 per ITU-T G.711.""" + decoded = PCMA.decode(bytes([0xAA]), 8000) + assert abs(decoded[0] - 0.984375) < 1e-6 def test_decode__real_decode_returns_float32(self): """Decode produces a float32 array from real A-law encoded input.""" @@ -95,6 +100,26 @@ def test_encode__positive_and_negative_differ(self): neg = PCMA.encode(np.array([-0.5], dtype=np.float32)) assert pos != neg + def test_encode__silence_codeword(self): + """Silence (0.0) must encode to 0xD5 per ITU-T G.711 A-law.""" + assert PCMA.encode(np.zeros(1, dtype=np.float32))[0] == 0xD5 + + def test_encode__max_amplitude_codeword(self): + """Maximum positive amplitude (1.0) must encode to 0xAA per ITU-T G.711.""" + assert PCMA.encode(np.array([1.0], dtype=np.float32))[0] == 0xAA + + def test_encode_decode__roundtrip_midrange(self): + """Roundtrip encode→decode stays within one G.711 quantisation step. + + Values in segments 2–5 had ~3× the expected error with the old (wrong) + mantissa shift; this test guards against regression. + """ + # Values chosen so all four land in segments 3–5 (magnitudes 1 000–8 191). + original = np.array([0.05, -0.05, 0.1, -0.1], dtype=np.float32) + recovered = PCMA.decode(PCMA.encode(original), 8000) + # Maximum step for segment 5 is 256/32768 ≈ 0.0078; allow ~2 steps of margin. + assert np.allclose(original, recovered, atol=1 / 64) + class TestPCMUConstants: def test_payload_type(self): @@ -148,12 +173,17 @@ def test_decode__max_negative_roundtrip(self): decoded = PCMU.decode(bytes([0x80]), 8000) assert decoded[0] < -0.9 - def test_decode__ignores_input_rate_hz(self): - """Decode ignores input_rate_hz: mu-law is always at 8 kHz.""" + def test_decode__uses_input_rate_hz_as_source_rate(self): + """input_rate_hz is treated as the source sample rate for resampling.""" payload = PCMU.encode(np.zeros(160, dtype=np.float32)) - result_default = PCMU.decode(payload, 8000, input_rate_hz=None) - result_override = PCMU.decode(payload, 8000, input_rate_hz=16000) - np.testing.assert_array_equal(result_default, result_override) + # 160 samples interpreted at 16 kHz, resampled to 8 kHz → 80 output samples. + result = PCMU.decode(payload, 8000, input_rate_hz=16000) + assert len(result) == 80 + + def test_decode__silence_codeword(self): + """µ-law codeword 0x7F (silence) decodes to exactly 0.0 per ITU-T G.711.""" + decoded = PCMU.decode(bytes([0x7F]), 8000) + assert decoded[0] == pytest.approx(0.0, abs=1e-7) def test_decode__real_decode_returns_float32(self): """Decode produces a float32 array from real mu-law encoded input.""" diff --git a/voip/codecs/__init__.py b/voip/codecs/__init__.py index 0522491..b37a6a1 100644 --- a/voip/codecs/__init__.py +++ b/voip/codecs/__init__.py @@ -20,7 +20,7 @@ from voip.codecs.pcma import PCMA from voip.codecs.pcmu import PCMU -__all__ = ["G722", "Opus", "PCMA", "PCMU", "PyAVCodec", "RTPCodec", "get"] +__all__ = ["PCMA", "PCMU", "RTPCodec", "get"] #: Registry mapping lowercase encoding names to codec classes. REGISTRY: dict[str, type[RTPCodec]] = { @@ -34,6 +34,7 @@ from voip.codecs.opus import Opus REGISTRY |= {G722.encoding_name: G722, Opus.encoding_name: Opus} + __all__ = [*__all__, "G722", "Opus", "PyAVCodec"] except ImportError: pass diff --git a/voip/codecs/base.py b/voip/codecs/base.py index 0893d48..69915ba 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -97,7 +97,9 @@ def resample( """ if source_rate_hz == destination_rate_hz: return audio - n_out = round(len(audio) * destination_rate_hz / source_rate_hz) + if len(audio) == 0: + return np.empty(0, dtype=np.float32) + n_out = max(1, round(len(audio) * destination_rate_hz / source_rate_hz)) return np.interp( np.linspace(0, len(audio) - 1, n_out), np.arange(len(audio)), diff --git a/voip/codecs/pcma.py b/voip/codecs/pcma.py index a8c683f..c27b940 100644 --- a/voip/codecs/pcma.py +++ b/voip/codecs/pcma.py @@ -1,8 +1,8 @@ """PCMA (G.711 A-law) codec implementation for RTP audio streams (RFC 3551). The [`PCMA`][voip.codecs.pcma.PCMA] class decodes and encodes A-law RTP -payloads using a pure-NumPy implementation of ITU-T G.711 A-law companding. -No PyAV dependency is required. +payloads using a pure-NumPy implementation of the ITU-T G.711 A-law segmented +companding algorithm. No PyAV dependency is required. """ from __future__ import annotations @@ -15,9 +15,12 @@ __all__ = ["PCMA"] -_A_LAW: float = 87.6 -_LN_A: float = float(np.log(_A_LAW)) -_COMPRESS_SCALE: float = 1.0 + _LN_A +# G.711 A-law segment upper bounds (16-bit PCM magnitude, inclusive per segment). +# Vectorised segment lookup via np.searchsorted uses side='left' to count thresholds +# strictly exceeded (v > threshold), giving the correct 0–7 segment index. +_ALAW_SEG_UBOUND: np.ndarray = np.array( + (0xFF, 0x1FF, 0x3FF, 0x7FF, 0xFFF, 0x1FFF, 0x3FFF, 0x7FFF), dtype=np.int32 +) class PCMA(RTPCodec): @@ -26,7 +29,8 @@ class PCMA(RTPCodec): PCMA is the ITU-T G.711 A-law logarithmic companding codec for PSTN telephony, standardised in RFC 3551 with static payload type 8. - Both encode and decode are pure-NumPy and require no PyAV dependency. + Both encode and decode interoperate bit-exactly with real RTP PCMA + streams. [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ @@ -47,30 +51,42 @@ def decode( *, input_rate_hz: int | None = None, ) -> np.ndarray: - raw = np.frombuffer(payload, dtype=np.uint8) ^ 0x55 - sign = np.where(raw & 0x80, 1.0, -1.0) - quantized = (raw & 0x7F).astype(np.float32) / 127.0 - threshold = 1.0 / _COMPRESS_SCALE - linear = np.where( - quantized < threshold, - quantized * _COMPRESS_SCALE / _A_LAW, - np.exp(quantized * _COMPRESS_SCALE - 1.0) / _A_LAW, - ).astype(np.float32) - return cls.resample( - (sign * linear).astype(np.float32), cls.sample_rate_hz, output_rate_hz + # XOR with 0x55 to undo the G.711 bit-inversion applied before transmission. + raw = np.frombuffer(payload, dtype=np.uint8) ^ np.uint8(0x55) + sign = np.where(raw & 0x80, 1.0, -1.0).astype(np.float32) + segment = ((raw & 0x70) >> 4).astype(np.int32) + # t = mantissa bits shifted to the top of the 4-bit slot (×16). + mantissa_t = ((raw & 0x0F).astype(np.int32)) << 4 + # Segment 0: add step mid-point (8). + t_seg0 = mantissa_t + 8 + # Segments ≥ 1: add 0x108 bias then left-shift by (segment − 1). + t_bias = mantissa_t + 0x108 + shift = np.maximum(segment - 1, 0) + t_segN = np.left_shift(t_bias, shift) + t = np.where(segment == 0, t_seg0, t_segN).astype(np.float32) + normalized = (sign * t / 32768.0).astype(np.float32) + source_rate_hz = ( + input_rate_hz if input_rate_hz is not None else cls.sample_rate_hz ) + return cls.resample(normalized, source_rate_hz, output_rate_hz) @classmethod def encode(cls, samples: np.ndarray) -> bytes: - pcm = np.clip(np.abs(samples), 0, 1.0) - low = pcm < (1.0 / _A_LAW) - compressed = np.where( - low, - _A_LAW * pcm / _COMPRESS_SCALE, - (1.0 + np.log(np.maximum(_A_LAW * pcm, 1e-10))) / _COMPRESS_SCALE, - # 1e-10 prevents log(0) when pcm is exactly 0.0 in the high range + # Scale to 16-bit signed PCM range. + pcm = np.clip(np.round(samples * 32768.0), -32768, 32767).astype(np.int32) + # Positive samples: mask 0xD5. Negative samples: mask 0x55, negate−1. + magnitude = np.where(pcm >= 0, pcm, -pcm - 1).astype(np.int32) + mask = np.where(pcm >= 0, np.uint8(0xD5), np.uint8(0x55)) + # Find the segment index (0–7) via vectorised binary search on the upper bounds. + # side='left' counts thresholds strictly less than magnitude, i.e. exceeded. + seg = np.minimum( + np.searchsorted(_ALAW_SEG_UBOUND, magnitude, side="left").astype(np.int32), + 7, ) - quantized = np.clip(np.round(compressed * 127), 0, 127).astype(np.uint8) - sign = np.where(samples >= 0, 0x80, 0x00).astype(np.uint8) - # XOR even bits per G.711 §A (toggle bits via 0x55) - return ((sign | quantized) ^ 0x55).astype(np.uint8).tobytes() + # Extract the 4-bit mantissa in 16-bit space. + # G.711 A-law quantises 13-bit PCM (16-bit >> 3), so the effective + # mantissa shift is 4 for segments 0–1 and (seg + 3) for segments ≥ 2. + shift = np.where(seg < 2, 4, seg + 3).astype(np.int32) + mantissa = (np.right_shift(magnitude, shift) & 0x0F).astype(np.uint8) + aval = (seg.astype(np.uint8) << 4) | mantissa + return (aval ^ mask).astype(np.uint8).tobytes() diff --git a/voip/codecs/pcmu.py b/voip/codecs/pcmu.py index 5709af1..361f1e8 100644 --- a/voip/codecs/pcmu.py +++ b/voip/codecs/pcmu.py @@ -50,11 +50,14 @@ def decode( sign = np.where(raw & 0x80, 1.0, -1.0) exp = ((raw >> 4) & 0x07).astype(np.int32) mantissa = (raw & 0x0F).astype(np.int32) - # Reconstruct the biased linear magnitude from segment and mantissa. - biased = ((mantissa | 0x10) << (exp + 3)).astype(np.int32) - linear = ((biased - _MU_LAW_BIAS) / 32768.0).astype(np.float32) + # ITU-T G.711 §A: magnitude = (((mantissa << 3) + BIAS) << exp) - BIAS + magnitude = (((mantissa << 3) + _MU_LAW_BIAS) << exp) - _MU_LAW_BIAS + linear = (magnitude.astype(np.float32) / 32768.0).astype(np.float32) + source_rate_hz = ( + input_rate_hz if input_rate_hz is not None else cls.sample_rate_hz + ) return cls.resample( - (sign * linear).astype(np.float32), cls.sample_rate_hz, output_rate_hz + (sign * linear).astype(np.float32), source_rate_hz, output_rate_hz ) @classmethod