Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This Source Code Form is subject to the terms of the Mozilla Public
2# License, v. 2.0. If a copy of the MPL was not distributed with this
3# file, You can obtain one at https://mozilla.org/MPL/2.0/.
4#
5# This file incorporates work covered by the following copyright and
6# permission notice:
7#
8# Copyright (c) Jeremy Lainé.
9# All rights reserved.
10#
11# Redistribution and use in source and binary forms, with or without
12# modification, are permitted provided that the following conditions are met:
13#
14# * Redistributions of source code must retain the above copyright notice,
15# this list of conditions and the following disclaimer.
16# * Redistributions in binary form must reproduce the above copyright notice,
17# this list of conditions and the following disclaimer in the documentation
18# and/or other materials provided with the distribution.
19# * Neither the name of aiortc nor the names of its contributors may
20# be used to endorse or promote products derived from this software without
21# specific prior written permission.
22#
23# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
24# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
25# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
26# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
27# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
28# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
29# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
32# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
34import random
35from struct import pack, unpack_from
36from typing import Optional, Type, TypeVar, Union
38from ..mediastreams import VIDEO_TIME_BASE, convert_timebase
39from .base import Decoder, Encoder, EncodedPacket
41DEFAULT_BITRATE = 500000 # 500 kbps
42MIN_BITRATE = 250000 # 250 kbps
43MAX_BITRATE = 1500000 # 1.5 Mbps
45PACKET_MAX = 1200
47DESCRIPTOR_T = TypeVar("DESCRIPTOR_T", bound="VpxPayloadDescriptor")
50class VpxPayloadDescriptor:
51 def __init__(
52 self,
53 partition_start: int,
54 partition_id: int,
55 picture_id: Optional[int] = None,
56 tl0picidx: Optional[int] = None,
57 tid: Optional[tuple[int, int]] = None,
58 keyidx: Optional[int] = None,
59 ) -> None:
60 self.partition_start = partition_start
61 self.partition_id = partition_id
62 self.picture_id = picture_id
63 self.tl0picidx = tl0picidx
64 self.tid = tid
65 self.keyidx = keyidx
67 def __bytes__(self) -> bytes:
68 octet = (self.partition_start << 4) | self.partition_id
70 ext_octet = 0
71 if self.picture_id is not None:
72 ext_octet |= 1 << 7
73 if self.tl0picidx is not None:
74 ext_octet |= 1 << 6
75 if self.tid is not None:
76 ext_octet |= 1 << 5
77 if self.keyidx is not None:
78 ext_octet |= 1 << 4
80 if ext_octet:
81 data = pack("!BB", (1 << 7) | octet, ext_octet)
82 if self.picture_id is not None:
83 if self.picture_id < 128:
84 data += pack("!B", self.picture_id)
85 else:
86 data += pack("!H", (1 << 15) | self.picture_id)
87 if self.tl0picidx is not None:
88 data += pack("!B", self.tl0picidx)
89 if self.tid is not None or self.keyidx is not None:
90 t_k = 0
91 if self.tid is not None:
92 t_k |= (self.tid[0] << 6) | (self.tid[1] << 5)
93 if self.keyidx is not None:
94 t_k |= self.keyidx
95 data += pack("!B", t_k)
96 else:
97 data = pack("!B", octet)
99 return data
101 def __repr__(self) -> str:
102 return (
103 f"VpxPayloadDescriptor(S={self.partition_start}, "
104 f"PID={self.partition_id}, pic_id={self.picture_id})"
105 )
107 @classmethod
108 def parse(cls: Type[DESCRIPTOR_T], data: bytes) -> tuple[DESCRIPTOR_T, bytes]:
109 if len(data) < 1:
110 raise ValueError("VPX descriptor is too short")
112 # first byte
113 octet = data[0]
114 extended = octet >> 7
115 partition_start = (octet >> 4) & 1
116 partition_id = octet & 0xF
117 picture_id = None
118 tl0picidx = None
119 tid = None
120 keyidx = None
121 pos = 1
123 # extended control bits
124 if extended:
125 if len(data) < pos + 1:
126 raise ValueError("VPX descriptor has truncated extended bits")
128 octet = data[pos]
129 ext_I = (octet >> 7) & 1
130 ext_L = (octet >> 6) & 1
131 ext_T = (octet >> 5) & 1
132 ext_K = (octet >> 4) & 1
133 pos += 1
135 # picture id
136 if ext_I:
137 if len(data) < pos + 1:
138 raise ValueError("VPX descriptor has truncated PictureID")
140 if data[pos] & 0x80:
141 if len(data) < pos + 2:
142 raise ValueError("VPX descriptor has truncated long PictureID")
144 picture_id = unpack_from("!H", data, pos)[0] & 0x7FFF
145 pos += 2
146 else:
147 picture_id = data[pos]
148 pos += 1
150 # unused
151 if ext_L:
152 if len(data) < pos + 1:
153 raise ValueError("VPX descriptor has truncated TL0PICIDX")
155 tl0picidx = data[pos]
156 pos += 1
157 if ext_T or ext_K:
158 if len(data) < pos + 1:
159 raise ValueError("VPX descriptor has truncated T/K")
161 t_k = data[pos]
162 if ext_T:
163 tid = ((t_k >> 6) & 3, (t_k >> 5) & 1)
164 if ext_K:
165 keyidx = t_k & 0x1F
166 pos += 1
168 obj = cls(
169 partition_start=partition_start,
170 partition_id=partition_id,
171 picture_id=picture_id,
172 tl0picidx=tl0picidx,
173 tid=tid,
174 keyidx=keyidx,
175 )
176 return obj, data[pos:]
179class Vp8Decoder(Decoder):
180 """VP8 receive codec. The webcam uplink is decoded by pixelflux off the GIL,
181 so this registry entry carries no libav decode of its own."""
184class Vp8Encoder(Encoder):
185 def __init__(self) -> None:
186 self.picture_id = random.randint(0, (1 << 15) - 1)
187 self.__target_bitrate = DEFAULT_BITRATE
189 def pack(self, packet: EncodedPacket) -> tuple[list[bytes], int, bool]:
190 frame = memoryview(packet.data)
191 payloads = self._packetize(frame, self.picture_id)
192 timestamp = convert_timebase(packet.pts, packet.time_base, VIDEO_TIME_BASE)
193 self.picture_id = (self.picture_id + 1) % (1 << 15)
194 # The frame tag's lowest bit is clear on a key frame.
195 return payloads, timestamp, len(frame) > 0 and frame[0] & 1 == 0
197 @property
198 def target_bitrate(self) -> int:
199 """
200 Target bitrate in bits per second.
201 """
202 return self.__target_bitrate
204 @target_bitrate.setter
205 def target_bitrate(self, bitrate: int) -> None:
206 bitrate = max(MIN_BITRATE, min(bitrate, MAX_BITRATE))
207 self.__target_bitrate = bitrate
209 @classmethod
210 def _packetize(cls, buffer: Union[bytes, memoryview], picture_id: int) -> list[bytes]:
211 payloads = []
212 descr = VpxPayloadDescriptor(
213 partition_start=1, partition_id=0, picture_id=picture_id
214 )
215 length = len(buffer)
216 pos = 0
217 while pos < length:
218 descr_bytes = bytes(descr)
219 size = min(length - pos, PACKET_MAX - len(descr_bytes))
220 # join takes the buffer slice zero-copy and materializes only this
221 # payload's bytes; the frame is never copied whole.
222 payloads.append(b"".join((descr_bytes, buffer[pos : pos + size])))
223 descr.partition_start = 0
224 pos += size
225 return payloads
228def vp8_depayload(payload: bytes) -> bytes:
229 descriptor, data = VpxPayloadDescriptor.parse(payload)
230 return data