Skip to content

Instantly share code, notes, and snippets.

@mathieugouin
Forked from bmcfarland-rldrake/SCTE35Decoder.py
Created December 10, 2016 02:40
Show Gist options
  • Save mathieugouin/39146df66e0bb930364c5bf84db66b2d to your computer and use it in GitHub Desktop.
Save mathieugouin/39146df66e0bb930364c5bf84db66b2d to your computer and use it in GitHub Desktop.
SCTE-35 Parser/Decoder in Python
#!/usr/bin/python
'''
SCTE-35 Decoder
Modifications Copyright (c) 2016 RL Drake Holdings, LLC.
All modifications are also available for use under the MIT license as stated
below.
The MIT License (MIT)
Copyright (c) 2014 Al McCormack
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
'''
import bitstring
from datetime import timedelta
import sys
class MixinDictRepr(object):
def __repr__(self, *args, **kwargs):
return repr(self.__dict__)
class SCTE35_ParseException(Exception):
pass
class SCTE35_SpliceInfoSectionException(Exception):
pass
class SCTE35_SpliceInfoSection(MixinDictRepr):
def __init__(self, table_id):
if table_id != 0xfc:
raise SCTE35_SpliceInfoSectionException("%d valid. Should be 0xfc" %
table_id)
class MPEG_Time(long):
""" Relative time represented by 90kHz clock """
@property
def seconds(self):
return self / 90000.0
@property
def timedelta(self):
return timedelta(seconds=self.seconds)
def __repr__(self, *args, **kwargs):
return "%d (seconds: %f, time: %s)" % (self, self.seconds, self.timedelta)
class Splice_Time(MixinDictRepr):
def __init__(self):
self.time_specified_flag = None
self.pts_time = None
class SCTE35_SegmentationDescriptor(MixinDictRepr):
class Component(MixinDictRepr):
def __init__(self, tag, pts_offset):
self.pts_offset = pts_offset
self.tag = tag
pass
class SCTE35_AvailDescriptor(MixinDictRepr):
pass
class SCTE35_TimeSignal(MixinDictRepr):
pass
class SCTE35_SpliceInsert(MixinDictRepr):
class Component(MixinDictRepr):
def __init__(self, tag):
self.tag = tag
self.splice_time = None
class BreakDuration(MixinDictRepr):
def __init__(self):
self.auto_return = None
self.duration = None
def __init__(self, splice_id):
self.splice_id = splice_id
self.components = []
class SCTE35_SpliceDescriptor(MixinDictRepr):
pass
class SCTE35_Parser(object):
def parse(self, input_bytes):
input_bitarray = bitstring.BitString(bytes=input_bytes)
table_id = input_bitarray.read("uint:8")
splice_info_section = SCTE35_SpliceInfoSection(table_id)
splice_info_section.section_syntax_indicator = input_bitarray.read("bool")
splice_info_section.private = input_bitarray.read("bool")
#private
input_bitarray.pos += 2
splice_info_section.section_length = input_bitarray.read("uint:12")
splice_info_section.protocol_version = input_bitarray.read("uint:8")
splice_info_section.encrypted_packet = input_bitarray.read("bool")
splice_info_section.encryption_algorithm = input_bitarray.read("uint:6")
splice_info_section.pts_adjustment = input_bitarray.read("uint:33")
splice_info_section.cw_index = input_bitarray.read("uint:8")
splice_info_section.tier = input_bitarray.read("hex:12")
splice_info_section.splice_command_length = input_bitarray.read("uint:12")
splice_info_section.splice_command_type = input_bitarray.read("uint:8")
# splice command type parsing
if splice_info_section.splice_command_type == 5:
splice_info_section.splice_command = self.__parse_splice_insert(input_bitarray)
elif splice_info_section.splice_command_type == 6:
splice_info_section.splice_command = self.__parse_time_signal(input_bitarray)
else:
raise SCTE35_SpliceInfoSectionException("splice_command_type: %d not yet supported" % splice_info_section.splice_command_type)
splice_info_section.splice_descriptor_loop_length = input_bitarray.read("uint:16")
splice_info_section.splice_descriptors = None
if splice_info_section.splice_descriptor_loop_length > 0:
splice_info_section.splice_descriptors = \
self.__parse_splice_descriptors(input_bitarray,
splice_info_section.splice_descriptor_loop_length)
return splice_info_section
def __parse_splice_time(self, bitarray):
splice_time = Splice_Time()
splice_time.time_specified_flag = bitarray.read("bool")
if splice_time.time_specified_flag:
# Reserved for 6 bits
bitarray.pos += 6
splice_time.pts_time = MPEG_Time(bitarray.read("uint:33"))
else:
bitarray.pos += 7
return splice_time
def __parse_break_duration(self, bitarray):
break_duration = SCTE35_SpliceInsert.BreakDuration()
break_duration.auto_return = bitarray.read("bool")
bitarray.pos += 6
break_duration.duration = MPEG_Time(bitarray.read("uint:33"))
return break_duration
def __parse_splice_insert(self, bitarray):
splice_event_id = bitarray.read("uint:32")
ssi = SCTE35_SpliceInsert(splice_event_id)
ssi.splice_event_cancel_indicator = bitarray.read("bool")
bitarray.pos += 7
if not ssi.splice_event_cancel_indicator:
ssi.out_of_network_indicator = bitarray.read("bool")
ssi.program_splice_flag = bitarray.read("bool")
ssi.duration_flag = bitarray.read("bool")
ssi.splice_immediate_flag = bitarray.read("bool")
# Next 4 bits are reserved
bitarray.pos += 4
if ssi.program_splice_flag and not ssi.splice_immediate_flag:
ssi.splice_time = self.__parse_splice_time(bitarray)
if not ssi.program_splice_flag:
ssi.component_count = bitarray.read("uint:8")
for i in xrange(0, ssi.component_count):
component = SCTE35_SpliceInsert.Component(bitarray.read("uint:8"))
if ssi.splice_immediate_flag:
component.splice_time = self.__parse_splice_time(bitarray)
ssi.components.append(component)
if ssi.duration_flag:
ssi.break_duration = self.__parse_break_duration(bitarray)
ssi.unique_program_id = bitarray.read("uint:16")
ssi.avail_num = bitarray.read("uint:8")
ssi.avails_expected = bitarray.read("uint:8")
return ssi
def __parse_time_signal(self, bitarray):
ssi = SCTE35_TimeSignal()
ssi.splice_time = self.__parse_splice_time(bitarray)
return ssi
def __parse_avail_descriptor(self, bitarray, length):
desc = SCTE35_AvailDescriptor()
desc.identifier = bitarray.read('uint:32')
desc.providier_avail_id = bitarray.read('uint:32')
return desc
def __parse_segmentation_descriptor(self, bitarray, length):
sd = SCTE35_SegmentationDescriptor()
sd.identifier = bitarray.read('uint:32')
sd.segmentation_event_id = bitarray.read('uint:32')
sd.segmentation_event_cancel_indicator = bitarray.read('bool')
# next 7 bits reserved
bitarray.pos += 7
if sd.segmentation_event_cancel_indicator == False:
sd.program_segmentation_flag = bitarray.read('bool')
sd.segmentation_duration_flag = bitarray.read('bool')
sd.delivery_not_restricted_flag = bitarray.read('bool')
if sd.delivery_not_restricted_flag == False:
sd.web_delivery_allawed_flag = bitarray.read('bool')
sd.no_regional_blackout_flag = bitarray.read('bool')
sd.archive_allowed_flag = bitarray.read('bool')
sd.device_restrictions = bitarray.read('uint:2')
else:
bitarray.pos += 5
if sd.program_segmentation_flag == False:
sd.component_count = bitarray.read('uint:8')
sd.components = []
for x in xrange(sd.component_count):
tag = bitarray.read('uint:8')
bitarray.pos += 7
pts_offset = bitarray.read('uint:33')
comp = SCTE35_SegmentationDescriptor.Component(tag, pts_offset)
sd.components.append(comp)
if sd.segmentation_duration_flag == True:
sd.segmentation_duration = bitarray.read('uint:40')
sd.segmentation_upid_type = bitarray.read('uint:8')
sd.segmentation_upid_length = bitarray.read('uint:8')
sd.segmentation_upid = bitarray.read('hex:%d' %(8*sd.segmentation_upid_length))
sd.segment_type_id = bitarray.read('uint:8')
sd.segment_num = bitarray.read('uint:8')
sd.segments_expected = bitarray.read('uint:8')
return sd
def __parse_splice_descriptors(self, bitarray, length):
results = []
while length > 2 and bitarray.pos < (bitarray.len - 16):
desc_tag = bitarray.read('uint:8')
desc_len = bitarray.read('uint:8')
length -= 2
length -= desc_len
if desc_tag == 0x0:
splice_desc = self.__parse_avail_descriptor(bitarray, desc_len)
elif desc_tag == 0x2:
splice_desc = self.__parse_segmentation_descriptor(bitarray, desc_len)
else:
splice_desc = SCTE35_SpliceDescriptor();
if desc_len > 0:
splice_desc.raw = bitarray.read("hex:%d" %(desc_len * 8))
splice_desc.splice_descriptor_tag = desc_tag
splice_desc.descriptor_length = desc_len
results.append( splice_desc )
return results
if __name__ == "__main__":
import base64
import argparse
import pprint
import json
class SCTE35JsonEncoder(json.JSONEncoder):
def default(self, o):
return o.__dict__
description = """ Parse SCTE-35 markers from Base64 Strings.
Example String: "/DAlAAAAAAAAAP/wFAUAAAABf+/+LRQrAP4BI9MIAAEBAQAAfxV6SQ==" """
parser = argparse.ArgumentParser(description=description)
parser.add_argument('base64_scte35', metavar='SCTE35_Marker',
help='Base64 encoded SCTE-35 marker OR "-" to read stdin')
args = parser.parse_args()
if args.base64_scte35 != '-':
myinput = base64.standard_b64decode(args.base64_scte35)
section = SCTE35_Parser().parse(myinput)
print json.dumps(section, cls=SCTE35JsonEncoder, indent=2)
else:
for line in sys.stdin:
if line and len(line) > 0 :
myinput = base64.standard_b64decode( line )
section = SCTE35_Parser().parse(myinput)
print json.dumps(section, cls=SCTE35JsonEncoder, indent=2)
else:
print '-EOF-'
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment