mirror of
https://github.com/ApfelTeeSaft/restream_playout.git
synced 2026-08-26 19:33:32 +00:00
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
import struct
|
|
from pathlib import Path
|
|
|
|
from restream_playout.filler import AAC_SILENCE, AVCC, AVC_SAMPLE
|
|
|
|
|
|
def box(kind: bytes, payload: bytes) -> bytes:
|
|
return (len(payload) + 8).to_bytes(4, "big") + kind + payload
|
|
|
|
|
|
def full_box(kind: bytes, payload: bytes, version: int = 0, flags: int = 0) -> bytes:
|
|
return box(kind, bytes((version,)) + flags.to_bytes(3, "big") + payload)
|
|
|
|
|
|
def descriptor(tag: int, payload: bytes) -> bytes:
|
|
if len(payload) >= 128:
|
|
raise ValueError("test descriptor is too large")
|
|
return bytes((tag, len(payload))) + payload
|
|
|
|
|
|
def stbl(sample_entry: bytes, sample: bytes, chunk_offset: int, duration: int, *, sync: bool) -> bytes:
|
|
stsd = full_box(b"stsd", b"\x00\x00\x00\x01" + sample_entry)
|
|
stts = full_box(b"stts", b"\x00\x00\x00\x01" + struct.pack(">II", 1, duration))
|
|
stsc = full_box(b"stsc", b"\x00\x00\x00\x01" + struct.pack(">III", 1, 1, 1))
|
|
stsz = full_box(b"stsz", b"\x00\x00\x00\x00\x00\x00\x00\x01" + len(sample).to_bytes(4, "big"))
|
|
stco = full_box(b"stco", b"\x00\x00\x00\x01" + chunk_offset.to_bytes(4, "big"))
|
|
stss = full_box(b"stss", b"\x00\x00\x00\x01\x00\x00\x00\x01") if sync else b""
|
|
return box(b"stbl", stsd + stts + stsc + stsz + stco + stss)
|
|
|
|
|
|
def video_entry() -> bytes:
|
|
payload = (
|
|
b"\x00" * 6
|
|
+ b"\x00\x01"
|
|
+ b"\x00" * 16
|
|
+ struct.pack(">HH", 16, 16)
|
|
+ b"\x00\x48\x00\x00" * 2
|
|
+ b"\x00\x00\x00\x00"
|
|
+ b"\x00\x01"
|
|
+ b"\x00" * 32
|
|
+ b"\x00\x18\xff\xff"
|
|
+ box(b"avcC", AVCC)
|
|
)
|
|
return box(b"avc1", payload)
|
|
|
|
|
|
def audio_entry() -> bytes:
|
|
asc = b"\x12\x10"
|
|
decoder_config = b"\x40\x15\x00\x00\x00\x00\x00\x00\x00\x00\x00" + descriptor(0x05, asc)
|
|
es_descriptor = b"\x00\x01\x00" + descriptor(0x04, decoder_config) + descriptor(0x06, b"\x02")
|
|
esds = full_box(b"esds", descriptor(0x03, es_descriptor))
|
|
payload = (
|
|
b"\x00" * 6
|
|
+ b"\x00\x01"
|
|
+ b"\x00" * 8
|
|
+ b"\x00\x02"
|
|
+ b"\x00\x10"
|
|
+ b"\x00\x00\x00\x00"
|
|
+ (44100 << 16).to_bytes(4, "big")
|
|
+ esds
|
|
)
|
|
return box(b"mp4a", payload)
|
|
|
|
|
|
def track(kind: bytes, timescale: int, table: bytes) -> bytes:
|
|
mdhd = full_box(
|
|
b"mdhd",
|
|
b"\x00" * 8 + timescale.to_bytes(4, "big") + timescale.to_bytes(4, "big") + b"\x00\x00\x00\x00",
|
|
)
|
|
hdlr = full_box(b"hdlr", b"\x00\x00\x00\x00" + kind + b"\x00" * 12)
|
|
minf = box(b"minf", table)
|
|
return box(b"trak", box(b"mdia", mdhd + hdlr + minf))
|
|
|
|
|
|
def write_test_mp4(path: Path) -> None:
|
|
ftyp = box(b"ftyp", b"isom\x00\x00\x02\x00isomavc1")
|
|
video_offset = len(ftyp) + 8
|
|
audio_offset = video_offset + len(AVC_SAMPLE)
|
|
mdat = box(b"mdat", AVC_SAMPLE + AAC_SILENCE)
|
|
video_table = stbl(video_entry(), AVC_SAMPLE, video_offset, 3000, sync=True)
|
|
audio_table = stbl(audio_entry(), AAC_SILENCE, audio_offset, 1024, sync=False)
|
|
moov = box(b"moov", track(b"vide", 90000, video_table) + track(b"soun", 44100, audio_table))
|
|
path.write_bytes(ftyp + mdat + moov)
|
|
|