pytoolbox.network.rtp module¶
-
class
pytoolbox.network.rtp.RtpPacket(data, length)[source]¶ Bases:
objectThis represent a real-time transport protocol (RTP) packet.
Packet header
- RFC 3550 page 13
0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |V=2|P|X| CC |M| PT | sequence number | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | timestamp | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | synchronization source (SSRC) identifier | +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ | contributing source (CSRC) identifiers | | .... | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Extension header
- RFC 3550 page 19
0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | defined by profile | length | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | header extension | | .... | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-
ER_VERSION= 'RTP Header : Version must be set to 2'¶
-
ER_PADDING_LENGTH= 'RTP Header : Bad padding length'¶
-
ER_EXTENSION_LENGTH= 'RTP Header : Bad extension length'¶
-
ER_PAYLOAD= 'RTP packet must have a payload'¶
-
HEADER_LENGTH= 12¶
-
V_MASK= 192¶
-
V_SHIFT= 6¶
-
P_MASK= 32¶
-
X_MASK= 16¶
-
CC_MASK= 15¶
-
M_MASK= 128¶
-
PT_MASK= 127¶
-
DYNAMIC_PT= 96¶
-
MP2T_PT= 33¶
-
MP2T_CLK= 90000¶
-
S_MASK= 65535¶
-
TS_MASK= 4294967295¶
-
valid¶ Returns True if this packet is a valid RTP packet.
-
validMP2T¶ Returns True if this packet is a valid RTP packet containing a MPEG2-TS payload.
-
errors¶ Returns an array containing any errors.
Returns: array of error message(s). Example usage
Testing invalid header:
>>> from pytoolbox.unittest import asserts >>> rtp = RtpPacket(bytearray(RtpPacket.HEADER_LENGTH-1), RtpPacket.HEADER_LENGTH-1) >>> asserts.list_equal(rtp.errors, [ ... 'RTP Header : Version must be set to 2', ... 'RTP packet must have a payload' ... ])
Testing a valid RTP packet with a MPEG2-TS payload:
>>> rtp = RtpPacket.create(6, 777, RtpPacket.MP2T_PT, 'salut') >>> asserts.list_equal(rtp.errors, [])
-
clock_rate¶ Return the MPEG2-TS clock rate of a MPEG2-TS payload or 1 if this is not.
-
header_size¶ Returns the length (aka size) of the header.
Example usage
>>> rtp = RtpPacket.create(6, 777, RtpPacket.MP2T_PT, 'salut') >>> print(rtp.header_size) 12
-
payload_size¶ Returns the length (aka size) of the payload.
Example usage
>>> rtp = RtpPacket.create(6, 777, RtpPacket.MP2T_PT, 'salut') >>> print(rtp.payload_size) 5
-
time¶ Return computed time (timestamp / clock rate).
-
header_bytes¶ Return the RTP header bytes.
Example usage
>>> rtp = RtpPacket.create(6, 777, RtpPacket.MP2T_PT, bytearray.fromhex('00 01 02 03')) >>> print(rtp) version = 2 errors = [] padding = False extension = False marker = False payload type = 33 sequence = 6 timestamp = 777 clock rate = 90000 time = 0 ssrc = 0 csrc count = 0 payload size = 4 >>> header = rtp.header_bytes >>> assert len(header) == 12 >>> print(''.join(' %02x' % b for b in header)) 80 21 00 06 00 00 03 09 00 00 00 00 >>> header += rtp.payload >>> assert rtp == RtpPacket(header, len(header))
>>> rtp = RtpPacket.create(0xffffffff, 0xffffffffff, RtpPacket.DYNAMIC_PT, bytearray(1023)) >>> print(rtp) version = 2 errors = [] padding = False extension = False marker = False payload type = 96 sequence = 65535 timestamp = 4294967295 clock rate = 1 time = 4294967295 ssrc = 0 csrc count = 0 payload size = 1023 >>> header = rtp.header_bytes >>> assert len(header) == 12 >>> print(''.join(' %02x' % b for b in header)) 80 60 ff ff ff ff ff ff 00 00 00 00 >>> header += rtp.payload >>> assert rtp == RtpPacket(header, len(header))
-
bytes¶ Return the RTP packet header and payload bytes.
-
__init__(data, length)[source]¶ This constructor will parse input bytes array to fill packet’s fields. In case of error (e.g. bad version number) the constructor will abort filling fields and un-updated fields are set to their corresponding default value.
Parameters: Example usage
Testing invalid headers:
>>> rtp = RtpPacket(bytearray(RtpPacket.HEADER_LENGTH-1), RtpPacket.HEADER_LENGTH-1) >>> rtp.valid # Bad length False >>> rtp = RtpPacket(bytearray(RtpPacket.HEADER_LENGTH), RtpPacket.HEADER_LENGTH) >>> rtp.valid # Bad version False >>> bytes = bytearray(RtpPacket.HEADER_LENGTH) >>> bytes[0] = 0xa0 >>> rtp = RtpPacket(bytes, RtpPacket.HEADER_LENGTH) >>> rtp.valid # Padding enabled but not present False
Testing header fields value:
>>> bytes = bytes.fromhex('80 a1 a4 25 ca fe b5 04 b0 60 5e bb 12 34') >>> rtp = RtpPacket(bytes, len(bytes)) >>> rtp.valid True >>> print(rtp) version = 2 errors = [] padding = False extension = False marker = True payload type = 33 sequence = 42021 timestamp = 3405690116 clock rate = 90000 time = 37841 ssrc = 2959105723 csrc count = 0 payload size = 2 >>> rtp.csrc [] >>> rtp.payload[0] 18 >>> rtp.payload[1] 52
Testing header fields value (with padding, extension and ccrc):
>>> bytes = bytes.fromhex('b5a1a401 cafea421 b0605ebb 11111111 22222222 33333333 ' ... '44444444 55555555 00000004 87654321 12340002') >>> rtp = RtpPacket(bytes, len(bytes)) >>> rtp.valid True >>> rtp.version 2 >>> rtp.padding True >>> rtp.extension True >>> rtp.marker True >>> rtp.payload_type 33 >>> rtp.sequence 41985 >>> rtp.timestamp 3405685793 >>> rtp.clock_rate 90000 >>> rtp.ssrc 2959105723 >>> len(rtp.csrc) 5 >>> rtp.csrc [286331153, 572662306, 858993459, 1145324612, 1431655765] >>> rtp.payload bytearray(b'\x124')
-
classmethod
create(sequence, timestamp, payload_type, payload)[source]¶ Create a valid RTP packet with a given payload.
Parameters: Example usage
>>> p = RtpPacket.create(10, 1024, RtpPacket.MP2T_PT, 'The payload string') >>> q = RtpPacket.create(11, 1028, RtpPacket.MP2T_PT, bytearray.fromhex('00 11 22 33')) >>> r = RtpPacket.create(11, 1028, RtpPacket.DYNAMIC_PT, bytearray.fromhex('cc aa ff ee')) >>> assert p.validMP2T and q.validMP2T and r.valid