diff options
Diffstat (limited to 'python/examples/kaitai/rtcp_payload.py')
| -rw-r--r-- | python/examples/kaitai/rtcp_payload.py | 578 |
1 files changed, 0 insertions, 578 deletions
diff --git a/python/examples/kaitai/rtcp_payload.py b/python/examples/kaitai/rtcp_payload.py deleted file mode 100644 index c10066d1..00000000 --- a/python/examples/kaitai/rtcp_payload.py +++ /dev/null @@ -1,578 +0,0 @@ -# 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() - - - |