diff --git a/src/tscut/h264/avc.py b/src/tscut/h264/avc.py new file mode 100644 index 0000000..125728b --- /dev/null +++ b/src/tscut/h264/avc.py @@ -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 diff --git a/src/tscut/h264/parameters.py b/src/tscut/h264/parameters.py index b61c99b..d6236e0 100644 --- a/src/tscut/h264/parameters.py +++ b/src/tscut/h264/parameters.py @@ -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 diff --git a/src/tscut/matroska/__init__.py b/src/tscut/matroska/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/tscut/matroska/ebml.py b/src/tscut/matroska/ebml.py new file mode 100644 index 0000000..78b54db --- /dev/null +++ b/src/tscut/matroska/ebml.py @@ -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 diff --git a/src/tscut/tscut.py b/src/tscut/tscut.py index 5ab00c4..f0e57e5 100755 --- a/src/tscut/tscut.py +++ b/src/tscut/tscut.py @@ -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: