monkee.ch/SickGear
1
0
Fork 0
mirror of https://github.com/SickGear/SickGear.git synced 2024-11-17 18:35:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
SickGear/lib/hachoir/parser/video/mpeg_video.py

592 lines
22 KiB
Python
Raw Normal View History

Change bump to major version 3.xx to signal that this branch supports Python3+ only. Initial SickGear for Python 3.
2023-01-12 01:04:47 +00:00
"""
Moving Picture Experts Group (MPEG) video version 1 and 2 parser.
Information:
- http://www.mpucoder.com/DVD/
- http://dvd.sourceforge.net/dvdinfo/
- http://www.mit.jyu.fi/mweber/leffakone/software/parsempegts/
- http://homepage.mac.com/rnc/EditMpegHeaderIFO.html
- http://standards.iso.org/ittf/PubliclyAvailableStandards/c025029_ISO_IEC_TR_11172-5_1998(E)_Software_Simulation.zip
This is a sample encoder/decoder implementation for MPEG-1.
Author: Victor Stinner
Creation date: 15 september 2006
"""
from hachoir.parser import Parser
from hachoir.parser.audio.mpeg_audio import MpegAudioFile
from hachoir.field import (FieldSet,
FieldError, ParserError,
Bit, Bits, Bytes, RawBits, PaddingBits, NullBits,
UInt8, UInt16,
RawBytes, PaddingBytes,
Enum, CustomFragment)
from hachoir.core.endian import BIG_ENDIAN
from hachoir.core.text_handler import textHandler, hexadecimal
class Timestamp(FieldSet):
static_size = 36
def createFields(self):
yield Bits(self, "ts_32_30", 3)
yield Bit(self, "sync[]") # =True
yield Bits(self, "ts_29_15", 15)
yield Bit(self, "sync[]") # =True
yield Bits(self, "ts_14_0", 15)
yield Bit(self, "sync[]") # =True
def createValue(self):
return ((self["ts_32_30"].value << 30)
+ (self["ts_29_15"].value << 15)
+ self["ts_14_0"].value)
class SCRExt(FieldSet):
static_size = 46
def createFields(self):
yield Bits(self, "scr_32_30", 3)
yield Bit(self, "sync[]") # =True
yield Bits(self, "scr_29_15", 15)
yield Bit(self, "sync[]") # =True
yield Bits(self, "scr_14_0", 15)
yield Bit(self, "sync[]") # =True
yield Bits(self, "scr_ext", 9)
yield Bit(self, "sync[]") # =True
def createDescription(self):
return "System Clock Reference (90kHz per tick), with extended 27MHz resolution"
def createValue(self):
return ((self["scr_32_30"].value << 30)
+ (self["scr_29_15"].value << 15)
+ self["scr_14_0"].value
+ (self["scr_ext"].value / 300.0))
class PackHeader(FieldSet):
def createFields(self):
if self.stream.readBits(self.absolute_address, 2, self.endian) == 1:
# MPEG version 2
yield Bits(self, "sync[]", 2)
yield SCRExt(self, "scr")
yield Bits(self, "mux_rate", 22)
yield Bits(self, "sync[]", 2)
yield PaddingBits(self, "reserved", 5, pattern=1)
yield Bits(self, "stuffing_length", 3)
count = self["stuffing_length"].value
if count:
yield PaddingBytes(self, "stuffing", count, pattern="\xff")
else:
# MPEG version 1
yield Bits(self, "sync[]", 4)
yield Timestamp(self, "scr")
yield Bit(self, "sync[]")
yield Bits(self, "mux_rate", 22)
yield Bit(self, "sync[]")
def validate(self):
if self["mux_rate"].value == 0:
return "Invalid mux rate"
sync0 = self["sync[0]"]
if (sync0.size == 2 and sync0.value == 1):
# MPEG2
if self["sync[1]"].value != 3:
return "Invalid synchronisation bits"
elif (sync0.size == 4 and sync0.value == 2):
# MPEG1
if not self["sync[1]"].value \
or not self["sync[2]"].value:
return "Invalid synchronisation bits"
else:
return "Unknown version"
return True
class SystemHeader(FieldSet):
def createFields(self):
yield Bits(self, "marker[]", 1)
yield Bits(self, "rate_bound", 22)
yield Bits(self, "marker[]", 1)
yield Bits(self, "audio_bound", 6)
yield Bit(self, "fixed_bitrate")
yield Bit(self, "csps", description="Constrained system parameter stream")
yield Bit(self, "audio_lock")
yield Bit(self, "video_lock")
yield Bits(self, "marker[]", 1)
yield Bits(self, "video_bound", 5)
length = self['../length'].value - 5
if length:
yield RawBytes(self, "raw[]", length)
class defaultParser(FieldSet):
def createFields(self):
yield RawBytes(self, "data", self["../length"].value)
class Padding(FieldSet):
def createFields(self):
yield PaddingBytes(self, "data", self["../length"].value)
class VideoExtension2(FieldSet):
def createFields(self):
yield Bit(self, "sync[]") # =True
yield Bits(self, "ext_length", 7)
yield NullBits(self, "reserved[]", 8)
size = self["ext_length"].value
if size:
yield RawBytes(self, "ext_bytes", size)
class VideoExtension1(FieldSet):
def createFields(self):
yield Bit(self, "has_private")
yield Bit(self, "has_pack_lgth")
yield Bit(self, "has_pack_seq")
yield Bit(self, "has_pstd_buffer")
yield Bits(self, "sync[]", 3) # =7
yield Bit(self, "has_extension2")
if self["has_private"].value:
yield RawBytes(self, "private", 16)
if self["has_pack_lgth"].value:
yield UInt8(self, "pack_lgth")
if self["has_pack_seq"].value:
yield Bit(self, "sync[]") # =True
yield Bits(self, "pack_seq_counter", 7)
yield Bit(self, "sync[]") # =True
yield Bit(self, "mpeg12_id")
yield Bits(self, "orig_stuffing_length", 6)
if self["has_pstd_buffer"].value:
yield Bits(self, "sync[]", 2) # =1
yield Enum(Bit(self, "pstd_buffer_scale"),
{True: "128 bytes", False: "1024 bytes"})
yield Bits(self, "pstd_size", 13)
class VideoSeqHeader(FieldSet):
ASPECT = ["forbidden", "1.0000 (VGA etc.)", "0.6735",
"0.7031 (16:9, 625line)", "0.7615", "0.8055",
"0.8437 (16:9, 525line)", "0.8935",
"0.9157 (CCIR601, 625line)", "0.9815", "1.0255", "1.0695",
"1.0950 (CCIR601, 525line)", "1.1575", "1.2015", "reserved"]
FRAMERATE = ["forbidden", "23.976 fps", "24 fps", "25 fps", "29.97 fps",
"30 fps", "50 fps", "59.94 fps", "60 fps"]
def createFields(self):
yield Bits(self, "width", 12)
yield Bits(self, "height", 12)
yield Enum(Bits(self, "aspect", 4), self.ASPECT)
yield Enum(Bits(self, "frame_rate", 4), self.FRAMERATE)
yield Bits(self, "bit_rate", 18, "Bit rate in units of 50 bytes")
yield Bits(self, "sync[]", 1) # =1
yield Bits(self, "vbv_size", 10, "Video buffer verifier size, in units of 16768")
yield Bit(self, "constrained_params_flag")
yield Bit(self, "has_intra_quantizer")
if self["has_intra_quantizer"].value:
for i in range(64):
yield Bits(self, "intra_quantizer[]", 8)
yield Bit(self, "has_non_intra_quantizer")
if self["has_non_intra_quantizer"].value:
for i in range(64):
yield Bits(self, "non_intra_quantizer[]", 8)
class GroupStart(FieldSet):
def createFields(self):
yield Bit(self, "drop_frame")
yield Bits(self, "time_hh", 5)
yield Bits(self, "time_mm", 6)
yield PaddingBits(self, "time_pad[]", 1)
yield Bits(self, "time_ss", 6)
yield Bits(self, "time_ff", 6)
yield Bit(self, "closed_group")
yield Bit(self, "broken_group")
yield PaddingBits(self, "pad[]", 5)
class PacketElement(FieldSet):
def createFields(self):
yield Bits(self, "sync[]", 2) # =2
if self["sync[0]"].value != 2:
raise ParserError("Unknown video elementary data")
yield Bits(self, "is_scrambled", 2)
yield Bits(self, "priority", 1)
yield Bit(self, "alignment")
yield Bit(self, "is_copyrighted")
yield Bit(self, "is_original")
yield Bit(self, "has_pts", "Presentation Time Stamp")
yield Bit(self, "has_dts", "Decode Time Stamp")
yield Bit(self, "has_escr", "Elementary Stream Clock Reference")
yield Bit(self, "has_es_rate", "Elementary Stream rate")
yield Bit(self, "dsm_trick_mode")
yield Bit(self, "has_copy_info")
yield Bit(self, "has_prev_crc", "If True, previous PES packet CRC follows")
yield Bit(self, "has_extension")
yield UInt8(self, "size")
# Time stamps
if self["has_pts"].value:
yield Bits(self, "sync[]", 4) # =2, or 3 if has_dts=True
yield Timestamp(self, "pts")
if self["has_dts"].value:
if not(self["has_pts"].value):
raise ParserError("Invalid PTS/DTS values")
yield Bits(self, "sync[]", 4) # =1
yield Timestamp(self, "dts")
if self["has_escr"].value:
yield Bits(self, "sync[]", 2) # =0
yield SCRExt(self, "escr")
if self["has_es_rate"].value:
yield Bit(self, "sync[]") # =True
yield Bits(self, "es_rate", 14) # in units of 50 bytes/second
yield Bit(self, "sync[]") # =True
if self["has_copy_info"].value:
yield Bit(self, "sync[]") # =True
yield Bits(self, "copy_info", 7)
if self["has_prev_crc"].value:
yield textHandler(UInt16(self, "prev_crc"), hexadecimal)
# --- Extension ---
if self["has_extension"].value:
yield VideoExtension1(self, "extension")
if self["extension/has_extension2"].value:
yield VideoExtension2(self, "extension2")
class VideoExtension(FieldSet):
EXT_TYPE = {1: 'Sequence', 2: 'Sequence Display', 8: 'Picture Coding'}
def createFields(self):
yield Enum(Bits(self, "ext_type", 4), self.EXT_TYPE)
ext_type = self['ext_type'].value
if ext_type == 1:
# Sequence extension
yield Bits(self, 'profile_and_level', 8)
yield Bit(self, 'progressive_sequence')
yield Bits(self, 'chroma_format', 2)
yield Bits(self, 'horiz_size_ext', 2)
yield Bits(self, 'vert_size_ext', 2)
yield Bits(self, 'bit_rate_ext', 12)
yield Bits(self, 'pad[]', 1)
yield Bits(self, 'vbv_buffer_size_ext', 8)
yield Bit(self, 'low_delay')
yield Bits(self, 'frame_rate_ext_n', 2)
yield Bits(self, 'frame_rate_ext_d', 5)
elif ext_type == 2:
# Sequence Display extension
yield Bits(self, 'video_format', 3)
yield Bit(self, 'color_desc_present')
if self['color_desc_present'].value:
yield UInt8(self, 'color_primaries')
yield UInt8(self, 'transfer_characteristics')
yield UInt8(self, 'matrix_coeffs')
yield Bits(self, 'display_horiz_size', 14)
yield Bits(self, 'pad[]', 1)
yield Bits(self, 'display_vert_size', 14)
yield NullBits(self, 'pad[]', 3)
elif ext_type == 8:
yield Bits(self, 'f_code[0][0]', 4, description="forward horizontal")
yield Bits(self, 'f_code[0][1]', 4, description="forward vertical")
yield Bits(self, 'f_code[1][0]', 4, description="backward horizontal")
yield Bits(self, 'f_code[1][1]', 4, description="backward vertical")
yield Bits(self, 'intra_dc_precision', 2)
yield Bits(self, 'picture_structure', 2)
yield Bit(self, 'top_field_first')
yield Bit(self, 'frame_pred_frame_dct')
yield Bit(self, 'concealment_motion_vectors')
yield Bit(self, 'q_scale_type')
yield Bit(self, 'intra_vlc_format')
yield Bit(self, 'alternate_scan')
yield Bit(self, 'repeat_first_field')
yield Bit(self, 'chroma_420_type')
yield Bit(self, 'progressive_frame')
yield Bit(self, 'composite_display')
if self['composite_display'].value:
yield Bit(self, 'v_axis')
yield Bits(self, 'field_sequence', 3)
yield Bit(self, 'sub_carrier')
yield Bits(self, 'burst_amplitude', 7)
yield Bits(self, 'sub_carrier_phase', 8)
yield NullBits(self, 'pad[]', 2)
else:
yield NullBits(self, 'pad[]', 6)
else:
yield RawBits(self, "raw[]", 4)
class VideoPicture(FieldSet):
CODING_TYPE = ["forbidden", "intra-coded (I)",
"predictive-coded (P)",
"bidirectionally-predictive-coded (B)",
"dc intra-coded (D)", "reserved",
"reserved", "reserved"]
def createFields(self):
yield Bits(self, "temporal_ref", 10)
yield Enum(Bits(self, "coding_type", 3), self.CODING_TYPE)
yield Bits(self, "vbv_delay", 16)
if self['coding_type'].value in (2, 3):
# predictive coding
yield Bit(self, 'full_pel_fwd_vector')
yield Bits(self, 'forward_f_code', 3)
if self['coding_type'].value == 3:
# bidi predictive coding
yield Bit(self, 'full_pel_back_vector')
yield Bits(self, 'backward_f_code', 3)
yield Bits(self, "padding", 8 - (self.current_size % 8))
class VideoSlice(FieldSet):
def createFields(self):
yield Bits(self, "quantizer_scale", 5)
start = self.absolute_address + self.current_size + 3
# seek forward by at most 1MB
pos = self.stream.searchBytes(
b'\0\0\1', start, start + 1024 * 1024 * 8)
if pos is None:
pos = self.root.size
yield RawBits(self, "data", pos - start + 3)
class VideoChunk(FieldSet):
tag_info = {
0x00: ("pict_start[]", VideoPicture, "Picture start"),
0xB2: ("data_start[]", None, "Data start"),
0xB3: ("seq_hdr[]", VideoSeqHeader, "Sequence header"),
0xB4: ("seq_err[]", None, "Sequence error"),
0xB5: ("ext_start[]", VideoExtension, "Extension start"),
0xB7: ("seq_end[]", None, "Sequence end"),
0xB8: ("group_start[]", GroupStart, "Group start"),
}
def __init__(self, *args):
FieldSet.__init__(self, *args)
tag = self["tag"].value
if tag in self.tag_info:
self._name, self.parser, self._description = self.tag_info[tag]
if not self.parser:
self.parser = defaultParser
elif 0x01 <= tag <= 0xaf:
self._name, self.parser, self._description = (
'slice[]', VideoSlice, 'Picture slice')
else:
self.parser = defaultParser
def createFields(self):
yield Bytes(self, "sync", 3)
yield textHandler(UInt8(self, "tag"), hexadecimal)
if self.parser and self['tag'].value != 0xb7:
yield self.parser(self, "content")
class VideoStream(Parser):
endian = BIG_ENDIAN
def createFields(self):
while self.current_size < self.size:
# seek forward by at most 1MB
pos = self.stream.searchBytes(
b'\0\0\1', self.current_size, self.current_size + 1024 * 1024 * 8)
if pos is not None:
padsize = pos - self.current_size
if padsize:
yield PaddingBytes(self, "pad[]", padsize // 8)
yield VideoChunk(self, "chunk[]")
class Stream(FieldSet):
def createFields(self):
padding = 0
position = 0
streamlength = self["../length"].value
while position < streamlength * 8:
next = ord(self.parent.stream.readBytes(
self.absolute_address + self.current_size + position, 1))
if next == 0xff:
padding += 1
position += 8
elif padding:
yield PaddingBytes(self, "pad[]", padding)
padding = None
position = 0
elif 0x40 <= next <= 0x7f:
yield Bits(self, "scale_marker", 2) # 1
yield Bit(self, "scale")
scale = self['scale'].value
if scale:
scaleval = 1024
else:
scaleval = 128
yield textHandler(Bits(self, "size", 13), lambda field: str(field.value * scaleval))
elif 0x00 <= next <= 0x3f:
yield Bits(self, "ts_marker", 2) # 0
yield Bit(self, "has_pts")
yield Bit(self, "has_dts")
if self['has_pts'].value:
yield Timestamp(self, "pts")
if self['has_dts'].value:
yield PaddingBits(self, "pad[]", 4)
yield Timestamp(self, "dts")
if self.current_size % 8 == 4:
yield PaddingBits(self, "pad[]", 4)
break
elif 0x80 <= next <= 0xbf:
# MPEG-2 extension
yield PacketElement(self, "pkt")
break
else:
# 0xc0 - 0xfe: unknown
break
length = self["../length"].value - self.current_size // 8
if length:
tag = self['../tag'].value
group = self.root.streamgroups[tag]
parname = self.parent._name
if parname.startswith('audio'):
frag = CustomFragment(
self, "data", length * 8, MpegAudioFile, group=group)
elif parname.startswith('video'):
frag = CustomFragment(
self, "data", length * 8, VideoStream, group=group)
else:
frag = CustomFragment(
self, "data", length * 8, None, group=group)
self.root.streamgroups[tag] = frag.group
yield frag
class Chunk(FieldSet):
ISO_END_CODE = 0xB9
tag_info = {
0xB9: ("end", None, "End"),
0xBA: ("pack_start[]", PackHeader, "Pack start"),
0xBB: ("system_start[]", SystemHeader, "System start"),
# streams
0xBD: ("private[]", Stream, "Private elementary"),
0xBE: ("padding[]", Stream, "Padding"),
# 0xC0 to 0xFE handled specially
0xFF: ("directory[]", Stream, "Program Stream Directory"),
}
def __init__(self, *args):
FieldSet.__init__(self, *args)
if not hasattr(self.root, 'streamgroups'):
self.root.streamgroups = {}
for tag in range(0xBC, 0x100):
self.root.streamgroups[tag] = None
tag = self["tag"].value
if tag in self.tag_info:
self._name, self.parser, self._description = self.tag_info[tag]
elif 0xBC <= tag <= 0xFF:
if 0xC0 <= tag < 0xE0:
# audio
streamid = tag - 0xC0
self._name, self.parser, self._description = (
"audio[%i][]" % streamid, Stream, "Audio Stream %i Packet" % streamid)
elif 0xE0 <= tag < 0xF0:
# video
streamid = tag - 0xE0
self._name, self.parser, self._description = (
"video[%i][]" % streamid, Stream, "Video Stream %i Packet" % streamid)
else:
self._name, self.parser, self._description = (
"stream[]", Stream, "Data Stream Packet")
else:
self.parser = defaultParser
if not self.parser:
self.parser = defaultParser
elif self.parser != PackHeader and "length" in self:
self._size = (6 + self["length"].value) * 8
def createFields(self):
yield Bytes(self, "sync", 3)
yield textHandler(UInt8(self, "tag"), hexadecimal)
if self.parser:
if self.parser != PackHeader:
yield UInt16(self, "length")
if not self["length"].value:
return
yield self.parser(self, "content")
def createDescription(self):
return "Chunk: tag %s" % self["tag"].display
class MPEGVideoFile(Parser):
PARSER_TAGS = {
"id": "mpeg_video",
"category": "video",
"file_ext": ("mpeg", "mpg", "mpe", "vob"),
"mime": ("video/mpeg", "video/mp2p"),
"min_size": 12 * 8,
# TODO: "magic": xxx,
"description": "MPEG video, version 1 or 2"
}
endian = BIG_ENDIAN
version = None
def createFields(self):
while self.current_size < self.size:
# seek forward by at most 1MB
pos = self.stream.searchBytes(
b'\0\0\1', self.current_size, self.current_size + 1024 * 1024 * 8)
if pos is not None:
padsize = pos - self.current_size
if padsize:
yield PaddingBytes(self, "pad[]", padsize // 8)
chunk = Chunk(self, "chunk[]")
try:
# force chunk to be processed, so that CustomFragments are
# complete
chunk['content/data']
except Exception:
pass
yield chunk
def validate(self):
try:
pack = self[0]
except FieldError:
return "Unable to create first chunk"
if pack.name != "pack_start[0]":
return "Invalid first chunk"
if pack["sync"].value != b"\0\0\1":
return "Invalid synchronisation"
return pack["content"].validate()
def getVersion(self):
if not self.version:
if self["pack_start[0]/content/sync[0]"].size == 2:
self.version = 2
else:
self.version = 1
return self.version
def createDescription(self):
if self.getVersion() == 2:
return "MPEG-2 video"
else:
return "MPEG-1 video"
Reference in a new issue Copy permalink