"""Utilities to help convert mp4s to fmp4s."""
from __future__ import annotations
from collections.abc import Generator
from typing import TYPE_CHECKING
from homeassistant.exceptions import HomeAssistantError
from .core import Orientation
if TYPE_CHECKING:
from io import BufferedIOBase
def find_box(
mp4_bytes: bytes, target_type: bytes, box_start: int = 0
) -> Generator[int, None, None]:
"""Find location of first box (or sub box if box_start provided) of given type."""
if box_start == 0:
index = 0
box_end = len(mp4_bytes)
else:
box_end = box_start + int.from_bytes(
mp4_bytes[box_start : box_start + 4], byteorder="big"
)
index = box_start + 8
while 1:
if index > box_end - 8: # End of box, not found
break
box_header = mp4_bytes[index : index + 8]
if box_header[4:8] == target_type:
yield index
index += int.from_bytes(box_header[0:4], byteorder="big")
def get_codec_string(mp4_bytes: bytes) -> str:
"""Get RFC 6381 codec string."""
codecs = []
# Find moov
moov_location = next(find_box(mp4_bytes, b"moov"))
# Find tracks
for trak_location in find_box(mp4_bytes, b"trak", moov_location):
# Drill down to media info
mdia_location = next(find_box(mp4_bytes, b"mdia", trak_location))
minf_location = next(find_box(mp4_bytes, b"minf", mdia_location))
stbl_location = next(find_box(mp4_bytes, b"stbl", minf_location))
stsd_location = next(find_box(mp4_bytes, b"stsd", stbl_location))
# Get stsd box
stsd_length = int.from_bytes(
mp4_bytes[stsd_location : stsd_location + 4], byteorder="big"
)
stsd_box = mp4_bytes[stsd_location : stsd_location + stsd_length]
# Base Codec
codec = stsd_box[20:24].decode("utf-8")
# Handle H264
if (
codec in ("avc1", "avc2", "avc3", "avc4")
and stsd_length > 110
and stsd_box[106:110] == b"avcC"
):
profile = stsd_box[111:112].hex()
compatibility = stsd_box[112:113].hex()
# Cap level at 4.1 for compatibility with some Google Cast devices
level = hex(min(stsd_box[113], 41))[2:]
codec += "." + profile + compatibility + level
# Handle H265
elif (
codec in ("hev1", "hvc1")
and stsd_length > 110
and stsd_box[106:110] == b"hvcC"
):
tmp_byte = int.from_bytes(stsd_box[111:112], byteorder="big")
# Profile Space
codec += "."
profile_space_map = {0: "", 1: "A", 2: "B", 3: "C"}
profile_space = tmp_byte >> 6
codec += profile_space_map[profile_space]
general_profile_idc = tmp_byte & 31
codec += str(general_profile_idc)
# Compatibility
codec += "."
general_profile_compatibility = int.from_bytes(
stsd_box[112:116], byteorder="big"
)
reverse = 0
for i in range(0, 32):
reverse |= general_profile_compatibility & 1
if i == 31:
break
reverse <<= 1
general_profile_compatibility >>= 1
codec += hex(reverse)[2:]
# Tier Flag
if (tmp_byte & 32) >> 5 == 0:
codec += ".L"
else:
codec += ".H"
codec += str(int.from_bytes(stsd_box[122:123], byteorder="big"))
# Constraint String
has_byte = False
constraint_string = ""
for i in range(121, 115, -1):
gci = int.from_bytes(stsd_box[i : i + 1], byteorder="big")
if gci or has_byte:
constraint_string = "." + hex(gci)[2:] + constraint_string
has_byte = True
codec += constraint_string
# Handle Audio
elif codec == "mp4a":
oti = None
dsi = None
# Parse ES Descriptors
oti_loc = stsd_box.find(b"\x04\x80\x80\x80")
if oti_loc > 0:
oti = stsd_box[oti_loc + 5 : oti_loc + 6].hex()
codec += f".{oti}"
dsi_loc = stsd_box.find(b"\x05\x80\x80\x80")
if dsi_loc > 0:
dsi_length = int.from_bytes(
stsd_box[dsi_loc + 4 : dsi_loc + 5], byteorder="big"
)
dsi_data = stsd_box[dsi_loc + 5 : dsi_loc + 5 + dsi_length]
dsi0 = int.from_bytes(dsi_data[0:1], byteorder="big")
dsi = (dsi0 & 248) >> 3
if dsi == 31 and len(dsi_data) >= 2:
dsi1 = int.from_bytes(dsi_data[1:2], byteorder="big")
dsi = 32 + ((dsi0 & 7) << 3) + ((dsi1 & 224) >> 5)
codec += f".{dsi}"
codecs.append(codec)
return ",".join(codecs)
def find_moov(mp4_io: BufferedIOBase) -> int:
"""Find location of moov atom in a BufferedIOBase mp4."""
index = 0
while 1:
mp4_io.seek(index)
box_header = mp4_io.read(8)
if len(box_header) != 8:
raise HomeAssistantError("moov atom not found")
if box_header[4:8] == b"moov":
return index
index += int.from_bytes(box_header[0:4], byteorder="big")
def read_init(bytes_io: BufferedIOBase) -> bytes:
"""Read the init from a mp4 file."""
moov_loc = find_moov(bytes_io)
bytes_io.seek(moov_loc)
moov_len = int.from_bytes(bytes_io.read(4), byteorder="big")
bytes_io.seek(0)
return bytes_io.read(moov_loc + moov_len)
ZERO32 = b"\x00\x00\x00\x00"
ONE32 = b"\x00\x01\x00\x00"
NEGONE32 = b"\xFF\xFF\x00\x00"
XYW_ROW = ZERO32 + ZERO32 + b"\x40\x00\x00\x00"
ROTATE_RIGHT = (ZERO32 + ONE32 + ZERO32) + (NEGONE32 + ZERO32 + ZERO32)
ROTATE_LEFT = (ZERO32 + NEGONE32 + ZERO32) + (ONE32 + ZERO32 + ZERO32)
ROTATE_180 = (NEGONE32 + ZERO32 + ZERO32) + (ZERO32 + NEGONE32 + ZERO32)
MIRROR = (NEGONE32 + ZERO32 + ZERO32) + (ZERO32 + ONE32 + ZERO32)
FLIP = (ONE32 + ZERO32 + ZERO32) + (ZERO32 + NEGONE32 + ZERO32)
# The two below do not seem to get applied properly
ROTATE_LEFT_FLIP = (ZERO32 + NEGONE32 + ZERO32) + (NEGONE32 + ZERO32 + ZERO32)
ROTATE_RIGHT_FLIP = (ZERO32 + ONE32 + ZERO32) + (ONE32 + ZERO32 + ZERO32)
TRANSFORM_MATRIX_TOP = (
# The index into this tuple corresponds to the EXIF orientation tag
# Only index values of 2 through 8 are used
# The first two entries are just to keep everything aligned
b"", # 0
b"", # 1
MIRROR, # 2
ROTATE_180, # 3
FLIP, # 4
ROTATE_LEFT_FLIP, # 5
ROTATE_LEFT, # 6
ROTATE_RIGHT_FLIP, # 7
ROTATE_RIGHT, # 8
)
def transform_init(init: bytes, orientation: Orientation) -> bytes:
"""Change the transformation matrix in the header."""
if orientation == Orientation.NO_TRANSFORM:
return init
# Find moov
moov_location = next(find_box(init, b"moov"))
mvhd_location = next(find_box(init, b"trak", moov_location))
tkhd_location = next(find_box(init, b"tkhd", mvhd_location))
tkhd_length = int.from_bytes(
init[tkhd_location : tkhd_location + 4], byteorder="big"
)
return (
init[: tkhd_location + tkhd_length - 44]
+ TRANSFORM_MATRIX_TOP[orientation]
+ XYW_ROW
+ init[tkhd_location + tkhd_length - 8 :]
)