Deduplicate codec base classes

.github/workflows/ci.yml

@@ -51,6 +51,7 @@ jobs:
           - "3.14"
         extra:
           - audio
+          - hd-audio
           - cli
           - pygments
     runs-on: ${{ matrix.os }}

docs/codecs.md

@@ -0,0 +1,52 @@
+# Codecs
+
+## Overview
+
+VoIP ships two tiers of audio codecs:
+
+| Extra required                           | Codecs available                       |
+| ---------------------------------------- | -------------------------------------- |
+| `audio` (includes [numpy])               | PCMA (G.711 A-law), PCMU (G.711 µ-law) |
+| `hd-audio` (includes [numpy] and [pyav]) | + G.722, Opus                          |
+
+Install the minimal tier for pure-Python telephony deployments:
+
+```bash
+pip install voip[audio]
+```
+
+Install the full tier for wideband / Opus support via [FFmpeg]:
+
+```bash
+pip install voip[hd-audio]
+```
+
+## SD audio
+
+These codecs work without PyAV and require only `numpy`.
+
+::: voip.codecs.pcma.PCMA
+
+::: voip.codecs.pcmu.PCMU
+
+## HD audio
+
+These codecs require the `pyav` extra (`pip install voip[pyav]`).
+
+::: voip.codecs.g722.G722
+
+::: voip.codecs.opus.Opus
+
+## Registry
+
+::: voip.codecs.get
+
+## Base classes
+
+::: voip.codecs.base.RTPCodec
+
+::: voip.codecs.av.PyAVCodec
+
+[ffmpeg]: https://ffmpeg.org/
+[numpy]: https://numpy.org/
+[pyav]: https://pyav.org/

docs/rtp.md

@@ -5,3 +5,7 @@
 ## Encryption
 
 ::: voip.srtp
+
+## NAT Traversal
+
+::: voip.stun

mkdocs.yml

@@ -12,6 +12,7 @@ nav:
       - RFC Implementation Status: rfc_status.md
   - API Reference:
       - Calls: calls.md
+      - Codecs: codecs.md
       - RTP: rtp.md
       - SDP: sdp.md
       - SIP: sip.md

pyproject.toml

@@ -39,9 +39,10 @@ requires-python = ">=3.13"
 dependencies = ["cryptography"]
 
 [project.optional-dependencies]
-audio = ["numpy", "av"]
+audio = ["numpy"]
+hd-audio = ["numpy", "av"]
 ai = ["faster-whisper", "numpy", "av", "ollama", "pocket-tts"]
-cli = ["click", "pygments", "faster-whisper", "numpy", "av", "ollama", "pocket-tts"]
+cli = ["click", "pygments", "faster-whisper", "numpy", "ollama", "pocket-tts"]
 pygments = ["Pygments"]
 
 [project.scripts]

tests/codecs/test_av.py

