# This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild from pkg_resources import parse_version from .kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO from enum import Enum import collections if parse_version(ks_version) < parse_version('0.7'): raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version)) class RtcpPayload(KaitaiStruct): """RTCP is the Real-Time Control Protocol. .. seealso:: Source - https://tools.ietf.org/html/rfc3550 """ class PayloadType(Enum): fir = 192 nack = 193 ij = 195 sr = 200 rr = 201 sdes = 202 bye = 203 app = 204 rtpfb = 205 psfb = 206 xr = 207 avb = 208 rsi = 209 class SdesSubtype(Enum): pad = 0 cname = 1 name = 2 email = 3 phone = 4 loc = 5 tool = 6 note = 7 priv = 8 class PsfbSubtype(Enum): pli = 1 sli = 2 rpsi = 3 fir = 4 tstr = 5 tstn = 6 vbcm = 7 afb = 15 class RtpfbSubtype(Enum): nack = 1 tmmbr = 3 tmmbn = 4 rrr = 5 transport_feedback = 15 SEQ_FIELDS = ["rtcp_packets"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['rtcp_packets']['start'] = self._io.pos() self.rtcp_packets = [] i = 0 while not self._io.is_eof(): if not 'arr' in self._debug['rtcp_packets']: self._debug['rtcp_packets']['arr'] = [] self._debug['rtcp_packets']['arr'].append({'start': self._io.pos()}) _t_rtcp_packets = self._root.RtcpPacket(self._io, self, self._root) _t_rtcp_packets._read() self.rtcp_packets.append(_t_rtcp_packets) self._debug['rtcp_packets']['arr'][len(self.rtcp_packets) - 1]['end'] = self._io.pos() i += 1 self._debug['rtcp_packets']['end'] = self._io.pos() class PsfbAfbRembPacket(KaitaiStruct): SEQ_FIELDS = ["num_ssrc", "br_exp", "br_mantissa", "ssrc_list"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['num_ssrc']['start'] = self._io.pos() self.num_ssrc = self._io.read_u1() self._debug['num_ssrc']['end'] = self._io.pos() self._debug['br_exp']['start'] = self._io.pos() self.br_exp = self._io.read_bits_int(6) self._debug['br_exp']['end'] = self._io.pos() self._debug['br_mantissa']['start'] = self._io.pos() self.br_mantissa = self._io.read_bits_int(18) self._debug['br_mantissa']['end'] = self._io.pos() self._io.align_to_byte() self._debug['ssrc_list']['start'] = self._io.pos() self.ssrc_list = [None] * (self.num_ssrc) for i in range(self.num_ssrc): if not 'arr' in self._debug['ssrc_list']: self._debug['ssrc_list']['arr'] = [] self._debug['ssrc_list']['arr'].append({'start': self._io.pos()}) self.ssrc_list[i] = self._io.read_u4be() self._debug['ssrc_list']['arr'][i]['end'] = self._io.pos() self._debug['ssrc_list']['end'] = self._io.pos() @property def max_total_bitrate(self): if hasattr(self, '_m_max_total_bitrate'): return self._m_max_total_bitrate if hasattr(self, '_m_max_total_bitrate') else None self._m_max_total_bitrate = (self.br_mantissa * (1 << self.br_exp)) return self._m_max_total_bitrate if hasattr(self, '_m_max_total_bitrate') else None class SrPacket(KaitaiStruct): SEQ_FIELDS = ["ssrc", "ntp_msw", "ntp_lsw", "rtp_timestamp", "sender_packet_count", "sender_octet_count", "report_block"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc']['start'] = self._io.pos() self.ssrc = self._io.read_u4be() self._debug['ssrc']['end'] = self._io.pos() self._debug['ntp_msw']['start'] = self._io.pos() self.ntp_msw = self._io.read_u4be() self._debug['ntp_msw']['end'] = self._io.pos() self._debug['ntp_lsw']['start'] = self._io.pos() self.ntp_lsw = self._io.read_u4be() self._debug['ntp_lsw']['end'] = self._io.pos() self._debug['rtp_timestamp']['start'] = self._io.pos() self.rtp_timestamp = self._io.read_u4be() self._debug['rtp_timestamp']['end'] = self._io.pos() self._debug['sender_packet_count']['start'] = self._io.pos() self.sender_packet_count = self._io.read_u4be() self._debug['sender_packet_count']['end'] = self._io.pos() self._debug['sender_octet_count']['start'] = self._io.pos() self.sender_octet_count = self._io.read_u4be() self._debug['sender_octet_count']['end'] = self._io.pos() self._debug['report_block']['start'] = self._io.pos() self.report_block = [None] * (self._parent.subtype) for i in range(self._parent.subtype): if not 'arr' in self._debug['report_block']: self._debug['report_block']['arr'] = [] self._debug['report_block']['arr'].append({'start': self._io.pos()}) _t_report_block = self._root.ReportBlock(self._io, self, self._root) _t_report_block._read() self.report_block[i] = _t_report_block self._debug['report_block']['arr'][i]['end'] = self._io.pos() self._debug['report_block']['end'] = self._io.pos() @property def ntp(self): if hasattr(self, '_m_ntp'): return self._m_ntp if hasattr(self, '_m_ntp') else None self._m_ntp = ((self.ntp_msw << 32) & self.ntp_lsw) return self._m_ntp if hasattr(self, '_m_ntp') else None class RrPacket(KaitaiStruct): SEQ_FIELDS = ["ssrc", "report_block"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc']['start'] = self._io.pos() self.ssrc = self._io.read_u4be() self._debug['ssrc']['end'] = self._io.pos() self._debug['report_block']['start'] = self._io.pos() self.report_block = [None] * (self._parent.subtype) for i in range(self._parent.subtype): if not 'arr' in self._debug['report_block']: self._debug['report_block']['arr'] = [] self._debug['report_block']['arr'].append({'start': self._io.pos()}) _t_report_block = self._root.ReportBlock(self._io, self, self._root) _t_report_block._read() self.report_block[i] = _t_report_block self._debug['report_block']['arr'][i]['end'] = self._io.pos() self._debug['report_block']['end'] = self._io.pos() class RtcpPacket(KaitaiStruct): SEQ_FIELDS = ["version", "padding", "subtype", "payload_type", "length", "body"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['version']['start'] = self._io.pos() self.version = self._io.read_bits_int(2) self._debug['version']['end'] = self._io.pos() self._debug['padding']['start'] = self._io.pos() self.padding = self._io.read_bits_int(1) != 0 self._debug['padding']['end'] = self._io.pos() self._debug['subtype']['start'] = self._io.pos() self.subtype = self._io.read_bits_int(5) self._debug['subtype']['end'] = self._io.pos() self._io.align_to_byte() self._debug['payload_type']['start'] = self._io.pos() self.payload_type = KaitaiStream.resolve_enum(self._root.PayloadType, self._io.read_u1()) self._debug['payload_type']['end'] = self._io.pos() self._debug['length']['start'] = self._io.pos() self.length = self._io.read_u2be() self._debug['length']['end'] = self._io.pos() self._debug['body']['start'] = self._io.pos() _on = self.payload_type if _on == self._root.PayloadType.sr: self._raw_body = self._io.read_bytes((4 * self.length)) io = KaitaiStream(BytesIO(self._raw_body)) self.body = self._root.SrPacket(io, self, self._root) self.body._read() elif _on == self._root.PayloadType.psfb: self._raw_body = self._io.read_bytes((4 * self.length)) io = KaitaiStream(BytesIO(self._raw_body)) self.body = self._root.PsfbPacket(io, self, self._root) self.body._read() elif _on == self._root.PayloadType.rr: self._raw_body = self._io.read_bytes((4 * self.length)) io = KaitaiStream(BytesIO(self._raw_body)) self.body = self._root.RrPacket(io, self, self._root) self.body._read() elif _on == self._root.PayloadType.rtpfb: self._raw_body = self._io.read_bytes((4 * self.length)) io = KaitaiStream(BytesIO(self._raw_body)) self.body = self._root.RtpfbPacket(io, self, self._root) self.body._read() elif _on == self._root.PayloadType.sdes: self._raw_body = self._io.read_bytes((4 * self.length)) io = KaitaiStream(BytesIO(self._raw_body)) self.body = self._root.SdesPacket(io, self, self._root) self.body._read() else: self.body = self._io.read_bytes((4 * self.length)) self._debug['body']['end'] = self._io.pos() class SdesTlv(KaitaiStruct): SEQ_FIELDS = ["type", "length", "value"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['type']['start'] = self._io.pos() self.type = KaitaiStream.resolve_enum(self._root.SdesSubtype, self._io.read_u1()) self._debug['type']['end'] = self._io.pos() if self.type != self._root.SdesSubtype.pad: self._debug['length']['start'] = self._io.pos() self.length = self._io.read_u1() self._debug['length']['end'] = self._io.pos() if self.type != self._root.SdesSubtype.pad: self._debug['value']['start'] = self._io.pos() self.value = self._io.read_bytes(self.length) self._debug['value']['end'] = self._io.pos() class ReportBlock(KaitaiStruct): SEQ_FIELDS = ["ssrc_source", "lost_val", "highest_seq_num_received", "interarrival_jitter", "lsr", "dlsr"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc_source']['start'] = self._io.pos() self.ssrc_source = self._io.read_u4be() self._debug['ssrc_source']['end'] = self._io.pos() self._debug['lost_val']['start'] = self._io.pos() self.lost_val = self._io.read_u1() self._debug['lost_val']['end'] = self._io.pos() self._debug['highest_seq_num_received']['start'] = self._io.pos() self.highest_seq_num_received = self._io.read_u4be() self._debug['highest_seq_num_received']['end'] = self._io.pos() self._debug['interarrival_jitter']['start'] = self._io.pos() self.interarrival_jitter = self._io.read_u4be() self._debug['interarrival_jitter']['end'] = self._io.pos() self._debug['lsr']['start'] = self._io.pos() self.lsr = self._io.read_u4be() self._debug['lsr']['end'] = self._io.pos() self._debug['dlsr']['start'] = self._io.pos() self.dlsr = self._io.read_u4be() self._debug['dlsr']['end'] = self._io.pos() @property def fraction_lost(self): if hasattr(self, '_m_fraction_lost'): return self._m_fraction_lost if hasattr(self, '_m_fraction_lost') else None self._m_fraction_lost = (self.lost_val >> 24) return self._m_fraction_lost if hasattr(self, '_m_fraction_lost') else None @property def cumulative_packets_lost(self): if hasattr(self, '_m_cumulative_packets_lost'): return self._m_cumulative_packets_lost if hasattr(self, '_m_cumulative_packets_lost') else None self._m_cumulative_packets_lost = (self.lost_val & 16777215) return self._m_cumulative_packets_lost if hasattr(self, '_m_cumulative_packets_lost') else None class RtpfbTransportFeedbackPacket(KaitaiStruct): SEQ_FIELDS = ["base_sequence_number", "packet_status_count", "b4", "remaining"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['base_sequence_number']['start'] = self._io.pos() self.base_sequence_number = self._io.read_u2be() self._debug['base_sequence_number']['end'] = self._io.pos() self._debug['packet_status_count']['start'] = self._io.pos() self.packet_status_count = self._io.read_u2be() self._debug['packet_status_count']['end'] = self._io.pos() self._debug['b4']['start'] = self._io.pos() self.b4 = self._io.read_u4be() self._debug['b4']['end'] = self._io.pos() self._debug['remaining']['start'] = self._io.pos() self.remaining = self._io.read_bytes_full() self._debug['remaining']['end'] = self._io.pos() @property def reference_time(self): if hasattr(self, '_m_reference_time'): return self._m_reference_time if hasattr(self, '_m_reference_time') else None self._m_reference_time = (self.b4 >> 8) return self._m_reference_time if hasattr(self, '_m_reference_time') else None @property def fb_pkt_count(self): if hasattr(self, '_m_fb_pkt_count'): return self._m_fb_pkt_count if hasattr(self, '_m_fb_pkt_count') else None self._m_fb_pkt_count = (self.b4 & 255) return self._m_fb_pkt_count if hasattr(self, '_m_fb_pkt_count') else None @property def packet_status(self): if hasattr(self, '_m_packet_status'): return self._m_packet_status if hasattr(self, '_m_packet_status') else None self._debug['_m_packet_status']['start'] = self._io.pos() self._m_packet_status = self._io.read_bytes(0) self._debug['_m_packet_status']['end'] = self._io.pos() return self._m_packet_status if hasattr(self, '_m_packet_status') else None @property def recv_delta(self): if hasattr(self, '_m_recv_delta'): return self._m_recv_delta if hasattr(self, '_m_recv_delta') else None self._debug['_m_recv_delta']['start'] = self._io.pos() self._m_recv_delta = self._io.read_bytes(0) self._debug['_m_recv_delta']['end'] = self._io.pos() return self._m_recv_delta if hasattr(self, '_m_recv_delta') else None class PsfbPacket(KaitaiStruct): SEQ_FIELDS = ["ssrc", "ssrc_media_source", "fci_block"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc']['start'] = self._io.pos() self.ssrc = self._io.read_u4be() self._debug['ssrc']['end'] = self._io.pos() self._debug['ssrc_media_source']['start'] = self._io.pos() self.ssrc_media_source = self._io.read_u4be() self._debug['ssrc_media_source']['end'] = self._io.pos() self._debug['fci_block']['start'] = self._io.pos() _on = self.fmt if _on == self._root.PsfbSubtype.afb: self._raw_fci_block = self._io.read_bytes_full() io = KaitaiStream(BytesIO(self._raw_fci_block)) self.fci_block = self._root.PsfbAfbPacket(io, self, self._root) self.fci_block._read() else: self.fci_block = self._io.read_bytes_full() self._debug['fci_block']['end'] = self._io.pos() @property def fmt(self): if hasattr(self, '_m_fmt'): return self._m_fmt if hasattr(self, '_m_fmt') else None self._m_fmt = KaitaiStream.resolve_enum(self._root.PsfbSubtype, self._parent.subtype) return self._m_fmt if hasattr(self, '_m_fmt') else None class SourceChunk(KaitaiStruct): SEQ_FIELDS = ["ssrc", "sdes_tlv"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc']['start'] = self._io.pos() self.ssrc = self._io.read_u4be() self._debug['ssrc']['end'] = self._io.pos() self._debug['sdes_tlv']['start'] = self._io.pos() self.sdes_tlv = [] i = 0 while not self._io.is_eof(): if not 'arr' in self._debug['sdes_tlv']: self._debug['sdes_tlv']['arr'] = [] self._debug['sdes_tlv']['arr'].append({'start': self._io.pos()}) _t_sdes_tlv = self._root.SdesTlv(self._io, self, self._root) _t_sdes_tlv._read() self.sdes_tlv.append(_t_sdes_tlv) self._debug['sdes_tlv']['arr'][len(self.sdes_tlv) - 1]['end'] = self._io.pos() i += 1 self._debug['sdes_tlv']['end'] = self._io.pos() class SdesPacket(KaitaiStruct): SEQ_FIELDS = ["source_chunk"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['source_chunk']['start'] = self._io.pos() self.source_chunk = [None] * (self.source_count) for i in range(self.source_count): if not 'arr' in self._debug['source_chunk']: self._debug['source_chunk']['arr'] = [] self._debug['source_chunk']['arr'].append({'start': self._io.pos()}) _t_source_chunk = self._root.SourceChunk(self._io, self, self._root) _t_source_chunk._read() self.source_chunk[i] = _t_source_chunk self._debug['source_chunk']['arr'][i]['end'] = self._io.pos() self._debug['source_chunk']['end'] = self._io.pos() @property def source_count(self): if hasattr(self, '_m_source_count'): return self._m_source_count if hasattr(self, '_m_source_count') else None self._m_source_count = self._parent.subtype return self._m_source_count if hasattr(self, '_m_source_count') else None class RtpfbPacket(KaitaiStruct): SEQ_FIELDS = ["ssrc", "ssrc_media_source", "fci_block"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['ssrc']['start'] = self._io.pos() self.ssrc = self._io.read_u4be() self._debug['ssrc']['end'] = self._io.pos() self._debug['ssrc_media_source']['start'] = self._io.pos() self.ssrc_media_source = self._io.read_u4be() self._debug['ssrc_media_source']['end'] = self._io.pos() self._debug['fci_block']['start'] = self._io.pos() _on = self.fmt if _on == self._root.RtpfbSubtype.transport_feedback: self._raw_fci_block = self._io.read_bytes_full() io = KaitaiStream(BytesIO(self._raw_fci_block)) self.fci_block = self._root.RtpfbTransportFeedbackPacket(io, self, self._root) self.fci_block._read() else: self.fci_block = self._io.read_bytes_full() self._debug['fci_block']['end'] = self._io.pos() @property def fmt(self): if hasattr(self, '_m_fmt'): return self._m_fmt if hasattr(self, '_m_fmt') else None self._m_fmt = KaitaiStream.resolve_enum(self._root.RtpfbSubtype, self._parent.subtype) return self._m_fmt if hasattr(self, '_m_fmt') else None class PacketStatusChunk(KaitaiStruct): SEQ_FIELDS = ["t", "s2", "s1", "rle", "symbol_list"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['t']['start'] = self._io.pos() self.t = self._io.read_bits_int(1) != 0 self._debug['t']['end'] = self._io.pos() if int(self.t) == 0: self._debug['s2']['start'] = self._io.pos() self.s2 = self._io.read_bits_int(2) self._debug['s2']['end'] = self._io.pos() if int(self.t) == 1: self._debug['s1']['start'] = self._io.pos() self.s1 = self._io.read_bits_int(1) != 0 self._debug['s1']['end'] = self._io.pos() if int(self.t) == 0: self._debug['rle']['start'] = self._io.pos() self.rle = self._io.read_bits_int(13) self._debug['rle']['end'] = self._io.pos() if int(self.t) == 1: self._debug['symbol_list']['start'] = self._io.pos() self.symbol_list = self._io.read_bits_int(14) self._debug['symbol_list']['end'] = self._io.pos() @property def s(self): if hasattr(self, '_m_s'): return self._m_s if hasattr(self, '_m_s') else None self._m_s = (self.s2 if int(self.t) == 0 else (1 if int(self.s1) == 0 else 0)) return self._m_s if hasattr(self, '_m_s') else None class PsfbAfbPacket(KaitaiStruct): SEQ_FIELDS = ["uid", "contents"] def __init__(self, _io, _parent=None, _root=None): self._io = _io self._parent = _parent self._root = _root if _root else self self._debug = collections.defaultdict(dict) def _read(self): self._debug['uid']['start'] = self._io.pos() self.uid = self._io.read_u4be() self._debug['uid']['end'] = self._io.pos() self._debug['contents']['start'] = self._io.pos() _on = self.uid if _on == 1380273474: self._raw_contents = self._io.read_bytes_full() io = KaitaiStream(BytesIO(self._raw_contents)) self.contents = self._root.PsfbAfbRembPacket(io, self, self._root) self.contents._read() else: self.contents = self._io.read_bytes_full() self._debug['contents']['end'] = self._io.pos()