Add support for G.722 audio

jlaine · jlaine · commit d0a57d1a2560 · 2025-05-22T21:44:45.000+02:00
The G.722 codec is "special" because the clock rate is 8kHz even though
the sampling rate is 16kHz.
diff --git a/src/aiortc/codecs/__init__.py b/src/aiortc/codecs/__init__.py
@@ -11,10 +11,16 @@
 )
 from .base import Decoder, Encoder
 from .g711 import PcmaDecoder, PcmaEncoder, PcmuDecoder, PcmuEncoder
+from .g722 import G722Decoder, G722Encoder
 from .h264 import H264Decoder, H264Encoder, h264_depayload
 from .opus import OpusDecoder, OpusEncoder
 from .vpx import Vp8Decoder, Vp8Encoder, vp8_depayload
 
+# The clockrate for G.722 is 8kHz even though the sampling rate is 16kHz.
+# See https://datatracker.ietf.org/doc/html/rfc3551
+G722_CODEC = RTCRtpCodecParameters(
+    mimeType="audio/G722", clockRate=8000, channels=1, payloadType=9
+)
 PCMU_CODEC = RTCRtpCodecParameters(
     mimeType="audio/PCMU", clockRate=8000, channels=1, payloadType=0
 )
@@ -27,6 +33,7 @@
         RTCRtpCodecParameters(
             mimeType="audio/opus", clockRate=48000, channels=2, payloadType=96
         ),
+        G722_CODEC,
         PCMU_CODEC,
         PCMA_CODEC,
     ],
@@ -141,7 +148,9 @@ def get_capabilities(kind: str) -> RTCRtpCapabilities:
 def get_decoder(codec: RTCRtpCodecParameters) -> Decoder:
     mimeType = codec.mimeType.lower()
 
-    if mimeType == "audio/opus":
+    if mimeType == "audio/g722":
+        return G722Decoder()
+    elif mimeType == "audio/opus":
         return OpusDecoder()
     elif mimeType == "audio/pcma":
         return PcmaDecoder()
@@ -158,7 +167,9 @@ def get_decoder(codec: RTCRtpCodecParameters) -> Decoder:
 def get_encoder(codec: RTCRtpCodecParameters) -> Encoder:
     mimeType = codec.mimeType.lower()
 
-    if mimeType == "audio/opus":
+    if mimeType == "audio/g722":
+        return G722Encoder()
+    elif mimeType == "audio/opus":
         return OpusEncoder()
     elif mimeType == "audio/pcma":
         return PcmaEncoder()
diff --git a/src/aiortc/codecs/g722.py b/src/aiortc/codecs/g722.py
@@ -0,0 +1,75 @@
+import fractions
+from typing import cast
+
+from av import AudioFrame, AudioResampler, CodecContext
+from av.frame import Frame
+from av.packet import Packet
+
+from ..jitterbuffer import JitterFrame
+from ..mediastreams import convert_timebase
+from .base import Decoder, Encoder
+
+SAMPLE_RATE = 16000
+SAMPLE_WIDTH = 2
+SAMPLES_PER_FRAME = 320
+CLOCK_BASE = fractions.Fraction(1, 8000)
+TIME_BASE = fractions.Fraction(1, 16000)
+
+
+class G722Decoder(Decoder):
+    def __init__(self) -> None:
+        self.codec = CodecContext.create("g722", "r")
+        self.codec.format = "s16"
+        self.codec.layout = "mono"
+        self.codec.sample_rate = SAMPLE_RATE
+
+    def decode(self, encoded_frame: JitterFrame) -> list[Frame]:
+        packet = Packet(encoded_frame.data)
+        packet.pts = encoded_frame.timestamp
+        packet.time_base = TIME_BASE
+        return cast(list[Frame], self.codec.decode(packet))
+
+
+class G722Encoder(Encoder):
+    def __init__(self) -> None:
+        self.codec = CodecContext.create("g722", "w")
+        self.codec.format = "s16"
+        self.codec.layout = "mono"
+        self.codec.sample_rate = SAMPLE_RATE
+        self.codec.time_base = TIME_BASE
+        self.first_pts = None
+
+        # Create our own resampler to control the frame size.
+        self.resampler = AudioResampler(
+            format="s16",
+            layout="mono",
+            rate=SAMPLE_RATE,
+            frame_size=SAMPLES_PER_FRAME,
+        )
+
+    def encode(
+        self, frame: Frame, force_keyframe: bool = False
+    ) -> tuple[list[bytes], int]:
+        assert isinstance(frame, AudioFrame)
+        assert frame.format.name == "s16"
+        assert frame.layout.name in ["mono", "stereo"]
+
+        # Send frame through resampler and encoder.
+        packets = []
+        for frame in self.resampler.resample(frame):
+            packets += self.codec.encode(frame)
+
+        if packets:
+            # Packets were returned.
+            if self.first_pts is None:
+                self.first_pts = packets[0].pts
+            # Even though the sample rate is 16kHz, the clockrate is defined as 8kHz.
+            timestamp = (packets[0].pts - self.first_pts) // 2
+            return [bytes(p) for p in packets], timestamp
+        else:
+            # No packets were returned due to buffering.
+            return [], None
+
+    def pack(self, packet: Packet) -> tuple[list[bytes], int]:
+        timestamp = convert_timebase(packet.pts, packet.time_base, CLOCK_BASE)
+        return [bytes(packet)], timestamp
diff --git a/tests/codecs.py b/tests/codecs.py
@@ -1,4 +1,5 @@
 import fractions
