from pathlib import Path import struct data = Path("first_moof.bin").read_bytes() MDAT_ABSOLUTE_OFFSET = 4848 MDAT_HEADER_SIZE = 8 def uint32(offset): return struct.unpack_from(">I", data, offset)[0] def uint64(offset): return struct.unpack_from(">Q", data, offset)[0] def int32(offset): return struct.unpack_from(">i", data, offset)[0] def box_header(offset): size32 = uint32(offset) box_type = data[offset + 4:offset + 8].decode( "latin-1", errors="replace", ) if size32 == 1: size = uint64(offset + 8) header_size = 16 elif size32 == 0: size = len(data) - offset header_size = 8 else: size = size32 header_size = 8 return ( box_type, size, header_size, offset + size, ) def children(start, end): offset = start while offset + 8 <= end: box_type, size, header_size, box_end = box_header( offset, ) if size < header_size or box_end > end: raise RuntimeError( "Box non valido: " f"type={box_type} " f"offset={offset} " f"size={size}" ) yield ( box_type, offset, size, header_size, box_end, ) offset = box_end if offset != end: raise RuntimeError( f"Byte residui nel container: offset={offset} end={end}" ) moof_type, moof_size, moof_header, moof_end = box_header(0) assert moof_type == "moof" assert moof_size == len(data) print(f"moofSize={moof_size}") for traf_index, box in enumerate( children(moof_header, moof_end) ): ( box_type, traf_offset, traf_size, traf_header, traf_end, ) = box if box_type != "traf": continue tfhd_info = None trun_info = None for child in children( traf_offset + traf_header, traf_end, ): ( child_type, child_offset, child_size, child_header, child_end, ) = child payload_offset = child_offset + child_header if payload_offset + 4 > child_end: raise RuntimeError( f"FullBox incompleto: {child_type}" ) version = data[payload_offset] flags = int.from_bytes( data[payload_offset + 1:payload_offset + 4], "big", ) cursor = payload_offset + 4 if child_type == "tfhd": track_id = uint32(cursor) cursor += 4 base_data_offset = None default_sample_duration = None default_sample_size = None default_sample_flags = None if flags & 0x000001: base_data_offset = uint64(cursor) cursor += 8 if flags & 0x000002: cursor += 4 if flags & 0x000008: default_sample_duration = uint32(cursor) cursor += 4 if flags & 0x000010: default_sample_size = uint32(cursor) cursor += 4 if flags & 0x000020: default_sample_flags = uint32(cursor) cursor += 4 tfhd_info = { "track_id": track_id, "base_data_offset": base_data_offset, "default_sample_duration": default_sample_duration, "default_sample_size": default_sample_size, "default_sample_flags": default_sample_flags, } elif child_type == "trun": sample_count = uint32(cursor) cursor += 4 data_offset = None first_sample_flags = None if flags & 0x000001: data_offset = int32(cursor) cursor += 4 if flags & 0x000004: first_sample_flags = uint32(cursor) cursor += 4 samples = [] for sample_index in range(sample_count): duration = None size = None sample_flags = None composition_offset = 0 if flags & 0x000100: duration = uint32(cursor) cursor += 4 if flags & 0x000200: size = uint32(cursor) cursor += 4 if flags & 0x000400: sample_flags = uint32(cursor) cursor += 4 if flags & 0x000800: if version == 0: composition_offset = uint32(cursor) else: composition_offset = int32(cursor) cursor += 4 samples.append( { "index": sample_index, "duration": duration, "size": size, "flags": sample_flags, "composition_offset": composition_offset, } ) trun_info = { "version": version, "flags": flags, "sample_count": sample_count, "data_offset": data_offset, "first_sample_flags": first_sample_flags, "samples": samples, "parsed_end": cursor, "box_end": child_end, } if tfhd_info is None or trun_info is None: raise RuntimeError( f"traf incompleto: index={traf_index}" ) track_id = tfhd_info["track_id"] samples = trun_info["samples"] durations = [sample["duration"] for sample in samples] sizes = [sample["size"] for sample in samples] flag_values = [sample["flags"] for sample in samples] if any(value is None for value in durations): raise RuntimeError( f"Durata sample assente nel track {track_id}" ) if any(value is None for value in sizes): raise RuntimeError( f"Dimensione sample assente nel track {track_id}" ) base_data_offset = tfhd_info["base_data_offset"] data_offset = trun_info["data_offset"] if base_data_offset is None or data_offset is None: first_sample_absolute_offset = None else: first_sample_absolute_offset = ( base_data_offset + data_offset ) non_sync_count = sum( 1 for value in flag_values if value is not None and (value & 0x00010000) != 0 ) sync_count = len(samples) - non_sync_count expected_mdat_payload_offset = ( MDAT_ABSOLUTE_OFFSET + MDAT_HEADER_SIZE ) print() print(f"=== TRAF {traf_index} ===") print(f"trackId={track_id}") print( f"baseDataOffset={base_data_offset}" ) print(f"dataOffset={data_offset}") print( "firstSampleAbsoluteOffset=" f"{first_sample_absolute_offset}" ) print( "expectedMdatPayloadOffset=" f"{expected_mdat_payload_offset}" ) print(f"sampleCount={len(samples)}") print(f"totalDuration={sum(durations)}") print(f"minimumDuration={min(durations)}") print(f"maximumDuration={max(durations)}") print(f"totalSampleBytes={sum(sizes)}") print(f"minimumSampleSize={min(sizes)}") print(f"maximumSampleSize={max(sizes)}") print(f"syncSamples={sync_count}") print(f"nonSyncSamples={non_sync_count}") print( "trunParsedExactly=" f"{trun_info['parsed_end'] == trun_info['box_end']}" ) print("firstSamples:") decode_time = 0 running_offset = first_sample_absolute_offset for sample in samples[:8]: flags_value = sample["flags"] is_non_sync = ( flags_value is not None and (flags_value & 0x00010000) != 0 ) flags_text = ( f"0x{flags_value:08x}" if flags_value is not None else "None" ) print( f" index={sample['index']} " f"offset={running_offset} " f"size={sample['size']} " f"duration={sample['duration']} " f"dts={decode_time} " f"pts={decode_time + sample['composition_offset']} " f"flags={flags_text} " f"sync={not is_non_sync}" ) if running_offset is not None: running_offset += sample["size"] decode_time += sample["duration"]