1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
import asyncio
import fractions
import time
from typing import Tuple
from unittest import TestCase
from aiortc.mediastreams import (
VIDEO_CLOCK_RATE,
VIDEO_PTIME,
VIDEO_TIME_BASE,
AudioStreamTrack,
MediaStreamTrack,
VideoStreamTrack,
)
from av.packet import Packet
class VideoPacketStreamTrack(MediaStreamTrack):
"""
A dummy video native track which reads green frames.
"""
kind = "video"
_start: float
_timestamp: int
async def next_timestamp(self) -> Tuple[int, fractions.Fraction]:
if hasattr(self, "_timestamp"):
self._timestamp += int(VIDEO_PTIME * VIDEO_CLOCK_RATE)
wait = self._start + (self._timestamp / VIDEO_CLOCK_RATE) - time.time()
await asyncio.sleep(wait)
else:
self._start = time.time()
self._timestamp = 0
return self._timestamp, VIDEO_TIME_BASE
async def recv(self) -> Packet:
"""
Receive the next :class:`~av.packet.Packet`.
The base implementation dummy packet h264 for tests
"""
pts, time_base = await self.next_timestamp()
header = [0, 0, 0, 1]
buffer = header + [0] * 1020
packet = Packet(len(buffer))
packet.update(bytes(buffer))
packet.pts = pts
packet.time_base = time_base
return packet
class MediaStreamTrackTest(TestCase):
def test_audio(self):
track = AudioStreamTrack()
self.assertEqual(track.kind, "audio")
self.assertEqual(len(track.id), 36)
def test_video(self):
track = VideoStreamTrack()
self.assertEqual(track.kind, "video")
self.assertEqual(len(track.id), 36)
def test_native_video(self):
track = VideoPacketStreamTrack()
self.assertEqual(track.kind, "video")
self.assertEqual(len(track.id), 36)
|
