Separate MKV class and functions in a separate module.
This commit is contained in:
Frédéric Tronel
@@ -0,0 +1,332 @@
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
#
|
||||
# Copyright (C) 2026 Frédéric Tronel
|
||||
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from math import floor
|
||||
from typing import IO
|
||||
|
||||
|
||||
from typeguard import typechecked
|
||||
import hexdump
|
||||
|
||||
from tscut.h264.bitstream import (
|
||||
read_bit,
|
||||
read_bits,
|
||||
read_byte,
|
||||
read_long,
|
||||
read_word,
|
||||
write_bits,
|
||||
write_byte,
|
||||
write_word,
|
||||
sodb_to_rbsp,
|
||||
rbsp_to_sodb
|
||||
)
|
||||
|
||||
from tscut.h264.parameters import (
|
||||
SPS,
|
||||
PPS,
|
||||
)
|
||||
|
||||
from tscut.matroska.ebml import get_ebml_length
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class AVCDecoderConfiguration:
|
||||
configuration_version:int=1 # u(8)
|
||||
avc_profile_indication:int=0 # u(8)
|
||||
profile_compatibility:int=0 # u(8)
|
||||
avc_level_indication:int=0 # u(8)
|
||||
length_size_minus_one:int=0 # u(2) (0,1 or 3)
|
||||
num_of_sequence_parameter_sets:int=0 # u(5)
|
||||
sps:dict = field(default_factory=dict)
|
||||
num_of_picture_parameter_sets:int=0 #u(8)
|
||||
pps:dict = field(default_factory=dict)
|
||||
chroma_format:int=0 # u(2)
|
||||
bit_depth_luma_minus8:int=0 # u(3)
|
||||
bit_depth_chroma_minus8:int=0 # u(3)
|
||||
num_of_sequence_parameter_set_ext:int=0 # u(8)
|
||||
spsext:dict = field(default_factory=dict)
|
||||
|
||||
def __init__(self):
|
||||
self.sps = {}
|
||||
self.spsext = {}
|
||||
self.pps = {}
|
||||
|
||||
def from_bytes(self, buf):
|
||||
logger.debug('Parsing: %s', (hexdump.dump(buf,sep=':')))
|
||||
bit_position = 0
|
||||
bit_position, self.configuration_version = read_byte(buf, bit_position)
|
||||
bit_position, self.avc_profile_indication = read_byte(buf, bit_position)
|
||||
bit_position, self.profile_compatibility = read_byte(buf, bit_position)
|
||||
bit_position, self.avc_level_indication = read_byte(buf, bit_position)
|
||||
bit_position, v = read_bits(buf, bit_position, 6)
|
||||
if v != 0b111111:
|
||||
raise ValueError(f'Reserved bits are not equal to 0b111111: {v:x}')
|
||||
bit_position, self.length_size_minus_one = read_bits(buf, bit_position, 2)
|
||||
bit_position, v = read_bits(buf, bit_position, 3)
|
||||
if v != 0b111:
|
||||
raise ValueError(f'Reserved bits are not equal to 0b111: {v:x}')
|
||||
bit_position, self.num_of_sequence_parameter_sets= read_bits(buf, bit_position, 5)
|
||||
logger.debug('Number of SPS: %d', self.num_of_sequence_parameter_sets)
|
||||
for _ in range(self.num_of_sequence_parameter_sets):
|
||||
bit_position, length = read_word(buf, bit_position)
|
||||
if bit_position % 8 != 0:
|
||||
raise ValueError(f'SPS is not located at a byte boundary: {bit_position:d}')
|
||||
|
||||
sps = SPS()
|
||||
sodb = rbsp_to_sodb(buf[floor(bit_position/8):])
|
||||
bit_length = sps.from_bytes(sodb)
|
||||
spsid = sps.seq_parameter_set_id
|
||||
self.sps[spsid] = sps
|
||||
|
||||
parsed_length = floor(bit_length/8)
|
||||
logger.debug('Expected length of SPS: %d bytes. Parsed: %d bytes', length,
|
||||
parsed_length)
|
||||
# Parse length can be shorter than length because of rewriting from RBSP to SODB
|
||||
# (that is shorter).
|
||||
# So we advance of indicated length.
|
||||
bit_position+=length*8
|
||||
|
||||
logger.debug('Bit position:%d. Reading one byte of: %s', bit_position,
|
||||
hexdump.dump(buf[floor(bit_position/8):], sep=':'))
|
||||
bit_position, self.num_of_picture_parameter_sets = read_byte(buf, bit_position)
|
||||
logger.debug('Number of PPS: %d', self.num_of_picture_parameter_sets)
|
||||
for _ in range(self.num_of_picture_parameter_sets):
|
||||
bit_position, length = read_word(buf, bit_position)
|
||||
if bit_position % 8 != 0:
|
||||
raise ValueError('PPS is not located at a byte boundary: {bit_position:d}')
|
||||
|
||||
pps = PPS()
|
||||
sodb = rbsp_to_sodb(buf[floor(bit_position/8):])
|
||||
bit_length = pps.from_bytes(sodb, self.chroma_format)
|
||||
ppsid = pps.pic_parameter_set_id
|
||||
self.pps[ppsid] = pps
|
||||
|
||||
parsed_length = floor(bit_length/8)
|
||||
logger.debug('Expected length of PPS: %d bytes. Parsed: %d bytes', length,
|
||||
parsed_length)
|
||||
# Parse length can be shorter than length because of rewriting from RBSP to SODB
|
||||
# (that is shorter).
|
||||
# So we advance of indicated length.
|
||||
bit_position+=length*8
|
||||
|
||||
logger.debug('Remaining bits: %s', hexdump.dump(buf[floor(bit_position/8):]))
|
||||
|
||||
if self.avc_profile_indication in [100, 110, 122, 144]:
|
||||
bit_position, reserved = read_bits(buf, bit_position, 6)
|
||||
if reserved != 0b111111:
|
||||
raise ValueError(f'Reserved bits are different from 111111: {reserved:x}')
|
||||
bit_position, self.chroma_format = read_bits(buf, bit_position, 2)
|
||||
bit_position, reserved = read_bits(buf, bit_position, 5)
|
||||
if reserved != 0b11111:
|
||||
raise ValueError(f'Reserved bits are different from 11111: {reserved:x}')
|
||||
bit_position, self.bit_depth_luma_minus8 = read_bits(buf, bit_position, 3)
|
||||
bit_position, reserved = read_bits(buf, bit_position, 5)
|
||||
if reserved != 0b11111:
|
||||
raise ValueError(f'Reserved bits are different from 11111: {reserved:x}')
|
||||
bit_position, self.bit_depth_chroma_minus8 = read_bits(buf, bit_position, 3)
|
||||
bit_position, self.num_of_sequence_parameter_set_ext = read_byte(buf, bit_position)
|
||||
for _ in range(self.num_of_sequence_parameter_set_ext):
|
||||
# TODO: parse SPSextended
|
||||
logger.error('Parsing of SPS extended not yet implemented !')
|
||||
pass
|
||||
|
||||
|
||||
|
||||
def to_bytes(self):
|
||||
buf = bytearray()
|
||||
bit_position = 0
|
||||
bit_position = write_byte(buf, bit_position, self.configuration_version)
|
||||
bit_position = write_byte(buf, bit_position, self.avc_profile_indication)
|
||||
bit_position = write_byte(buf, bit_position, self.profile_compatibility)
|
||||
bit_position = write_byte(buf, bit_position, self.avc_level_indication)
|
||||
bit_position = write_bits(buf, bit_position, 0b111111, 6)
|
||||
bit_position = write_bits(buf, bit_position, self.length_size_minus_one, 2)
|
||||
bit_position = write_bits(buf, bit_position, 0b111, 3)
|
||||
bit_position = write_bits(buf, bit_position, self.num_of_sequence_parameter_sets, 5)
|
||||
for spsid, sps in self.sps.items():
|
||||
sodb = sps.to_bytes()
|
||||
sodb_length = len(sodb)
|
||||
rbsp = sodb_to_rbsp(sodb)
|
||||
rbsp_length = len(rbsp)
|
||||
|
||||
logger.debug('SODB length: %d RBSP length:%d', sodb_length, rbsp_length)
|
||||
|
||||
bit_position = write_word(buf, bit_position, rbsp_length)
|
||||
buf.extend(rbsp)
|
||||
bit_position+=rbsp_length*8
|
||||
|
||||
logger.debug('2. Buffer: %s', hexdump.dump(buf, sep=':'))
|
||||
|
||||
bit_position = write_byte(buf, bit_position, self.num_of_picture_parameter_sets)
|
||||
for ppsid, lpps in self.pps.items():
|
||||
logger.debug('Writing PPS: %d', ppsid)
|
||||
# TODO: does chroma_format should come from self ?
|
||||
sodb = lpps.to_bytes(self.chroma_format)
|
||||
sodb_length = len(sodb)
|
||||
rbsp = sodb_to_rbsp(sodb)
|
||||
rbsp_length = len(rbsp)
|
||||
|
||||
logger.debug('SODB length: %d RBSP length:%d', sodb_length, rbsp_length)
|
||||
|
||||
bit_position = write_word(buf, bit_position, rbsp_length)
|
||||
buf.extend(rbsp)
|
||||
bit_position+=rbsp_length*8
|
||||
|
||||
if self.avc_profile_indication in [ 100, 110, 122, 144]:
|
||||
bit_position = write_bits(buf, bit_position, 0b111111, 6)
|
||||
bit_position = write_bits(buf, bit_position, self.chroma_format, 2)
|
||||
bit_position = write_bits(buf, bit_position, 0b11111, 5)
|
||||
bit_position = write_bits(buf, bit_position, self.bit_depth_luma_minus8, 3)
|
||||
bit_position = write_bits(buf, bit_position, 0b11111, 5)
|
||||
bit_position = write_bits(buf, bit_position, self.bit_depth_chroma_minus8, 3)
|
||||
bit_position = write_byte(buf, bit_position, self.num_of_sequence_parameter_set_ext)
|
||||
for _ in range(self.num_of_sequence_parameter_set_ext):
|
||||
# TODO: dump SPSextended
|
||||
logger.error('Dumping SPS extended not yet implemented')
|
||||
pass
|
||||
|
||||
return buf
|
||||
|
||||
def merge(self, config):
|
||||
# Check config compatibility
|
||||
if self.configuration_version != config.configuration_version:
|
||||
raise ValueError(f'Configuration versions are different: {self.configuration_version:d}\
|
||||
vs {config.configuration_version:d}')
|
||||
if self.avc_profile_indication != config.avc_profile_indication:
|
||||
raise ValueError(f'AVC profiles are different: {self.avc_profile_indication:d} vs \
|
||||
{config.avc_profile_indication:d}')
|
||||
if self.profile_compatibility != config.profile_compatibility:
|
||||
raise ValueError(f'Profile compatilities are different: {self.profile_compatibility:d} \
|
||||
vs {config.profile_compatibility:d}')
|
||||
if self.avc_level_indication != config.avc_level_indication:
|
||||
raise ValueError(f'Level indications are different: {self.avc_level_indication:d} vs \
|
||||
{config.avc_level_indication:d}')
|
||||
if self.length_size_minus_one != config.length_size_minus_one:
|
||||
raise ValueError(f'Length units are different: {self.length_size_minus_one:d} vs \
|
||||
{config.length_size_minus_one:d}')
|
||||
if self.chroma_format != config.chroma_format:
|
||||
raise ValueError(f'Colour format are different: {self.chroma_format:d} vs \
|
||||
{config.chroma_format:d}')
|
||||
if self.bit_depth_luma_minus8 != config.bit_depth_luma_minus8:
|
||||
raise ValueError(f'Depth of luminance are different: {self.bit_depth_luma_minus8:d} vs \
|
||||
{config.bit_depth_luma_minus8:d}')
|
||||
if self.bit_depth_chroma_minus8 != config.bit_depth_chroma_minus8:
|
||||
raise ValueError(f'Depth of chromaticity are different: \
|
||||
{self.bit_depth_chroma_minus8:d} vs {config.bit_depth_luma_minus8:d}')
|
||||
|
||||
for spsid in config.sps:
|
||||
sps = config.sps[spsid]
|
||||
if spsid in self.sps:
|
||||
localsps = self.sps[spsid]
|
||||
if sps!=localsps:
|
||||
raise ValueError(f'Profile are not compatible. They contain two different SPS\
|
||||
with the same identifier ({spsid:d}): {localsps}\n{sps}\n')
|
||||
self.sps[spsid] = sps
|
||||
|
||||
self.num_of_sequence_parameter_sets = len(self.sps)
|
||||
|
||||
for ppsid in config.pps:
|
||||
pps = config.pps[ppsid]
|
||||
if ppsid in self.pps:
|
||||
localpps = self.pps[ppsid]
|
||||
if pps!=localpps:
|
||||
raise ValueError(f'Profile are not compatible. They contain two different PPS\
|
||||
with the same identifier ({ppsid:d}): {localpps}\n{pps}\n')
|
||||
self.pps[ppsid] = pps
|
||||
|
||||
self.num_of_picture_parameter_sets = len(self.pps)
|
||||
|
||||
# TODO: do the same with extended SPS !
|
||||
|
||||
@typechecked
|
||||
def parse_codec_private(codec_private_data: bytes) -> AVCDecoderConfiguration:
|
||||
if codec_private_data[0] != 0x63:
|
||||
raise ValueError(f'Matroska header is wrong: {codec_private_data[0]:x}')
|
||||
if codec_private_data[1] != 0xA2:
|
||||
raise ValueError(f'Matroska header is wrong: {codec_private_data[1]:x}')
|
||||
length = codec_private_data[2]
|
||||
if length == 0:
|
||||
raise ValueError('Matroska length cannot start with zero byte.')
|
||||
for nb_zeroes in range(8):
|
||||
b = read_bit(codec_private_data[2:], nb_zeroes)
|
||||
if b != 0:
|
||||
break
|
||||
mask = 2^(7-nb_zeroes)-1
|
||||
length = codec_private_data[2] and mask
|
||||
for i in range(nb_zeroes):
|
||||
length*=256
|
||||
length+=(codec_private_data[3+i])
|
||||
byte_position = 3+nb_zeroes
|
||||
avcconfig = AVCDecoderConfiguration()
|
||||
avcconfig.from_bytes(codec_private_data[byte_position:])
|
||||
|
||||
return avcconfig
|
||||
|
||||
@typechecked
|
||||
def get_avc_config_from_h264(input_file: IO[bytes]) -> AVCDecoderConfiguration:
|
||||
# TODO: improve this ...
|
||||
rbsp = input_file.read(1000)
|
||||
sodb = rbsp_to_sodb(rbsp)
|
||||
|
||||
bit_position = 0
|
||||
bit_position, start_code = read_long(sodb, bit_position)
|
||||
if start_code != 1:
|
||||
raise ValueError(f'Starting code not detected: {start_code:x}')
|
||||
sps = SPS()
|
||||
bit_length = sps.from_bytes(sodb[4:])
|
||||
bit_position+=bit_length
|
||||
|
||||
bit_position, start_code = read_long(sodb, bit_position)
|
||||
if start_code != 1:
|
||||
raise ValueError(f'Starting code not detected: {start_code:x}')
|
||||
pps = PPS()
|
||||
bit_length = pps.from_bytes(sodb[floor(bit_position/8):], sps.chroma_format_idc)
|
||||
logger.debug(pps)
|
||||
|
||||
avcconfig = AVCDecoderConfiguration()
|
||||
avcconfig.configuration_version = 1
|
||||
avcconfig.avc_profile_indication = sps.profile_idc
|
||||
avcconfig.profile_compatibility = 0
|
||||
avcconfig.avc_level_indication = sps.level_idc
|
||||
avcconfig.length_size_minus_one = 3
|
||||
avcconfig.num_of_sequence_parameter_sets = 1
|
||||
avcconfig.num_of_picture_parameter_sets = 1
|
||||
avcconfig.num_of_sequence_parameter_set_ext = 0
|
||||
avcconfig.chroma_format = sps.chroma_format_idc
|
||||
avcconfig.bit_depth_chroma_minus8 = sps.bit_depth_chroma_minus8
|
||||
avcconfig.bit_depth_luma_minus8 = sps.bit_depth_luma_minus8
|
||||
avcconfig.sps[sps.seq_parameter_set_id] = sps
|
||||
avcconfig.pps[pps.pic_parameter_set_id] = pps
|
||||
|
||||
return avcconfig
|
||||
|
||||
# Unused ?
|
||||
@typechecked
|
||||
def get_codec_private_data_from_h264(input_file: IO[bytes]) -> AVCDecoderConfiguration:
|
||||
avcconfig = get_avc_config_from_h264(input_file)
|
||||
res = dump_codec_private_data(avcconfig)
|
||||
|
||||
return res
|
||||
|
||||
@typechecked
|
||||
def dump_codec_private_data(avc_decoder_configuration: AVCDecoderConfiguration) -> bytearray:
|
||||
# Rebuild a Matroska Codec Private Element
|
||||
res = bytearray()
|
||||
# Code private element
|
||||
res.extend(b'\x63\xA2')
|
||||
buf = avc_decoder_configuration.to_bytes()
|
||||
logger.debug('AVC configuration bitstream: %s (length: %d))', hexdump.dump(buf, sep=':'),
|
||||
len(buf))
|
||||
|
||||
embl_length = get_ebml_length(len(buf))
|
||||
logger.debug('EMBL encoded length: %s', hexdump.dump(embl_length, sep=':'))
|
||||
res.extend(embl_length)
|
||||
res.extend(buf)
|
||||
|
||||
return res
|
||||
@@ -1,3 +1,7 @@
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
#
|
||||
# Copyright (C) 2026 Frédéric Tronel
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from math import floor, log, ceil
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
# SPDX-License-Identifier: GPL-2.0-or-later
|
||||
#
|
||||
# Copyright (C) 2026 Frédéric Tronel
|
||||
|
||||
import logging
|
||||
from os import (
|
||||
SEEK_SET,
|
||||
fstat,
|
||||
ftruncate,
|
||||
lseek,
|
||||
read,
|
||||
write,
|
||||
)
|
||||
from typing import IO
|
||||
|
||||
from typeguard import typechecked
|
||||
import hexdump
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#MKV is formatted as an EBML file (Extended Binary Markup Langage).
|
||||
# cf http://matroska-org.github.io/libebml/specs.html
|
||||
# It is a Type, Length, Value (TLV) kind of binary file.
|
||||
# Types are encoded as follows:
|
||||
# 1xxx xxxx - Class A IDs (2^7 -1 possible values)
|
||||
# 01xx xxxx xxxx xxxx - Class B IDs (2^14-1 possible values)
|
||||
# 001x xxxx xxxx xxxx xxxx xxxx - Class C IDs (2^21-1 possible values)
|
||||
# 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx - Class D IDs (2^28-1 possible values)
|
||||
# Lengths are encoded as follows:
|
||||
# 1xxx xxxx
|
||||
# value 0 to 2^7-2
|
||||
# 01xx xxxx xxxx xxxx
|
||||
# value 0 to 2^14-2
|
||||
# 001x xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^21-2
|
||||
# 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^28-2
|
||||
# 0000 1xxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^35-2
|
||||
# 0000 01xx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^42-2
|
||||
# 0000 001x xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^49-2
|
||||
# 0000 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^56-2
|
||||
|
||||
@typechecked
|
||||
def get_ebml_length(length:int) -> bytes|None:
|
||||
if 0 <= length <= 2**7-2:
|
||||
size = 1
|
||||
elif length <= 2**14-2:
|
||||
size = 2
|
||||
elif length <= 2**21-2:
|
||||
size = 3
|
||||
elif length <= 2**28-2:
|
||||
size = 4
|
||||
elif length <= 2**35-2:
|
||||
size = 5
|
||||
elif length <= 2**42-2:
|
||||
size = 6
|
||||
elif length <= 2**49-2:
|
||||
size = 7
|
||||
elif length <= 2**56-2:
|
||||
size = 8
|
||||
elif length < 0:
|
||||
logger.error('Impossible to encode a negative length with EBML.')
|
||||
return None
|
||||
else:
|
||||
logger.error('Impossible to encode a length larger than 2^56-2 with EBML.')
|
||||
return None
|
||||
|
||||
encoded_length = length + ((128>>(size-1))<<((size-1)*8))
|
||||
res = (encoded_length).to_bytes(size, byteorder='big')
|
||||
return res
|
||||
|
||||
|
||||
@typechecked
|
||||
def change_ebml_element_size(input_file: IO[bytes], position:int, addendum:int) -> int:
|
||||
initial_position = position
|
||||
infd = input_file.fileno()
|
||||
lseek(infd, position, SEEK_SET)
|
||||
|
||||
buf = read(infd, 1)
|
||||
element_type = int.from_bytes(buf, byteorder='big')
|
||||
mask=128
|
||||
found = False
|
||||
for i in range(1,5):
|
||||
if element_type&mask:
|
||||
type_size = i
|
||||
found = True
|
||||
break
|
||||
mask = mask>>1
|
||||
|
||||
if not found:
|
||||
logger.error('Size of element type cannot be determined: %d', element_type)
|
||||
raise ValueError()
|
||||
|
||||
# We seek to size
|
||||
position+=type_size
|
||||
lseek(infd, position, SEEK_SET)
|
||||
|
||||
buf = read(infd, 1)
|
||||
size_head = int.from_bytes(buf, byteorder='big')
|
||||
logger.info('First byte of size: %x', size_head)
|
||||
mask=128
|
||||
found = False
|
||||
for i in range(1,9):
|
||||
if size_head&mask:
|
||||
size_of_data_size = i
|
||||
found = True
|
||||
break
|
||||
mask = mask>>1
|
||||
|
||||
if not found:
|
||||
logger.error('Size of data size cannot be determined: %d', size_head)
|
||||
raise ValueError()
|
||||
logger.info('Size of data size: %d.', size_of_data_size)
|
||||
|
||||
lseek(infd, position, SEEK_SET)
|
||||
old_size_buf = read(infd, size_of_data_size)
|
||||
max_size = 2**(size_of_data_size*7)-2
|
||||
size_of_data = int.from_bytes(old_size_buf, byteorder='big')
|
||||
logger.info('Size of data with mask: %x mask: %d.', size_of_data, mask)
|
||||
size_of_data-= (mask<<((size_of_data_size-1)*8))
|
||||
logger.info('Found element at position: %d, size of type: %d size of data: %d \
|
||||
maximal size: %d.', initial_position, type_size, size_of_data, max_size)
|
||||
|
||||
new_size = size_of_data+addendum
|
||||
delta = 0
|
||||
if new_size > max_size:
|
||||
# TODO: Test this code ...
|
||||
new_encoded_size = get_ebml_length(new_size)
|
||||
size_of_new_encoded_size = len(new_encoded_size)
|
||||
if size_of_new_encoded_size <= size_of_data_size:
|
||||
logger.error('New encoded size is smaller (%d) or equal than previous size (%d).\
|
||||
This should not happen.', size_of_new_encoded_size, size_of_data_size)
|
||||
raise ValueError()
|
||||
# The difference of length between old size field and new one.
|
||||
delta = size_of_new_encoded_size - size_of_data_size
|
||||
file_length = fstat(infd).st_size
|
||||
# We seek after actual length field
|
||||
lseek(infd, position+size_of_data_size, SEEK_SET)
|
||||
# We read the rest of file
|
||||
tail = read(infd, file_length-(position+size_of_data_size))
|
||||
# We increase file length
|
||||
ftruncate(infd, file_length+delta)
|
||||
# We go to the beginning of length field
|
||||
lseek(infd, position, SEEK_SET)
|
||||
# We write the new length field
|
||||
write(infd, new_encoded_size)
|
||||
# We overwrite the rest of file with its previous content that has been offset.
|
||||
write(infd, tail)
|
||||
else:
|
||||
size = new_size + ((128>>(size_of_data_size-1))<<((size_of_data_size-1)*8))
|
||||
new_size_buf = (size).to_bytes(size_of_data_size, byteorder='big')
|
||||
|
||||
logger.info('Old encoded size: %s New encoded size: %s', hexdump.dump(old_size_buf,sep=':'),
|
||||
hexdump.dump(new_size_buf, sep=':'))
|
||||
lseek(infd, position, SEEK_SET)
|
||||
write(infd, new_size_buf)
|
||||
|
||||
# We return the potential increase in size of the file if the length field had to be increased.
|
||||
return delta
|
||||
+3
-471
@@ -6,7 +6,6 @@ import json
|
||||
import logging
|
||||
import os.path
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import timedelta
|
||||
from enum import IntEnum, unique
|
||||
from io import BytesIO, TextIOWrapper
|
||||
@@ -34,22 +33,9 @@ from iso639.exceptions import InvalidLanguageValue
|
||||
from tqdm import tqdm
|
||||
from typeguard import typechecked
|
||||
|
||||
from tscut.h264.bitstream import (
|
||||
read_bit,
|
||||
read_bits,
|
||||
read_byte,
|
||||
read_word,
|
||||
read_long,
|
||||
write_bits,
|
||||
write_byte,
|
||||
write_word,
|
||||
sodb_to_rbsp,
|
||||
rbsp_to_sodb,
|
||||
)
|
||||
from tscut.h264.parameters import SPS, PPS
|
||||
|
||||
# Local modules
|
||||
# TODO: create local modules for MP4, MKV
|
||||
from tscut.h264.avc import (dump_codec_private_data,
|
||||
get_avc_config_from_h264,
|
||||
parse_codec_private)
|
||||
|
||||
# Useful SPS/PPS discussion.
|
||||
# https://copyprogramming.com/howto/including-sps-and-pps-in-a-raw-h264-track
|
||||
@@ -455,294 +441,6 @@ def get_codec_private_data_from_mkv(mkvinfo_path:str,
|
||||
|
||||
return None, None
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class AVCDecoderConfiguration:
|
||||
configuration_version:int=1 # u(8)
|
||||
avc_profile_indication:int=0 # u(8)
|
||||
profile_compatibility:int=0 # u(8)
|
||||
avc_level_indication:int=0 # u(8)
|
||||
length_size_minus_one:int=0 # u(2) (0,1 or 3)
|
||||
num_of_sequence_parameter_sets:int=0 # u(5)
|
||||
sps:dict = field(default_factory=dict)
|
||||
num_of_picture_parameter_sets:int=0 #u(8)
|
||||
pps:dict = field(default_factory=dict)
|
||||
chroma_format:int=0 # u(2)
|
||||
bit_depth_luma_minus8:int=0 # u(3)
|
||||
bit_depth_chroma_minus8:int=0 # u(3)
|
||||
num_of_sequence_parameter_set_ext:int=0 # u(8)
|
||||
spsext:dict = field(default_factory=dict)
|
||||
|
||||
def __init__(self):
|
||||
self.sps = {}
|
||||
self.spsext = {}
|
||||
self.pps = {}
|
||||
|
||||
def from_bytes(self, buf):
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.debug('Parsing: %s', (hexdump.dump(buf,sep=':')))
|
||||
bit_position = 0
|
||||
bit_position, self.configuration_version = read_byte(buf, bit_position)
|
||||
bit_position, self.avc_profile_indication = read_byte(buf, bit_position)
|
||||
bit_position, self.profile_compatibility = read_byte(buf, bit_position)
|
||||
bit_position, self.avc_level_indication = read_byte(buf, bit_position)
|
||||
bit_position, v = read_bits(buf, bit_position, 6)
|
||||
if v != 0b111111:
|
||||
raise ValueError(f'Reserved bits are not equal to 0b111111: {v:x}')
|
||||
bit_position, self.length_size_minus_one = read_bits(buf, bit_position, 2)
|
||||
bit_position, v = read_bits(buf, bit_position, 3)
|
||||
if v != 0b111:
|
||||
raise ValueError(f'Reserved bits are not equal to 0b111: {v:x}')
|
||||
bit_position, self.num_of_sequence_parameter_sets= read_bits(buf, bit_position, 5)
|
||||
logger.debug('Number of SPS: %d', self.num_of_sequence_parameter_sets)
|
||||
for _ in range(self.num_of_sequence_parameter_sets):
|
||||
bit_position, length = read_word(buf, bit_position)
|
||||
if bit_position % 8 != 0:
|
||||
raise ValueError(f'SPS is not located at a byte boundary: {bit_position:d}')
|
||||
|
||||
sps = SPS()
|
||||
sodb = rbsp_to_sodb(buf[floor(bit_position/8):])
|
||||
bit_length = sps.from_bytes(sodb)
|
||||
spsid = sps.seq_parameter_set_id
|
||||
self.sps[spsid] = sps
|
||||
|
||||
parsed_length = floor(bit_length/8)
|
||||
logger.debug('Expected length of SPS: %d bytes. Parsed: %d bytes', length,
|
||||
parsed_length)
|
||||
# Parse length can be shorter than length because of rewriting from RBSP to SODB
|
||||
# (that is shorter).
|
||||
# So we advance of indicated length.
|
||||
bit_position+=length*8
|
||||
|
||||
logger.debug('Bit position:%d. Reading one byte of: %s', bit_position,
|
||||
hexdump.dump(buf[floor(bit_position/8):], sep=':'))
|
||||
bit_position, self.num_of_picture_parameter_sets = read_byte(buf, bit_position)
|
||||
logger.debug('Number of PPS: %d', self.num_of_picture_parameter_sets)
|
||||
for _ in range(self.num_of_picture_parameter_sets):
|
||||
bit_position, length = read_word(buf, bit_position)
|
||||
if bit_position % 8 != 0:
|
||||
raise ValueError('PPS is not located at a byte boundary: {bit_position:d}')
|
||||
|
||||
pps = PPS()
|
||||
sodb = rbsp_to_sodb(buf[floor(bit_position/8):])
|
||||
bit_length = pps.from_bytes(sodb, self.chroma_format)
|
||||
ppsid = pps.pic_parameter_set_id
|
||||
self.pps[ppsid] = pps
|
||||
|
||||
parsed_length = floor(bit_length/8)
|
||||
logger.debug('Expected length of PPS: %d bytes. Parsed: %d bytes', length,
|
||||
parsed_length)
|
||||
# Parse length can be shorter than length because of rewriting from RBSP to SODB
|
||||
# (that is shorter).
|
||||
# So we advance of indicated length.
|
||||
bit_position+=length*8
|
||||
|
||||
logger.debug('Remaining bits: %s', hexdump.dump(buf[floor(bit_position/8):]))
|
||||
|
||||
if self.avc_profile_indication in [100, 110, 122, 144]:
|
||||
bit_position, reserved = read_bits(buf, bit_position, 6)
|
||||
if reserved != 0b111111:
|
||||
raise ValueError(f'Reserved bits are different from 111111: {reserved:x}')
|
||||
bit_position, self.chroma_format = read_bits(buf, bit_position, 2)
|
||||
bit_position, reserved = read_bits(buf, bit_position, 5)
|
||||
if reserved != 0b11111:
|
||||
raise ValueError(f'Reserved bits are different from 11111: {reserved:x}')
|
||||
bit_position, self.bit_depth_luma_minus8 = read_bits(buf, bit_position, 3)
|
||||
bit_position, reserved = read_bits(buf, bit_position, 5)
|
||||
if reserved != 0b11111:
|
||||
raise ValueError(f'Reserved bits are different from 11111: {reserved:x}')
|
||||
bit_position, self.bit_depth_chroma_minus8 = read_bits(buf, bit_position, 3)
|
||||
bit_position, self.num_of_sequence_parameter_set_ext = read_byte(buf, bit_position)
|
||||
for _ in range(self.num_of_sequence_parameter_set_ext):
|
||||
# TODO: parse SPSextended
|
||||
logger.error('Parsing of SPS extended not yet implemented !')
|
||||
pass
|
||||
|
||||
|
||||
|
||||
def to_bytes(self):
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
buf = bytearray()
|
||||
bit_position = 0
|
||||
bit_position = write_byte(buf, bit_position, self.configuration_version)
|
||||
bit_position = write_byte(buf, bit_position, self.avc_profile_indication)
|
||||
bit_position = write_byte(buf, bit_position, self.profile_compatibility)
|
||||
bit_position = write_byte(buf, bit_position, self.avc_level_indication)
|
||||
bit_position = write_bits(buf, bit_position, 0b111111, 6)
|
||||
bit_position = write_bits(buf, bit_position, self.length_size_minus_one, 2)
|
||||
bit_position = write_bits(buf, bit_position, 0b111, 3)
|
||||
bit_position = write_bits(buf, bit_position, self.num_of_sequence_parameter_sets, 5)
|
||||
for spsid, sps in self.sps.items():
|
||||
sodb = sps.to_bytes()
|
||||
sodb_length = len(sodb)
|
||||
rbsp = sodb_to_rbsp(sodb)
|
||||
rbsp_length = len(rbsp)
|
||||
|
||||
logger.debug('SODB length: %d RBSP length:%d', sodb_length, rbsp_length)
|
||||
|
||||
bit_position = write_word(buf, bit_position, rbsp_length)
|
||||
buf.extend(rbsp)
|
||||
bit_position+=rbsp_length*8
|
||||
|
||||
logger.debug('2. Buffer: %s', hexdump.dump(buf, sep=':'))
|
||||
|
||||
bit_position = write_byte(buf, bit_position, self.num_of_picture_parameter_sets)
|
||||
for ppsid, lpps in self.pps.items():
|
||||
logger.debug('Writing PPS: %d', ppsid)
|
||||
# TODO: does chroma_format should come from self ?
|
||||
sodb = lpps.to_bytes(self.chroma_format)
|
||||
sodb_length = len(sodb)
|
||||
rbsp = sodb_to_rbsp(sodb)
|
||||
rbsp_length = len(rbsp)
|
||||
|
||||
logger.debug('SODB length: %d RBSP length:%d', sodb_length, rbsp_length)
|
||||
|
||||
bit_position = write_word(buf, bit_position, rbsp_length)
|
||||
buf.extend(rbsp)
|
||||
bit_position+=rbsp_length*8
|
||||
|
||||
if self.avc_profile_indication in [ 100, 110, 122, 144]:
|
||||
bit_position = write_bits(buf, bit_position, 0b111111, 6)
|
||||
bit_position = write_bits(buf, bit_position, self.chroma_format, 2)
|
||||
bit_position = write_bits(buf, bit_position, 0b11111, 5)
|
||||
bit_position = write_bits(buf, bit_position, self.bit_depth_luma_minus8, 3)
|
||||
bit_position = write_bits(buf, bit_position, 0b11111, 5)
|
||||
bit_position = write_bits(buf, bit_position, self.bit_depth_chroma_minus8, 3)
|
||||
bit_position = write_byte(buf, bit_position, self.num_of_sequence_parameter_set_ext)
|
||||
for _ in range(self.num_of_sequence_parameter_set_ext):
|
||||
# TODO: dump SPSextended
|
||||
logger.error('Dumping SPS extended not yet implemented')
|
||||
pass
|
||||
|
||||
return buf
|
||||
|
||||
def merge(self, config):
|
||||
# Check config compatibility
|
||||
if self.configuration_version != config.configuration_version:
|
||||
raise ValueError(f'Configuration versions are different: {self.configuration_version:d}\
|
||||
vs {config.configuration_version:d}')
|
||||
if self.avc_profile_indication != config.avc_profile_indication:
|
||||
raise ValueError(f'AVC profiles are different: {self.avc_profile_indication:d} vs \
|
||||
{config.avc_profile_indication:d}')
|
||||
if self.profile_compatibility != config.profile_compatibility:
|
||||
raise ValueError(f'Profile compatilities are different: {self.profile_compatibility:d} \
|
||||
vs {config.profile_compatibility:d}')
|
||||
if self.avc_level_indication != config.avc_level_indication:
|
||||
raise ValueError(f'Level indications are different: {self.avc_level_indication:d} vs \
|
||||
{config.avc_level_indication:d}')
|
||||
if self.length_size_minus_one != config.length_size_minus_one:
|
||||
raise ValueError(f'Length units are different: {self.length_size_minus_one:d} vs \
|
||||
{config.length_size_minus_one:d}')
|
||||
if self.chroma_format != config.chroma_format:
|
||||
raise ValueError(f'Colour format are different: {self.chroma_format:d} vs \
|
||||
{config.chroma_format:d}')
|
||||
if self.bit_depth_luma_minus8 != config.bit_depth_luma_minus8:
|
||||
raise ValueError(f'Depth of luminance are different: {self.bit_depth_luma_minus8:d} vs \
|
||||
{config.bit_depth_luma_minus8:d}')
|
||||
if self.bit_depth_chroma_minus8 != config.bit_depth_chroma_minus8:
|
||||
raise ValueError(f'Depth of chromaticity are different: \
|
||||
{self.bit_depth_chroma_minus8:d} vs {config.bit_depth_luma_minus8:d}')
|
||||
|
||||
for spsid in config.sps:
|
||||
sps = config.sps[spsid]
|
||||
if spsid in self.sps:
|
||||
localsps = self.sps[spsid]
|
||||
if sps!=localsps:
|
||||
raise ValueError(f'Profile are not compatible. They contain two different SPS\
|
||||
with the same identifier ({spsid:d}): {localsps}\n{sps}\n')
|
||||
self.sps[spsid] = sps
|
||||
|
||||
self.num_of_sequence_parameter_sets = len(self.sps)
|
||||
|
||||
for ppsid in config.pps:
|
||||
pps = config.pps[ppsid]
|
||||
if ppsid in self.pps:
|
||||
localpps = self.pps[ppsid]
|
||||
if pps!=localpps:
|
||||
raise ValueError(f'Profile are not compatible. They contain two different PPS\
|
||||
with the same identifier ({ppsid:d}): {localpps}\n{pps}\n')
|
||||
self.pps[ppsid] = pps
|
||||
|
||||
self.num_of_picture_parameter_sets = len(self.pps)
|
||||
|
||||
# TODO: do the same with extended SPS !
|
||||
|
||||
@typechecked
|
||||
def parse_codec_private(codec_private_data: bytes) -> AVCDecoderConfiguration:
|
||||
if codec_private_data[0] != 0x63:
|
||||
raise ValueError(f'Matroska header is wrong: {codec_private_data[0]:x}')
|
||||
if codec_private_data[1] != 0xA2:
|
||||
raise ValueError(f'Matroska header is wrong: {codec_private_data[1]:x}')
|
||||
length = codec_private_data[2]
|
||||
if length == 0:
|
||||
raise ValueError('Matroska length cannot start with zero byte.')
|
||||
for nb_zeroes in range(8):
|
||||
b = read_bit(codec_private_data[2:], nb_zeroes)
|
||||
if b != 0:
|
||||
break
|
||||
mask = 2^(7-nb_zeroes)-1
|
||||
length = codec_private_data[2] and mask
|
||||
for i in range(nb_zeroes):
|
||||
length*=256
|
||||
length+=(codec_private_data[3+i])
|
||||
byte_position = 3+nb_zeroes
|
||||
avcconfig = AVCDecoderConfiguration()
|
||||
avcconfig.from_bytes(codec_private_data[byte_position:])
|
||||
|
||||
return avcconfig
|
||||
|
||||
@typechecked
|
||||
def get_avc_config_from_h264(input_file: IO[bytes]) -> AVCDecoderConfiguration:
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# TODO: improve this ...
|
||||
rbsp = input_file.read(1000)
|
||||
sodb = rbsp_to_sodb(rbsp)
|
||||
|
||||
bit_position = 0
|
||||
bit_position, start_code = read_long(sodb, bit_position)
|
||||
if start_code != 1:
|
||||
raise ValueError(f'Starting code not detected: {start_code:x}')
|
||||
sps = SPS()
|
||||
bit_length = sps.from_bytes(sodb[4:])
|
||||
bit_position+=bit_length
|
||||
|
||||
bit_position, start_code = read_long(sodb, bit_position)
|
||||
if start_code != 1:
|
||||
raise ValueError(f'Starting code not detected: {start_code:x}')
|
||||
pps = PPS()
|
||||
bit_length = pps.from_bytes(sodb[floor(bit_position/8):], sps.chroma_format_idc)
|
||||
logger.debug(pps)
|
||||
|
||||
avcconfig = AVCDecoderConfiguration()
|
||||
avcconfig.configuration_version = 1
|
||||
avcconfig.avc_profile_indication = sps.profile_idc
|
||||
avcconfig.profile_compatibility = 0
|
||||
avcconfig.avc_level_indication = sps.level_idc
|
||||
avcconfig.length_size_minus_one = 3
|
||||
avcconfig.num_of_sequence_parameter_sets = 1
|
||||
avcconfig.num_of_picture_parameter_sets = 1
|
||||
avcconfig.num_of_sequence_parameter_set_ext = 0
|
||||
avcconfig.chroma_format = sps.chroma_format_idc
|
||||
avcconfig.bit_depth_chroma_minus8 = sps.bit_depth_chroma_minus8
|
||||
avcconfig.bit_depth_luma_minus8 = sps.bit_depth_luma_minus8
|
||||
avcconfig.sps[sps.seq_parameter_set_id] = sps
|
||||
avcconfig.pps[pps.pic_parameter_set_id] = pps
|
||||
|
||||
return avcconfig
|
||||
|
||||
# Unused ?
|
||||
@typechecked
|
||||
def get_codec_private_data_from_h264(input_file: IO[bytes]) -> AVCDecoderConfiguration:
|
||||
avcconfig = get_avc_config_from_h264(input_file)
|
||||
res = dump_codec_private_data(avcconfig)
|
||||
|
||||
return res
|
||||
|
||||
@typechecked
|
||||
def parse_mkv_tree(mkvinfo_path:str, input_file: IO[bytes]) -> dict[str,tuple[int,int]]:
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -798,173 +496,7 @@ def parse_mkv_tree(mkvinfo_path:str, input_file: IO[bytes]) -> dict[str,tuple[in
|
||||
mkvinfo.wait()
|
||||
return elements
|
||||
|
||||
# MKV is formatted as an EBML file (Extended Binary Markup Langage).
|
||||
# cf http://matroska-org.github.io/libebml/specs.html
|
||||
# It is a Type, Length, Value (TLV) kind of binary file.
|
||||
# Types are encoded as follows:
|
||||
# 1xxx xxxx - Class A IDs (2^7 -1 possible values)
|
||||
# 01xx xxxx xxxx xxxx - Class B IDs (2^14-1 possible values)
|
||||
# 001x xxxx xxxx xxxx xxxx xxxx - Class C IDs (2^21-1 possible values)
|
||||
# 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx - Class D IDs (2^28-1 possible values)
|
||||
# Lengths are encoded as follows:
|
||||
# 1xxx xxxx
|
||||
# value 0 to 2^7-2
|
||||
# 01xx xxxx xxxx xxxx
|
||||
# value 0 to 2^14-2
|
||||
# 001x xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^21-2
|
||||
# 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^28-2
|
||||
# 0000 1xxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^35-2
|
||||
# 0000 01xx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^42-2
|
||||
# 0000 001x xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^49-2
|
||||
# 0000 0001 xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx xxxx
|
||||
# value 0 to 2^56-2
|
||||
|
||||
@typechecked
|
||||
def get_ebml_length(length:int) -> bytes|None:
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
if 0 <= length <= 2**7-2:
|
||||
size = 1
|
||||
elif length <= 2**14-2:
|
||||
size = 2
|
||||
elif length <= 2**21-2:
|
||||
size = 3
|
||||
elif length <= 2**28-2:
|
||||
size = 4
|
||||
elif length <= 2**35-2:
|
||||
size = 5
|
||||
elif length <= 2**42-2:
|
||||
size = 6
|
||||
elif length <= 2**49-2:
|
||||
size = 7
|
||||
elif length <= 2**56-2:
|
||||
size = 8
|
||||
elif length < 0:
|
||||
logger.error('Impossible to encode a negative length with EBML.')
|
||||
return None
|
||||
else:
|
||||
logger.error('Impossible to encode a length larger than 2^56-2 with EBML.')
|
||||
return None
|
||||
|
||||
encoded_length = length + ((128>>(size-1))<<((size-1)*8))
|
||||
res = (encoded_length).to_bytes(size, byteorder='big')
|
||||
return res
|
||||
|
||||
@typechecked
|
||||
def dump_codec_private_data(avc_decoder_configuration: AVCDecoderConfiguration) -> bytearray:
|
||||
logger = logging.getLogger(__name__)
|
||||
# Rebuild a Matroska Codec Private Element
|
||||
res = bytearray()
|
||||
# Code private element
|
||||
res.extend(b'\x63\xA2')
|
||||
buf = avc_decoder_configuration.to_bytes()
|
||||
logger.debug('AVC configuration bitstream: %s (length: %d))', hexdump.dump(buf, sep=':'),
|
||||
len(buf))
|
||||
|
||||
embl_length = get_ebml_length(len(buf))
|
||||
logger.debug('EMBL encoded length: %s', hexdump.dump(embl_length, sep=':'))
|
||||
res.extend(embl_length)
|
||||
res.extend(buf)
|
||||
|
||||
return res
|
||||
|
||||
@typechecked
|
||||
def change_ebml_element_size(input_file: IO[bytes], position:int, addendum:int) -> int:
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
initial_position = position
|
||||
infd = input_file.fileno()
|
||||
lseek(infd, position, SEEK_SET)
|
||||
|
||||
buf = read(infd, 1)
|
||||
element_type = int.from_bytes(buf, byteorder='big')
|
||||
mask=128
|
||||
found = False
|
||||
for i in range(1,5):
|
||||
if element_type&mask:
|
||||
type_size = i
|
||||
found = True
|
||||
break
|
||||
else:
|
||||
mask = mask>>1
|
||||
|
||||
if not found:
|
||||
logger.error('Size of element type cannot be determined: %d', element_type)
|
||||
exit(-1)
|
||||
|
||||
# We seek to size
|
||||
position+=type_size
|
||||
lseek(infd, position, SEEK_SET)
|
||||
|
||||
buf = read(infd, 1)
|
||||
size_head = int.from_bytes(buf, byteorder='big')
|
||||
logger.info('First byte of size: %x', size_head)
|
||||
mask=128
|
||||
found = False
|
||||
for i in range(1,9):
|
||||
if size_head&mask:
|
||||
size_of_data_size = i
|
||||
found = True
|
||||
break
|
||||
else:
|
||||
mask = mask>>1
|
||||
|
||||
if not found:
|
||||
logger.error('Size of data size cannot be determined: %d', size_head)
|
||||
exit(-1)
|
||||
else:
|
||||
logger.info('Size of data size: %d.', size_of_data_size)
|
||||
|
||||
lseek(infd, position, SEEK_SET)
|
||||
old_size_buf = read(infd, size_of_data_size)
|
||||
max_size = 2**(size_of_data_size*7)-2
|
||||
size_of_data = int.from_bytes(old_size_buf, byteorder='big')
|
||||
logger.info('Size of data with mask: %x mask: %d.', size_of_data, mask)
|
||||
size_of_data-= (mask<<((size_of_data_size-1)*8))
|
||||
logger.info('Found element at position: %d, size of type: %d size of data: %d \
|
||||
maximal size: %d.', initial_position, type_size, size_of_data, max_size)
|
||||
|
||||
new_size = size_of_data+addendum
|
||||
delta = 0
|
||||
if new_size > max_size:
|
||||
# TODO: Test this code ...
|
||||
new_encoded_size = get_ebml_length(new_size)
|
||||
size_of_new_encoded_size = len(new_encoded_size)
|
||||
if size_of_new_encoded_size <= size_of_data_size:
|
||||
logger.error('New encoded size is smaller (%d) or equal than previous size (%d).\
|
||||
This should not happen.', size_of_new_encoded_size, size_of_data_size)
|
||||
exit(-1)
|
||||
# The difference of length between old size field and new one.
|
||||
delta = size_of_new_encoded_size - size_of_data_size
|
||||
file_length = fstat(infd).st_size
|
||||
# We seek after actual length field
|
||||
lseek(infd, position+size_of_data_size, SEEK_SET)
|
||||
# We read the rest of file
|
||||
tail = read(infd, file_length-(position+size_of_data_size))
|
||||
# We increase file length
|
||||
ftruncate(infd, file_length+delta)
|
||||
# We go to the beginning of length field
|
||||
lseek(infd, position, SEEK_SET)
|
||||
# We write the new length field
|
||||
write(infd, new_encoded_size)
|
||||
# We overwrite the rest of file with its previous content that has been offset.
|
||||
write(infd, tail)
|
||||
else:
|
||||
size = new_size + ((128>>(size_of_data_size-1))<<((size_of_data_size-1)*8))
|
||||
new_size_buf = (size).to_bytes(size_of_data_size, byteorder='big')
|
||||
|
||||
logger.info('Old encoded size: %s New encoded size: %s', hexdump.dump(old_size_buf,sep=':'),
|
||||
hexdump.dump(new_size_buf, sep=':'))
|
||||
lseek(infd, position, SEEK_SET)
|
||||
write(infd, new_size_buf)
|
||||
|
||||
# We return the potential increase in size of the file if the length field had to be increased.
|
||||
return delta
|
||||
|
||||
@typechecked
|
||||
def change_codec_private_data(mkvinfo_path:str, input_file: IO[bytes], codec_data:bytes) -> None:
|
||||
|
||||
Reference in New Issue
Block a user