Fix G.711 codec algorithms, resample edge cases, and conditional __all__

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"]
 

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
 
 

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)

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."""

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
 

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)),

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()

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