@@ -0,0 +1,83 @@
+"""Tests for the PyAVCodec base class (voip.codecs.av)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+np = pytest.importorskip("numpy")
+av = pytest.importorskip("av")
+
+from voip.codecs.av import PyAVCodec  # noqa: E402
+from voip.codecs.pcma import PCMA  # noqa: E402
+
+
+class TestDecodePCM:
+    def test_decode_pcm__alaw_returns_float32(self):
+        """decode_pcm decodes A-law bytes to a float32 numpy array."""
+        payload = PCMA.encode(np.zeros(160, dtype=np.float32))
+        result = PyAVCodec.decode_pcm(payload, "alaw", 8000, input_rate_hz=8000)
+        assert result.dtype == np.float32
+
+    def test_decode_pcm__resampler_flush_yields_frames(self):
+        """Include frames flushed from the resampler after the last input frame."""
+        pcm_array = np.zeros(16000, dtype=np.float32)
+        flush_frame = MagicMock()
+        flush_frame.to_ndarray.return_value = pcm_array
+        with patch("voip.codecs.av.av") as mock_av:
+            mock_resampler = MagicMock()
+            mock_resampler.resample.side_effect = [[], [flush_frame]]
+            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
+            mock_container = MagicMock()
+            mock_container.__enter__ = lambda s: s
+            mock_container.__exit__ = MagicMock(return_value=False)
+            mock_container.decode.return_value = [MagicMock()]
+            mock_av.open.return_value = mock_container
+            result = PyAVCodec.decode_pcm(b"fake", "alaw", 8000, input_rate_hz=8000)
+        assert result.dtype == np.float32
+        assert len(result) == len(pcm_array)
+
+    def test_decode_pcm__empty_result_when_no_frames(self):
+        """decode_pcm returns an empty float32 array when no audio frames are decoded."""
+        with patch("voip.codecs.av.av") as mock_av:
+            mock_resampler = MagicMock()
+            mock_resampler.resample.return_value = []
+            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
+            mock_container = MagicMock()
+            mock_container.__enter__ = lambda s: s
+            mock_container.__exit__ = MagicMock(return_value=False)
+            mock_container.decode.return_value = []
+            mock_av.open.return_value = mock_container
+            result = PyAVCodec.decode_pcm(b"fake", "alaw", 8000)
+        assert result.dtype == np.float32
+        assert len(result) == 0
+
+    def test_decode_pcm__without_input_rate_passes_no_options(self):
+        """decode_pcm passes no sample_rate option when input_rate_hz is None."""
+        with patch("voip.codecs.av.av") as mock_av:
+            mock_resampler = MagicMock()
+            mock_resampler.resample.return_value = []
+            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
+            mock_container = MagicMock()
+            mock_container.__enter__ = lambda s: s
+            mock_container.__exit__ = MagicMock(return_value=False)
+            mock_container.decode.return_value = []
+            mock_av.open.return_value = mock_container
+            PyAVCodec.decode_pcm(b"fake", "alaw", 8000, input_rate_hz=None)
+        call_kwargs = mock_av.open.call_args[1]
+        assert call_kwargs["options"] == {}
+
+
+class TestEncodePCM:
+    def test_encode_pcm__g722_returns_bytes(self):
+        """encode_pcm produces non-empty bytes for G.722."""
+        result = PyAVCodec.encode_pcm(np.zeros(320, dtype=np.float32), "g722", 16000)
+        assert isinstance(result, bytes)
+        assert len(result) > 0
+
+    def test_encode_pcm__opus_returns_bytes(self):
+        """encode_pcm produces non-empty bytes for Opus (libopus)."""
+        result = PyAVCodec.encode_pcm(np.zeros(960, dtype=np.float32), "libopus", 48000)
+        assert isinstance(result, bytes)
+        assert len(result) > 0

tests/codecs/test_base.py

@@ -1,87 +1,48 @@
-"""Tests for the PyAVCodec base class (voip.codecs.av)."""
+"""Tests for the RTPCodec base class (voip.codecs.base)."""
 
 from __future__ import annotations
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
 
 np = pytest.importorskip("numpy")
-av = pytest.importorskip("av")
 
 from voip.codecs.base import RTPCodec  # noqa: E402
 from voip.codecs.pcma import PCMA  # noqa: E402
 from voip.sdp.types import RTPPayloadFormat  # noqa: E402
 
 
-class TestDecodePCM:
-    def test_decode_pcm__alaw_returns_float32(self):
-        """decode_pcm decodes A-law bytes to a float32 numpy array."""
-        payload = PCMA.encode(np.zeros(160, dtype=np.float32))
-        result = PCMA.decode_pcm(payload, "alaw", 8000, input_rate_hz=8000)
-        assert result.dtype == np.float32
+class TestResample:
+    def test_resample__passthrough_when_rates_equal(self):
+        """Return the original array unchanged when source and destination rates are equal."""
+        audio = np.zeros(160, dtype=np.float32)
+        assert RTPCodec.resample(audio, 8000, 8000) is audio
 
-    def test_decode_pcm__resampler_flush_yields_frames(self):
-        """Include frames flushed from the resampler after the last input frame."""
-        pcm_array = np.zeros(16000, dtype=np.float32)
-        flush_frame = MagicMock()
-        flush_frame.to_ndarray.return_value = pcm_array
-        with patch("voip.codecs.base.av") as mock_av:
-            mock_resampler = MagicMock()
-            mock_resampler.resample.side_effect = [[], [flush_frame]]
-            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
-            mock_container = MagicMock()
-            mock_container.__enter__ = lambda s: s
-            mock_container.__exit__ = MagicMock(return_value=False)
-            mock_container.decode.return_value = [MagicMock()]
-            mock_av.open.return_value = mock_container
-            result = PCMA.decode_pcm(b"fake", "alaw", 8000, input_rate_hz=8000)
-        assert result.dtype == np.float32
-        assert len(result) == len(pcm_array)
-
-    def test_decode_pcm__empty_result_when_no_frames(self):
-        """decode_pcm returns an empty float32 array when no audio frames are decoded."""
-        with patch("voip.codecs.base.av") as mock_av:
-            mock_resampler = MagicMock()
-            mock_resampler.resample.return_value = []
-            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
-            mock_container = MagicMock()
-            mock_container.__enter__ = lambda s: s
-            mock_container.__exit__ = MagicMock(return_value=False)
-            mock_container.decode.return_value = []
-            mock_av.open.return_value = mock_container
-            result = PCMA.decode_pcm(b"fake", "alaw", 8000)
+    def test_resample__upsample_doubles_length(self):
+        """Upsampling from 8 kHz to 16 kHz produces twice as many samples."""
+        audio = np.zeros(160, dtype=np.float32)
+        result = RTPCodec.resample(audio, 8000, 16000)
+        assert len(result) == 320
+
+    def test_resample__downsample_halves_length(self):
+        """Downsampling from 16 kHz to 8 kHz halves the sample count."""
+        audio = np.zeros(320, dtype=np.float32)
+        result = RTPCodec.resample(audio, 16000, 8000)
+        assert len(result) == 160
+
+    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_decode_pcm__without_input_rate_passes_no_options(self):
-        """decode_pcm passes no sample_rate option when input_rate_hz is None."""
-        with patch("voip.codecs.base.av") as mock_av:
-            mock_resampler = MagicMock()
-            mock_resampler.resample.return_value = []
-            mock_av.audio.resampler.AudioResampler.return_value = mock_resampler
-            mock_container = MagicMock()
-            mock_container.__enter__ = lambda s: s
-            mock_container.__exit__ = MagicMock(return_value=False)
-            mock_container.decode.return_value = []
-            mock_av.open.return_value = mock_container
-            PCMA.decode_pcm(b"fake", "alaw", 8000, input_rate_hz=None)
-        call_kwargs = mock_av.open.call_args[1]
-        assert call_kwargs["options"] == {}
-
-
-class TestEncodePCM:
-    def test_encode_pcm__g722_returns_bytes(self):
-        """encode_pcm produces non-empty bytes for G.722."""
-        result = RTPCodec.encode_pcm(np.zeros(320, dtype=np.float32), "g722", 16000)
-        assert isinstance(result, bytes)
-        assert len(result) > 0
-
-    def test_encode_pcm__opus_returns_bytes(self):
-        """encode_pcm produces non-empty bytes for Opus (libopus)."""
-        result = RTPCodec.encode_pcm(np.zeros(960, dtype=np.float32), "libopus", 48000)
-        assert isinstance(result, bytes)
-        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
 
 
 class TestToPayloadFormat:
@@ -95,6 +56,7 @@ def test_to_payload_format__returns_rtp_payload_format(self):
 
     def test_to_payload_format__uses_rtp_clock_rate_for_sdp(self):
         """to_payload_format uses rtp_clock_rate_hz as the SDP sample_rate."""
+        pytest.importorskip("av")
         from voip.codecs.g722 import G722  # noqa: PLC0415
 
         result = G722.to_payload_format()
@@ -113,11 +75,11 @@ def test_packetize__default_encodes_per_frame(self):
 
 class TestAbstractMethods:
     def test_decode__raises_not_implemented(self):
-        """PyAVCodec.decode raises NotImplementedError."""
+        """RTPCodec.decode raises NotImplementedError."""
         with pytest.raises(NotImplementedError):
             RTPCodec.decode(b"data", 8000)
 
     def test_encode__raises_not_implemented(self):
-        """PyAVCodec.encode raises NotImplementedError."""
+        """RTPCodec.encode raises NotImplementedError."""
         with pytest.raises(NotImplementedError):
             RTPCodec.encode(np.zeros(160, dtype=np.float32))