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 new file mode 100644 index 0000000..ec99e09 --- /dev/null +++ b/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/ 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/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/pyproject.toml b/pyproject.toml index 163dbf4..ec86dd3 100644 --- a/pyproject.toml +++ b/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] 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..ef743a8 100644 --- a/tests/codecs/test_base.py +++ b/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)) diff --git a/tests/codecs/test_codecs.py b/tests/codecs/test_codecs.py index e00f297..c87e82d 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,57 @@ 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") + + +_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.""" + 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 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 + + 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) + + 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_opus.py b/tests/codecs/test_opus.py index 5dc3a4e..2b0a358 100644 --- a/tests/codecs/test_opus.py +++ b/tests/codecs/test_opus.py @@ -14,64 +14,64 @@ class TestOggCRC32: def test_ogg_crc32__empty_bytes(self): - """ogg_crc32 of empty bytes is zero.""" - assert Opus.ogg_crc32(b"") == 0 + """_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.""" - crc = Opus.ogg_crc32(b"OggS") + """_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'.""" - page = Opus.ogg_page(0x02, 0, 0x12345678, 0, [b"hello"]) + """_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.""" - page = Opus.ogg_page(0x02, 0, 0, 0, [b"payload"]) + """_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.""" - page = Opus.ogg_page(0x00, 0, 0, 0, [b"x" * 256]) + """_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 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") + """_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.""" - assert b"OpusHead" in Opus.ogg_container(b"packet") + """_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.""" - assert b"OpusTags" in Opus.ogg_container(b"packet") + """_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.""" - assert len(Opus.ogg_container(b"x" * 100)) > 100 + """_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.""" - result = Opus.ogg_container(b"") + """_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.""" - result = Opus.ogg_container(b"x" * 10) + """_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/tests/codecs/test_pcm.py b/tests/codecs/test_pcm.py index e45384b..d1657ba 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,45 @@ 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__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)) + # 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.""" @@ -87,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): @@ -115,29 +148,42 @@ 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__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)) + # 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/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/ai.py b/voip/ai.py index c52562b..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 Codec -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[Codec]]] = [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..a6eb8c3 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -21,11 +21,7 @@ import numpy as np import voip.codecs as codecs -from voip.codecs import Codec -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.codecs import RTPCodec from voip.rtp import RTPCall, RTPPacket from voip.sdp.types import MediaDescription @@ -55,7 +51,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[Codec]]] = [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 @@ -64,7 +66,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) @@ -264,7 +266,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 +276,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 61a084d..b37a6a1 100644 --- a/voip/codecs/__init__.py +++ b/voip/codecs/__init__.py @@ -1,126 +1,45 @@ """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 -- [`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 collections.abc import Iterator -from typing import TYPE_CHECKING, ClassVar, Protocol - -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`.""" +from voip.codecs.base import RTPCodec +from voip.codecs.pcma import PCMA +from voip.codecs.pcmu import PCMU - 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__ = ["PCMA", "PCMU", "RTPCodec", "get"] #: Registry mapping lowercase encoding names to codec classes. -REGISTRY: dict[str, type[Codec]] = { - codec.encoding_name: codec for codec in (Opus, G722, PCMA, PCMU) +REGISTRY: dict[str, type[RTPCodec]] = { + 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} + __all__ = [*__all__, "G722", "Opus", "PyAVCodec"] +except ImportError: + pass + -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/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 b665bde..e4e0ed0 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -1,41 +1,60 @@ -"""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]. + +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.PCMA], [`PCMU`][voip.codecs.pcmu.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 decode and encode via PyAV. + """Base class for RTP audio codecs. + + Concrete implementations: [`Opus`][voip.codecs.Opus], + [`G722`][voip.codecs.G722], [`PCMA`][voip.codecs.pcma.PCMA], + [`PCMU`][voip.codecs.pcmu.PCMU]. - 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]. + 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 override + [`decode`][voip.codecs.base.RTPCodec.decode], + [`encode`][voip.codecs.base.RTPCodec.encode], and optionally + [`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. + + Subclasses that require [PyAV][] additionally inherit from + [`PyAVCodec`][voip.codecs.av.PyAVCodec]. + + [PyAV]: https://pyav.basswood-io.com/ """ payload_type: ClassVar[int] @@ -60,80 +79,32 @@ 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. + """Resample *audio* from *source_rate_hz* to *destination_rate_hz*. - 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. + 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 + 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)), + audio, + ).astype(np.float32) @classmethod def to_payload_format(cls) -> RTPPayloadFormat: @@ -163,8 +134,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 9cfdd05..9c79263 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,20 +13,20 @@ 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 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/opus.py b/voip/codecs/opus.py index d80a9d4..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, @@ -68,7 +70,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 +81,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: @@ -101,23 +101,21 @@ 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 (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" @@ -138,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]), ] ) @@ -152,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..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 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 the ITU-T G.711 A-law segmented +companding algorithm. No PyAV dependency is required. """ from __future__ import annotations @@ -15,6 +15,13 @@ __all__ = ["PCMA"] +# 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): """G.711 A-law codec ([RFC 3551 §4.5.14][]). @@ -22,6 +29,9 @@ 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 interoperate bit-exactly with real RTP PCMA + streams. + [RFC 3551 §4.5.14]: https://datatracker.ietf.org/doc/html/rfc3551#section-4.5.14 """ @@ -41,27 +51,42 @@ 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, + # 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: - 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) - 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)), - # 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 620cd74..361f1e8 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,8 @@ 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 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 """ @@ -41,22 +46,25 @@ 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) + # 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), source_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 )