From 20d1987259b6b76f2c0e036bb8ff587a316039f7 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 21:29:14 +0100 Subject: [PATCH 1/9] Deduplicate codec base classes --- docs/codecs.md | 3 ++ mkdocs.yml | 1 + voip/ai.py | 4 +- voip/audio.py | 6 +-- voip/codecs/__init__.py | 110 ++++------------------------------------ voip/codecs/base.py | 23 ++++++--- voip/codecs/g722.py | 11 ++++ voip/codecs/opus.py | 8 +-- 8 files changed, 47 insertions(+), 119 deletions(-) create mode 100644 docs/codecs.md diff --git a/docs/codecs.md b/docs/codecs.md new file mode 100644 index 0000000..371fb5c --- /dev/null +++ b/docs/codecs.md @@ -0,0 +1,3 @@ +# Codecs + +::: voip.codecs diff --git a/mkdocs.yml b/mkdocs.yml index 4ae1f14..25d9246 100644 --- a/mkdocs.yml +++ b/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 diff --git a/voip/ai.py b/voip/ai.py index c52562b..789470e 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -20,7 +20,7 @@ from pocket_tts import TTSModel from voip.audio import VoiceActivityCall -from voip.codecs import Codec +from voip.codecs import RTPCodec from voip.codecs.g722 import G722 from voip.codecs.opus import Opus from voip.codecs.pcma import PCMA @@ -160,7 +160,7 @@ class AgentCall(TranscribeCall): "Keep your answers very brief and conversational." ) #: Preferred codecs in priority order (highest first). - PREFERRED_CODECS: ClassVar[list[type[Codec]]] = [Opus, G722, PCMU, PCMA] + PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [Opus, G722, PCMU, PCMA] #: Ollama model name for generating replies. ollama_model: str = dataclasses.field(default="llama3") diff --git a/voip/audio.py b/voip/audio.py index d9256ee..67ecb65 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -21,7 +21,7 @@ import numpy as np import voip.codecs as codecs -from voip.codecs import Codec +from voip.codecs import RTPCodec from voip.codecs.g722 import G722 # noqa: E402 from voip.codecs.opus import Opus # noqa: E402 from voip.codecs.pcma import PCMA # noqa: E402 @@ -55,7 +55,7 @@ class is stored on `codec` after `__post_init__` and used for all """ #: Preferred codecs in priority order (highest priority first). - PREFERRED_CODECS: ClassVar[list[type[Codec]]] = [Opus, G722, PCMA, PCMU] + PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [Opus, G722, PCMA, PCMU] #: Target sample rate for decoded audio delivered to `audio_received`. RESAMPLING_RATE_HZ: ClassVar[int] = 16000 @@ -64,7 +64,7 @@ class is stored on `codec` after `__post_init__` and used for all RTP_PACKET_DURATION_SECS: ClassVar[float] = 0.02 #: Resolved codec class for this call, set in `__post_init__`. - codec: type[Codec] = dataclasses.field(init=False, repr=False) + codec: type[RTPCodec] = dataclasses.field(init=False, repr=False) #: Outbound RTP sequence counter. rtp_sequence_number: int = dataclasses.field(init=False, repr=False, default=0) diff --git a/voip/codecs/__init__.py b/voip/codecs/__init__.py index 61a084d..21d5012 100644 --- a/voip/codecs/__init__.py +++ b/voip/codecs/__init__.py @@ -1,6 +1,6 @@ """Audio codec implementations for RTP streams. -Provides the [`Codec`][voip.codecs.Codec] structural interface and concrete +Provides the [`RTPCodec`][voip.codecs.base.RTPCodec] base class and concrete implementations for all supported RTP audio codecs: - [`Opus`][voip.codecs.Opus] — Opus (RFC 7587), PT 111 @@ -14,113 +14,21 @@ from __future__ import annotations -from collections.abc import Iterator -from typing import TYPE_CHECKING, ClassVar, Protocol +from voip.codecs.base import RTPCodec +from voip.codecs.g722 import G722 +from voip.codecs.opus import Opus +from voip.codecs.pcma import PCMA +from voip.codecs.pcmu import PCMU -if TYPE_CHECKING: - import numpy as np - - from voip.sdp.types import RTPPayloadFormat - -__all__ = ["Codec", "G722", "Opus", "PCMA", "PCMU", "get"] - - -class Codec(Protocol): - """Structural interface for RTP audio codec classes. - - Concrete implementations: [`Opus`][voip.codecs.Opus], - [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.PCMA], - [`PCMU`][voip.codecs.PCMU]. - - All codec implementations are stateless: every method is a classmethod - or staticmethod and codecs are referenced as `type[Codec]`, never - instantiated. - """ - - payload_type: ClassVar[int] - """RTP payload type number (static or dynamic).""" - - encoding_name: ClassVar[str] - """SDP encoding name in lowercase (e.g. `"opus"`, `"g722"`).""" - - sample_rate_hz: ClassVar[int] - """Actual audio sample rate in Hz.""" - - rtp_clock_rate_hz: ClassVar[int] - """RTP timestamp clock rate in Hz (may differ from `sample_rate_hz`).""" - - frame_size: ClassVar[int] - """Audio samples per 20 ms RTP frame at `sample_rate_hz`.""" - - timestamp_increment: ClassVar[int] - """RTP timestamp ticks per frame at `rtp_clock_rate_hz`.""" - - channels: ClassVar[int] - """Channel count (1 = mono, 2 = stereo).""" - - @classmethod - def decode( - cls, - payload: bytes, - output_rate_hz: int, - *, - input_rate_hz: int | None = None, - ) -> np.ndarray: - """Decode an RTP payload to float32 mono PCM. - - Args: - payload: Raw RTP payload bytes. - output_rate_hz: Target sample rate in Hz. - input_rate_hz: Optional input clock rate override in Hz. - - Returns: - Float32 mono PCM array at *output_rate_hz* Hz. - """ - - @classmethod - def encode(cls, samples: np.ndarray) -> bytes: - """Encode float32 mono PCM to an RTP payload. - - Args: - samples: Float32 mono PCM at `sample_rate_hz` Hz. - - Returns: - Encoded bytes for one RTP payload. - """ - - @classmethod - def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: - """Encode *audio* and yield one payload per 20 ms RTP frame. - - Args: - audio: Float32 mono PCM at `sample_rate_hz` Hz. - - Yields: - Encoded payload bytes, one per RTP packet. - """ - - @classmethod - def to_payload_format(cls) -> RTPPayloadFormat: - """Create an [`RTPPayloadFormat`][voip.sdp.types.RTPPayloadFormat] for SDP. - - Returns: - Payload format descriptor for this codec. - """ - - -# Concrete implementations — imported after the Protocol to avoid circularity. -from voip.codecs.g722 import G722 # noqa: E402 -from voip.codecs.opus import Opus # noqa: E402 -from voip.codecs.pcma import PCMA # noqa: E402 -from voip.codecs.pcmu import PCMU # noqa: E402 +__all__ = ["G722", "Opus", "PCMA", "PCMU", "RTPCodec", "get"] #: Registry mapping lowercase encoding names to codec classes. -REGISTRY: dict[str, type[Codec]] = { +REGISTRY: dict[str, type[RTPCodec]] = { codec.encoding_name: codec for codec in (Opus, G722, PCMA, PCMU) } -def get(encoding_name: str) -> type[Codec]: +def get(encoding_name: str) -> type[RTPCodec]: """Get a codec class by its SDP encoding name. Args: diff --git a/voip/codecs/base.py b/voip/codecs/base.py index b665bde..5cabccb 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -1,8 +1,9 @@ -"""PyAV-based codec base class for RTP audio streams. +"""Base class for RTP audio codecs. -All concrete codec classes in this package inherit from [`PyAVCodec`][voip.codecs.av.PyAVCodec], -which provides shared [`decode_pcm`][voip.codecs.av.PyAVCodec.decode_pcm] and -[`encode_pcm`][voip.codecs.av.PyAVCodec.encode_pcm] helpers backed by [PyAV][]. +All concrete codec classes in this package inherit from +[`RTPCodec`][voip.codecs.base.RTPCodec], which provides shared +[`decode_pcm`][voip.codecs.base.RTPCodec.decode_pcm] and +[`encode_pcm`][voip.codecs.base.RTPCodec.encode_pcm] helpers backed by [PyAV][]. [PyAV]: https://pyav.basswood-io.com/ """ @@ -28,10 +29,18 @@ class RTPCodec: """Base class for RTP audio codecs that decode and encode via PyAV. + Concrete implementations: [`Opus`][voip.codecs.Opus], + [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.PCMA], + [`PCMU`][voip.codecs.PCMU]. + + All codec implementations are stateless: every method is a classmethod or + staticmethod and codecs are referenced as `type[RTPCodec]`, never + instantiated. + Concrete subclasses define codec-specific class variables and may override - [`decode`][voip.codecs.av.PyAVCodec.decode], - [`encode`][voip.codecs.av.PyAVCodec.encode], and - [`packetize`][voip.codecs.av.PyAVCodec.packetize]. + [`decode`][voip.codecs.base.RTPCodec.decode], + [`encode`][voip.codecs.base.RTPCodec.encode], and + [`packetize`][voip.codecs.base.RTPCodec.packetize]. Subclasses that produce variable-length output across frames (e.g. G.722 ADPCM) should override `packetize` to encode the whole buffer at once and diff --git a/voip/codecs/g722.py b/voip/codecs/g722.py index 9cfdd05..ccdeb05 100644 --- a/voip/codecs/g722.py +++ b/voip/codecs/g722.py @@ -62,6 +62,17 @@ def encode(cls, samples: np.ndarray) -> bytes: @classmethod def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: + """Packetize a G.722 audio buffer into 160-byte RTP payloads. + + Encodes the entire *audio* buffer at once so that the ADPCM predictor + state is preserved across packet boundaries. + + Args: + audio: Float32 PCM samples at 16 000 Hz. + + Yields: + 160-byte G.722 encoded RTP payloads. + """ encoded = cls.encode(audio) # G.722 2:1 sample-to-byte ratio: frame_size (320) samples → 160 bytes. payload_size = cls.frame_size // 2 diff --git a/voip/codecs/opus.py b/voip/codecs/opus.py index d80a9d4..30ce95f 100644 --- a/voip/codecs/opus.py +++ b/voip/codecs/opus.py @@ -68,7 +68,7 @@ def ogg_page( sequence_number: int, packets: list[bytes], ) -> bytes: - """Build a single Ogg page ([RFC 3533][]). + """Build a single Ogg page ([RFC 3533](https://datatracker.ietf.org/doc/html/rfc3533)). Args: header_type: Page header type flags (e.g. `0x02` for BOS, `0x04` for EOS). @@ -79,8 +79,6 @@ def ogg_page( Returns: Complete Ogg page bytes including CRC. - - [RFC 3533]: https://datatracker.ietf.org/doc/html/rfc3533 """ lacing: list[int] = [] for packet in packets: @@ -109,15 +107,13 @@ def ogg_container(cls, packet: bytes) -> bytes: Produces a three-page Ogg stream: BOS (OpusHead), comment (OpusTags), and the single data page. Opus always uses 48 000 Hz - internally ([RFC 7587 §4][]). + internally ([RFC 7587 §4](https://datatracker.ietf.org/doc/html/rfc7587#section-4)). Args: packet: Raw Opus RTP payload bytes. Returns: Ogg Opus container bytes suitable for PyAV decoding. - - [RFC 7587 §4]: https://datatracker.ietf.org/doc/html/rfc7587#section-4 """ serial_number = int.from_bytes(os.urandom(4), "little") vendor = b"voip" From d046730fcf6dbf601dfc91fbc93dc1c8c4e3b852 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 21:39:11 +0100 Subject: [PATCH 2/9] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- voip/codecs/base.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/voip/codecs/base.py b/voip/codecs/base.py index 5cabccb..92cdef4 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -27,7 +27,7 @@ class RTPCodec: - """Base class for RTP audio codecs that decode and encode via PyAV. + """Base class for RTP audio codecs that provide PyAV-backed helpers. Concrete implementations: [`Opus`][voip.codecs.Opus], [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.PCMA], @@ -42,6 +42,11 @@ class RTPCodec: [`encode`][voip.codecs.base.RTPCodec.encode], and [`packetize`][voip.codecs.base.RTPCodec.packetize]. + Subclasses may use the shared PyAV-backed helpers or implement + [`decode`][voip.codecs.base.RTPCodec.decode] and + [`encode`][voip.codecs.base.RTPCodec.encode] using alternative backends + such as NumPy. + Subclasses that produce variable-length output across frames (e.g. G.722 ADPCM) should override `packetize` to encode the whole buffer at once and preserve predictor state. From d64e6ba7435145ee3d5dcdf128d1079cb33f712a Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 22:38:05 +0100 Subject: [PATCH 3/9] Update codec extras --- .github/workflows/ci.yml | 1 + docs/codecs.md | 49 ++++++++++++++- pyproject.toml | 7 ++- tests/codecs/test_av.py | 83 +++++++++++++++++++++++++ tests/codecs/test_base.py | 97 ++++++++--------------------- tests/codecs/test_codecs.py | 71 ++++++++++++++++++---- tests/codecs/test_opus.py | 22 +++---- tests/codecs/test_pcm.py | 114 +++++++++++++++++++--------------- tests/test_audio.py | 4 +- voip/audio.py | 29 ++++----- voip/codecs/__init__.py | 26 +++++--- voip/codecs/av.py | 113 ++++++++++++++++++++++++++++++++++ voip/codecs/base.py | 118 +++++++++++------------------------- voip/codecs/g722.py | 17 ++---- voip/codecs/opus.py | 22 ++++--- voip/codecs/pcma.py | 39 +++++++----- voip/codecs/pcmu.py | 33 ++++++---- 17 files changed, 540 insertions(+), 305 deletions(-) create mode 100644 tests/codecs/test_av.py create mode 100644 voip/codecs/av.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e7acc3..0aa9151 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -51,6 +51,7 @@ jobs: - "3.14" extra: - audio + - hd-audio - cli - pygments runs-on: ${{ matrix.os }} diff --git a/docs/codecs.md b/docs/codecs.md index 371fb5c..26ba115 100644 --- a/docs/codecs.md +++ b/docs/codecs.md @@ -1,3 +1,50 @@ # Codecs -::: voip.codecs +## Overview + +VoIP ships two tiers of audio codecs: + +| Extra required | Codecs available | +| ------------------------- | -------------------------------------- | +| `numpy` | PCMA (G.711 A-law), PCMU (G.711 µ-law) | +| `pyav` (includes `numpy`) | + 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] +``` + +## Base classes + +::: voip.codecs.base + +::: voip.codecs.av + +## Pure-NumPy codecs + +These codecs work without PyAV and require only `numpy`. + +::: voip.codecs.pcma + +::: voip.codecs.pcmu + +## PyAV codecs + +These codecs require the `pyav` extra (`pip install voip[pyav]`). + +::: voip.codecs.g722 + +::: voip.codecs.opus + +## Registry + +::: voip.codecs.get + +[ffmpeg]: https://ffmpeg.org/ diff --git a/pyproject.toml b/pyproject.toml index 163dbf4..612e0dd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,9 +39,10 @@ requires-python = ">=3.13" dependencies = ["cryptography"] [project.optional-dependencies] -audio = ["numpy", "av"] -ai = ["faster-whisper", "numpy", "av", "ollama", "pocket-tts"] -cli = ["click", "pygments", "faster-whisper", "numpy", "av", "ollama", "pocket-tts"] +audio = ["numpy"] +hd-audio = ["numpy", "av"] +ai = ["faster-whisper", "numpy", "ollama", "pocket-tts"] +cli = ["click", "pygments", "faster-whisper", "numpy", "ollama", "pocket-tts"] pygments = ["Pygments"] [project.scripts] diff --git a/tests/codecs/test_av.py b/tests/codecs/test_av.py new file mode 100644 index 0000000..3248ab1 --- /dev/null +++ b/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 diff --git a/tests/codecs/test_base.py b/tests/codecs/test_base.py index db680da..3f699f0 100644 --- a/tests/codecs/test_base.py +++ b/tests/codecs/test_base.py @@ -1,87 +1,41 @@ -"""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__returns_float32(self): + """Resample always returns a float32 array.""" + audio = np.ones(100, dtype=np.float64) + result = RTPCodec.resample(audio, 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 class TestToPayloadFormat: @@ -97,6 +51,7 @@ 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.""" from voip.codecs.g722 import G722 # noqa: PLC0415 + pytest.importorskip("av") result = G722.to_payload_format() assert result.sample_rate == G722.rtp_clock_rate_hz # 8000, not 16000 @@ -113,11 +68,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)) diff --git a/tests/codecs/test_codecs.py b/tests/codecs/test_codecs.py index e00f297..e37c80f 100644 --- a/tests/codecs/test_codecs.py +++ b/tests/codecs/test_codecs.py @@ -2,23 +2,17 @@ from __future__ import annotations +import importlib +import sys + import pytest pytest.importorskip("numpy") -pytest.importorskip("av") -from voip.codecs import G722, PCMA, PCMU, Opus, get # noqa: E402 +from voip.codecs import PCMA, PCMU, get # noqa: E402 class TestGet: - def test_get__opus(self): - """Get returns Opus for the encoding name 'opus'.""" - assert get("opus") is Opus - - def test_get__g722(self): - """Get returns G722 for the encoding name 'g722'.""" - assert get("g722") is G722 - def test_get__pcma(self): """Get returns PCMA for the encoding name 'pcma'.""" assert get("pcma") is PCMA @@ -27,10 +21,22 @@ def test_get__pcmu(self): """Get returns PCMU for the encoding name 'pcmu'.""" assert get("pcmu") is PCMU + def test_get__opus(self): + """Get returns Opus for the encoding name 'opus'.""" + pytest.importorskip("av") + from voip.codecs import Opus # noqa: PLC0415 + + assert get("opus") is Opus + + def test_get__g722(self): + """Get returns G722 for the encoding name 'g722'.""" + pytest.importorskip("av") + from voip.codecs import G722 # noqa: PLC0415 + + assert get("g722") is G722 + def test_get__case_insensitive(self): """Get normalises the encoding name to lowercase before lookup.""" - assert get("OPUS") is Opus - assert get("G722") is G722 assert get("PCMA") is PCMA assert get("PCMU") is PCMU @@ -38,3 +44,44 @@ def test_get__raise_not_implemented_error(self): """Get raises NotImplementedError for an unrecognised encoding name.""" with pytest.raises(NotImplementedError, match="Unsupported codec"): get("unknown") + + +class TestRegistry: + def test_registry__always_contains_numpy_codecs(self): + """REGISTRY always contains PCMA and PCMU regardless of PyAV availability.""" + import voip.codecs as m # noqa: PLC0415 + + assert "pcma" in m.REGISTRY + assert "pcmu" in m.REGISTRY + + 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", + } + ] + 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 + + assert "pcma" in fresh.REGISTRY + assert "pcmu" in fresh.REGISTRY + assert "g722" not in fresh.REGISTRY + assert "opus" not in fresh.REGISTRY + finally: + for k in keys_to_remove: + sys.modules.pop(k, None) + sys.modules.update(saved) + importlib.reload(target) diff --git a/tests/codecs/test_opus.py b/tests/codecs/test_opus.py index 5dc3a4e..d6b8105 100644 --- a/tests/codecs/test_opus.py +++ b/tests/codecs/test_opus.py @@ -15,28 +15,28 @@ class TestOggCRC32: def test_ogg_crc32__empty_bytes(self): """ogg_crc32 of empty bytes is zero.""" - assert Opus.ogg_crc32(b"") == 0 + assert Opus._ogg_crc32(b"") == 0 def test_ogg_crc32__known_value(self): """ogg_crc32 produces a deterministic 32-bit value.""" - crc = Opus.ogg_crc32(b"OggS") + crc = Opus._ogg_crc32(b"OggS") assert 0 <= crc <= 0xFFFFFFFF class TestOggPage: def test_ogg_page__starts_with_capture_pattern(self): """ogg_page output starts with the Ogg capture pattern 'OggS'.""" - page = Opus.ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) + page = Opus._ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) assert page[:4] == b"OggS" def test_ogg_page__contains_packet_data(self): """ogg_page embeds the provided packet bytes.""" - page = Opus.ogg_page(0x02, 0, 0, 0, [b"payload"]) + page = Opus._ogg_page(0x02, 0, 0, 0, [b"payload"]) assert b"payload" in page def test_ogg_page__large_packet_uses_255_lacing(self): """ogg_page correctly laces a packet exceeding 254 bytes.""" - page = Opus.ogg_page(0x00, 0, 0, 0, [b"x" * 256]) + page = Opus._ogg_page(0x00, 0, 0, 0, [b"x" * 256]) assert page[:4] == b"OggS" assert len(page) > 256 @@ -44,28 +44,28 @@ def test_ogg_page__large_packet_uses_255_lacing(self): class TestOggContainer: def test_ogg_container__starts_with_ogg_magic(self): """ogg_container output starts with the Ogg capture pattern 'OggS'.""" - assert Opus.ogg_container(b"packet").startswith(b"OggS") + assert Opus._ogg_container(b"packet").startswith(b"OggS") def test_ogg_container__contains_opus_head(self): """ogg_container includes the OpusHead identification header.""" - assert b"OpusHead" in Opus.ogg_container(b"packet") + assert b"OpusHead" in Opus._ogg_container(b"packet") def test_ogg_container__contains_opus_tags(self): """ogg_container includes the OpusTags comment header.""" - assert b"OpusTags" in Opus.ogg_container(b"packet") + assert b"OpusTags" in Opus._ogg_container(b"packet") def test_ogg_container__non_empty_for_single_packet(self): """ogg_container produces a non-empty Ogg container for a single Opus packet.""" - assert len(Opus.ogg_container(b"x" * 100)) > 100 + assert len(Opus._ogg_container(b"x" * 100)) > 100 def test_ogg_container__empty_payload(self): """ogg_container produces a valid Ogg container even for empty payload.""" - result = Opus.ogg_container(b"") + result = Opus._ogg_container(b"") assert b"OggS" in result def test_ogg_container__produces_three_pages(self): """ogg_container produces exactly three Ogg pages: BOS, tags, and data.""" - result = Opus.ogg_container(b"x" * 10) + result = Opus._ogg_container(b"x" * 10) assert result.count(b"OggS") == 3 diff --git a/tests/codecs/test_pcm.py b/tests/codecs/test_pcm.py index e45384b..3255b3b 100644 --- a/tests/codecs/test_pcm.py +++ b/tests/codecs/test_pcm.py @@ -2,12 +2,9 @@ from __future__ import annotations -from unittest.mock import patch - import pytest np = pytest.importorskip("numpy") -av = pytest.importorskip("av") from voip.codecs.pcma import PCMA # noqa: E402 from voip.codecs.pcmu import PCMU # noqa: E402 @@ -40,29 +37,40 @@ def test_timestamp_increment(self): class TestPCMADecode: - def test_decode__uses_alaw_format(self): - """Decode calls decode_pcm with the alaw PyAV format string.""" - with patch.object( - PCMA, "decode_pcm", return_value=np.zeros(8000, dtype=np.float32) - ) as mock: - PCMA.decode(b"payload", 8000) - assert mock.call_args[0][1] == "alaw" - - def test_decode__passes_sample_rate_as_default(self): - """Decode passes sample_rate_hz as input_rate_hz when not overridden.""" - with patch.object( - PCMA, "decode_pcm", return_value=np.zeros(8000, dtype=np.float32) - ) as mock: - PCMA.decode(b"payload", 8000) - assert mock.call_args[1]["input_rate_hz"] == PCMA.sample_rate_hz - - def test_decode__uses_input_rate_hz_override(self): - """Decode passes the caller-supplied input_rate_hz when provided.""" - with patch.object( - PCMA, "decode_pcm", return_value=np.zeros(16000, dtype=np.float32) - ) as mock: - PCMA.decode(b"payload", 16000, input_rate_hz=16000) - assert mock.call_args[1]["input_rate_hz"] == 16000 + def test_decode__returns_float32(self): + """Decode produces float32 samples from A-law encoded input.""" + payload = PCMA.encode(np.zeros(160, dtype=np.float32)) + assert PCMA.decode(payload, 8000).dtype == np.float32 + + def test_decode__native_rate_preserves_sample_count(self): + """Decode at native 8 kHz returns one sample per byte.""" + payload = PCMA.encode(np.zeros(160, dtype=np.float32)) + assert len(PCMA.decode(payload, 8000)) == 160 + + def test_decode__resamples_to_output_rate(self): + """Decode resamples to the requested output rate.""" + payload = PCMA.encode(np.zeros(160, dtype=np.float32)) + assert len(PCMA.decode(payload, 16000)) == 320 + + def test_decode__silence_roundtrip(self): + """Encode then decode silence returns values near zero.""" + payload = PCMA.encode(np.zeros(160, dtype=np.float32)) + result = PCMA.decode(payload, 8000) + assert np.allclose(result, 0.0, atol=0.02) + + def test_decode__positive_and_negative_differ(self): + """Positive and negative input decode to values with opposite sign.""" + pos = PCMA.decode(PCMA.encode(np.full(1, 0.5, dtype=np.float32)), 8000) + neg = PCMA.decode(PCMA.encode(np.full(1, -0.5, dtype=np.float32)), 8000) + 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.""" + 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) def test_decode__real_decode_returns_float32(self): """Decode produces a float32 array from real A-law encoded input.""" @@ -115,29 +123,37 @@ def test_timestamp_increment(self): class TestPCMUDecode: - def test_decode__uses_mulaw_format(self): - """Decode calls decode_pcm with the mulaw PyAV format string.""" - with patch.object( - PCMU, "decode_pcm", return_value=np.zeros(8000, dtype=np.float32) - ) as mock: - PCMU.decode(b"payload", 8000) - assert mock.call_args[0][1] == "mulaw" - - def test_decode__passes_sample_rate_as_default(self): - """Decode passes sample_rate_hz as input_rate_hz when not overridden.""" - with patch.object( - PCMU, "decode_pcm", return_value=np.zeros(8000, dtype=np.float32) - ) as mock: - PCMU.decode(b"payload", 8000) - assert mock.call_args[1]["input_rate_hz"] == PCMU.sample_rate_hz - - def test_decode__uses_input_rate_hz_override(self): - """Decode passes the caller-supplied input_rate_hz when provided.""" - with patch.object( - PCMU, "decode_pcm", return_value=np.zeros(16000, dtype=np.float32) - ) as mock: - PCMU.decode(b"payload", 16000, input_rate_hz=16000) - assert mock.call_args[1]["input_rate_hz"] == 16000 + def test_decode__returns_float32(self): + """Decode produces float32 samples from mu-law encoded input.""" + payload = PCMU.encode(np.zeros(160, dtype=np.float32)) + assert PCMU.decode(payload, 8000).dtype == np.float32 + + def test_decode__native_rate_preserves_sample_count(self): + """Decode at native 8 kHz returns one sample per byte.""" + payload = PCMU.encode(np.zeros(160, dtype=np.float32)) + assert len(PCMU.decode(payload, 8000)) == 160 + + def test_decode__resamples_to_output_rate(self): + """Decode resamples to the requested output rate.""" + payload = PCMU.encode(np.zeros(160, dtype=np.float32)) + assert len(PCMU.decode(payload, 16000)) == 320 + + def test_decode__max_positive_roundtrip(self): + """Max positive amplitude (0x00) decodes to a large positive value.""" + decoded = PCMU.decode(bytes([0x00]), 8000) + assert decoded[0] > 0.9 + + def test_decode__max_negative_roundtrip(self): + """Max negative amplitude (0x80) decodes to a large negative value.""" + 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.""" + 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) def test_decode__real_decode_returns_float32(self): """Decode produces a float32 array from real mu-law encoded input.""" diff --git a/tests/test_audio.py b/tests/test_audio.py index 36d6815..50c1b27 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -251,11 +251,11 @@ class PCMAOnlyCall(AudioCall): assert result.fmt[0].payload_type == 8 def test_preferred_codecs__class_attribute(self): - """PREFERRED_CODECS is a class attribute on AudioCall with Opus first.""" + """PREFERRED_CODECS is a class attribute on AudioCall with Opus first when PyAV is available.""" codec_classes = AudioCall.PREFERRED_CODECS assert isinstance(codec_classes, list) pts = [c.payload_type for c in codec_classes] - assert pts[0] == 111 # Opus is highest priority + assert pts[0] == 111 # Opus is highest priority when PyAV is present assert 8 in pts # PCMA present assert 0 in pts # PCMU present diff --git a/voip/audio.py b/voip/audio.py index 67ecb65..5abe17f 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -16,19 +16,17 @@ import json import logging import secrets +import typing from typing import ClassVar -import numpy as np - import voip.codecs as codecs from voip.codecs import RTPCodec -from voip.codecs.g722 import G722 # noqa: E402 -from voip.codecs.opus import Opus # noqa: E402 -from voip.codecs.pcma import PCMA # noqa: E402 -from voip.codecs.pcmu import PCMU # noqa: E402 from voip.rtp import RTPCall, RTPPacket from voip.sdp.types import MediaDescription +if typing.TYPE_CHECKING: + import numpy as np + __all__ = ["AudioCall", "EchoCall", "VoiceActivityCall"] @@ -55,7 +53,13 @@ class is stored on `codec` after `__post_init__` and used for all """ #: Preferred codecs in priority order (highest priority first). - PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [Opus, G722, PCMA, PCMU] + #: Populated from [`voip.codecs.REGISTRY`][voip.codecs.REGISTRY] at import + #: time; falls back to pure-NumPy codecs when the ``pyav`` extra is absent. + PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [ + codecs.REGISTRY[name] + for name in ("opus", "g722", "pcma", "pcmu") + if name in codecs.REGISTRY + ] #: Target sample rate for decoded audio delivered to `audio_received`. RESAMPLING_RATE_HZ: ClassVar[int] = 16000 @@ -264,7 +268,7 @@ def resample( ) -> np.ndarray: """Resample *audio* from *source_rate_hz* to *destination_rate_hz*. - Uses linear interpolation via [`numpy.interp`][]. + Delegates to [`RTPCodec.resample`][voip.codecs.base.RTPCodec.resample]. Args: audio: Float32 mono PCM array. @@ -274,14 +278,7 @@ def resample( Returns: Resampled float32 array at *destination_rate_hz* Hz. """ - if source_rate_hz == destination_rate_hz: - return audio - n_out = round(len(audio) * destination_rate_hz / source_rate_hz) - return np.interp( - np.linspace(0, len(audio) - 1, n_out), - np.arange(len(audio)), - audio, - ).astype(np.float32) + return RTPCodec.resample(audio, source_rate_hz, destination_rate_hz) @dataclasses.dataclass(kw_only=True) diff --git a/voip/codecs/__init__.py b/voip/codecs/__init__.py index 21d5012..0522491 100644 --- a/voip/codecs/__init__.py +++ b/voip/codecs/__init__.py @@ -3,30 +3,40 @@ Provides the [`RTPCodec`][voip.codecs.base.RTPCodec] base class and concrete implementations for all supported RTP audio codecs: -- [`Opus`][voip.codecs.Opus] — Opus (RFC 7587), PT 111 -- [`G722`][voip.codecs.G722] — G.722 (RFC 3551), PT 9 -- [`PCMA`][voip.codecs.PCMA] — G.711 A-law (RFC 3551), PT 8 -- [`PCMU`][voip.codecs.PCMU] — G.711 mu-law (RFC 3551), PT 0 +- [`PCMA`][voip.codecs.PCMA] — G.711 A-law (RFC 3551), PT 8 *(pure NumPy)* +- [`PCMU`][voip.codecs.PCMU] — G.711 mu-law (RFC 3551), PT 0 *(pure NumPy)* +- [`G722`][voip.codecs.G722] — G.722 (RFC 3551), PT 9 *(requires* ``pyav`` *extra)* +- [`Opus`][voip.codecs.Opus] — Opus (RFC 7587), PT 111 *(requires* ``pyav`` *extra)* Use [`get`][voip.codecs.get] to look up a codec class by its SDP encoding name (case-insensitive). + +When the ``pyav`` extra is not installed only PCMA and PCMU are registered. """ from __future__ import annotations from voip.codecs.base import RTPCodec -from voip.codecs.g722 import G722 -from voip.codecs.opus import Opus from voip.codecs.pcma import PCMA from voip.codecs.pcmu import PCMU -__all__ = ["G722", "Opus", "PCMA", "PCMU", "RTPCodec", "get"] +__all__ = ["G722", "Opus", "PCMA", "PCMU", "PyAVCodec", "RTPCodec", "get"] #: Registry mapping lowercase encoding names to codec classes. REGISTRY: dict[str, type[RTPCodec]] = { - codec.encoding_name: codec for codec in (Opus, G722, PCMA, PCMU) + PCMA.encoding_name: PCMA, + PCMU.encoding_name: PCMU, } +try: + from voip.codecs.av import PyAVCodec + from voip.codecs.g722 import G722 + from voip.codecs.opus import Opus + + REGISTRY |= {G722.encoding_name: G722, Opus.encoding_name: Opus} +except ImportError: + pass + def get(encoding_name: str) -> type[RTPCodec]: """Get a codec class by its SDP encoding name. diff --git a/voip/codecs/av.py b/voip/codecs/av.py new file mode 100644 index 0000000..005c262 --- /dev/null +++ b/voip/codecs/av.py @@ -0,0 +1,113 @@ +"""PyAV-backed RTP codec base class. + +[`PyAVCodec`][voip.codecs.av.PyAVCodec] extends +[`RTPCodec`][voip.codecs.base.RTPCodec] with +[`decode_pcm`][voip.codecs.av.PyAVCodec.decode_pcm] and +[`encode_pcm`][voip.codecs.av.PyAVCodec.encode_pcm] helpers that use +[PyAV][] for container-aware decode and codec-aware encode. + +Requires the ``pyav`` extra: ``pip install voip[pyav]``. + +Concrete subclasses: [`Opus`][voip.codecs.Opus], [`G722`][voip.codecs.G722]. + +[PyAV]: https://pyav.basswood-io.com/ +""" + +from __future__ import annotations + +import io +from typing import cast + +import av +import av.audio.resampler +import numpy as np + +from voip.codecs.base import RTPCodec + +__all__ = ["PyAVCodec"] + + +class PyAVCodec(RTPCodec): + """RTP codec that decodes and encodes audio via [PyAV][]. + + Concrete implementations: [`Opus`][voip.codecs.Opus], + [`G722`][voip.codecs.G722]. + + [PyAV]: https://pyav.basswood-io.com/ + """ + + @classmethod + def decode_pcm( + cls, + data: bytes, + av_format: str, + output_rate_hz: int, + *, + input_rate_hz: int | None = None, + ) -> np.ndarray: + """Decode raw audio bytes via PyAV into float32 mono PCM. + + Args: + data: Raw audio bytes in the codec's wire format. + av_format: PyAV format string (e.g. `"ogg"`, `"alaw"`). + output_rate_hz: Target sample rate in Hz. + input_rate_hz: Input clock rate hint for the PyAV decoder, or + `None` for self-describing formats like Ogg. + + Returns: + Float32 mono PCM array at *output_rate_hz* Hz. + """ + resampler = av.audio.resampler.AudioResampler( + format="fltp", layout="mono", rate=output_rate_hz + ) + frames: list[np.ndarray] = [] + with av.open( + io.BytesIO(data), + mode="r", + format=av_format, + options=( + {"sample_rate": str(input_rate_hz)} if input_rate_hz is not None else {} + ), + ) as container: + for frame in container.decode(audio=0): + for resampled in resampler.resample(frame): + frames.append(resampled.to_ndarray().flatten()) + for resampled in resampler.resample(None): + frames.append(resampled.to_ndarray().flatten()) + return np.concatenate(frames) if frames else np.array([], dtype=np.float32) + + @classmethod + def encode_pcm( + cls, + samples: np.ndarray, + av_codec_name: str, + sample_rate_hz: int, + ) -> bytes: + """Encode float32 mono PCM to raw codec bytes via PyAV. + + Args: + samples: Float32 mono PCM array in the range `[-1, 1]`. + av_codec_name: PyAV codec name (e.g. `"g722"` or `"libopus"`). + sample_rate_hz: Sample rate of *samples* in Hz. + + Returns: + Encoded audio bytes. + """ + codec: av.AudioCodecContext = cast( + av.AudioCodecContext, av.CodecContext.create(av_codec_name, "w") + ) + codec.sample_rate = sample_rate_hz + codec.format = av.AudioFormat("s16") + codec.layout = av.AudioLayout("mono") + codec.open() + pcm = np.clip(np.round(samples * 32768.0), -32768, 32767).astype(np.int16) + frame = av.AudioFrame.from_ndarray( + pcm[np.newaxis, :], format="s16", layout="mono" + ) + frame.sample_rate = sample_rate_hz + frame.pts = 0 + return b"".join( + bytes(packet) + for segment in (codec.encode(frame), codec.encode(None)) + for packet in segment + ) diff --git a/voip/codecs/base.py b/voip/codecs/base.py index 92cdef4..0893d48 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -1,33 +1,33 @@ """Base class for RTP audio codecs. All concrete codec classes in this package inherit from -[`RTPCodec`][voip.codecs.base.RTPCodec], which provides shared -[`decode_pcm`][voip.codecs.base.RTPCodec.decode_pcm] and -[`encode_pcm`][voip.codecs.base.RTPCodec.encode_pcm] helpers backed by [PyAV][]. +[`RTPCodec`][voip.codecs.base.RTPCodec]. + +Codecs that require [PyAV][] for decode/encode additionally inherit from +[`PyAVCodec`][voip.codecs.av.PyAVCodec], which provides +[`decode_pcm`][voip.codecs.av.PyAVCodec.decode_pcm] and +[`encode_pcm`][voip.codecs.av.PyAVCodec.encode_pcm]. + +Pure-NumPy codecs ([`PCMA`][voip.codecs.PCMA], [`PCMU`][voip.codecs.PCMU]) +inherit directly from `RTPCodec` and require no PyAV dependency. [PyAV]: https://pyav.basswood-io.com/ """ from __future__ import annotations -import io from collections.abc import Iterator -from typing import TYPE_CHECKING, ClassVar, cast +from typing import ClassVar -import av -import av.audio.resampler import numpy as np from voip.sdp.types import RTPPayloadFormat -if TYPE_CHECKING: - pass - __all__ = ["RTPCodec"] class RTPCodec: - """Base class for RTP audio codecs that provide PyAV-backed helpers. + """Base class for RTP audio codecs. Concrete implementations: [`Opus`][voip.codecs.Opus], [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.PCMA], @@ -37,9 +37,9 @@ class RTPCodec: staticmethod and codecs are referenced as `type[RTPCodec]`, never instantiated. - Concrete subclasses define codec-specific class variables and may override + Concrete subclasses define codec-specific class variables and override [`decode`][voip.codecs.base.RTPCodec.decode], - [`encode`][voip.codecs.base.RTPCodec.encode], and + [`encode`][voip.codecs.base.RTPCodec.encode], and optionally [`packetize`][voip.codecs.base.RTPCodec.packetize]. Subclasses may use the shared PyAV-backed helpers or implement @@ -50,6 +50,11 @@ class RTPCodec: Subclasses that produce variable-length output across frames (e.g. G.722 ADPCM) should override `packetize` to encode the whole buffer at once and preserve predictor state. + + Subclasses that require [PyAV][] additionally inherit from + [`PyAVCodec`][voip.codecs.av.PyAVCodec]. + + [PyAV]: https://pyav.basswood-io.com/ """ payload_type: ClassVar[int] @@ -74,80 +79,30 @@ class RTPCodec: """Channel count (1 = mono, 2 = stereo).""" @classmethod - def decode_pcm( - cls, - data: bytes, - av_format: str, - output_rate_hz: int, - *, - input_rate_hz: int | None = None, + def resample( + cls, audio: np.ndarray, source_rate_hz: int, destination_rate_hz: int ) -> np.ndarray: - """Decode raw audio bytes via PyAV into float32 mono PCM. - - Args: - data: Raw audio bytes in the codec's wire format. - av_format: PyAV format string (e.g. `"ogg"`, `"alaw"`). - output_rate_hz: Target sample rate in Hz. - input_rate_hz: Input clock rate hint for the PyAV decoder, or - `None` for self-describing formats like Ogg. + """Resample *audio* from *source_rate_hz* to *destination_rate_hz*. - Returns: - Float32 mono PCM array at *output_rate_hz* Hz. - """ - resampler = av.audio.resampler.AudioResampler( - format="fltp", layout="mono", rate=output_rate_hz - ) - frames: list[np.ndarray] = [] - with av.open( - io.BytesIO(data), - mode="r", - format=av_format, - options=( - {"sample_rate": str(input_rate_hz)} if input_rate_hz is not None else {} - ), - ) as container: - for frame in container.decode(audio=0): - for resampled in resampler.resample(frame): - frames.append(resampled.to_ndarray().flatten()) - for resampled in resampler.resample(None): - frames.append(resampled.to_ndarray().flatten()) - return np.concatenate(frames) if frames else np.array([], dtype=np.float32) - - @classmethod - def encode_pcm( - cls, - samples: np.ndarray, - av_codec_name: str, - sample_rate_hz: int, - ) -> bytes: - """Encode float32 mono PCM to raw codec bytes via PyAV. + Uses linear interpolation via [`numpy.interp`][]. Args: - samples: Float32 mono PCM array in the range `[-1, 1]`. - av_codec_name: PyAV codec name (e.g. `"g722"` or `"libopus"`). - sample_rate_hz: Sample rate of *samples* in Hz. + audio: Float32 mono PCM array. + source_rate_hz: Sample rate of *audio* in Hz. + destination_rate_hz: Target sample rate in Hz. Returns: - Encoded audio bytes. + Resampled float32 array at *destination_rate_hz* Hz, or *audio* + unchanged when both rates are equal. """ - codec: av.AudioCodecContext = cast( - av.AudioCodecContext, av.CodecContext.create(av_codec_name, "w") - ) - codec.sample_rate = sample_rate_hz - codec.format = av.AudioFormat("s16") - codec.layout = av.AudioLayout("mono") - codec.open() - pcm = np.clip(np.round(samples * 32768.0), -32768, 32767).astype(np.int16) - frame = av.AudioFrame.from_ndarray( - pcm[np.newaxis, :], format="s16", layout="mono" - ) - frame.sample_rate = sample_rate_hz - frame.pts = 0 - return b"".join( - bytes(packet) - for segment in (codec.encode(frame), codec.encode(None)) - for packet in segment - ) + if source_rate_hz == destination_rate_hz: + return audio + n_out = round(len(audio) * destination_rate_hz / source_rate_hz) + return np.interp( + np.linspace(0, len(audio) - 1, n_out), + np.arange(len(audio)), + audio, + ).astype(np.float32) @classmethod def to_payload_format(cls) -> RTPPayloadFormat: @@ -177,8 +132,7 @@ def decode( ) -> np.ndarray: """Decode an RTP payload to float32 mono PCM. - Override in subclasses to wrap the payload in a container format - (e.g. Ogg for Opus) or select a codec-specific PyAV format string. + Override in subclasses to implement codec-specific decoding. Args: payload: Raw RTP payload bytes. diff --git a/voip/codecs/g722.py b/voip/codecs/g722.py index ccdeb05..a6f6ddf 100644 --- a/voip/codecs/g722.py +++ b/voip/codecs/g722.py @@ -2,6 +2,8 @@ The [`G722`][voip.codecs.g722.G722] class handles the RFC 3551 clock-rate quirk: SDP advertises 8 000 Hz but the actual audio runs at 16 000 Hz. + +Requires the ``pyav`` extra: ``pip install voip[pyav]``. """ from __future__ import annotations @@ -11,12 +13,12 @@ import numpy as np -from voip.codecs.base import RTPCodec +from voip.codecs.av import PyAVCodec __all__ = ["G722"] -class G722(RTPCodec): +class G722(PyAVCodec): """G.722 wideband audio codec ([RFC 3551 §4.5.2][]). G.722 is an ITU-T ADPCM wideband codec. Despite encoding audio at @@ -62,17 +64,6 @@ def encode(cls, samples: np.ndarray) -> bytes: @classmethod def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: - """Packetize a G.722 audio buffer into 160-byte RTP payloads. - - Encodes the entire *audio* buffer at once so that the ADPCM predictor - state is preserved across packet boundaries. - - Args: - audio: Float32 PCM samples at 16 000 Hz. - - Yields: - 160-byte G.722 encoded RTP payloads. - """ encoded = cls.encode(audio) # G.722 2:1 sample-to-byte ratio: frame_size (320) samples → 160 bytes. payload_size = cls.frame_size // 2 diff --git a/voip/codecs/opus.py b/voip/codecs/opus.py index 30ce95f..a8ef9ff 100644 --- a/voip/codecs/opus.py +++ b/voip/codecs/opus.py @@ -4,6 +4,8 @@ minimal [Ogg][] container before passing them to PyAV for decoding, and encodes float32 PCM via `libopus`. +Requires the ``pyav`` extra: ``pip install voip[pyav]``. + [Ogg]: https://wiki.xiph.org/Ogg """ @@ -15,12 +17,12 @@ import numpy as np -from voip.codecs.base import RTPCodec +from voip.codecs.av import PyAVCodec __all__ = ["Opus"] -class Opus(RTPCodec): +class Opus(PyAVCodec): """Opus audio codec ([RFC 7587][]). Opus is a highly flexible codec for interactive real-time speech and audio @@ -43,7 +45,7 @@ class Opus(RTPCodec): channels: ClassVar[int] = 2 @staticmethod - def ogg_crc32(data: bytes) -> int: + def _ogg_crc32(data: bytes) -> int: """Compute an Ogg CRC32 checksum (polynomial 0x04C11DB7). Args: @@ -60,7 +62,7 @@ def ogg_crc32(data: bytes) -> int: return crc & 0xFFFFFFFF @classmethod - def ogg_page( + def _ogg_page( cls, header_type: int, granule_position: int, @@ -99,10 +101,10 @@ def ogg_page( len(lacing), ) + bytes(lacing) page = header + b"".join(packets) - return page[:22] + struct.pack(" bytes: + def _ogg_container(cls, packet: bytes) -> bytes: """Wrap a raw Opus RTP payload in a minimal Ogg Opus container. Produces a three-page Ogg stream: BOS (OpusHead), comment @@ -134,9 +136,9 @@ def ogg_container(cls, packet: bytes) -> bytes: ) return b"".join( [ - cls.ogg_page(0x02, 0, serial_number, 0, [opus_head]), # BOS - cls.ogg_page(0x00, 0, serial_number, 1, [opus_tags]), - cls.ogg_page(0x04, 0, serial_number, 2, [packet]), + cls._ogg_page(0x02, 0, serial_number, 0, [opus_head]), # BOS + cls._ogg_page(0x00, 0, serial_number, 1, [opus_tags]), + cls._ogg_page(0x04, 0, serial_number, 2, [packet]), ] ) @@ -148,7 +150,7 @@ def decode( *, input_rate_hz: int | None = None, ) -> np.ndarray: - return cls.decode_pcm(cls.ogg_container(payload), "ogg", output_rate_hz) + return cls.decode_pcm(cls._ogg_container(payload), "ogg", output_rate_hz) @classmethod def encode(cls, samples: np.ndarray) -> bytes: diff --git a/voip/codecs/pcma.py b/voip/codecs/pcma.py index 963fbe9..f006b40 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 A-law RTP payloads via -PyAV and encodes float32 PCM using a pure-NumPy implementation of -ITU-T G.711 A-law companding. +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. """ from __future__ import annotations @@ -15,6 +15,10 @@ __all__ = ["PCMA"] +_A_LAW: float = 87.6 +_LN_A: float = float(np.log(_A_LAW)) +_COMPRESS_SCALE: float = 1.0 + _LN_A + class PCMA(RTPCodec): """G.711 A-law codec ([RFC 3551 §4.5.14][]). @@ -22,6 +26,10 @@ 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`][voip.codecs.pcma.PCMA.encode] and + [`decode`][voip.codecs.pcma.PCMA.decode] are pure-NumPy and require no + PyAV dependency. + [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ @@ -41,24 +49,27 @@ def decode( *, input_rate_hz: int | None = None, ) -> np.ndarray: - return cls.decode_pcm( - payload, - "alaw", - output_rate_hz, - input_rate_hz=input_rate_hz - if input_rate_hz is not None - else cls.sample_rate_hz, + 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 ) @classmethod def encode(cls, samples: np.ndarray) -> bytes: - a_law = 87.6 # G.711 A-law compression parameter pcm = np.clip(np.abs(samples), 0, 1.0) - low = pcm < (1.0 / a_law) + low = pcm < (1.0 / _A_LAW) compressed = np.where( low, - a_law * pcm / (1.0 + np.log(a_law)), - (1.0 + np.log(np.maximum(a_law * pcm, 1e-10))) / (1.0 + np.log(a_law)), + _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 ) quantized = np.clip(np.round(compressed * 127), 0, 127).astype(np.uint8) diff --git a/voip/codecs/pcmu.py b/voip/codecs/pcmu.py index 620cd74..c4b96a5 100644 --- a/voip/codecs/pcmu.py +++ b/voip/codecs/pcmu.py @@ -1,8 +1,8 @@ """PCMU (G.711 mu-law) codec implementation for RTP audio streams (RFC 3551). -The [`PCMU`][voip.codecs.pcmu.PCMU] class decodes mu-law RTP payloads via -PyAV and encodes float32 PCM using a pure-NumPy implementation of -ITU-T G.711 mu-law companding. +The [`PCMU`][voip.codecs.pcmu.PCMU] class decodes and encodes mu-law RTP +payloads using a pure-NumPy implementation of ITU-T G.711 mu-law companding. +No PyAV dependency is required. """ from __future__ import annotations @@ -15,6 +15,9 @@ __all__ = ["PCMU"] +_MU_LAW_BIAS: int = 0x84 # 132: G.711 mu-law bias constant +_MU_LAW_CLIP: int = 32635 # maximum biased magnitude (14-bit saturate) + class PCMU(RTPCodec): """G.711 mu-law codec ([RFC 3551 §4.5.14][]). @@ -22,6 +25,10 @@ class PCMU(RTPCodec): PCMU is the ITU-T G.711 mu-law logarithmic companding codec for PSTN telephony, standardised in RFC 3551 with static payload type 0. + Both [`encode`][voip.codecs.pcmu.PCMU.encode] and + [`decode`][voip.codecs.pcmu.PCMU.decode] are pure-NumPy and require no + PyAV dependency. + [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ @@ -41,22 +48,22 @@ def decode( *, input_rate_hz: int | None = None, ) -> np.ndarray: - return cls.decode_pcm( - payload, - "mulaw", - output_rate_hz, - input_rate_hz=input_rate_hz - if input_rate_hz is not None - else cls.sample_rate_hz, + raw = (~np.frombuffer(payload, dtype=np.uint8)).astype(np.uint8) + 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) + return cls.resample( + (sign * linear).astype(np.float32), cls.sample_rate_hz, output_rate_hz ) @classmethod def encode(cls, samples: np.ndarray) -> bytes: - BIAS = 0x84 # 132: G.711 mu-law bias constant - CLIP = 32635 # maximum biased magnitude (14-bit saturate) pcm = np.clip(np.round(samples * 32768.0), -32768, 32767).astype(np.int32) sign = np.where(pcm >= 0, 0x80, 0x00).astype(np.uint8) - biased = np.minimum(np.abs(pcm) + BIAS, CLIP) + biased = np.minimum(np.abs(pcm) + _MU_LAW_BIAS, _MU_LAW_CLIP) exp = np.clip( np.floor(np.log2(np.maximum(biased, 1))).astype(np.int32) - 7, 0, 7 ) From 8619c6d48a7aceb366bdefa8eb6b6aee93666b2d Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 22:47:47 +0100 Subject: [PATCH 4/9] Update the docs --- docs/codecs.md | 34 ++++++++++++++++++---------------- docs/rtp.md | 4 ++++ tests/codecs/test_opus.py | 24 ++++++++++++------------ voip/codecs/g722.py | 4 ++-- voip/codecs/pcma.py | 4 +--- voip/codecs/pcmu.py | 4 +--- 6 files changed, 38 insertions(+), 36 deletions(-) diff --git a/docs/codecs.md b/docs/codecs.md index 26ba115..ec99e09 100644 --- a/docs/codecs.md +++ b/docs/codecs.md @@ -4,10 +4,10 @@ VoIP ships two tiers of audio codecs: -| Extra required | Codecs available | -| ------------------------- | -------------------------------------- | -| `numpy` | PCMA (G.711 A-law), PCMU (G.711 µ-law) | -| `pyav` (includes `numpy`) | + G.722, Opus | +| 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: @@ -21,30 +21,32 @@ Install the full tier for wideband / Opus support via [FFmpeg]: pip install voip[hd-audio] ``` -## Base classes - -::: voip.codecs.base - -::: voip.codecs.av - -## Pure-NumPy codecs +## SD audio These codecs work without PyAV and require only `numpy`. -::: voip.codecs.pcma +::: voip.codecs.pcma.PCMA -::: voip.codecs.pcmu +::: voip.codecs.pcmu.PCMU -## PyAV codecs +## HD audio These codecs require the `pyav` extra (`pip install voip[pyav]`). -::: voip.codecs.g722 +::: voip.codecs.g722.G722 -::: voip.codecs.opus +::: 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/ diff --git a/docs/rtp.md b/docs/rtp.md index 6946aa4..ec3f42e 100644 --- a/docs/rtp.md +++ b/docs/rtp.md @@ -5,3 +5,7 @@ ## Encryption ::: voip.srtp + +## NAT Traversal + +::: voip.stun diff --git a/tests/codecs/test_opus.py b/tests/codecs/test_opus.py index d6b8105..2b0a358 100644 --- a/tests/codecs/test_opus.py +++ b/tests/codecs/test_opus.py @@ -14,28 +14,28 @@ class TestOggCRC32: def test_ogg_crc32__empty_bytes(self): - """ogg_crc32 of empty bytes is zero.""" + """_ogg_crc32 of empty bytes is zero.""" assert Opus._ogg_crc32(b"") == 0 def test_ogg_crc32__known_value(self): - """ogg_crc32 produces a deterministic 32-bit value.""" + """_ogg_crc32 produces a deterministic 32-bit value.""" crc = Opus._ogg_crc32(b"OggS") assert 0 <= crc <= 0xFFFFFFFF class TestOggPage: def test_ogg_page__starts_with_capture_pattern(self): - """ogg_page output starts with the Ogg capture pattern 'OggS'.""" + """_ogg_page output starts with the Ogg capture pattern 'OggS'.""" page = Opus._ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) assert page[:4] == b"OggS" def test_ogg_page__contains_packet_data(self): - """ogg_page embeds the provided packet bytes.""" + """_ogg_page embeds the provided packet bytes.""" page = Opus._ogg_page(0x02, 0, 0, 0, [b"payload"]) assert b"payload" in page def test_ogg_page__large_packet_uses_255_lacing(self): - """ogg_page correctly laces a packet exceeding 254 bytes.""" + """_ogg_page correctly laces a packet exceeding 254 bytes.""" page = Opus._ogg_page(0x00, 0, 0, 0, [b"x" * 256]) assert page[:4] == b"OggS" assert len(page) > 256 @@ -43,35 +43,35 @@ def test_ogg_page__large_packet_uses_255_lacing(self): class TestOggContainer: def test_ogg_container__starts_with_ogg_magic(self): - """ogg_container output starts with the Ogg capture pattern 'OggS'.""" + """_ogg_container output starts with the Ogg capture pattern 'OggS'.""" assert Opus._ogg_container(b"packet").startswith(b"OggS") def test_ogg_container__contains_opus_head(self): - """ogg_container includes the OpusHead identification header.""" + """_ogg_container includes the OpusHead identification header.""" assert b"OpusHead" in Opus._ogg_container(b"packet") def test_ogg_container__contains_opus_tags(self): - """ogg_container includes the OpusTags comment header.""" + """_ogg_container includes the OpusTags comment header.""" assert b"OpusTags" in Opus._ogg_container(b"packet") def test_ogg_container__non_empty_for_single_packet(self): - """ogg_container produces a non-empty Ogg container for a single Opus packet.""" + """_ogg_container produces a non-empty Ogg container for a single Opus packet.""" assert len(Opus._ogg_container(b"x" * 100)) > 100 def test_ogg_container__empty_payload(self): - """ogg_container produces a valid Ogg container even for empty payload.""" + """_ogg_container produces a valid Ogg container even for empty payload.""" result = Opus._ogg_container(b"") assert b"OggS" in result def test_ogg_container__produces_three_pages(self): - """ogg_container produces exactly three Ogg pages: BOS, tags, and data.""" + """_ogg_container produces exactly three Ogg pages: BOS, tags, and data.""" result = Opus._ogg_container(b"x" * 10) assert result.count(b"OggS") == 3 class TestOpusDecode: def test_decode__wraps_in_ogg_format(self): - """Decode passes the payload through ogg_container before calling decode_pcm.""" + """Decode passes the payload through _ogg_container before calling decode_pcm.""" with patch.object( Opus, "decode_pcm", return_value=np.zeros(16000, dtype=np.float32) ) as mock_decode_pcm: diff --git a/voip/codecs/g722.py b/voip/codecs/g722.py index a6f6ddf..9c79263 100644 --- a/voip/codecs/g722.py +++ b/voip/codecs/g722.py @@ -25,8 +25,8 @@ class G722(PyAVCodec): 16 000 Hz, the RTP timestamp clock runs at 8 000 Hz per RFC 3551 — a well-known quirk of the original specification. - [`packetize`][voip.codecs.g722.G722.packetize] encodes the entire buffer - at once to preserve the ADPCM predictor state across packet boundaries. + The entire buffer is encoded at once to preserve the ADPCM predictor + state across packet boundaries. [RFC 3551 §4.5.2]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.2 """ diff --git a/voip/codecs/pcma.py b/voip/codecs/pcma.py index f006b40..a8c683f 100644 --- a/voip/codecs/pcma.py +++ b/voip/codecs/pcma.py @@ -26,9 +26,7 @@ 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`][voip.codecs.pcma.PCMA.encode] and - [`decode`][voip.codecs.pcma.PCMA.decode] are pure-NumPy and require no - PyAV dependency. + Both encode and decode are pure-NumPy and require no PyAV dependency. [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ diff --git a/voip/codecs/pcmu.py b/voip/codecs/pcmu.py index c4b96a5..5709af1 100644 --- a/voip/codecs/pcmu.py +++ b/voip/codecs/pcmu.py @@ -25,9 +25,7 @@ class PCMU(RTPCodec): PCMU is the ITU-T G.711 mu-law logarithmic companding codec for PSTN telephony, standardised in RFC 3551 with static payload type 0. - Both [`encode`][voip.codecs.pcmu.PCMU.encode] and - [`decode`][voip.codecs.pcmu.PCMU.decode] are pure-NumPy and require no - PyAV dependency. + Both encode and decode are pure-NumPy and require no PyAV dependency. [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ From 5ebb7f69016c2619ef80c084976793b6fa7ba5e1 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 22:53:25 +0100 Subject: [PATCH 5/9] Fix tests --- tests/codecs/test_base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/codecs/test_base.py b/tests/codecs/test_base.py index 3f699f0..40776b0 100644 --- a/tests/codecs/test_base.py +++ b/tests/codecs/test_base.py @@ -49,9 +49,9 @@ 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 - pytest.importorskip("av") result = G722.to_payload_format() assert result.sample_rate == G722.rtp_clock_rate_hz # 8000, not 16000 From da8e446e37e2ec89351471c33c48a894eab190b2 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 23:02:30 +0100 Subject: [PATCH 6/9] Fix tests --- voip/audio.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/voip/audio.py b/voip/audio.py index 5abe17f..a6eb8c3 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -16,17 +16,15 @@ import json import logging import secrets -import typing from typing import ClassVar +import numpy as np + import voip.codecs as codecs from voip.codecs import RTPCodec from voip.rtp import RTPCall, RTPPacket from voip.sdp.types import MediaDescription -if typing.TYPE_CHECKING: - import numpy as np - __all__ = ["AudioCall", "EchoCall", "VoiceActivityCall"] From c45ca709065ab1bbbd64875ec0dbcd8b2dc2f7a0 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 15 Mar 2026 23:18:34 +0100 Subject: [PATCH 7/9] Drop redundant type imports --- voip/ai.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/voip/ai.py b/voip/ai.py index 789470e..86766a9 100644 --- a/voip/ai.py +++ b/voip/ai.py @@ -12,7 +12,7 @@ import asyncio import dataclasses import logging -from typing import Any, ClassVar +from typing import Any import numpy as np import ollama @@ -20,11 +20,6 @@ from pocket_tts import TTSModel from voip.audio import VoiceActivityCall -from voip.codecs import RTPCodec -from voip.codecs.g722 import G722 -from voip.codecs.opus import Opus -from voip.codecs.pcma import PCMA -from voip.codecs.pcmu import PCMU __all__ = ["TranscribeCall", "AgentCall"] @@ -159,8 +154,6 @@ class AgentCall(TranscribeCall): "You are a person on a phone call. " "Keep your answers very brief and conversational." ) - #: Preferred codecs in priority order (highest first). - PREFERRED_CODECS: ClassVar[list[type[RTPCodec]]] = [Opus, G722, PCMU, PCMA] #: Ollama model name for generating replies. ollama_model: str = dataclasses.field(default="llama3") From 49a7158970dcf428be63b2fc015dde7ac8d62107 Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Mon, 16 Mar 2026 00:36:47 +0100 Subject: [PATCH 8/9] Fix G.711 codec algorithms, resample edge cases, and conditional `__all__` (#41) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses several correctness issues in the G.711 codec layer introduced during the codec base-class refactor: wrong µ-law/A-law decode formulas, `input_rate_hz` ignored as source rate, `__all__` exporting undefined names without PyAV, and `resample()` crashing on edge inputs. ## PCMU (µ-law) **Decode formula**: Replaced `(mantissa | 0x10) << (exp + 3)` with the canonical ITU-T G.711 expansion: ```python magnitude = (((mantissa << 3) + _MU_LAW_BIAS) << exp) - _MU_LAW_BIAS ``` Silence codeword `0x7F` now decodes to exactly `0.0` instead of a small non-zero value. ## PCMA (A-law) **Segmented algorithm**: Replaced the continuous companding approximation (`log`/`exp`) with the standard ITU-T G.711 segmented codec (Sun g711.c, 16-bit scale). Codeword `0xAA` now decodes to exactly `0.984375` per spec, enabling bit-exact interop with real RTP streams. **Vectorized segment lookup**: Segment index determination uses `np.searchsorted` instead of an 8-step Python loop. ## Both PCMA and PCMU **`input_rate_hz` as source rate**: `decode()` now uses `input_rate_hz` (when provided) as the resampling source rate, correctly handling non-standard G.711 clock rates advertised in SDP (e.g. `PCMA/16000`). ## `voip/codecs/__init__.py` **Conditional `__all__`**: `G722`, `Opus`, `PyAVCodec` are now appended to `__all__` only when the `try` import succeeds, preventing `from voip.codecs import *` from raising `AttributeError` in environments without PyAV. ## `RTPCodec.resample()` **Edge cases**: Returns an empty `float32` array for empty input; clamps `n_out` to a minimum of `1` for non-empty input, avoiding `numpy.interp` failures during aggressive downsampling. ## Tests - Golden codeword assertions: `PCMU 0x7F → 0.0`, `PCMA 0xAA → 0.984375`, `PCMA 1.0 → 0xAA`, `PCMA 0.0 → 0xD5` - `input_rate_hz`-as-source-rate tests for both codecs - `resample()` empty-input and single-sample heavy-downsample edge cases - `__all__` conditional test (no-PyAV scenario) --- 📱 Kick off Copilot coding agent tasks wherever you are with [GitHub Mobile](https://gh.io/cca-mobile-docs), available on iOS and Android. --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: codingjoe <1772890+codingjoe@users.noreply.github.com> Co-authored-by: Johannes Maron Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- pyproject.toml | 2 +- tests/codecs/test_base.py | 15 +++++--- tests/codecs/test_codecs.py | 37 ++++++++++++------- tests/codecs/test_pcm.py | 50 ++++++++++++++++++++------ voip/codecs/__init__.py | 3 +- voip/codecs/base.py | 4 ++- voip/codecs/pcma.py | 72 ++++++++++++++++++++++--------------- voip/codecs/pcmu.py | 11 +++--- 8 files changed, 133 insertions(+), 61 deletions(-) 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 From b4a7fa0910160b1dd96e19da76abaa499697e608 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Mon, 16 Mar 2026 09:44:01 +0100 Subject: [PATCH 9/9] fix docs --- voip/codecs/base.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/voip/codecs/base.py b/voip/codecs/base.py index 69915ba..e4e0ed0 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -8,7 +8,7 @@ [`decode_pcm`][voip.codecs.av.PyAVCodec.decode_pcm] and [`encode_pcm`][voip.codecs.av.PyAVCodec.encode_pcm]. -Pure-NumPy codecs ([`PCMA`][voip.codecs.PCMA], [`PCMU`][voip.codecs.PCMU]) +Pure-NumPy codecs ([`PCMA`][voip.codecs.pcma.PCMA], [`PCMU`][voip.codecs.pcmu.PCMU]) inherit directly from `RTPCodec` and require no PyAV dependency. [PyAV]: https://pyav.basswood-io.com/ @@ -30,8 +30,8 @@ class RTPCodec: """Base class for RTP audio codecs. Concrete implementations: [`Opus`][voip.codecs.Opus], - [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.PCMA], - [`PCMU`][voip.codecs.PCMU]. + [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.pcma.PCMA], + [`PCMU`][voip.codecs.pcmu.PCMU]. All codec implementations are stateless: every method is a classmethod or staticmethod and codecs are referenced as `type[RTPCodec]`, never