+from typing import Optional
 
 from aiortc.codecs import depayload, get_decoder, get_encoder
 from aiortc.jitterbuffer import JitterFrame
@@ -20,7 +21,7 @@ def assertAudioFrame(
         pts: int,
         samples: int,
         sample_rate: int,
-        data: bytes,
+        data: Optional[bytes],
     ) -> None:
         assert isinstance(frame, AudioFrame)
         self.assertEqual(frame.format.name, "s16")
@@ -30,8 +31,9 @@ def assertAudioFrame(
         self.assertEqual(frame.sample_rate, sample_rate)
         self.assertEqual(frame.time_base, fractions.Fraction(1, sample_rate))
 
-        plane_data = bytes(frame.planes[0])
-        self.assertEqual(plane_data[: len(data)], data)
+        if data is not None:
+            plane_data = bytes(frame.planes[0])
+            self.assertEqual(plane_data[: len(data)], data)
 
     def create_audio_frame(
         self, samples: int, pts: int, layout: str = "mono", sample_rate: int = 48000
@@ -117,6 +119,7 @@ def roundtrip_audio(
         codec: RTCRtpCodecParameters,
         output_layout: str,
         output_sample_rate: int,
+        output_clock_rate: Optional[int] = None,
         input_layout: str = "mono",
         input_sample_rate: int = 8000,
         drop: list[int] = [],
@@ -131,7 +134,11 @@ def roundtrip_audio(
             layout=input_layout, sample_rate=input_sample_rate, count=10
         )
 
-        output_sample_count = int(output_sample_rate * AUDIO_PTIME)
+        # Usually the clock rate matches the sample rate, but not for G722.
+        if output_clock_rate is not None:
+            output_frame_pts = int(output_clock_rate * AUDIO_PTIME)
+        else:
+            output_frame_pts = int(output_sample_rate * AUDIO_PTIME)
 
         for i, frame in enumerate(input_frames):
             # encode
@@ -151,7 +158,7 @@ def roundtrip_audio(
                 self.assertEqual(frames[0].layout.name, output_layout)
                 self.assertEqual(frames[0].samples, output_sample_rate * AUDIO_PTIME)
                 self.assertEqual(frames[0].sample_rate, output_sample_rate)
-                self.assertEqual(frames[0].pts, i * output_sample_count)
+                self.assertEqual(frames[0].pts, i * output_frame_pts)
                 self.assertEqual(
                     frames[0].time_base, fractions.Fraction(1, output_sample_rate)
                 )
diff --git a/tests/test_g722.py b/tests/test_g722.py
@@ -0,0 +1,106 @@
+import sys
+
+from aiortc.codecs import G722_CODEC, get_decoder, get_encoder
+from aiortc.codecs.g722 import G722Decoder, G722Encoder
+from aiortc.jitterbuffer import JitterFrame
+
+from .codecs import CodecTestCase
+
+# silence
+G722_PAYLOAD = b"\xfa" * 160
+
+
+class G722Test(CodecTestCase):
+    def test_decoder(self) -> None:
+        decoder = get_decoder(G722_CODEC)
+        self.assertIsInstance(decoder, G722Decoder)
+
+        frames = decoder.decode(JitterFrame(data=G722_PAYLOAD, timestamp=0))
+        self.assertEqual(len(frames), 1)
+        frame = frames[0]
+        self.assertAudioFrame(
+            frame,
+            data=None,
+            layout="mono",
+            pts=0,
+            samples=320,
+            sample_rate=16000,
+        )
+
+    def test_encoder_mono_16khz(self) -> None:
+        encoder = get_encoder(G722_CODEC)
+        self.assertIsInstance(encoder, G722Encoder)
+
+        for frame in self.create_audio_frames(
+            layout="mono", sample_rate=16000, count=10
+        ):
+            payloads, timestamp = encoder.encode(frame)
+            self.assertEqual(len(payloads), 1)
+            self.assertEqual(len(payloads[0]), 160)
+            self.assertEqual(timestamp, frame.pts // 2)
+
+    def test_encoder_stereo_16khz(self) -> None:
+        encoder = get_encoder(G722_CODEC)
+        self.assertIsInstance(encoder, G722Encoder)
+
+        for frame in self.create_audio_frames(
+            layout="stereo", sample_rate=16000, count=10
+        ):
+            payloads, timestamp = encoder.encode(frame)
+            self.assertEqual(len(payloads), 1)
+            self.assertEqual(len(payloads[0]), 160)
+            self.assertEqual(timestamp, frame.pts // 2)
+
+    def test_encoder_stereo_48khz(self) -> None:
+        encoder = get_encoder(G722_CODEC)
+        self.assertIsInstance(encoder, G722Encoder)
+
+        output = [
+            encoder.encode(frame)
+            for frame in self.create_audio_frames(
+                layout="stereo", sample_rate=48000, count=10
+            )
+        ]
+        self.assertEqual(
+            [([len(p) for p in payloads], timestamp) for payloads, timestamp in output],
+            [
+                ([], None),  # No output due to buffering.
+                ([160], 0),
+                ([160], 160),
+                ([160], 320),
+                ([160], 480),
+                ([160], 640),
+                ([160], 800),
+                ([160], 960),
+                ([160], 1120),
+                ([160], 1280),
+            ],
+        )
+
+    def test_encoder_pack(self) -> None:
+        encoder = get_encoder(G722_CODEC)
+        self.assertTrue(isinstance(encoder, G722Encoder))
+
+        packet = self.create_packet(payload=G722_PAYLOAD, pts=1)
+        payloads, timestamp = encoder.pack(packet)
+        self.assertEqual(payloads, [G722_PAYLOAD])
+        self.assertEqual(timestamp, 8)
+
+    def test_roundtrip(self) -> None:
+        self.roundtrip_audio(
+            G722_CODEC,
+            input_sample_rate=16000,
+            output_clock_rate=8000,
+            output_layout="mono",
+            output_sample_rate=16000,
+        )
+
+    def test_roundtrip_with_loss(self) -> None:
+        self.roundtrip_audio(
+            G722_CODEC,
+            input_sample_rate=16000,
+            output_clock_rate=8000,
+            output_layout="mono",
+            output_sample_rate=16000,
+            drop=[1],
+        )
diff --git a/tests/test_rtcpeerconnection.py b/tests/test_rtcpeerconnection.py
@@ -825,6 +825,7 @@ async def _test_connect_audio_bidirectional(
         self.assertTrue(
             lf2crlf(
                 """a=rtpmap:96 opus/48000/2
+a=rtpmap:9 G722/8000
 a=rtpmap:0 PCMU/8000
 a=rtpmap:8 PCMA/8000
 """
@@ -863,6 +864,7 @@ async def _test_connect_audio_bidirectional(
         self.assertTrue(
             lf2crlf(
                 """a=rtpmap:96 opus/48000/2
+a=rtpmap:9 G722/8000
 a=rtpmap:0 PCMU/8000
 a=rtpmap:8 PCMA/8000
 """
