diff --git a/docs/codecs.md b/docs/codecs.md index ec99e09..34b5c64 100644 --- a/docs/codecs.md +++ b/docs/codecs.md @@ -31,10 +31,12 @@ These codecs work without PyAV and require only `numpy`. ## HD audio -These codecs require the `pyav` extra (`pip install voip[pyav]`). +These codecs require the `hd-audio` extra (`pip install voip[hd-audio]`). ::: voip.codecs.g722.G722 +::: voip.codecs.g722.G722Decoder + ::: voip.codecs.opus.Opus ## Registry @@ -45,6 +47,10 @@ These codecs require the `pyav` extra (`pip install voip[pyav]`). ::: voip.codecs.base.RTPCodec +::: voip.codecs.base.PayloadDecoder + +::: voip.codecs.base.PerPacketDecoder + ::: voip.codecs.av.PyAVCodec [ffmpeg]: https://ffmpeg.org/ diff --git a/tests/codecs/test_base.py b/tests/codecs/test_base.py index ef743a8..57791f4 100644 --- a/tests/codecs/test_base.py +++ b/tests/codecs/test_base.py @@ -83,3 +83,40 @@ def test_encode__raises_not_implemented(self): """RTPCodec.encode raises NotImplementedError.""" with pytest.raises(NotImplementedError): RTPCodec.encode(np.zeros(160, dtype=np.float32)) + + +class TestCreateDecoder: + def test_create_decoder__returns_per_packet_decoder(self): + """RTPCodec.create_decoder returns a PerPacketDecoder for stateless codecs.""" + from voip.codecs.base import PerPacketDecoder # noqa: PLC0415 + + decoder = PCMA.create_decoder(16000) + assert isinstance(decoder, PerPacketDecoder) + + def test_create_decoder__stores_codec_and_rates(self): + """PerPacketDecoder holds the codec class and both rate parameters.""" + from voip.codecs.base import PerPacketDecoder # noqa: PLC0415 + + decoder = PCMA.create_decoder(16000, input_rate_hz=8000) + assert isinstance(decoder, PerPacketDecoder) + assert decoder.codec is PCMA + assert decoder.output_rate_hz == 16000 + assert decoder.input_rate_hz == 8000 + + def test_create_decoder__input_rate_hz_defaults_to_none(self): + """input_rate_hz defaults to None when not specified.""" + from voip.codecs.base import PerPacketDecoder # noqa: PLC0415 + + decoder = PCMA.create_decoder(16000) + assert isinstance(decoder, PerPacketDecoder) + assert decoder.input_rate_hz is None + + def test_per_packet_decoder__delegates_to_codec_decode(self): + """PerPacketDecoder.decode calls codec.decode with stored rates.""" + with patch.object( + PCMA, "decode", return_value=np.zeros(160, dtype=np.float32) + ) as mock_decode: + decoder = PCMA.create_decoder(16000, input_rate_hz=8000) + result = decoder.decode(b"payload") + mock_decode.assert_called_once_with(b"payload", 16000, input_rate_hz=8000) + assert result.dtype == np.float32 diff --git a/tests/codecs/test_g722.py b/tests/codecs/test_g722.py index 1e77ba9..8a3e4dc 100644 --- a/tests/codecs/test_g722.py +++ b/tests/codecs/test_g722.py @@ -111,3 +111,168 @@ def test_packetize__yields_160_byte_chunks(self): packets = list(G722.packetize(audio)) assert len(packets) == 1 assert len(packets[0]) == payload_size + + +class TestG722CreateDecoder: + def test_create_decoder__returns_g722_decoder(self): + """create_decoder returns a G722Decoder instance.""" + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + decoder = G722.create_decoder(16000) + assert isinstance(decoder, G722Decoder) + + def test_create_decoder__output_rate_hz_set(self): + """create_decoder stores the output_rate_hz on the returned decoder.""" + decoder = G722.create_decoder(16000) + assert decoder.output_rate_hz == 16000 + + def test_create_decoder__ignores_input_rate_hz(self): + """create_decoder ignores input_rate_hz (G.722 always decodes at 16 kHz).""" + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + decoder = G722.create_decoder(16000, input_rate_hz=8000) + assert isinstance(decoder, G722Decoder) + assert decoder.output_rate_hz == 16000 + + +class TestG722Decoder: + def _make_encoded_packets(self, packet_count: int = 3) -> list[bytes]: + """Encode *packet_count* 20 ms G.722 packets from a continuous sine wave.""" + import av # noqa: PLC0415 + + encoder = av.CodecContext.create("g722", "w") + encoder.sample_rate = 16000 + encoder.format = av.AudioFormat("s16") + encoder.layout = av.AudioLayout("mono") + encoder.open() + t = np.linspace( + 0, packet_count * 0.02, packet_count * G722.frame_size, endpoint=False + ) + signal = (np.sin(2 * np.pi * 440 * t) * 0.5).astype(np.float32) + pcm = np.clip(np.round(signal * 32768.0), -32768, 32767).astype(np.int16) + frame = av.AudioFrame.from_ndarray( + pcm[np.newaxis, :], format="s16", layout="mono" + ) + frame.sample_rate = 16000 + frame.pts = 0 + return [bytes(p) for p in encoder.encode(frame)] + + def test_decoder__initialises_codec_context(self): + """G722Decoder creates a persistent PyAV codec context on init.""" + import av # noqa: PLC0415 + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + decoder = G722Decoder(16000) + assert isinstance(decoder.codec_context, av.CodecContext) + + def test_decoder__initialises_resampler(self): + """G722Decoder creates an AudioResampler targeting output_rate_hz.""" + import av # noqa: PLC0415 + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + decoder = G722Decoder(8000) + assert isinstance(decoder.resampler, av.audio.resampler.AudioResampler) + + def test_decoder__decode_returns_float32(self): + """Decode produces a float32 array for a real G.722 encoded packet.""" + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + packet = self._make_encoded_packets(1)[0] + decoder = G722Decoder(16000) + result = decoder.decode(packet) + assert result.dtype == np.float32 + + def test_decoder__decode_yields_correct_sample_count(self): + """Decode returns 320 float32 samples for a 160-byte G.722 packet at 16 kHz.""" + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + packet = self._make_encoded_packets(1)[0] + assert len(packet) == 160 + decoder = G722Decoder(16000) + result = decoder.decode(packet) + assert len(result) == G722.frame_size # 320 + + def test_decoder__preserves_adpcm_state_across_packets(self): + """Stateful G722Decoder matches decoding all bytes together (reference). + + A per-packet stateless decoder resets the ADPCM predictor and produces + near-silent output for packets 1+. A stateful G722Decoder feeds each + packet into the same persistent context and matches the reference (all + packets decoded together in one container). + """ + import io # noqa: PLC0415 + + import av # noqa: PLC0415 + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + packets = self._make_encoded_packets(3) + + # Reference: decode all bytes in one container (correct, stateful). + resampler = av.audio.resampler.AudioResampler( + format="fltp", layout="mono", rate=16000 + ) + ref_frames: list[np.ndarray] = [] + with av.open( + io.BytesIO(b"".join(packets)), + mode="r", + format="g722", + options={"sample_rate": "8000"}, + ) as container: + for f in container.decode(audio=0): + for rs in resampler.resample(f): + ref_frames.append(rs.to_ndarray().flatten()) + reference = np.concatenate(ref_frames) + + # Stateful G722Decoder: should match the reference exactly. + stateful_decoder = G722Decoder(16000) + stateful_parts = [stateful_decoder.decode(p) for p in packets] + stateful = np.concatenate(stateful_parts) + + assert len(stateful) == len(reference) + assert np.allclose(stateful, reference, atol=1e-5), ( + "Stateful G722Decoder output differs from reference: " + "ADPCM state may not be preserved across packets" + ) + + def test_stateless_decode__diverges_after_first_packet(self): + """Per-packet stateless decoding diverges from reference for packets 1+. + + This test documents the original bug: resetting ADPCM state each + packet causes the decoded signal to be near-silent for all but the + first packet, making the echo 'too short' and 'robotic'. + """ + import io # noqa: PLC0415 + + import av # noqa: PLC0415 + + packets = self._make_encoded_packets(3) + + # Reference: decode all bytes together. + resampler = av.audio.resampler.AudioResampler( + format="fltp", layout="mono", rate=16000 + ) + ref_frames: list[np.ndarray] = [] + with av.open( + io.BytesIO(b"".join(packets)), + mode="r", + format="g722", + options={"sample_rate": "8000"}, + ) as container: + for f in container.decode(audio=0): + for rs in resampler.resample(f): + ref_frames.append(rs.to_ndarray().flatten()) + reference = np.concatenate(ref_frames) + + # Stateless (original buggy behaviour): fresh context per packet. + stateless_parts = [G722.decode(p, 16000) for p in packets] + + # Packet 0 is identical (both start from zero state). + assert np.allclose(stateless_parts[0], reference[: G722.frame_size], atol=1e-5) + # Packets 1+ diverge: stateless is near-silent, reference has full signal. + for i, part in enumerate(stateless_parts[1:], start=1): + ref_segment = reference[i * G722.frame_size : (i + 1) * G722.frame_size] + mse = float(np.mean((part - ref_segment) ** 2)) + assert mse > 0.01, ( # near-silence vs full-amplitude signal + f"Packet {i}: expected stateless decoder to diverge from reference " + f"(ADPCM state reset), but MSE={mse:.6f} is too low" + ) diff --git a/tests/test_audio.py b/tests/test_audio.py index 50c1b27..a3f5967 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -311,7 +311,7 @@ class TestDecodePayload: """Tests for AudioCall.decode_payload.""" def test_decode_payload__delegates_to_codec(self): - """decode_payload calls self.codec.decode with output and input rates.""" + """decode_payload routes through PerPacketDecoder which calls codec.decode.""" call = make_audio_call(media=PCMA_MEDIA) with patch.object( PCMA, "decode", return_value=np.zeros(16000, dtype=np.float32) @@ -353,6 +353,20 @@ def test_decode_payload__raises_for_unsupported_codec(self): with pytest.raises(NotImplementedError, match="Unsupported codec"): make_audio_call(media=media) + def test_decode_payload__g722_uses_stateful_decoder(self): + """G.722 AudioCall uses a G722Decoder that preserves ADPCM state.""" + from voip.codecs.g722 import G722Decoder # noqa: PLC0415 + + call = make_audio_call(media=G722_MEDIA) + assert isinstance(call.payload_decoder, G722Decoder) + + def test_decode_payload__pcma_uses_per_packet_decoder(self): + """PCMA AudioCall uses a PerPacketDecoder (stateless).""" + from voip.codecs.base import PerPacketDecoder # noqa: PLC0415 + + call = make_audio_call(media=PCMA_MEDIA) + assert isinstance(call.payload_decoder, PerPacketDecoder) + class TestAudioCallInit: def test_init__raises_value_error_for_none_encoding_name(self): diff --git a/voip/audio.py b/voip/audio.py index a6eb8c3..c8a9aa6 100644 --- a/voip/audio.py +++ b/voip/audio.py @@ -22,6 +22,7 @@ import voip.codecs as codecs from voip.codecs import RTPCodec +from voip.codecs.base import PayloadDecoder from voip.rtp import RTPCall, RTPPacket from voip.sdp.types import MediaDescription @@ -68,6 +69,10 @@ class is stored on `codec` after `__post_init__` and used for all #: Resolved codec class for this call, set in `__post_init__`. codec: type[RTPCodec] = dataclasses.field(init=False, repr=False) + #: Per-call payload decoder, set in `__post_init__`. + #: Stateful for ADPCM codecs (e.g. G.722), stateless for others. + payload_decoder: PayloadDecoder = dataclasses.field(init=False, repr=False) + #: Outbound RTP sequence counter. rtp_sequence_number: int = dataclasses.field(init=False, repr=False, default=0) #: Outbound RTP timestamp counter. @@ -82,6 +87,9 @@ def __post_init__(self) -> None: if fmt.encoding_name is None: raise ValueError(f"No encoding name for payload type {fmt.payload_type}") self.codec = codecs.get(fmt.encoding_name) + self.payload_decoder = self.codec.create_decoder( + self.RESAMPLING_RATE_HZ, input_rate_hz=self.sample_rate + ) logger.info( json.dumps( { @@ -194,10 +202,11 @@ async def emit_audio(self, packet: RTPPacket) -> None: def decode_payload(self, payload: bytes) -> np.ndarray: """Decode an RTP payload to float32 PCM at `RESAMPLING_RATE_HZ`. - Delegates to the negotiated `codec`, - passing the SDP-negotiated `sample_rate` as the input rate hint so - that non-standard variants (e.g. wideband PCMA at 16 000 Hz) are - handled correctly. + Delegates to `payload_decoder`, which is either a + [`PerPacketDecoder`][voip.codecs.base.PerPacketDecoder] (for stateless + codecs such as PCMA, PCMU, Opus) or a + [`G722Decoder`][voip.codecs.g722.G722Decoder] (for G.722, which + preserves ADPCM predictor state across consecutive packets). Args: payload: Raw RTP payload bytes. @@ -205,9 +214,7 @@ def decode_payload(self, payload: bytes) -> np.ndarray: Returns: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz. """ - return self.codec.decode( - payload, self.RESAMPLING_RATE_HZ, input_rate_hz=self.sample_rate - ) + return self.payload_decoder.decode(payload) def audio_received(self, *, audio: np.ndarray, rms: float) -> None: """Handle decoded audio. Override in subclasses. diff --git a/voip/codecs/base.py b/voip/codecs/base.py index e4e0ed0..10ee694 100644 --- a/voip/codecs/base.py +++ b/voip/codecs/base.py @@ -16,14 +16,37 @@ from __future__ import annotations +import dataclasses from collections.abc import Iterator -from typing import ClassVar +from typing import ClassVar, Protocol import numpy as np from voip.sdp.types import RTPPayloadFormat -__all__ = ["RTPCodec"] +__all__ = ["PayloadDecoder", "PerPacketDecoder", "RTPCodec"] + + +class PayloadDecoder(Protocol): + """Protocol for per-call RTP payload decoders. + + Implementations decode raw RTP payload bytes to float32 mono PCM. + Stateful implementations (e.g. [`G722Decoder`][voip.codecs.g722.G722Decoder]) + preserve codec predictor state across successive + [`decode`][voip.codecs.base.PayloadDecoder.decode] calls within a single + call session. + """ + + def decode(self, payload: bytes) -> np.ndarray: + """Decode one RTP payload to float32 mono PCM. + + Args: + payload: Raw RTP payload bytes for a single packet. + + Returns: + Float32 mono PCM array. + """ + ... class RTPCodec: @@ -33,9 +56,11 @@ class RTPCodec: [`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 - instantiated. + Codec classes are stateless; every method is a classmethod or staticmethod + and codecs are referenced as `type[RTPCodec]`, never instantiated. + Per-call decoder state (required for ADPCM codecs such as G.722) is + managed by [`PayloadDecoder`][voip.codecs.base.PayloadDecoder] instances + returned by [`create_decoder`][voip.codecs.base.RTPCodec.create_decoder]. Concrete subclasses define codec-specific class variables and override [`decode`][voip.codecs.base.RTPCodec.decode], @@ -147,6 +172,28 @@ def decode( """ raise NotImplementedError(f"{cls.__name__} does not implement decode.") + @classmethod + def create_decoder( + cls, output_rate_hz: int, *, input_rate_hz: int | None = None + ) -> PayloadDecoder: + """Create a stateless per-call payload decoder for this codec. + + Override in subclasses that require stateful decoding across RTP + packets (e.g. G.722 ADPCM — see + [`G722.create_decoder`][voip.codecs.g722.G722.create_decoder]). + + Args: + output_rate_hz: Target PCM sample rate in Hz for decoded audio. + input_rate_hz: Input clock rate override, or `None` to use the + codec default. + + Returns: + A [`PayloadDecoder`][voip.codecs.base.PayloadDecoder] that, by + default, is a [`PerPacketDecoder`][voip.codecs.base.PerPacketDecoder] + delegating each call to [`decode`][voip.codecs.base.RTPCodec.decode]. + """ + return PerPacketDecoder(cls, output_rate_hz, input_rate_hz) + @classmethod def encode(cls, samples: np.ndarray) -> bytes: """Encode float32 mono PCM to an RTP payload. @@ -177,3 +224,36 @@ def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: """ for i in range(0, len(audio), cls.frame_size): yield cls.encode(audio[i : i + cls.frame_size]) + + +@dataclasses.dataclass(frozen=True) +class PerPacketDecoder: + """Stateless payload decoder that processes each RTP packet independently. + + Delegate each call to + [`RTPCodec.decode`][voip.codecs.base.RTPCodec.decode], decoding each + payload independently without preserving cross-packet state. Suitable for + stateless codecs such as PCMA, PCMU, and Opus. + + Attributes: + codec: Codec class to delegate decoding to. + output_rate_hz: Target PCM sample rate in Hz. + input_rate_hz: Input clock rate override, or `None` to use the codec default. + """ + + codec: type[RTPCodec] + output_rate_hz: int + input_rate_hz: int | None = None + + def decode(self, payload: bytes) -> np.ndarray: + """Decode one RTP payload to float32 PCM. + + Args: + payload: Raw RTP payload bytes. + + Returns: + Float32 mono PCM array at `output_rate_hz` Hz. + """ + return self.codec.decode( + payload, self.output_rate_hz, input_rate_hz=self.input_rate_hz + ) diff --git a/voip/codecs/g722.py b/voip/codecs/g722.py index 9c79263..deab23e 100644 --- a/voip/codecs/g722.py +++ b/voip/codecs/g722.py @@ -3,19 +3,28 @@ 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]``. +Use [`G722Decoder`][voip.codecs.g722.G722Decoder] (via +[`G722.create_decoder`][voip.codecs.g722.G722.create_decoder]) for per-call +stateful decoding that preserves the ADPCM predictor state across consecutive +RTP packets. + +Requires the ``hd-audio`` extra: ``pip install voip[hd-audio]``. """ from __future__ import annotations +import dataclasses +import typing from collections.abc import Iterator from typing import ClassVar +import av +import av.audio.resampler import numpy as np from voip.codecs.av import PyAVCodec -__all__ = ["G722"] +__all__ = ["G722", "G722Decoder"] class G722(PyAVCodec): @@ -69,3 +78,84 @@ def packetize(cls, audio: np.ndarray) -> Iterator[bytes]: payload_size = cls.frame_size // 2 for i in range(0, len(encoded), payload_size): yield encoded[i : i + payload_size] + + @classmethod + def create_decoder( + cls, output_rate_hz: int, *, input_rate_hz: int | None = None + ) -> G722Decoder: + """Create a stateful per-call G.722 decoder. + + Returns a [`G722Decoder`][voip.codecs.g722.G722Decoder] that preserves + the ADPCM predictor state across consecutive RTP packets. Pass the + returned decoder to + [`AudioCall`][voip.audio.AudioCall] (via the `create_decoder` + factory) to avoid the per-packet state reset that causes robotic + audio artefacts. + + The *input_rate_hz* parameter is accepted for API consistency with + [`RTPCodec.create_decoder`][voip.codecs.base.RTPCodec.create_decoder] + but is not used; G.722 always decodes at 16 000 Hz internally. + + Args: + output_rate_hz: Target PCM sample rate in Hz for decoded audio. + input_rate_hz: Ignored. G.722 always decodes at `sample_rate_hz`. + + Returns: + A new [`G722Decoder`][voip.codecs.g722.G722Decoder] instance. + """ + return G722Decoder(output_rate_hz) + + +@dataclasses.dataclass(slots=True) +class G722Decoder: + """Stateful G.722 decoder that preserves ADPCM predictor state across packets. + + Creates a single persistent + [`av.CodecContext`](https://pyav.basswood-io.com/docs/stable/api/codec.html#av.codec.context.CodecContext) + for the life of the decoder and feeds each incoming RTP packet to the + same context. This eliminates the per-packet predictor reset that causes + robotic artefacts when decoding a G.722 stream with independent codec + contexts. + + Use [`G722.create_decoder`][voip.codecs.g722.G722.create_decoder] rather + than instantiating this class directly. + + Attributes: + output_rate_hz: Target PCM sample rate in Hz for decoded audio. + codec_context: Persistent G.722 decoder context + shared across all [decode][voip.codecs.g722.G722Decoder.decode] + calls on this instance. + resampler: Sampler targeting `output_rate_hz` Hz. + """ + + output_rate_hz: int + codec_context: av.AudioCodecContext = dataclasses.field(init=False, repr=False) + resampler: av.audio.resampler.AudioResampler = dataclasses.field( + init=False, repr=False + ) + + def __post_init__(self) -> None: + self.codec_context = typing.cast( + av.AudioCodecContext, av.CodecContext.create("g722", "r") + ) + self.codec_context.sample_rate = G722.sample_rate_hz + self.codec_context.open() + self.resampler = av.audio.resampler.AudioResampler( + format="fltp", layout="mono", rate=self.output_rate_hz + ) + + def decode(self, payload: bytes) -> np.ndarray: + """Decode one G.722 RTP payload, preserving ADPCM state from prior packets. + + Args: + payload: Raw G.722 RTP payload bytes (160 bytes per 20 ms frame). + + Returns: + Float32 mono PCM array at `output_rate_hz` Hz. + """ + frames = [ + resampled.to_ndarray().flatten() + for frame in self.codec_context.decode(av.Packet(payload)) + for resampled in self.resampler.resample(frame) + ] + return np.concatenate(frames) if frames else np.array([], dtype=np.float32)