Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_encoder_stereo_48khz(self):
encoder = get_encoder(PCMU_CODEC)
self.assertTrue(isinstance(encoder, PcmuEncoder))
for frame in self.create_audio_frames(
layout="stereo", sample_rate=48000, count=10
):
payloads, timestamp = encoder.encode(frame)
self.assertEqual(payloads, [b"\xff" * 160])
self.assertEqual(timestamp, frame.pts // 6)
def test_encoder_stereo_48khz(self):
encoder = get_encoder(PCMA_CODEC)
self.assertTrue(isinstance(encoder, PcmaEncoder))
for frame in self.create_audio_frames(
layout="stereo", sample_rate=48000, count=10
):
payloads, timestamp = encoder.encode(frame)
self.assertEqual(payloads, [b"\xd5" * 160])
self.assertEqual(timestamp, frame.pts // 6)
def roundtrip_audio(self, codec, output_layout, output_sample_rate, drop=[]):
"""
Round-trip an AudioFrame through encoder then decoder.
"""
encoder = get_encoder(codec)
decoder = get_decoder(codec)
input_frames = self.create_audio_frames(
layout="mono", sample_rate=8000, count=10
)
output_sample_count = int(output_sample_rate * AUDIO_PTIME)
for i, frame in enumerate(input_frames):
# encode
packages, timestamp = encoder.encode(frame)
if i not in drop:
# depacketize
data = b""
for package in packages:
def test_encoder_stereo_8khz(self):
encoder = get_encoder(OPUS_CODEC)
self.assertTrue(isinstance(encoder, OpusEncoder))
frames = self.create_audio_frames(layout="stereo", sample_rate=8000, count=2)
# first frame
payloads, timestamp = encoder.encode(frames[0])
self.assertEqual(payloads, [b"\xfc\xff\xfe"])
self.assertEqual(timestamp, 0)
# second frame
payloads, timestamp = encoder.encode(frames[1])
self.assertEqual(timestamp, 960)
def test_encoder_mono_8hz(self):
encoder = get_encoder(PCMA_CODEC)
self.assertTrue(isinstance(encoder, PcmaEncoder))
for frame in self.create_audio_frames(
layout="mono", sample_rate=8000, count=10
):
payloads, timestamp = encoder.encode(frame)
self.assertEqual(payloads, [b"\xd5" * 160])
self.assertEqual(timestamp, frame.pts)
def roundtrip_video(self, codec, width, height, time_base=VIDEO_TIME_BASE):
"""
Round-trip a VideoFrame through encoder then decoder.
"""
encoder = get_encoder(codec)
decoder = get_decoder(codec)
input_frames = self.create_video_frames(
width=width, height=height, count=30, time_base=time_base
)
for i, frame in enumerate(input_frames):
# encode
packages, timestamp = encoder.encode(frame)
# depacketize
data = b""
for package in packages:
data += depayload(codec, package)
# decode
frames = decoder.decode(JitterFrame(data=data, timestamp=timestamp))
def test_encoder(self):
encoder = get_encoder(H264_CODEC)
self.assertTrue(isinstance(encoder, H264Encoder))
frame = self.create_video_frame(width=640, height=480, pts=0)
packages, timestamp = encoder.encode(frame)
self.assertGreaterEqual(len(packages), 1)
def test_encoder_rgb(self):
encoder = get_encoder(VP8_CODEC)
self.assertTrue(isinstance(encoder, Vp8Encoder))
frame = self.create_video_frame(width=640, height=480, pts=0, format="rgb24")
payloads, timestamp = encoder.encode(frame)
self.assertEqual(len(payloads), 1)
self.assertTrue(len(payloads[0]) < 1300)
self.assertEqual(timestamp, 0)
def test_encoder_target_bitrate(self):
encoder = get_encoder(VP8_CODEC)
self.assertTrue(isinstance(encoder, Vp8Encoder))
self.assertEqual(encoder.target_bitrate, 500000)
frame = self.create_video_frame(width=640, height=480, pts=0)
payloads, timestamp = encoder.encode(frame)
self.assertEqual(len(payloads), 1)
self.assertTrue(len(payloads[0]) < 1300)
self.assertEqual(timestamp, 0)
# change target bitrate
encoder.target_bitrate = 600000
self.assertEqual(encoder.target_bitrate, 600000)
frame = self.create_video_frame(width=640, height=480, pts=3000)
payloads, timestamp = encoder.encode(frame)
self.assertEqual(len(payloads), 1)
def test_encoder_mono_8hz(self):
encoder = get_encoder(PCMU_CODEC)
self.assertTrue(isinstance(encoder, PcmuEncoder))
for frame in self.create_audio_frames(
layout="mono", sample_rate=8000, count=10
):
payloads, timestamp = encoder.encode(frame)
self.assertEqual(payloads, [b"\xff" * 160])
self.assertEqual(timestamp, frame.pts)