Fix OAMD variant resolution.

This commit is contained in:
2026-09-08 21:05:26 +08:00
parent c8676d7aa3
commit cf50beafcb
+439 -235
View File
@@ -1,12 +1,31 @@
"""OAMD 位载荷 → 16 个对象槽的 q1/q2/q3 增量状态。 """OAMD 位载荷 → 16 个对象槽的 q1/q2/q3 增量状态。
槽 0 是 bed/LFE;槽 1..15 对应输出 ch1..15 的对象元数据。 解析依据 ETSI TS 103 420 V1.2.1(Backwards-compatible object audio carriage
using Enhanced AC-3)clause 5:
* 5.5.2 ``object_audio_metadata_payload()``:版本、对象数、program assignment、
element 目录;
* 5.5.3 ``program_assignment()``:bed / ISF / dynamic 三类对象及其数量;
* 5.5.4 ``oa_element_md()``:element id、字节长度、alternate data id;
* 5.5.5/5.5.6/5.5.7 object_element/md_update_info/block_update_info:
``start_sample = sample_offset + 32 * block_offset_factor``;
* 5.5.9/5.5.10/5.5.11 object_info_block/object_basic_info/object_render_info:
逐对象的位置字段;bed 与 ISF 对象不携带 render info;
* 5.6.1.1.8~5.6.1.1.11 pos3D_X/Y/Z:横向/纵向 62 格、高度 15 格 + 符号位。
槽 0 是 bed/LFE;槽 1..15 对应输出 ch1..15 的对象元数据。bed/ISF 对象与
``b_object_not_active`` 对象没有位置字段(5.5.9),保持上一帧位置。
element 目录由声明长度驱动,因此 alternate_object_data_present、任意对象数、
多 element(trim/extended/未知 id 按声明边界跳过)都能解析。个别编码器写出的
``oa_element_size`` 比实际内容短(例如 Dolby 测试信号
``Audio_ID_..._special_6ch_..._ddp_joc.mp4`` 的 ID11 少 2 字节),此时以结构解析
出的实际位置为准,并把差异放进 ``diagnostics``,不当作变体错误。
""" """
import numpy as np import numpy as np
from variant_error import UnsupportedVariantError, bytes_descriptor from variant_error import UnsupportedVariantError, bytes_descriptor
Q1_OFF = 192
N_Q12 = 62 N_Q12 = 62
N_Q3 = 15 N_Q3 = 15
SAMPLE_OFFSET_INDEX = (8, 16, 18, 24) SAMPLE_OFFSET_INDEX = (8, 16, 18, 24)
@@ -16,9 +35,20 @@ RAMP_DURATION_INDEX = (
1024, 1600, 1601, 1602, 1920, 2000, 2002, 2048, 1024, 1600, 1601, 1602, 1920, 2000, 2002, 2048,
) )
OBJECT_ELEMENT_ID = 1 OBJECT_ELEMENT_ID = 1
TRIM_ELEMENT_ID = 2 # 5.6.0.11 Table 11b:ISF 类型 → 对象数。
EXTENDED_OBJECT_ELEMENT_ID = 5 ISF_OBJECT_COUNTS = (4, 8, 10, 14, 15, 30)
POSITION_WINDOW_END_BIT = 112 + 31 * (15 - 3) + 24 # 5.6.1.1.4 Table 12:10 bit 标准 bed 掩码按位对应的声道(LSB = RC_L/RC_R)。
BED_CHANNEL_LABELS = (
"RC_L/RC_R", "RC_C", "RC_LFE", "RC_LS/RC_RS", "RC_LB/RC_RB",
"RC_TFL/RC_TFR", "RC_TSL/RC_TSR", "RC_TBL/RC_TBR", "RC_LW/RC_RW", "RC_LFE2",
)
# 5.6.1.1.5 Table 13:17 bit 非标准 bed 掩码按位对应的声道标签。
NONSTD_BED_LABELS = (
"RC_LFE2", "RC_RW", "RC_LW", "RC_TBR", "RC_TBL", "RC_TSR", "RC_TSL",
"RC_TFR", "RC_TFL", "RC_RB", "RC_LB", "RC_RS", "RC_LS", "RC_LFE", "RC_C",
"RC_R", "RC_L",
)
MAX_SLOTS = 16
def q_of(k, n): def q_of(k, n):
@@ -91,14 +121,23 @@ class _BitReader:
self.read(count) self.read(count)
def _variable_bits(reader, width, max_groups=5): def _variable_bits_max(reader, width, max_groups):
value = 0 """5.5.1 ``variable_bits_max(n, max_num_groups)``。"""
for _ in range(max_groups + 1): value = reader.read(width)
value += reader.read(width) more = reader.read(1)
if not reader.read(1): num_group = 1
return value if max_groups > num_group:
value = (value + 1) << width if more:
raise ValueError(f"OAMD variable_bits({width}) 延伸组过多") value = (value + 1) << width
while more:
value += reader.read(width)
more = reader.read(1)
if num_group >= max_groups:
break
if more:
value = (value + 1) << width
num_group += 1
return value
def _element_details(elements): def _element_details(elements):
@@ -109,166 +148,185 @@ def _element_details(elements):
"header_start_bit": element["header_start_bit"], "header_start_bit": element["header_start_bit"],
"body_start_bit": element["body_start_bit"], "body_start_bit": element["body_start_bit"],
"body_end_bit": element["body_end_bit"], "body_end_bit": element["body_end_bit"],
"parsed_end_bit": element.get("parsed_end_bit"),
"discard_unknown": element["discard_unknown"], "discard_unknown": element["discard_unknown"],
"alternate_data_id": element["alternate_data_id"],
} for element in elements] } for element in elements]
def _parse_elements(bits, header, header_end, raw_payload): def _syntax_error(variant, message, raw_payload, exc=None, **details):
"""建立 OAMD element 目录;element size 表示其后 body 的字节数。""" payload = {"payload": bytes_descriptor(raw_payload)}
reader = _BitReader(bits, header_end) if exc is not None:
elements = [] payload["parser_error"] = str(exc)
for ordinal in range(header["element_count"]): payload.update(details)
header_start = reader.position return UnsupportedVariantError("oamd", variant, message, details=payload)
try:
element_id = reader.read(4)
size_bytes = _variable_bits(reader, 4) + 1
except ValueError as exc:
raise UnsupportedVariantError(
"oamd", "element_header_truncated",
"OAMD element header 不完整",
details={
"element_ordinal": ordinal,
"header_start_bit": header_start,
"payload": bytes_descriptor(raw_payload),
"parser_error": str(exc),
}) from exc
body_start = reader.position
body_end = body_start + size_bytes * 8 def _parse_program_assignment(reader, raw_payload):
if body_end > len(bits): """5.5.3 ``program_assignment()``:bed / ISF / dynamic 对象数量。"""
raise UnsupportedVariantError( program = {
"oamd", "element_bounds", "dynamic_object_only": bool(reader.read(1)),
"OAMD element 声明长度超过 payload 边界", "lfe_present": False,
details={ "content_description": None,
"element_ordinal": ordinal, "bed_assignments": [],
"element_id": element_id, "num_bed_objects": 0,
"size_bytes": size_bytes, "isf_idx": None,
"body_start_bit": body_start, "num_isf_objects": 0,
"body_end_bit": body_end, "num_dynamic_objects": None,
"payload_bits": len(bits), }
"payload": bytes_descriptor(raw_payload), if program["dynamic_object_only"]:
program["lfe_present"] = bool(reader.read(1))
# 5.6.4.8:对象顺序为 bed → ISF → dynamic,LFE 属于 bed,排在最前。
program["num_bed_objects"] = 1 if program["lfe_present"] else 0
return program
mask = reader.read(4)
program["content_description"] = {
"reserved": bool(mask & 0x8),
"dynamic": bool(mask & 0x4),
"isf": bool(mask & 0x2),
"bed": bool(mask & 0x1),
}
if mask & 0x1:
program["b_bed_chan_distribute"] = bool(reader.read(1))
num_instances = (reader.read(3) + 2) if reader.read(1) else 1
for instance in range(num_instances):
if reader.read(1): # b_lfe_only
program["bed_assignments"].append({
"instance": instance, "lfe_only": True, "channels": ["RC_LFE"],
}) })
continue
control_bits = 5 if header["alternate_object_present"] else 1 if reader.read(1): # b_standard_chan_assign
if body_start + control_bits > body_end: bed_mask = reader.read(10)
raise UnsupportedVariantError( channels = []
"oamd", "element_control_bounds", for bit in range(10):
"OAMD element 太短,无法容纳控制字段", if (bed_mask >> bit) & 1:
details={ channels.extend(BED_CHANNEL_LABELS[bit].split("/"))
"element_ordinal": ordinal, program["bed_assignments"].append({
"element_id": element_id, "instance": instance, "lfe_only": False, "standard": True,
"size_bytes": size_bytes, "mask": bed_mask, "channels": channels,
"control_bits": control_bits,
"payload": bytes_descriptor(raw_payload),
}) })
control = _BitReader(bits, body_start, body_end) else:
alternate_data_id = (control.read(4) bed_mask = reader.read(17)
if header["alternate_object_present"] else None) channels = [NONSTD_BED_LABELS[bit]
discard_unknown = bool(control.read(1)) for bit in range(17) if (bed_mask >> bit) & 1]
elements.append({ program["bed_assignments"].append({
"ordinal": ordinal, "instance": instance, "lfe_only": False, "standard": False,
"element_id": element_id, "mask": bed_mask, "channels": channels,
"size_bytes": size_bytes,
"header_start_bit": header_start,
"body_start_bit": body_start,
"data_start_bit": control.position,
"body_end_bit": body_end,
"alternate_data_id": alternate_data_id,
"discard_unknown": discard_unknown,
})
reader.position = body_end
padding = bits[reader.position:]
if len(padding) > 7:
raise UnsupportedVariantError(
"oamd", "trailing_payload_data",
"OAMD element 结束后仍有超过一个字节的未声明数据",
details={
"elements": _element_details(elements),
"trailing_bits": len(padding),
"payload": bytes_descriptor(raw_payload),
})
if np.any(padding):
raise UnsupportedVariantError(
"oamd", "nonzero_padding",
"OAMD payload 尾部 padding 含非零位",
details={
"elements": _element_details(elements),
"padding_start_bit": reader.position,
"padding_bits": "".join(str(int(bit)) for bit in padding),
"payload": bytes_descriptor(raw_payload),
})
object_elements = [element for element in elements
if element["element_id"] == OBJECT_ELEMENT_ID]
if len(object_elements) != 1:
raise UnsupportedVariantError(
"oamd", "object_element_count",
"OAMD 必须包含且只能包含一个 object element",
details={
"object_element_count": len(object_elements),
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
})
object_element = object_elements[0]
if object_element["ordinal"] != 0 or object_element["header_start_bit"] != header_end:
raise UnsupportedVariantError(
"oamd", "object_element_order",
"object element 不在当前固定位置窗口支持的首个 element 位置",
details={
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
"repair_hint": "以 object element 的实际位置为基准重新定位对象窗口",
})
for element in elements:
element_id = element["element_id"]
if element_id in (OBJECT_ELEMENT_ID, TRIM_ELEMENT_ID):
continue
if element_id == EXTENDED_OBJECT_ELEMENT_ID:
raise UnsupportedVariantError(
"oamd", "extended_object_element",
"OAMD 含可能改变坐标语义的 extended object element",
details={
"element": _element_details([element])[0],
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
"repair_hint": "解析 divergence/extended-precision position 后再应用轨迹",
}) })
raise UnsupportedVariantError( program["num_bed_objects"] = sum(
"oamd", f"unsupported_element_{element_id}", len(assignment["channels"]) for assignment in program["bed_assignments"])
f"OAMD 含当前未覆盖的 element id {element_id}", if mask & 0x2:
details={ isf_idx = reader.read(3)
"element": _element_details([element])[0], program["isf_idx"] = isf_idx
"elements": _element_details(elements), program["num_isf_objects"] = (ISF_OBJECT_COUNTS[isf_idx]
"payload": bytes_descriptor(raw_payload), if isf_idx < len(ISF_OBJECT_COUNTS) else 0)
}) if mask & 0x4:
return object_element, elements count = reader.read(5)
if count == 0x1F:
count += reader.read(7)
program["num_dynamic_objects"] = count + 1
if mask & 0x8:
# 5.6.4.1:reserved_data_size = reserved_data_size_bits + 1(字节)。
reader.skip((reader.read(4) + 1) * 8)
return program
def _update_timing(bits, object_element, raw_payload): def _parse_object_info_block(reader, object_index, in_bed_or_isf, raw_payload):
"""从已定位的 object element 读取位置块开始偏移和 ramp 时长。""" """5.5.9 ``object_info_block(0)``:一个对象的属性更新。"""
reader = _BitReader( info = {
bits, object_element["data_start_bit"], object_element["body_end_bit"]) "object": object_index,
"in_bed_or_isf": bool(in_bed_or_isf),
"not_active": bool(reader.read(1)),
}
# blk == 0 且对象激活时 basic info 恒为 full update(status 0b01)。
basic_status = 0 if info["not_active"] else 1
info["basic_status"] = basic_status
if basic_status in (1, 3):
if basic_status == 1:
# 5.5.10:object_basic_info[] = {true, true};
# 5.6.4.12:array[1] = object_gain_idx,array[0] = b_default_object_priority。
gain_present = priority_present = True
else:
flags = reader.read(2)
gain_present = bool(flags & 1)
priority_present = bool(flags & 2)
if gain_present:
gain_idx = reader.read(2)
info["gain_idx"] = gain_idx
if gain_idx == 2:
info["gain_bits"] = reader.read(6)
if priority_present:
default_priority = bool(reader.read(1))
info["default_priority"] = default_priority
if not default_priority:
info["priority_bits"] = reader.read(5)
render_status = 0 if (info["not_active"] or info["in_bed_or_isf"]) else 1
info["render_status"] = render_status
if render_status in (1, 3):
if render_status == 1:
# 5.5.11:obj_render_info[] = {true, true, true, true}。
position_present = zone_present = size_present = screen_present = True
else:
flags = reader.read(4)
position_present = bool(flags & 0x1)
zone_present = bool(flags & 0x2)
size_present = bool(flags & 0x4)
screen_present = bool(flags & 0x8)
if position_present:
# blk == 0 时 b_differential_position_specified 恒为 FALSE。
x = reader.read(6)
y = reader.read(6)
z_sign = reader.read(1)
z = reader.read(4)
info["position"] = (x, y, z if z_sign else -z)
if reader.read(1): # b_object_distance_specified
if not reader.read(1): # b_object_at_infinity
reader.read(4) # distance_factor_idx
if zone_present:
info["zone_constraints_idx"] = reader.read(3)
info["enable_elevation"] = bool(reader.read(1))
if size_present:
size_idx = reader.read(2)
info["object_size_idx"] = size_idx
if size_idx == 1:
reader.read(5)
elif size_idx == 2:
reader.read(15)
if screen_present:
if reader.read(1): # b_object_use_screen_ref
reader.read(3) # screen_factor_bits
reader.read(2) # depth_factor_idx
info["snap"] = bool(reader.read(1))
if reader.read(1): # b_additional_table_data_exists
info["additional_table_bytes"] = reader.read(4) + 1
reader.skip(info["additional_table_bytes"] * 8)
return info
def _parse_object_element(reader, element, raw_payload):
"""5.5.5 ``object_element()``:时间信息 + 每个对象一条 object_info_block。"""
object_count = element["object_count"]
try: try:
offset_code = reader.read(2) sample_offset_code = reader.read(2)
if offset_code == 0: if sample_offset_code == 0:
sample_offset = 0 sample_offset = 0
elif offset_code == 1: elif sample_offset_code == 1:
sample_offset = SAMPLE_OFFSET_INDEX[reader.read(2)] sample_offset = SAMPLE_OFFSET_INDEX[reader.read(2)]
elif offset_code == 2: elif sample_offset_code == 2:
sample_offset = reader.read(5) sample_offset = reader.read(5)
else: else:
raise UnsupportedVariantError( raise UnsupportedVariantError(
"oamd", "md_sample_offset_mode", "oamd", "md_sample_offset_mode",
"OAMD 使用了当前未覆盖的 MD sample-offset 模式", "OAMD 使用了当前未覆盖的 MD sample-offset 模式",
details={"payload": bytes_descriptor(raw_payload)}) details={
"sample_offset_code": sample_offset_code,
"payload": bytes_descriptor(raw_payload),
})
block_count = reader.read(3) + 1 block_count = reader.read(3) + 1
blocks = [] blocks = []
for _block in range(block_count): for _block in range(block_count):
block_offset_factor = reader.read(6) block_offset_factor = reader.read(6)
block_offset = sample_offset + block_offset_factor * 32
ramp_code = reader.read(2) ramp_code = reader.read(2)
if ramp_code == 3: if ramp_code == 3:
if reader.read(1): if reader.read(1):
@@ -277,107 +335,253 @@ def _update_timing(bits, object_element, raw_payload):
ramp_duration = reader.read(11) ramp_duration = reader.read(11)
else: else:
ramp_duration = RAMP_DURATIONS[ramp_code] ramp_duration = RAMP_DURATIONS[ramp_code]
blocks.append((block_offset, ramp_duration)) blocks.append({
"block_offset_samples": sample_offset + block_offset_factor * 32,
"ramp_duration_samples": ramp_duration,
})
reserved_data_not_present = bool(reader.read(1))
if not reserved_data_not_present:
reader.read(5)
objects = [
_parse_object_info_block(
reader, index, index < element["bed_isf_objects"], raw_payload)
for index in range(object_count)
]
except UnsupportedVariantError: except UnsupportedVariantError:
raise raise
except ValueError as exc: except ValueError as exc:
raise UnsupportedVariantError( raise _syntax_error(
"oamd", "object_element_syntax", "object_element_syntax",
"OAMD object element 的 timing 字段越界或不完整", "OAMD object element 字段越界或不完整",
details={ raw_payload, exc,
"element": _element_details([object_element])[0], element=_element_details([element])[0],
"payload": bytes_descriptor(raw_payload), object_count=object_count,
"parser_error": str(exc), ) from exc
}) from exc return {
"sample_offset": sample_offset,
"blocks": blocks,
"reserved_data_not_present": reserved_data_not_present,
"objects": objects,
}
def _parse_elements(reader, alternate_present, object_count, program, raw_payload):
"""5.5.4 ``oa_element_md()`` 目录;对象 element 按结构解析,其余按声明长度跳过。"""
try:
element_count = reader.read(4)
if element_count == 0xF:
element_count += reader.read(5)
except ValueError as exc:
raise _syntax_error(
"element_count_truncated", "OAMD element 数量字段不完整",
raw_payload, exc) from exc
if element_count == 0:
raise UnsupportedVariantError(
"oamd", "missing_object_element",
"OAMD 没有声明任何 element",
details={"payload": bytes_descriptor(raw_payload)})
bed_isf_objects = program["num_bed_objects"] + program["num_isf_objects"]
elements = []
object_element = None
for ordinal in range(element_count):
header_start = reader.position
try:
element_id = reader.read(4)
size_bytes = _variable_bits_max(reader, 4, 4) + 1
except ValueError as exc:
raise _syntax_error(
"element_header_truncated", "OAMD element header 不完整",
raw_payload, exc,
element_ordinal=ordinal, header_start_bit=header_start) from exc
region_start = reader.position
region_end = region_start + size_bytes * 8
if region_end > reader.limit:
raise UnsupportedVariantError(
"oamd", "element_bounds",
"OAMD element 声明长度超过 payload 边界",
details={
"element_ordinal": ordinal,
"element_id": element_id,
"size_bytes": size_bytes,
"body_start_bit": region_start,
"body_end_bit": region_end,
"payload_bits": reader.limit,
"payload": bytes_descriptor(raw_payload),
})
try:
alternate_data_id = (reader.read(4)
if alternate_present else None)
discard_unknown = bool(reader.read(1))
except ValueError as exc:
raise _syntax_error(
"element_control_bounds", "OAMD element 太短,无法容纳控制字段",
raw_payload, exc,
element_ordinal=ordinal, element_id=element_id,
size_bytes=size_bytes) from exc
element = {
"ordinal": ordinal,
"element_id": element_id,
"size_bytes": size_bytes,
"header_start_bit": header_start,
"body_start_bit": region_start,
"body_end_bit": region_end,
"data_start_bit": reader.position,
"alternate_data_id": alternate_data_id,
"discard_unknown": discard_unknown,
"object_count": object_count,
"bed_isf_objects": bed_isf_objects,
}
if element_id == OBJECT_ELEMENT_ID:
if object_element is not None:
raise UnsupportedVariantError(
"oamd", "multiple_object_elements",
"OAMD 含多个 object element",
details={
"elements": _element_details(elements + [element]),
"payload": bytes_descriptor(raw_payload),
})
object_element = _parse_object_element(
reader, element, raw_payload)
element["parsed_end_bit"] = reader.position
element["object_element"] = object_element
# 个别编码器声明的 oa_element_size 比实际内容短;以结构解析结果为准。
reader.position = max(region_end, reader.position)
else:
# trim / extended / 未知 element:按声明长度整体跳过(5.5.4)。
element["parsed_end_bit"] = region_end
reader.position = region_end
elements.append(element)
if object_element is None:
raise UnsupportedVariantError(
"oamd", "missing_object_element",
"OAMD 缺少 object element",
details={
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
})
padding = reader.bits[reader.position:]
if np.any(padding):
raise UnsupportedVariantError(
"oamd", "nonzero_padding",
"OAMD payload 尾部 padding 含非零位",
details={
"elements": _element_details(elements),
"padding_start_bit": reader.position,
"padding_bits": "".join(str(int(bit)) for bit in padding[:64]),
"payload": bytes_descriptor(raw_payload),
})
return object_element, elements
def frame_update(bits_one):
"""单帧 OAMD → 位置字段增量及其 sample offset/ramp duration。"""
bits, raw_payload = _payload_bits(bits_one)
try:
version = bits[0] << 1 | bits[1]
except IndexError:
raise UnsupportedVariantError(
"oamd", "header_truncated",
"OAMD payload 不足以容纳 header",
details={"payload_bits": len(bits),
"payload": bytes_descriptor(raw_payload)}) from None
reader = _BitReader(bits, 2)
try:
if version == 3:
version += reader.read(3)
object_count_bits = reader.read(5)
if object_count_bits == 0x1F:
object_count_bits += reader.read(7)
object_count = object_count_bits + 1
program = _parse_program_assignment(reader, raw_payload)
alternate_present = bool(reader.read(1))
object_element, elements = _parse_elements(
reader, alternate_present, object_count, program, raw_payload)
except UnsupportedVariantError:
raise
except ValueError as exc:
raise _syntax_error(
"header_truncated", "OAMD header 字段越界或不完整",
raw_payload, exc, payload_bits=len(bits)) from exc
if version != 0:
raise UnsupportedVariantError(
"oamd", "oamd_version",
"OAMD 使用了当前未覆盖的 syntax version",
details={
"version": version,
"payload": bytes_descriptor(raw_payload),
})
if object_count > MAX_SLOTS:
raise UnsupportedVariantError(
"oamd", "object_count",
"OAMD 对象数超过当前 16 槽对象模型",
details={
"object_count": object_count,
"max_slots": MAX_SLOTS,
"payload": bytes_descriptor(raw_payload),
})
blocks = object_element["blocks"]
if len(blocks) != 1: if len(blocks) != 1:
raise UnsupportedVariantError( raise UnsupportedVariantError(
"oamd", "multiple_position_blocks", "oamd", "multiple_position_blocks",
"OAMD 一帧含多个对象位置更新块,固定位置窗口不能安全套用", "OAMD 一帧含多个对象位置更新块,单次 frame_update 无法表达",
details={ details={
"block_count": len(blocks), "block_count": len(blocks),
"blocks": blocks, "blocks": blocks,
"payload": bytes_descriptor(raw_payload), "payload": bytes_descriptor(raw_payload),
"repair_hint": "按 ObjectInfoBlock 顺序逐块解析坐标,再生成分段 ADM ramp", "repair_hint": "按 ObjectInfoBlock 顺序逐块解析坐标,再生成分段 ADM ramp",
}) })
return blocks[0]
# 对外契约:始终给出 16 个槽;本帧没有覆盖到的槽保持上一帧位置。
out = {(slot, field): None for slot in range(MAX_SLOTS)
for field in ("q1", "q2", "q3")}
for info in object_element["objects"]:
slot = info["object"]
position = info.get("position")
if position is None:
# bed/ISF 对象与未激活对象没有位置字段:保持上一帧位置。
out[(slot, "q1")] = None
out[(slot, "q2")] = None
out[(slot, "q3")] = None
continue
x, y, z = position
out[(slot, "q1")] = q_of(x, N_Q12) if 0 <= x <= N_Q12 else None
out[(slot, "q2")] = q_of(y, N_Q12) if 0 <= y <= N_Q12 else None
# 5.6.1.1.10/5.6.1.1.11:pos3D_Z 带符号;当前 16 槽模型只表达非负高度。
out[(slot, "q3")] = q_of(z, N_Q3) if 0 <= z <= N_Q3 else (
0 if z < 0 else None)
def frame_update(bits_one): size_mismatch = [
"""单帧 OAMD → 位置字段增量及其 sample offset/ramp duration。""" {
bits, raw_payload = _payload_bits(bits_one) "element_ordinal": element["ordinal"],
if len(bits) < 14: "element_id": element["element_id"],
raise UnsupportedVariantError( "declared_size_bytes": element["size_bytes"],
"oamd", "header_truncated", "declared_end_bit": element["body_end_bit"],
"OAMD payload 不足以容纳受支持的 header", "parsed_end_bit": element["parsed_end_bit"],
details={"payload_bits": len(bits), "payload": bytes_descriptor(raw_payload)}) }
header = { for element in elements
"version": int((bits[0] << 1) | bits[1]), if element["parsed_end_bit"] > element["body_end_bit"]
"objects_minus_one": int(sum(int(bits[2 + i]) << (4 - i) for i in range(5))), ]
"dynamic_object_only": int(bits[7]),
"lfe_present": int(bits[8]),
"alternate_object_present": int(bits[9]),
"element_count": int(sum(int(bits[10 + i]) << (3 - i) for i in range(4))),
}
if raw_payload[:2] != b"\x1f\x88":
raise UnsupportedVariantError(
"oamd", f"header_signature_{raw_payload[:2].hex()}",
"OAMD header 与当前位置字段布局不一致",
details={
"supported_header_prefix_hex": "1f88",
"header_probe": header,
"payload": bytes_descriptor(raw_payload),
"repair_hint": "按新 header 的 program assignment 和 element 布局重新定位对象位置字段",
})
object_element, elements = _parse_elements(bits, header, 14, raw_payload)
if object_element["body_end_bit"] < POSITION_WINDOW_END_BIT:
raise UnsupportedVariantError(
"oamd", "object_element_too_short",
"OAMD object element 无法容纳当前固定位置窗口",
details={
"element": _element_details([object_element])[0],
"required_position_end_bit": POSITION_WINDOW_END_BIT,
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
})
block_offset, ramp_duration = _update_timing(bits, object_element, raw_payload)
weights = 1 << np.arange(7, -1, -1)
out = {}
for obj in range(16):
start = 112 + 31 * (obj - 3)
wq1 = int((bits[start:start + 8] * weights).sum())
wq2 = int((bits[start + 8:start + 16] * weights).sum())
wq3 = int((bits[start + 16:start + 24] * weights).sum())
if obj and (wq1 >> 6 != 3 or wq2 & 2 != 2 or wq3 & 0x1F != 1):
raise UnsupportedVariantError(
"oamd", "position_layout_signature",
"OAMD 对象位置字段标记或位偏移发生变化",
details={
"object_slot": obj,
"position_start_bit": start,
"q1_window_hex": f"{wq1:02x}",
"q2_window_hex": f"{wq2:02x}",
"q3_window_hex": f"{wq3:02x}",
"header_probe": header,
"elements": _element_details(elements),
"payload": bytes_descriptor(raw_payload),
"repair_hint": "解析 OAMD element 可选字段并更新每个对象的位置窗口偏移",
})
k1 = wq1 - Q1_OFF
if obj == 0:
out[(0, "q1")] = None
out[(0, "q2")] = None
else:
out[(obj, "q1")] = q_of(k1, N_Q12) if 0 <= k1 <= N_Q12 else None
k2 = wq2 >> 2
if obj != 0:
out[(obj, "q2")] = q_of(k2, N_Q12) if 0 <= k2 <= N_Q12 else None
k3 = (wq2 & 1) * 8 + (wq3 >> 5)
out[(obj, "q3")] = q_of(k3, N_Q3) if 0 <= k3 <= N_Q3 else None
return { return {
"values": out, "values": out,
"block_offset_samples": block_offset, "block_offset_samples": blocks[0]["block_offset_samples"],
"ramp_duration_samples": ramp_duration, "ramp_duration_samples": blocks[0]["ramp_duration_samples"],
"object_count": object_count,
"program": {
"dynamic_object_only": program["dynamic_object_only"],
"lfe_present": program["lfe_present"],
"content_description": program["content_description"],
"num_bed_objects": program["num_bed_objects"],
"num_isf_objects": program["num_isf_objects"],
"num_dynamic_objects": program["num_dynamic_objects"],
"bed_channels": [list(assignment["channels"])
for assignment in program["bed_assignments"]],
},
"elements": _element_details(elements),
"objects": object_element["objects"],
"size_mismatch": size_mismatch,
} }