339 lines
8.4 KiB
Python
339 lines
8.4 KiB
Python
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"]
|